import type {
  ApplicationStoredEvent,
  OrchestrationV2ArchivedShellStreamItem,
  OrchestrationV2ShellSnapshot,
  OrchestrationV2ThreadShellSnapshot,
  OrchestrationV2ShellStreamItem,
  OrchestrationV2StoredEvent,
} from "@t3tools/contracts";
import * as Stream from "effect/Stream";

/** Keep only the newest shell-relevant event per project/thread aggregate. */
export function coalesceShellApplicationEvents(
  events: ReadonlyArray<ApplicationStoredEvent>,
): ReadonlyArray<ApplicationStoredEvent> {
  const latestByAggregate = new Map<string, ApplicationStoredEvent>();
  for (const stored of events) {
    const key =
      "aggregateKind" in stored
        ? `project:${stored.aggregateId}`
        : `thread:${stored.event.threadId}`;
    latestByAggregate.set(key, stored);
  }
  return Array.from(latestByAggregate.values()).sort(
    (left, right) => left.sequence - right.sequence,
  );
}

/**
 * Emit the initial shell prefix strictly first, then merge the post-prefix
 * tail with enrichment refreshes. Prevents a newer marked enrichment from
 * landing before the unmarked authoritative initial snapshot.
 */
export function composeShellStreamWithEnrichment<A, E, R, A2, E2, R2, A3, E3, R3>(input: {
  readonly initial: Stream.Stream<A, E, R>;
  readonly tail: Stream.Stream<A2, E2, R2>;
  readonly enrichment: Stream.Stream<A3, E3, R3>;
}): Stream.Stream<A | A2 | A3, E | E2 | E3, R | R2 | R3> {
  return Stream.concat(input.initial, Stream.merge(input.tail, input.enrichment));
}

/** Build a shell snapshot stream item for a batched enrichment completion. */
export function shellStreamItemFromEnrichmentRefresh(input: {
  readonly snapshot: OrchestrationV2ShellSnapshot;
  readonly changes: ReadonlyArray<{ readonly workspaceRoot: string }>;
}): Extract<OrchestrationV2ShellStreamItem, { readonly kind: "snapshot" }> {
  return {
    kind: "snapshot",
    snapshot: input.snapshot,
    resolvedRepositoryIdentityRoots: [
      ...new Set(input.changes.map((change) => change.workspaceRoot)),
    ],
  };
}

/**
 * Initial subscribe frames: always emit the unmarked authoritative snapshot,
 * then a same-sequence enrichment frame only when some roots already resolved.
 */
export function shellStreamItemsFromInitialSnapshot(input: {
  readonly snapshot: OrchestrationV2ShellSnapshot;
  readonly resolvedRepositoryIdentityRoots: ReadonlyArray<string>;
}): ReadonlyArray<Extract<OrchestrationV2ShellStreamItem, { readonly kind: "snapshot" }>> {
  const authoritative = {
    kind: "snapshot" as const,
    snapshot: input.snapshot,
  };
  if (input.resolvedRepositoryIdentityRoots.length === 0) {
    return [authoritative];
  }
  return [
    authoritative,
    {
      kind: "snapshot" as const,
      snapshot: input.snapshot,
      resolvedRepositoryIdentityRoots: [...new Set(input.resolvedRepositoryIdentityRoots)],
    },
  ];
}

/** Converts a committed event and its resulting shell snapshot into one delta. */
export function shellStreamItemFromSnapshot(input: {
  readonly stored: OrchestrationV2StoredEvent;
  readonly snapshot: OrchestrationV2ThreadShellSnapshot;
}): Exclude<OrchestrationV2ShellStreamItem, { readonly kind: "snapshot" }> {
  const active = input.snapshot.threads.find((thread) => thread.id === input.stored.event.threadId);
  if (active !== undefined) {
    return {
      kind: "thread.updated",
      sequence: input.stored.sequence,
      location: "active",
      thread: active,
    };
  }

  const archived = input.snapshot.archivedThreads.find(
    (thread) => thread.id === input.stored.event.threadId,
  );
  if (archived !== undefined) {
    return {
      kind: "thread.updated",
      sequence: input.stored.sequence,
      location: "archive",
      thread: archived,
    };
  }

  return {
    kind: "thread.removed",
    sequence: input.stored.sequence,
    location:
      input.stored.event.type === "thread.deleted" && input.stored.event.payload.archivedAt !== null
        ? "archive"
        : "active",
    threadId: input.stored.event.threadId,
  };
}

/** Converts a committed event into an archive-only delta when it changes archive membership. */
export function archivedShellStreamItemFromSnapshot(input: {
  readonly stored: OrchestrationV2StoredEvent;
  readonly snapshot: OrchestrationV2ThreadShellSnapshot;
}): Exclude<OrchestrationV2ArchivedShellStreamItem, { readonly kind: "snapshot" }> | null {
  const archived = input.snapshot.archivedThreads.find(
    (thread) => thread.id === input.stored.event.threadId,
  );
  if (archived !== undefined) {
    return {
      kind: "thread.updated",
      sequence: input.stored.sequence,
      thread: archived,
    };
  }
  if (
    input.stored.event.type === "thread.unarchived" ||
    (input.stored.event.type === "thread.deleted" && input.stored.event.payload.archivedAt !== null)
  ) {
    return {
      kind: "thread.removed",
      sequence: input.stored.sequence,
      threadId: input.stored.event.threadId,
    };
  }
  return null;
}