import {
  ApplicationEventMetadata,
  ApplicationProjectEvent,
  type ApplicationStoredEvent,
  CommandId,
  EventId,
  IsoDateTime,
  isKnownOrchestrationV2EventType,
  NonNegativeInt,
  OrchestrationV2DomainEventJson,
  OrchestrationV2StoredEvent,
  ProjectId,
  StoredProjectIcon,
  ThreadId,
  type OrchestrationV2DomainEvent,
} from "@t3tools/contracts";
import * as Context from "effect/Context";
import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
import * as SqlClient from "effect/sql/SqlClient";
import * as SqlSchema from "effect/sql/SqlSchema";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as PubSub from "effect/PubSub";
import * as Option from "effect/Option";
import * as Schema from "effect/Schema";
import * as Stream from "effect/Stream";

import { replayAndBufferProjectedLiveEvents } from "../orchestration-v2/LiveStreamBudget.ts";

import {
  toPersistenceDecodeError,
  toPersistenceSqlError,
  type OrchestrationEventStoreError,
} from "./Errors.ts";

/** Projection metadata rows: events at or below the pruned-through sequence may be gone. */
export const EVENT_LOG_PRUNED_THROUGH = "event-log-pruned-through";
export const EVENT_LOG_PRUNE_MARK = "event-log-prune-mark";

export type UnsequencedProjectEvent = ApplicationProjectEvent extends infer Event
  ? Event extends ApplicationProjectEvent
    ? Omit<Event, "sequence">
    : never
  : never;

type ProjectedApplicationStreamInput<A, E> = {
  readonly afterSequence?: number;
  readonly project: (event: ApplicationStoredEvent) => A;
  readonly filter?: (event: ApplicationStoredEvent) => boolean;
  readonly onReplayComplete?: Effect.Effect<void, E>;
};

export class OrchestrationEventStore extends Context.Service<
  OrchestrationEventStore,
  {
    readonly appendProjectEvent: (
      event: UnsequencedProjectEvent,
    ) => Effect.Effect<ApplicationProjectEvent, OrchestrationEventStoreError>;

    readonly appendAgentEvents: (input: {
      readonly commandId?: CommandId;
      readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
    }) => Effect.Effect<ReadonlyArray<OrchestrationV2StoredEvent>, OrchestrationEventStoreError>;

    /** Unknown event types fail unless skipUnknownEventTypes logs and drops them; only client replay may opt in. */
    readonly readAgentEvents: (input?: {
      readonly afterSequence?: number;
      readonly throughSequence?: number;
      readonly threadId?: ThreadId;
      readonly commandId?: CommandId;
      readonly eventType?: OrchestrationV2DomainEvent["type"];
      readonly limit?: number;
      readonly skipUnknownEventTypes?: boolean;
    }) => Stream.Stream<OrchestrationV2StoredEvent, OrchestrationEventStoreError>;

    readonly getAgentReplayStats: (input: {
      readonly threadId: ThreadId;
      readonly afterSequence: number;
      readonly throughSequence: number;
      readonly maxEvents: number;
    }) => Effect.Effect<
      {
        readonly eventCount: number;
        readonly rawPayloadBytes: number;
        readonly hasCreateEvent: boolean;
      },
      OrchestrationEventStoreError
    >;

    readonly getReplayStats: (input: {
      readonly afterSequence: number;
      readonly throughSequence: number;
    }) => Effect.Effect<
      {
        readonly eventCount: number;
        readonly rawPayloadBytes: number;
      },
      OrchestrationEventStoreError
    >;

    readonly latestAgentSequence: (
      threadId?: ThreadId,
    ) => Effect.Effect<number, OrchestrationEventStoreError>;

    readonly latestApplicationSequence: Effect.Effect<number, OrchestrationEventStoreError>;

    /** Replay from a cursor below this misses pruned events; resync from a snapshot instead. */
    readonly prunedThroughSequence: Effect.Effect<number, OrchestrationEventStoreError>;

    /** Unknown event types fail unless skipUnknownEventTypes logs and drops them; only client replay may opt in. */
    readonly readApplicationEvents: (input: {
      readonly afterSequence: number;
      readonly throughSequence: number;
      readonly skipUnknownEventTypes?: boolean;
    }) => Stream.Stream<ApplicationStoredEvent, OrchestrationEventStoreError>;

    /** Publish only after the surrounding event/projection transaction commits. */
    readonly publishCommitted: (
      events: ReadonlyArray<ApplicationStoredEvent>,
    ) => Effect.Effect<void>;

    readonly streamApplicationEvents: (input?: {
      readonly afterSequence?: number;
    }) => Stream.Stream<ApplicationStoredEvent, OrchestrationEventStoreError>;
    readonly streamProjectedApplicationEvents: <A extends { readonly sequence: number }, E = never>(
      input: ProjectedApplicationStreamInput<A, E>,
    ) => Stream.Stream<A, OrchestrationEventStoreError | E>;
  }
