import { CommandId, EventId, ProviderSessionId, ThreadId } from "@t3tools/contracts";
import { assert, it } from "@effect/vitest";
import * as DateTime from "effect/DateTime";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Schema from "effect/Schema";
import * as Stream from "effect/Stream";
import * as Tracer from "effect/Tracer";
import * as SqlClient from "effect/unstable/sql/SqlClient";

import { runMigrations } from "../Migrations.ts";
import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
import { OrchestrationEventStore } from "../Services/OrchestrationEventStore.ts";
import { OrchestrationEventStoreLive } from "./OrchestrationEventStore.ts";
import { SqlitePersistenceMemory } from "./Sqlite.ts";

const eventStoreLayer = OrchestrationEventStoreLive.pipe(
  Layer.provideMerge(SqlitePersistenceMemory),
);
const occurredAt = "2026-09-03T00:00:00.000Z";
const encodeJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));

function eventRow(
  ordinal: number,
  aggregateKind: "project" | "thread",
  version: number,
  streamId = `${aggregateKind}:${ordinal}`,
  commandId: string | null = null,
) {
  return {
    event_id: `event:${ordinal}`,
    aggregate_kind: aggregateKind,
    stream_id: streamId,
    stream_version: ordinal,
    event_type: aggregateKind === "project" ? "project.created" : "provider-session.detached",
    occurred_at: occurredAt,
    command_id: commandId,
    causation_event_id: null,
    correlation_id: null,
    actor_kind: "server",
    payload_json: encodeJson(
      aggregateKind === "project"
        ? {
            projectId: streamId,
            title: "Sequence fixture",
            workspaceRoot: "/tmp/sequence-fixture",
            defaultModelSelection: null,
            scripts: [],
            createdAt: occurredAt,
            updatedAt: occurredAt,
          }
        : { providerSessionId: `session:${ordinal}`, detachedAt: occurredAt },
    ),
    metadata_json: "{}",
    application_event_version: version,
  };
}

const seedEvents = Effect.fn("test.seedSequenceEvents")(function* (
  rows: ReadonlyArray<ReturnType<typeof eventRow>>,
) {
  const sql = yield* SqlClient.SqlClient;
  return yield* sql<{
    readonly sequence: number;
    readonly aggregate_kind: string;
    readonly application_event_version: number;
    readonly stream_id: string;
    readonly command_id: string | null;
  }>`
    INSERT INTO orchestration_events ${sql.insert(rows)}
    RETURNING sequence, aggregate_kind, application_event_version, stream_id, command_id
  `;
});

it.effect("keeps application and scoped agent high-water marks separate from legacy history", () =>
  Effect.gen(function* () {
    const store = yield* OrchestrationEventStore;
    const sql = yield* SqlClient.SqlClient;
    const target = ThreadId.make("target");
    assert.equal(yield* store.latestApplicationSequence, 0);
    assert.equal(yield* store.latestAgentSequence(), 0);
    assert.equal(yield* store.latestAgentSequence(target), 0);

    yield* seedEvents([eventRow(1, "thread", 1, "legacy-only")]);
    assert.equal(yield* store.latestApplicationSequence, 0);
    assert.equal(yield* store.latestAgentSequence(ThreadId.make("legacy-only")), 0);

    const [legacyProject, targetEvent, otherEvent] = yield* seedEvents([
      eventRow(2, "project", 1),
      eventRow(3, "thread", 2, target),
      eventRow(4, "thread", 2, "other"),
    ]);
    assert.ok(legacyProject);
    assert.ok(targetEvent);
    assert.ok(otherEvent);
    assert.equal(yield* store.latestApplicationSequence, otherEvent.sequence);
    assert.equal(yield* store.latestAgentSequence(), otherEvent.sequence);
    assert.equal(yield* store.latestAgentSequence(target), targetEvent.sequence);

    const [projectEvent] = yield* seedEvents([
      eventRow(5, "project", 2),
      eventRow(6, "thread", 1, target),
      eventRow(7, "thread", 3, target),
    ]);
    assert.ok(projectEvent);
    assert.equal(yield* store.latestApplicationSequence, projectEvent.sequence);
    assert.equal(yield* store.latestAgentSequence(), otherEvent.sequence);
    assert.equal(yield* store.latestAgentSequence(target), targetEvent.sequence);
    assert.equal(yield* store.latestAgentSequence(ThreadId.make("missing")), 0);

    yield* sql`DELETE FROM orchestration_events WHERE sequence > ${legacyProject.sequence}`;
    assert.equal(yield* store.latestApplicationSequence, legacyProject.sequence);
    assert.equal(yield* store.latestAgentSequence(), 0);
    assert.equal(yield* store.latestAgentSequence(target), 0);
  }).pipe(Effect.provide(Layer.fresh(eventStoreLayer))),
);