>()("t3/persistence/OrchestrationEventStore") {}

const encodeProjectIcon = Schema.encodeSync(StoredProjectIcon);
const decodeProjectEvent = Schema.decodeUnknownEffect(ApplicationProjectEvent);
const UnknownFromJsonString = Schema.fromJsonString(Schema.Unknown);
const EventMetadataFromJsonString = Schema.fromJsonString(ApplicationEventMetadata);
const ProjectEventType = Schema.Literals([
  "project.created",
  "project.meta-updated",
  "project.deleted",
]);
const ActorKind = Schema.Literals(["client", "server", "provider"]);

const AppendProjectEventRequestSchema = Schema.Struct({
  eventId: EventId,
  streamId: ProjectId,
  type: ProjectEventType,
  causationEventId: Schema.NullOr(EventId),
  correlationId: Schema.NullOr(CommandId),
  actorKind: ActorKind,
  occurredAt: IsoDateTime,
  commandId: Schema.NullOr(CommandId),
  payloadJson: UnknownFromJsonString,
  metadataJson: EventMetadataFromJsonString,
});

const ProjectEventPersistedRowSchema = Schema.Struct({
  sequence: NonNegativeInt,
  eventId: EventId,
  type: ProjectEventType,
  aggregateKind: Schema.Literal("project"),
  aggregateId: ProjectId,
  occurredAt: IsoDateTime,
  commandId: Schema.NullOr(CommandId),
  causationEventId: Schema.NullOr(EventId),
  correlationId: Schema.NullOr(CommandId),
  payload: UnknownFromJsonString,
  metadata: EventMetadataFromJsonString,
});

const READ_PAGE_SIZE = 500;

interface ApplicationEventRow {
  readonly sequence: number;
  readonly event_id: string;
  readonly command_id: string | null;
  readonly aggregate_kind: "project" | "thread";
  readonly stream_id: string;
  readonly event_type: string;
  readonly occurred_at: string;
  readonly payload_json: string;
  readonly metadata_json: string;
  readonly application_event_version: number;
  readonly causation_event_id: string | null;
  readonly correlation_id: string | null;
}

const decodeV2EventJson = Schema.decodeUnknownEffect(OrchestrationV2DomainEventJson);
const encodeV2EventJson = Schema.encodeEffect(OrchestrationV2DomainEventJson);
const decodeV2StoredEvent = Schema.decodeUnknownEffect(OrchestrationV2StoredEvent);
const decodeJson = Schema.decodeUnknownEffect(Schema.fromJsonString(Schema.Unknown));
const encodeJson = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown));

function metadataForV2Event(event: OrchestrationV2DomainEvent): Record<string, unknown> {
  return {
    ...(event.runId === undefined ? {} : { runId: event.runId }),
    ...(event.nodeId === undefined ? {} : { nodeId: event.nodeId }),
    ...(event.driver === undefined ? {} : { driver: event.driver }),
    ...(event.providerInstanceId === undefined
      ? {}
      : { providerInstanceId: event.providerInstanceId }),
    ...(event.rawEventId === undefined ? {} : { rawEventId: event.rawEventId }),
  };
}

const rowToV2StoredEvent = Effect.fn("OrchestrationEventStore.rowToV2StoredEvent")(function* (
  row: ApplicationEventRow,
) {
  const payload = yield* decodeJson(row.payload_json);
  const metadata = yield* decodeJson(row.metadata_json);
  const values =
    typeof metadata === "object" && metadata !== null ? (metadata as Record<string, unknown>) : {};
  const event = yield* decodeV2EventJson({
    id: row.event_id,
    threadId: row.stream_id,
    type: row.event_type,
    occurredAt: row.occurred_at,
    payload,
    ...(values.runId === undefined ? {} : { runId: values.runId }),
    ...(values.nodeId === undefined ? {} : { nodeId: values.nodeId }),
    ...(values.driver === undefined ? {} : { driver: values.driver }),
    ...(values.providerInstanceId === undefined
      ? {}
      : { providerInstanceId: values.providerInstanceId }),
    ...(values.rawEventId === undefined ? {} : { rawEventId: values.rawEventId }),
  });
  return yield* decodeV2StoredEvent({
    sequence: row.sequence,
    commandId: row.command_id,
    event,
  });
});

const rowToProjectEvent = Effect.fn("OrchestrationEventStore.rowToProjectEvent")(function* (
  row: ApplicationEventRow,
) {
  return yield* decodeProjectEvent({
    sequence: row.sequence,
    eventId: row.event_id,
    type: row.event_type,
    aggregateKind: row.aggregate_kind,
    aggregateId: row.stream_id,
    occurredAt: row.occurred_at,
    commandId: row.command_id,
    causationEventId: row.causation_event_id,
    correlationId: row.correlation_id,
    payload: yield* decodeJson(row.payload_json),
    metadata: yield* decodeJson(row.metadata_json),
  });
});

function rowToApplicationStoredEvent(
  row: ApplicationEventRow,
): Effect.Effect<ApplicationStoredEvent, Schema.SchemaError> {
  return row.aggregate_kind === "project" ? rowToProjectEvent(row) : rowToV2StoredEvent(row);
}

const isKnownProjectEventType = Schema.is(ProjectEventType);

const keepReplayRow = (row: ApplicationEventRow, skipUnknownEventTypes?: boolean) =>
  skipUnknownEventTypes !== true ||
  (row.aggregate_kind === "project"
    ? isKnownProjectEventType(row.event_type)
    : isKnownOrchestrationV2EventType(row.event_type))
    ? Effect.succeed(true)
    : Effect.logWarning("Skipping an application event row with an unknown event type", {
        aggregateKind: row.aggregate_kind,
        sequence: row.sequence,
        eventTypeLength: row.event_type.length,
      }).pipe(Effect.as(false));

function inferActorKind(event: UnsequencedProjectEvent): typeof ActorKind.Type {
  if (event.commandId !== null && event.commandId.startsWith("provider:")) {
    return "provider";
  }
  if (event.commandId !== null && event.commandId.startsWith("server:")) {
    return "server";
  }
  if (
    event.metadata.providerTurnId !== undefined ||
    event.metadata.providerItemId !== undefined ||
    event.metadata.adapterKey !== undefined
  ) {
    return "provider";
  }
  if (event.commandId === null) {
    return "server";
  }
  return "client";
}

function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) {
  return (cause: unknown): OrchestrationEventStoreError =>
    Schema.isSchemaError(cause)
      ? toPersistenceDecodeError(decodeOperation)(cause)
      : toPersistenceSqlError(sqlOperation)(cause);
}