it.effect(
  "preserves mixed application paging, scoped replay, and the catch-up to live boundary",
  () =>
    Effect.gen(function* () {
      const store = yield* OrchestrationEventStore;
      const rows = yield* seedEvents(
        Array.from({ length: 1_560 }, (_, index) => {
          const kind = index % 3 === 1 ? "project" : "thread";
          return eventRow(
            index,
            kind,
            index % 3 === 0 ? 1 : 2,
            kind === "project" ? `project:${index}` : index % 2 === 0 ? "target" : "other",
            index % 5 === 0 ? "selected-command" : "other-command",
          );
        }),
      );
      const afterSequence = rows[25]!.sequence;
      const throughSequence = rows[1_525]!.sequence;
      const retained = rows.filter(
        (row) => row.aggregate_kind === "project" || row.application_event_version === 2,
      );
      const replay = yield* store
        .readApplicationEvents({ afterSequence, throughSequence })
        .pipe(Stream.runCollect);
      assert.deepEqual(
        replay.map((event) => event.sequence),
        retained
          .filter((row) => row.sequence > afterSequence && row.sequence <= throughSequence)
          .map((row) => row.sequence),
      );

      const scoped = yield* store
        .readAgentEvents({
          afterSequence,
          throughSequence,
          threadId: ThreadId.make("target"),
          commandId: CommandId.make("selected-command"),
          limit: 17,
        })
        .pipe(Stream.runCollect);
      assert.deepEqual(
        scoped.map((event) => event.sequence),
        retained
          .filter(
            (row) =>
              row.sequence > afterSequence &&
              row.sequence <= throughSequence &&
              row.aggregate_kind === "thread" &&
              row.stream_id === "target" &&
              row.command_id === "selected-command",
          )
          .slice(0, 17)
          .map((row) => row.sequence),
      );

      const catchUp = retained.filter((row) => row.sequence > afterSequence);
      const highWater = catchUp.at(-1)!.sequence;
      let liveSequence = 0;
      const streamed = yield* store.streamApplicationEvents({ afterSequence }).pipe(
        Stream.tap((event) =>
          event.sequence !== highWater
            ? Effect.void
            : Effect.gen(function* () {
                const [stored] = yield* store.appendAgentEvents({
                  events: [
                    {
                      id: EventId.make("live-after-catch-up"),
                      type: "provider-session.detached",
                      threadId: ThreadId.make("target"),
                      occurredAt: DateTime.makeUnsafe(occurredAt),
                      payload: {
                        providerSessionId: ProviderSessionId.make("live-session"),
                        detachedAt: DateTime.makeUnsafe(occurredAt),
                      },
                    },
                  ],
                });
                assert.ok(stored);
                liveSequence = stored.sequence;
                yield* store.publishCommitted([stored]);
              }),
        ),
        Stream.take(catchUp.length + 1),
        Stream.runCollect,
      );
      assert.deepEqual(
        streamed.map((event) => event.sequence),
        [...catchUp.map((row) => row.sequence), liveSequence],
      );
      assert.isAbove(liveSequence, highWater);
    }).pipe(Effect.provide(Layer.fresh(eventStoreLayer))),
);

it.effect("upgrades populated history to indexed high-water lookups without OR scans", () =>
  Effect.gen(function* () {
    const store = yield* OrchestrationEventStore;
    const sql = yield* SqlClient.SqlClient;
    yield* runMigrations({ toMigrationInclusive: 56 });
    yield* sql`
      WITH RECURSIVE history(n) AS (
        SELECT 1 UNION ALL SELECT n + 1 FROM history WHERE n < 25000
      )
      INSERT INTO orchestration_events (
        event_id, aggregate_kind, stream_id, stream_version, event_type,
        occurred_at, actor_kind, payload_json, metadata_json, application_event_version
      )
      SELECT 'history:' || n, 'thread',
        CASE WHEN n = 1 THEN 'target' ELSE 'other' END,
        n, 'provider-session.detached', ${occurredAt}, 'server', '{}', '{}',
        CASE WHEN n % 3 = 0 THEN 1 ELSE 2 END
      FROM history
    `;
    yield* runMigrations({ toMigrationInclusive: 57 });

    const statements: Array<string> = [];
    const tracer = Tracer.make({
      span(options) {
        const span = new Tracer.NativeSpan(options);
        const end = span.end.bind(span);
        span.end = (endTime, exit) => {
          end(endTime, exit);
          const query = span.attributes.get("db.query.text");
          if (typeof query === "string") statements.push(query);
        };
        return span;
      },
    });
    assert.equal(yield* store.latestApplicationSequence.pipe(Effect.withTracer(tracer)), 25_000);
    assert.equal(
      yield* store.latestAgentSequence(ThreadId.make("target")).pipe(Effect.withTracer(tracer)),
      1,
    );
    assert.equal(statements.length, 2);
    const applicationPlan = yield* sql.unsafe<{ readonly detail: string }>(
      `EXPLAIN QUERY PLAN ${statements[0]}`,
    );
    const agentPlan = yield* sql.unsafe<{ readonly detail: string }>(
      `EXPLAIN QUERY PLAN ${statements[1]}`,
      ["target"],
    );
    assert.match(
      applicationPlan.map((row) => row.detail).join("\n"),
      /SEARCH orchestration_events USING INDEX idx_orchestration_events_application_high_water/,
    );
    assert.match(
      agentPlan.map((row) => row.detail).join("\n"),
      /USING COVERING INDEX idx_orchestration_events_agent_stream_sequence \(stream_id=\?\)/,
    );
    for (const row of [...applicationPlan, ...agentPlan]) {
      assert.notMatch(row.detail, /MULTI-INDEX OR|TEMP B-TREE/);
    }
  }).pipe(
    Effect.provide(
      OrchestrationEventStoreLive.pipe(Layer.provideMerge(NodeSqliteClient.layerMemory())),
    ),
  ),
);