const makeEventStore = Effect.gen(function* () {
  const sql = yield* SqlClient.SqlClient;
  const committedEvents = yield* PubSub.unbounded<ApplicationStoredEvent>();
  const promptHints = new Map<string, number>();
  const restoredPayloadJson = sql`CASE
    WHEN orchestration_events.event_type IN ('subagent.updated', 'turn-item.updated')
      AND json_type(orchestration_events.payload_json, '$.promptSequence') = 'integer'
    THEN json_set(
      json_remove(orchestration_events.payload_json, '$.promptSequence'),
      '$.prompt',
      (SELECT json_extract(anchor.payload_json, '$.prompt')
       FROM orchestration_events AS anchor
       WHERE anchor.sequence = json_extract(orchestration_events.payload_json, '$.promptSequence'))
    )
    ELSE orchestration_events.payload_json
  END`;

  const appendProjectEventRow = SqlSchema.findOne({
    Request: AppendProjectEventRequestSchema,
    Result: ProjectEventPersistedRowSchema,
    execute: (request) =>
      sql`
        INSERT INTO orchestration_events (
          event_id,
          aggregate_kind,
          stream_id,
          stream_version,
          event_type,
          occurred_at,
          command_id,
          causation_event_id,
          correlation_id,
          actor_kind,
          payload_json,
          metadata_json,
          application_event_version
        )
        VALUES (
          ${request.eventId},
          'project',
          ${request.streamId},
          COALESCE(
            (
              SELECT stream_version + 1
              FROM orchestration_events
              WHERE aggregate_kind = 'project'
                AND stream_id = ${request.streamId}
              ORDER BY stream_version DESC
              LIMIT 1
            ),
            0
          ),
          ${request.type},
          ${request.occurredAt},
          ${request.commandId},
          ${request.causationEventId},
          ${request.correlationId},
          ${request.actorKind},
          ${request.payloadJson},
          ${request.metadataJson},
          2
        )
        RETURNING
          sequence,
          event_id AS "eventId",
          event_type AS "type",
          aggregate_kind AS "aggregateKind",
          stream_id AS "aggregateId",
          occurred_at AS "occurredAt",
          command_id AS "commandId",
          causation_event_id AS "causationEventId",
          correlation_id AS "correlationId",
          payload_json AS "payload",
          metadata_json AS "metadata"
      `,
  });

  const appendProjectEvent: OrchestrationEventStore["Service"]["appendProjectEvent"] = (event) =>
    appendProjectEventRow({
      eventId: event.eventId,
      streamId: event.aggregateId,
      type: event.type,
      causationEventId: event.causationEventId,
      correlationId: event.correlationId,
      actorKind: inferActorKind(event),
      occurredAt: event.occurredAt,
      commandId: event.commandId,
      payloadJson:
        event.type === "project.deleted" || !event.payload.projectIcon
          ? event.payload
          : { ...event.payload, projectIcon: encodeProjectIcon(event.payload.projectIcon) },
      metadataJson: event.metadata,
    }).pipe(
      Effect.flatMap((row) => decodeProjectEvent(row)),
      Effect.mapError(
        toPersistenceSqlOrDecodeError(
          "OrchestrationEventStore.appendProjectEvent:insert",
          "OrchestrationEventStore.appendProjectEvent:decode",
        ),
      ),
    );

  const readApplicationRows = (input: {
    readonly afterSequence: number;
    readonly throughSequence?: number;
    readonly threadId?: ThreadId;
    readonly commandId?: CommandId;
    readonly eventType?: OrchestrationV2DomainEvent["type"];
    readonly onlyAgentEvents?: boolean;
    readonly limit: number;
  }) =>
    sql<ApplicationEventRow>`
      SELECT
        sequence,
        event_id,
        command_id,
        aggregate_kind,
        stream_id,
        event_type,
        occurred_at,
        ${restoredPayloadJson} AS payload_json,
        metadata_json,
        application_event_version,
        causation_event_id,
        correlation_id
      FROM orchestration_events
      ${
        // Otherwise the sequence index wins and command lookups scan every V2 event.
        input.commandId !== undefined
          ? sql`INDEXED BY idx_orch_events_command_id`
          : input.onlyAgentEvents === true
            ? sql``
            : sql`INDEXED BY idx_orchestration_events_application_high_water`
      }
      WHERE sequence > ${input.afterSequence}
        AND sequence <= ${input.throughSequence ?? Number.MAX_SAFE_INTEGER}
        AND (
          ${
            input.onlyAgentEvents === true
              ? sql`application_event_version = 2 AND aggregate_kind = 'thread'`
              : sql`aggregate_kind = 'project'
                    OR (application_event_version = 2 AND aggregate_kind = 'thread')`
          }
        )
        AND ${sql.and([
          ...(input.threadId === undefined ? [] : [sql`stream_id = ${input.threadId}`]),
          ...(input.commandId === undefined ? [] : [sql`command_id = ${input.commandId}`]),
          ...(input.eventType === undefined ? [] : [sql`event_type = ${input.eventType}`]),
        ])}
      ORDER BY sequence ASC
      LIMIT ${input.limit}
    `.pipe(NodeSqliteClient.readOnly);

  const appendAgentEvents: OrchestrationEventStore["Service"]["appendAgentEvents"] = (input) =>
    Effect.forEach(
      input.events,
      (event) =>
        Effect.gen(function* () {
          const encoded = yield* encodeV2EventJson(event);
          const subagent =
            encoded.type === "subagent.updated"
              ? encoded.payload
              : encoded.type === "turn-item.updated" && encoded.payload.type === "subagent"
                ? encoded.payload
                : undefined;
          const promptKey = subagent === undefined ? undefined : `${event.type}:${subagent.id}`;
          const hint = promptKey === undefined ? undefined : promptHints.get(promptKey);
          const payloadJson = yield* encodeJson(encoded.payload);
          // Deduplicate repeated progress prompts via anchor rows; pruning keeps anchors that retained rows reference.
          const storedPayload =
            hint !== undefined && subagent !== undefined
              ? sql`CASE WHEN EXISTS (
                  SELECT 1 FROM orchestration_events
                  WHERE sequence = ${hint}
                    AND json_type(payload_json, '$.promptSequence') IS NULL
                    AND json_extract(payload_json, '$.prompt') = ${subagent.prompt}
                ) THEN json_set(${payloadJson}, '$.prompt', '', '$.promptSequence', CAST(${hint} AS INTEGER))
                ELSE ${payloadJson} END`
              : sql`${payloadJson}`;
          const rows = yield* sql<{
            readonly sequence: number;
            readonly stores_full_prompt?: number;
          }>`
            INSERT INTO orchestration_events (
              event_id,
              aggregate_kind,
              stream_id,
              stream_version,
              event_type,
              occurred_at,
              command_id,
              causation_event_id,
              correlation_id,
              actor_kind,
              payload_json,
              metadata_json,
              application_event_version
            )
            VALUES (
              ${event.id},
              'thread',
              ${event.threadId},
              COALESCE(
                (
                  SELECT MAX(stream_version) + 1
                  FROM orchestration_events
                  WHERE aggregate_kind = 'thread' AND stream_id = ${event.threadId}
                ),
                0
              ),
              ${event.type},
              ${encoded.occurredAt},
              ${input.commandId ?? null},
              NULL,
              ${input.commandId ?? null},
              ${event.rawEventId === undefined ? "server" : "provider"},
              ${storedPayload},
              ${yield* encodeJson(metadataForV2Event(event))},
              2
            )
            RETURNING sequence${
              hint === undefined
                ? sql``
                : sql`, json_type(payload_json, '$.promptSequence') IS NULL AS stores_full_prompt`
            }
          `;
          const row = rows[0];
          if (promptKey !== undefined && row !== undefined) {
            promptHints.delete(promptKey);
            promptHints.set(
              promptKey,
              hint === undefined || row.stores_full_prompt === 1 ? row.sequence : hint,
            );
            if (promptHints.size > 4096) {
              const oldest = promptHints.keys().next().value;
              if (oldest !== undefined) promptHints.delete(oldest);
            }
          }
          return yield* decodeV2StoredEvent({
            sequence: rows[0]?.sequence,
            commandId: input.commandId ?? null,
            event,
          });
        }),
      { concurrency: 1 },
    ).pipe(
      Effect.mapError(
        toPersistenceSqlOrDecodeError(
          "OrchestrationEventStore.appendAgentEvents:insert",
          "OrchestrationEventStore.appendAgentEvents:decode",
        ),
      ),
    );

  const readAgentEvents: OrchestrationEventStore["Service"]["readAgentEvents"] = (input) => {
    const totalLimit =
      input?.limit === undefined ? Number.MAX_SAFE_INTEGER : Math.max(0, Math.floor(input.limit));
    if (totalLimit === 0) {
      return Stream.empty;
    }
    return Stream.paginate(
      { cursor: input?.afterSequence ?? 0, remaining: totalLimit },
      ({ cursor, remaining }) => {
        const pageLimit = Math.min(remaining, READ_PAGE_SIZE);
        return readApplicationRows({
          afterSequence: cursor,
          ...(input?.throughSequence === undefined
            ? {}
            : { throughSequence: input.throughSequence }),
          ...(input?.threadId === undefined ? {} : { threadId: input.threadId }),
          ...(input?.commandId === undefined ? {} : { commandId: input.commandId }),
          ...(input?.eventType === undefined ? {} : { eventType: input.eventType }),
          onlyAgentEvents: true,
          limit: pageLimit,
        }).pipe(
          Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.readAgentEvents:query")),
          Effect.map((rows) => {
            const last = rows.at(-1);
            const nextRemaining = remaining - rows.length;
            return [
              rows,
              last === undefined ||
              rows.length < pageLimit ||
              nextRemaining <= 0 ||
              (input?.throughSequence !== undefined && last.sequence >= input.throughSequence)
                ? Option.none()
                : Option.some({ cursor: last.sequence, remaining: nextRemaining }),
            ] as const;
          }),
        );
      },
    ).pipe(
      Stream.filterEffect((row) => keepReplayRow(row, input?.skipUnknownEventTypes)),
      Stream.mapEffect((row) =>
        rowToV2StoredEvent(row).pipe(
          Effect.mapError(
            toPersistenceDecodeError("OrchestrationEventStore.readAgentEvents:decode"),
          ),
        ),
      ),
    );
  };

  const getAgentReplayStats: OrchestrationEventStore["Service"]["getAgentReplayStats"] = (input) =>
    sql<{
      readonly eventCount: number;
      readonly rawPayloadBytes: number;
      readonly hasCreateEvent: number;
    }>`
      SELECT
        COUNT(*) AS "eventCount",
        COALESCE(SUM(octet_length(payload_json)), 0) AS "rawPayloadBytes",
        COALESCE(MAX(event_type = 'thread.created'), 0) AS "hasCreateEvent"
      FROM (
        SELECT ${restoredPayloadJson} AS payload_json, event_type
        FROM orchestration_events
        WHERE aggregate_kind = 'thread'
          AND stream_id = ${input.threadId}
          AND application_event_version = 2
          AND sequence > ${input.afterSequence}
          AND sequence <= ${input.throughSequence}
        ORDER BY sequence ASC
        LIMIT ${Math.max(0, Math.floor(input.maxEvents)) + 1}
      )
    `.pipe(
      NodeSqliteClient.readOnly,
      Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.getAgentReplayStats:query")),
      Effect.map((rows) => ({
        eventCount: rows[0]?.eventCount ?? 0,
        rawPayloadBytes: rows[0]?.rawPayloadBytes ?? 0,
        hasCreateEvent: (rows[0]?.hasCreateEvent ?? 0) !== 0,
      })),
    );

  const getReplayStats: OrchestrationEventStore["Service"]["getReplayStats"] = (input) =>
    sql<{ readonly eventCount: number; readonly rawPayloadBytes: number }>`
      SELECT
        COUNT(*) AS "eventCount",
        COALESCE(SUM(octet_length(${restoredPayloadJson})), 0) AS "rawPayloadBytes"
      FROM orchestration_events INDEXED BY idx_orchestration_events_application_high_water
      WHERE sequence > ${input.afterSequence}
        AND sequence <= ${input.throughSequence}
        AND (
          aggregate_kind = 'project'
          OR (application_event_version = 2 AND aggregate_kind = 'thread')
        )
    `.pipe(
      NodeSqliteClient.readOnly,
      Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.getReplayStats:query")),
      Effect.map((rows) => ({
        eventCount: rows[0]?.eventCount ?? 0,
        rawPayloadBytes: rows[0]?.rawPayloadBytes ?? 0,
      })),
    );

  const latestAgentSequence: OrchestrationEventStore["Service"]["latestAgentSequence"] = (
    threadId,
  ) =>
    sql<{ readonly sequence: number | null }>`
      SELECT MAX(sequence) AS sequence
      FROM orchestration_events
      ${
        threadId === undefined
          ? // The planner can otherwise pick the covering stream index and scan every agent event.
            sql`INDEXED BY idx_orchestration_events_application_sequence`
          : sql`INDEXED BY idx_orchestration_events_agent_stream_sequence`
      }
      WHERE application_event_version = 2
        AND aggregate_kind = 'thread'
        ${threadId === undefined ? sql`` : sql`AND stream_id = ${threadId}`}
    `.pipe(
      NodeSqliteClient.readOnly,
      Effect.map((rows) => rows[0]?.sequence ?? 0),
      Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.latestAgentSequence:query")),
    );

  // The OR planner otherwise scans every V2 event instead of seeking the final sequence.
  const latestApplicationSequence = sql<{ readonly sequence: number | null }>`
    SELECT MAX(sequence) AS sequence
    FROM orchestration_events INDEXED BY idx_orchestration_events_application_high_water
    WHERE aggregate_kind = 'project'
      OR (application_event_version = 2 AND aggregate_kind = 'thread')
  `.pipe(
    NodeSqliteClient.readOnly,
    Effect.map((rows) => rows[0]?.sequence ?? 0),
    Effect.mapError(
      toPersistenceSqlError("OrchestrationEventStore.latestApplicationSequence:query"),
    ),
  );

  const prunedThroughSequence = sql<{ readonly last_sequence: number }>`
    SELECT last_sequence FROM orchestration_v2_projection_metadata
    WHERE projection_name = ${EVENT_LOG_PRUNED_THROUGH}
  `.pipe(
    NodeSqliteClient.readOnly,
    Effect.map((rows) => rows[0]?.last_sequence ?? 0),
    Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.prunedThroughSequence:query")),
  );

  // Advance over raw rows so even a full page of unknown types cannot end replay early.
  const catchUpApplicationEvents = (input: {
    readonly afterSequence: number;
    readonly throughSequence: number;
    readonly skipUnknownEventTypes?: boolean;
  }): Stream.Stream<ApplicationStoredEvent, OrchestrationEventStoreError> =>
    Stream.paginate(input.afterSequence, (afterSequence) =>
      readApplicationRows({
        afterSequence,
        throughSequence: input.throughSequence,
        limit: READ_PAGE_SIZE,
      }).pipe(
        Effect.mapError(
          toPersistenceSqlError("OrchestrationEventStore.readApplicationEvents:query"),
        ),
        Effect.map((rows) => {
          const last = rows.at(-1);
          return [
            rows,
            last === undefined ||
            rows.length < READ_PAGE_SIZE ||
            last.sequence >= input.throughSequence
              ? Option.none()
              : Option.some(last.sequence),
          ] as const;
        }),
      ),
    ).pipe(
      Stream.filterEffect((row) => keepReplayRow(row, input.skipUnknownEventTypes)),
      Stream.mapEffect((row) =>
        rowToApplicationStoredEvent(row).pipe(
          Effect.mapError(
            toPersistenceDecodeError("OrchestrationEventStore.readApplicationEvents:decode"),
          ),
        ),
      ),
    );

  const streamProjectedApplicationEvents = <A extends { readonly sequence: number }, E = never>(
    input: ProjectedApplicationStreamInput<A, E>,
  ) =>
    replayAndBufferProjectedLiveEvents<
      ApplicationStoredEvent,
      A,
      OrchestrationEventStoreError | E,
      never
    >({
      subscribe: PubSub.subscribe(committedEvents),
      latestSequence: latestApplicationSequence,
      afterSequence: input.afterSequence ?? 0,
      project: input.project,
      ...(input.filter === undefined ? {} : { filter: input.filter }),
      ...(input.onReplayComplete === undefined ? {} : { onReplayComplete: input.onReplayComplete }),
      replay: (throughSequence) =>
        catchUpApplicationEvents({
          afterSequence: input?.afterSequence ?? 0,
          throughSequence,
        }),
    }).pipe(
      Stream.catchTag("LiveStreamBufferError", (cause) =>
        Stream.fail(
          toPersistenceSqlError("OrchestrationEventStore.streamApplicationEvents:buffer")(cause),
        ),
      ),
    );

  const streamApplicationEvents: OrchestrationEventStore["Service"]["streamApplicationEvents"] = (
    input,
  ) => streamProjectedApplicationEvents({ ...input, project: (event) => event });

  return {
    appendProjectEvent,
    appendAgentEvents,
    readAgentEvents,
    getAgentReplayStats,
    getReplayStats,
    latestAgentSequence,
    latestApplicationSequence,
    prunedThroughSequence,
    readApplicationEvents: catchUpApplicationEvents,
    publishCommitted: (events) => PubSub.publishAll(committedEvents, events).pipe(Effect.asVoid),
    streamApplicationEvents,
    streamProjectedApplicationEvents,
  } satisfies OrchestrationEventStore["Service"];
});

export const layer = Layer.effect(OrchestrationEventStore, makeEventStore);