import { afterEach, beforeAll, afterAll, beforeEach, describe, expect, it, vi } from "vitest";
import { GenericContainer, Wait, type StartedTestContainer } from "testcontainers";
import { Client, type PoolConfig } from "pg";
import { createRequire } from "node:module";
import { execFileSync } from "node:child_process";
import { fileURLToPath } from "node:url";
import { z } from "zod";
import type {
  BatchResult,
  JsonObject,
  ReverseEtlContext,
  ReverseEtlWriter,
  WriteBatch,
} from "@jitsu/protocols/reverse-etl";
import { contentHash } from "@jitsu/destination-functions/src/reverse-etl/identity";
import { Database } from "./persistence";
import {
  openMirrorPersistence,
  runSnapshotMirror,
  resumeSnapshotMirror,
  type NewMirrorOptions,
  type MirrorSourceRecord,
  type SnapshotMirrorAdapter,
} from "./mirror";

import { MemoryObjects, persisted } from "./artifacts/test-support";
const objects = new MemoryObjects();
const storage = { objectStorage: { store: objects, signal: new AbortController().signal } };
const durable = () => persisted(admin, objects);
afterEach(() => vi.useRealTimers());

let container: StartedTestContainer;
let admin: Client;
let db: Database;
let runtimeConfig: PoolConfig;
type Session = Awaited<ReturnType<typeof openMirrorPersistence>>;
type SourceRow = { id: string; value: string };
const record = (key: string, id = key, value = "v"): MirrorSourceRecord => ({
  key: contentHash(key),
  row: { id, value },
});
const member = (id: string) => contentHash(id.toLowerCase());
const accepted = (batch: WriteBatch<JsonObject>): BatchResult => ({
  outcomes: batch.records.map(row => ({ operationId: row.operationId, status: "accepted" })),
});

beforeAll(async () => {
  container = await new GenericContainer("postgres:18-alpine")
    .withEnvironment({ POSTGRES_PASSWORD: "test", POSTGRES_DB: "mirror_test" })
    .withExposedPorts(5432)
    .withWaitStrategy(Wait.forLogMessage(/database system is ready to accept connections/, 2))
    .start();
  const config = {
    host: container.getHost(),
    port: container.getMappedPort(5432),
    database: "mirror_test",
    user: "postgres",
    password: "test",
  };
  admin = new Client(config);
  await admin.connect();
  const url = `postgresql://postgres:test@${config.host}:${config.port}/${config.database}?schema=newjitsu`;
  execFileSync(
    process.execPath,
    [
      createRequire(import.meta.url).resolve("prisma/build/index.js"),
      "db",
      "push",
      `--schema=${fileURLToPath(new URL("../../../webapps/console/prisma/schema.prisma", import.meta.url))}`,
      "--skip-generate",
    ],
    {
      // eslint-disable-next-line no-restricted-properties -- disposable test database only.
      env: { ...process.env, DATABASE_URL: url },
      stdio: "inherit",
    }
  );
  await admin.query(
    "CREATE ROLE mirror_runtime LOGIN PASSWORD 'runtime'; GRANT USAGE ON SCHEMA newjitsu TO mirror_runtime"
  );
  for (const table of ["control", "target_owner"])
    await admin.query(`GRANT SELECT,INSERT,UPDATE,DELETE ON newjitsu.reverse_sync_${table} TO mirror_runtime`);
  await admin.query("GRANT SELECT,INSERT,UPDATE,DELETE ON newjitsu.source_state TO mirror_runtime");
  runtimeConfig = { ...config, user: "mirror_runtime", password: "runtime" };
  db = new Database(runtimeConfig, storage);
}, 60_000);
afterAll(async () => {
  await db?.close();
  await admin?.end();
  await container?.stop();
});
beforeEach(async () => {
  await admin.query("TRUNCATE newjitsu.reverse_sync_control,newjitsu.reverse_sync_target_owner,newjitsu.source_state");
});
async function session(id = "run", taskId = id, database = db) {
  return openMirrorPersistence(database, {
    workspaceId: "workspace",
    syncId: "sync",
    logicalRunId: id,
    taskId,
    configRevision: "revision",
    targetIdentity: "test/account/audience",
    mode: "mirror",
    extraction: "full",
  });
}
async function takeover(run: Session) {
  return session(run.scope.logicalRunId, "recovery");
}
async function phase() {
  return (await admin.query("SELECT phase FROM newjitsu.reverse_sync_control ORDER BY run_order DESC LIMIT 1")).rows[0]
    .phase;
}
async function membership() {
  return (await durable()).members.map(row => row.effect.upsert);
}

function fixture() {
  const calls: string[] = [];
  const requests: { action: string; batch: WriteBatch<JsonObject> }[] = [];
  const remote = new Map<string, JsonObject>();
  const receipts = new Map<string, BatchResult>();
  let projections = 0;
  let onBatch: ((action: string, batch: WriteBatch<JsonObject>) => Promise<BatchResult>) | undefined;
  let afterBatch: (() => void | Promise<void>) | undefined;
  let finishResult: "accepted" | "pending" | "throw" = "accepted";
  const controller = new AbortController();
  const writer = (ctx: ReverseEtlContext<{}, {}>): ReverseEtlWriter<JsonObject> => {
    expect(ctx).not.toHaveProperty("snapshots");
    expect(ctx).not.toHaveProperty("core");
    expect(ctx).not.toHaveProperty("db");
    const send = async (action: string, batch: WriteBatch<JsonObject>) => {
      calls.push(action);
      requests.push({ action, batch: JSON.parse(JSON.stringify(batch)) });
      for (const row of batch.records) expect(row.row).not.toHaveProperty("identityHash");
      const result = onBatch ? await onBatch(action, batch) : accepted(batch);
      receipts.set(batch.batchId, result);
      for (const outcome of result.outcomes)
        if (outcome.status === "accepted") {
          const row = batch.records.find(row => row.operationId === outcome.operationId)!.row;
          if (action === "upsert") remote.set(row.member as string, row);
          else remote.delete(row.member as string);
        }
      ctx.store.set("lastBatch", batch.batchId);
      await afterBatch?.();
      return result;
    };
    return {
      init: async () => {
        calls.push("init");
        await ctx.delivery.saveProviderState({ session: "remote-session" });
      },
      upsert: batch => send("upsert", batch),
      remove: batch => send("remove", batch),
      finish: async () => {
        calls.push("finish");
        if (finishResult === "throw") throw new Error("private-provider-token");
        return finishResult === "pending" ? { delivery: "pending", remoteJobIds: ["job"] } : { delivery: "accepted" };
      },
      abort: async () => {
        calls.push("abort");
      },
    };
  };
  const adapter: SnapshotMirrorAdapter<{}, SourceRow, {}> = {
    batchDelivery: "accepted",
    projection: {
      rowType: z.object({ id: z.string(), value: z.string() }),
      project: row => {
        projections++;
        const identity = member(row.id);
        return [{ identity, upsert: { member: identity, value: row.value }, remove: { member: identity } }];
      },
    },
    stream: {
      name: "test",
      displayName: "Test",
      rowType: z.object({ member: z.string(), value: z.string() }).strict(),
      removeRowType: z.object({ member: z.string() }).strict(),
      options: z.object({}),
      batchSize: 2,
      capabilities: {
        supportsUpsert: true,
        supportsExplicitRemove: true,
        mirror: "snapshot-diff",
        replay: "idempotent-operation",
      },
      createWriter: async ctx => {
        calls.push("create");
        return writer(ctx);
      },
    },
  };
  const options = (run: Session, rows: MirrorSourceRecord[]): NewMirrorOptions<{}, SourceRow, {}> => ({
    persistence: run,
    adapter,
    targetBaseline: "tracked",
    mapping: { id: "id", value: "value" },
    sourcePageSize: 2,
    context: { credentials: {}, options: {}, signal: controller.signal, log: {} as any, fetch: fetch as any },
    source: async function* () {
      calls.push("source");
      yield* rows;
    },
  });
  const recovery = {
    attachWriter: async (ctx: ReverseEtlContext<{}, {}>) => {
      calls.push("attach");
      return writer(ctx);
    },
    reconcileBatch: async (
      batch: WriteBatch<JsonObject>,
      _action: string,
      _saved: BatchResult | undefined,
      ctx: ReverseEtlContext<{}, {}>
    ) => {
      calls.push("reconcile");
      ctx.store.set("reconciled", true);
      return receipts.get(batch.batchId)!;
    },
    reconcileFinish: async (_saved: unknown, ctx: ReverseEtlContext<{}, {}>) => {
      calls.push("reconcileFinish");
      ctx.store.set("final", true);
      return { delivery: "accepted" as const };
    },
  };
  return {
    adapter,
    options,
    recovery,
    calls,
    requests,
    remote,
    receipts,
    controller,
    projections: () => projections,
    setBatch: (fn: typeof onBatch) => {
      onBatch = fn;
    },
    setAfterBatch: (fn: typeof afterBatch) => {
      afterBatch = fn;
    },
    setFinish: (value: typeof finishResult) => {
      finishResult = value;
    },
  };
}

describe("core snapshot mirror lifecycle", () => {
  it("uploads a complete replacement, waits for every upload, and prunes only after durable cleanup acceptance", async () => {
    const baseline = fixture();
    await runSnapshotMirror(baseline.options(await session("baseline"), [record("stay"), record("old")]));
    const f = fixture();
    f.adapter.stream.capabilities.mirror = "native-replace";
    f.adapter.batchDelivery = "asynchronous";
    f.setBatch(async (_action, batch) => ({
      outcomes: batch.records.map(row => ({ operationId: row.operationId, status: "staged" })),
      remoteJobIds: [batch.batchId],
    }));
    f.setFinish("pending");
    const options = (run: Session, rows: MirrorSourceRecord[] = []) => ({
      ...f.options(run, rows),
      targetBaseline: "replace" as const,
    });
    const run = await session("replacement");
    expect(
      (await runSnapshotMirror(options(run, [record("stay"), record("new"), record("alias", "new")]))).delivery
    ).toBe("pending");
    expect(f.requests.flatMap(r => r.batch.records)).toHaveLength(2); // Includes unchanged "stay".
    expect(f.calls).not.toContain("finish");
    await expect(run.delivery.prepareFinish(2, {})).rejects.toThrow("Unaccepted uploads");
    await expect(run.snapshots.page("removals")).rejects.toThrow("does not use individual removals");
    const wrong = fixture();
    await expect(resumeSnapshotMirror(wrong.options(await takeover(run), []), wrong.recovery)).rejects.toThrow(
      "Snapshot strategy changed"
    );

    for (const { batch } of f.requests) f.receipts.set(batch.batchId, accepted(batch));
    expect((await resumeSnapshotMirror(options(await takeover(run)), f.recovery)).delivery).toBe("pending");
    expect(await phase()).toBe("finish_pending");
    expect((await durable()).head.snapshot?.replacementStatus).toBe("pending");
    expect(await membership()).toHaveLength(3); // Cleanup is not yet accepted: retain "old".
    expect(f.calls.filter(c => c === "source")).toHaveLength(1);
    expect(f.calls.filter(c => c === "finish")).toHaveLength(1);
    expect(f.requests.some(r => r.action === "remove")).toBe(false);

    expect((await resumeSnapshotMirror(options(await takeover(run)), f.recovery)).delivery).toBe("accepted");
    expect(await phase()).toBe("complete");
    expect((await durable()).head.snapshot?.replacementStatus).toBe("accepted");
    expect((await membership()).map(m => m.member).sort()).toEqual([member("stay"), member("new")].sort());
    const fresh = await session("next-replacement");
    expect([...fresh.core.local.memberPages()].flat()).toHaveLength(2);
    expect(f.calls.filter(c => c === "finish")).toHaveLength(1); // Status polling never resubmits cleanup.

    f.setBatch(undefined);
    f.setFinish("accepted");
    f.requests.length = 0;
    await runSnapshotMirror(options(fresh, [record("stay"), record("new")]));
    expect(f.requests.flatMap(r => r.batch.records)).toHaveLength(2); // Unchanged members upload again next run.
    f.requests.length = 0;
    await runSnapshotMirror(options(await session("empty-replacement")));
    expect(f.requests).toHaveLength(0);
    expect(await membership()).toHaveLength(0);
    expect([...(await session("after-empty")).core.local.memberPages()].flat()).toHaveLength(0);
  });
  it("never cleans a replacement after rejected uploads or incomplete extraction", async () => {
    const f = fixture();
    f.adapter.stream.capabilities.mirror = "native-replace";
    const run = await session();
    const input = { ...f.options(run, [record("a")]), targetBaseline: "replace" as const };
    f.setBatch(async (_action, batch) => ({
      outcomes: batch.records.map(row => ({
        operationId: row.operationId,
        status: "rejected",
        code: "invalid",
        safeReason: "Invalid",
      })),
    }));
    await expect(runSnapshotMirror(input)).rejects.toThrow();
    expect(f.calls).not.toContain("finish");
    const next = await session("incomplete");
    const incomplete = {
      ...f.options(next, []),
      targetBaseline: "replace" as const,
      source: async function* () {
        yield record("a");
        throw new Error("source failed");
      },
    };
    await expect(runSnapshotMirror(incomplete)).rejects.toThrow();
    expect(f.calls).not.toContain("finish");
  });
  it("retains an ambiguous replacement cleanup for reconciliation without resubmitting it", async () => {
    const f = fixture();
    f.adapter.stream.capabilities.mirror = "native-replace";
    f.setFinish("throw");
    const run = await session();
    const options = { ...f.options(run, [record("a")]), targetBaseline: "replace" as const };
    await expect(runSnapshotMirror(options)).rejects.toThrow();
    expect(await phase()).toBe("finish_prepared");
    await expect(
      resumeSnapshotMirror(
        { ...options, persistence: await takeover(run) },
        {
          ...f.recovery,
          reconcileFinish: async saved => {
            expect(saved).toBeUndefined();
            throw new Error("Receipt unavailable; do not replay");
          },
        }
      )
    ).rejects.toThrow();
    expect(f.calls.filter(c => c === "finish")).toHaveLength(1);
    expect((await durable()).head.snapshot?.replacementStatus).toBe("prepared");
  });
  it("stages a full page with bounded SQL and seals exact retry/deduplication counts", async () => {
    const run = await session();
    await run.delivery.prepareInit({});
    await run.delivery.acknowledgeInit({});
    await run.snapshots.start();
    const identity = { identity: "shared", upsert: { id: "shared" }, remove: { id: "shared" } };
    const page = Array.from({ length: 1000 }, (_, i) => ({ key: contentHash(i), identities: [identity] }));
    const query = vi.spyOn(Client.prototype, "query");
    try {
      await run.snapshots.append(page, 1);
      expect(query.mock.calls.length).toBeLessThanOrEqual(10);
    } finally {
      query.mockRestore();
    }
    const before = await run.snapshots.status();
    expect(before?.sourceKeyCount).toBe(1000);
    await run.snapshots.append(page, 1);
    expect(await run.snapshots.status()).toEqual(before);
    await run.snapshots.append([{ key: contentHash(1000), identities: [identity] }], 2);
    const after = await run.snapshots.status();
    expect(after?.sourceKeyCount).toBe(1001);
    await expect(
      run.snapshots.append([{ key: contentHash(1001), identities: [{ ...identity, remove: { id: "different" } }] }], 3)
    ).rejects.toThrow("Conflicting payloads");
    expect(await run.snapshots.status()).toEqual(after);
    expect(await run.snapshots.seal()).toEqual({ uniqueMembers: 1 });
    expect((await durable()).head.snapshot).toMatchObject({ sealed: true, keys: 1001, entries: 1001, page: 2 });
    expect(await run.snapshots.page("additions")).toHaveLength(1);
  });
  it("refreshes only due unchanged members, using acceptance time rather than source changes", async () => {
    const f = fixture();
    f.adapter.refreshAfterMs = 30 * 86400_000;
    const rows = [record("old"), record("fresh")];
    const now = Date.now();
    vi.useFakeTimers({ toFake: ["Date"] });
    vi.setSystemTime(now - 31 * 86400_000);
    await runSnapshotMirror(f.options(await session("baseline"), [record("old")]));
    vi.setSystemTime(now);
    // Add a fresh member without refreshing the old one yet.
    f.adapter.refreshAfterMs = 60 * 86400_000;
    await runSnapshotMirror(f.options(await session("add-fresh"), rows));
    f.requests.length = 0;
    await runSnapshotMirror(f.options(await session("unchanged"), rows));
    expect(f.requests).toHaveLength(0);
    f.adapter.refreshAfterMs = 30 * 86400_000;
    const refresh = await session("refresh");
    await runSnapshotMirror(f.options(refresh, rows));
    expect(f.requests.map(r => r.action)).toEqual(["upsert"]);
    expect(f.requests[0].batch.records.map(r => r.row.member)).toEqual([member("old")]);
    f.requests.length = 0;
    const next = await session("fresh-again");
    await runSnapshotMirror(f.options(next, rows));
    expect(f.requests).toHaveLength(0);
  });

  it("waits for refresh acceptance before removals and keeps its cutoff and payload on recovery", async () => {
    const f = fixture();
    f.adapter.refreshAfterMs = 30 * 86400_000;
    await runSnapshotMirror(f.options(await session("baseline"), [record("stay"), record("gone")]));
    vi.useFakeTimers({ toFake: ["Date"] });
    vi.setSystemTime(Date.now() + 31 * 86400_000);
    f.requests.length = 0;
    f.adapter.batchDelivery = "asynchronous";
    f.setBatch(async (_action, batch) => ({
      outcomes: batch.records.map(r => ({ operationId: r.operationId, status: "staged" })),
      remoteJobIds: [batch.batchId],
    }));
    const run = await session("refresh");
    expect((await runSnapshotMirror(f.options(run, [record("stay")]))).delivery).toBe("pending");
    const cutoff = (await durable()).head.snapshot!.refreshBefore;
    expect(f.requests.map(r => r.action)).toEqual(["upsert"]);
    const addition = f.requests[0].batch;
    // A later recovery must not advance the saved generation cutoff or reproject rows.
    f.adapter.refreshAfterMs = 1;
    const resumed = await takeover(run);
    f.calls.length = 0;
    expect((await resumeSnapshotMirror(f.options(resumed, []), f.recovery)).delivery).toBe("pending");
    expect(f.calls).toEqual(["reconcile"]);
    expect((await durable()).head.snapshot!.refreshBefore).toEqual(cutoff);
    expect(f.requests[0].batch).toEqual(addition);
    f.receipts.set(addition.batchId, accepted(addition));
    expect((await resumeSnapshotMirror(f.options(await takeover(run), []), f.recovery)).delivery).toBe("pending");
    expect(f.requests.map(r => r.action)).toEqual(["upsert", "remove"]);
    const removal = f.requests[1].batch;
    f.receipts.set(removal.batchId, accepted(removal));
    expect((await resumeSnapshotMirror(f.options(await takeover(run), []), f.recovery)).delivery).toBe("accepted");
    expect(f.projections()).toBe(3);
    expect(await membership()).toHaveLength(1);
  });

  it("reconciles asynchronous additions before removals and removals before finalization", async () => {
    const f = fixture();
    await runSnapshotMirror(f.options(await session("baseline"), [record("old")]));
    f.calls.length = 0;
    f.requests.length = 0;
    f.adapter.batchDelivery = "asynchronous";
    f.setBatch(async (_action, batch) => ({
      outcomes: batch.records.map(row => ({ operationId: row.operationId, status: "staged" })),
      remoteJobIds: [batch.batchId],
    }));
    const run = await session("async");
    const rows = [record("a"), record("b"), record("c")];
    expect(await runSnapshotMirror(f.options(run, rows))).toEqual({ delivery: "pending", sourceSequence: 3 });
    expect(f.calls).toEqual(["create", "init", "source", "upsert", "upsert"]);
    expect(await membership()).toHaveLength(1);
    expect(await phase()).toBe("running");
    expect((await run.snapshots.status())?.sealed).toBe(true);

    // Poll all durable requests without opening a source or reattaching a writer.
    f.calls.length = 0;
    expect((await resumeSnapshotMirror(f.options(await takeover(run), []), f.recovery)).delivery).toBe("pending");
    expect(f.calls).toEqual(["reconcile", "reconcile"]);
    expect(f.requests).toHaveLength(2);

    for (const request of f.requests) f.receipts.set(request.batch.batchId, accepted(request.batch));
    f.calls.length = 0;
    expect(await resumeSnapshotMirror(f.options(await takeover(run), []), f.recovery)).toEqual({
      delivery: "pending",
      sourceSequence: 4,
    });
    expect(f.calls).toEqual(["reconcile", "reconcile", "attach", "remove"]);
    expect(await membership()).toHaveLength(4);
    const removal = f.requests.at(-1)!;
    expect(removal.action).toBe("remove");
    f.receipts.set(removal.batch.batchId, accepted(removal.batch));
    f.calls.length = 0;
    expect((await resumeSnapshotMirror(f.options(await takeover(run), []), f.recovery)).delivery).toBe("accepted");
    expect(f.calls).toEqual(["reconcile", "attach", "finish"]);
    expect(await membership()).toHaveLength(3);
    expect(await phase()).toBe("complete");
    expect(f.projections()).toBe(4); // Once for baseline and once per new source row, never on recovery.
  });

  it.each([
    ["snapshot-diff", "staged"],
    ["snapshot-diff", "accepted"],
    ["native-replace", "staged"],
    ["native-replace", "accepted"],
  ] as const)(
    "resumes unsent %s uploads while earlier batches are pending and new batches are %s",
    async (strategy, outcome) => {
      const baseline = fixture();
      await runSnapshotMirror(baseline.options(await session("baseline"), [record("old")]));
      const f = fixture();
      f.adapter.stream.capabilities.mirror = strategy;
      f.adapter.batchDelivery = "asynchronous";
      f.setBatch(async (_action, batch) => ({
        outcomes: batch.records.map(row => ({ operationId: row.operationId, status: "staged" })),
        remoteJobIds: [batch.batchId],
      }));
      const options = (run: Session, rows: MirrorSourceRecord[] = []) => ({
        ...f.options(run, rows),
        targetBaseline: strategy === "native-replace" ? ("replace" as const) : ("tracked" as const),
      });
      const run = await session("interrupted");
      const acknowledge = run.delivery.acknowledge;
      let submitted = 0;
      run.delivery = {
        ...run.delivery,
        acknowledge: async (...args) => {
          await acknowledge(...args);
          if (++submitted === 2) throw new Error("worker stopped after second upload");
        },
      };
      const rows = Array.from({ length: 7 }, (_, i) => record(`new-${i}`));
      await expect(runSnapshotMirror(options(run, rows))).rejects.toThrow();
      expect(f.requests.map(r => r.batch.records.length)).toEqual([2, 2]);
      if (outcome === "accepted") f.setBatch(async (_action, batch) => accepted(batch));

      f.calls.length = 0;
      expect(await resumeSnapshotMirror(options(await takeover(run)), f.recovery)).toEqual({
        delivery: "pending",
        sourceSequence: 7,
      });
      expect(f.calls).toEqual(["reconcile", "reconcile", "attach", "upsert", "upsert"]);
      expect(f.requests.map(r => r.batch.records.length)).toEqual([2, 2, 2, 1]);
      const uploaded = f.requests.flatMap(r => r.batch.records);
      expect(new Set(uploaded.map(r => r.operationId)).size).toBe(7);
      expect(new Set(uploaded.map(r => r.row.member)).size).toBe(7);
      expect(uploaded.map(r => r.sourceSequence)).toEqual([1, 2, 3, 4, 5, 6, 7]);
      expect(await membership()).toContainEqual({ member: member("old"), value: "v" });
      expect(await phase()).toBe("running");

      // Repeated refreshes don't replay submitted rows or authorize deletions.
      f.calls.length = 0;
      expect((await resumeSnapshotMirror(options(await takeover(run)), f.recovery)).delivery).toBe("pending");
      expect(f.calls.every(call => call === "reconcile")).toBe(true);
      expect(f.requests).toHaveLength(4);

      for (const { batch } of f.requests) f.receipts.set(batch.batchId, accepted(batch));
      f.setBatch(async (_action, batch) => accepted(batch));
      expect((await resumeSnapshotMirror(options(await takeover(run)), f.recovery)).delivery).toBe("accepted");
      expect(await membership()).toHaveLength(7);
      expect(await membership()).not.toContainEqual({ member: member("old"), value: "v" });
      expect(f.projections()).toBe(7);
      expect(f.requests.filter(r => r.action === "upsert")).toHaveLength(4);
      expect(await phase()).toBe("complete");
    }
  );

  it.each(["unknown", "missing-job-id", "rejected"] as const)(
    "does not continue unsent uploads when reconciliation is %s",
    async status => {
      const f = fixture();
      f.adapter.batchDelivery = "asynchronous";
      f.setBatch(async (_action, batch) => ({
        outcomes: batch.records.map(row => ({ operationId: row.operationId, status: "staged" })),
        remoteJobIds: [batch.batchId],
      }));
      const run = await session();
      const acknowledge = run.delivery.acknowledge;
      run.delivery = {
        ...run.delivery,
        acknowledge: async (...args) => {
          await acknowledge(...args);
          throw new Error("worker stopped");
        },
      };
      await expect(runSnapshotMirror(f.options(run, [record("a"), record("b"), record("c")]))).rejects.toThrow();
      f.calls.length = 0;
      await expect(
        resumeSnapshotMirror(f.options(await takeover(run), []), {
          ...f.recovery,
          reconcileBatch: async batch => {
            if (status === "unknown") throw new Error("submission cannot be reconciled");
            return {
              outcomes: batch.records.map(row =>
                status === "rejected"
                  ? {
                      operationId: row.operationId,
                      status: "rejected" as const,
                      code: "invalid",
                      safeReason: "Invalid",
                    }
                  : { operationId: row.operationId, status: "staged" as const }
              ),
            };
          },
        })
      ).rejects.toThrow();
      expect(f.calls).toEqual([]);
      expect(f.requests).toHaveLength(1);
      expect(await phase()).toBe("running");
    }
  );

  it("stops async reconciliation on a permanent rejection and preserves known accepted effects", async () => {
    const f = fixture();
    f.adapter.batchDelivery = "asynchronous";
    f.setBatch(async (_action, batch) => ({
      outcomes: batch.records.map(row => ({ operationId: row.operationId, status: "staged" })),
      remoteJobIds: [batch.batchId],
    }));
    const run = await session();
    await runSnapshotMirror(f.options(run, [record("a"), record("b")]));
    const batch = f.requests[0].batch;
    f.receipts.set(batch.batchId, {
      outcomes: batch.records.map((row, i) =>
        i === 0
          ? { operationId: row.operationId, status: "accepted" }
          : { operationId: row.operationId, status: "rejected", code: "invalid", safeReason: "Invalid" }
      ),
    });
    f.calls.length = 0;
    await expect(resumeSnapshotMirror(f.options(await takeover(run), []), f.recovery)).rejects.toThrow();
    expect(f.calls).toEqual(["reconcile"]);
    expect(await membership()).toHaveLength(1);
    expect(await phase()).toBe("running");
  });

  it("does not resubmit an asynchronous request after an ambiguous acknowledgement", async () => {
    const f = fixture();
    f.adapter.batchDelivery = "asynchronous";
    f.setBatch(async (_action, batch) => ({
      outcomes: batch.records.map(row => ({ operationId: row.operationId, status: "staged" })),
      remoteJobIds: [batch.batchId],
    }));
    const run = await session();
    const acknowledge = run.delivery.acknowledge;
    run.delivery = {
      ...run.delivery,
      acknowledge: async (...args) => {
        await acknowledge(...args);
        throw new Error("lost database response");
      },
    };
    await expect(runSnapshotMirror(f.options(run, [record("a")]))).rejects.toThrow();
    f.calls.length = 0;
    expect((await resumeSnapshotMirror(f.options(await takeover(run), []), f.recovery)).delivery).toBe("pending");
    expect(f.calls).toEqual(["reconcile"]);
    expect(f.requests).toHaveLength(1);
  });

  it("keeps an async request without a recoverable ID blocked", async () => {
    const f = fixture();
    f.adapter.batchDelivery = "asynchronous";
    f.setBatch(async (_action, batch) => ({
      outcomes: batch.records.map(row => ({ operationId: row.operationId, status: "staged" })),
    }));
    const run = await session();
    await expect(runSnapshotMirror(f.options(run, [record("a")]))).rejects.toThrow();
    expect(f.calls).not.toContain("remove");
    expect(f.calls).not.toContain("finish");
    expect(f.calls).not.toContain("abort");
    expect(await phase()).toBe("running");
  });

  it("collects all input before bounded writes and final promotion", async () => {
    const f = fixture(),
      run = await session();
    expect(await runSnapshotMirror(f.options(run, [record("a"), record("b"), record("c")]))).toEqual({
      delivery: "accepted",
      sourceSequence: 3,
    });
    expect(f.calls).toEqual(["create", "init", "source", "upsert", "upsert", "finish"]);
    expect(f.requests.map(r => r.batch.records.length)).toEqual([2, 1]);
    expect(f.projections()).toBe(3);
    expect(await membership()).toHaveLength(3);
    expect(await phase()).toBe("complete");
    expect(
      (
        await admin.query(
          "SELECT committed_generation FROM newjitsu.reverse_sync_control ORDER BY run_order DESC LIMIT 1"
        )
      ).rows[0].committed_generation
    ).toBe("run");
  });
  it("handles unchanged, shared and changed identities across generations without rehashing", async () => {
    const f = fixture(),
      first = await session();
    await runSnapshotMirror(f.options(first, [record("key1", "Shared"), record("key2", "SHARED"), record("old")]));
    expect(f.requests.flatMap(r => r.batch.records.map(row => row.row.member)).sort()).toEqual(
      [member("shared"), member("old")].sort()
    );
    const second = await session("second");
    f.calls.length = 0;
    f.requests.length = 0;
    await runSnapshotMirror(f.options(second, [record("key2", "shared"), record("old", "new")]));
    expect(f.requests.map(r => [r.action, r.batch.records.map(row => row.row.member)])).toEqual([
      ["upsert", [member("new")]],
      ["remove", [member("old")]],
    ]);
    expect(await membership()).toHaveLength(2);
    const third = await session("third");
    f.requests.length = 0;
    await runSnapshotMirror(f.options(third, []));
    expect(f.requests.map(r => r.action)).toEqual(["remove"]);
    expect(await membership()).toEqual([]);
    expect(f.remote.size).toBe(0);
  });
  it("excludes empty projections, retains shared members, and removes them only after their last projection disappears", async () => {
    const f = fixture();
    const project = f.adapter.projection.project;
    f.adapter.projection.project = row => (row.value === "excluded" ? [] : project(row));
    await runSnapshotMirror(f.options(await session(), [record("a", "shared"), record("b", "SHARED"), record("old")]));
    f.requests.length = 0;
    const rows = [
      record("a", "shared", "excluded"),
      record("b", "SHARED"),
      record("old", "old", "excluded"),
      record("new"),
    ];
    expect(await runSnapshotMirror(f.options(await session("second"), rows))).toEqual({
      delivery: "accepted",
      sourceSequence: 2,
    });
    expect(f.requests.map(r => [r.action, r.batch.records.map(row => row.row.member)])).toEqual([
      ["upsert", [member("new")]],
      ["remove", [member("old")]],
    ]);
    expect([...f.remote.keys()].sort()).toEqual([member("shared"), member("new")].sort());
    expect((await durable()).head.snapshot!.keys).toBe(4);
    const third = await session("third");
    f.requests.length = 0;
    const excluded = rows.map(row => ({ ...row, row: { ...row.row, value: "excluded" } }));
    expect(await runSnapshotMirror(f.options(third, excluded))).toEqual({ delivery: "accepted", sourceSequence: 2 });
    expect(f.requests.map(r => r.action)).toEqual(["remove"]);
    expect(f.remote.size).toBe(0);
    expect(await membership()).toEqual([]);
    expect(await phase()).toBe("complete");
  });
  it("finishes an all-excluded initial snapshot without provider batches", async () => {
    const f = fixture();
    f.adapter.projection.project = () => [];
    expect(await runSnapshotMirror(f.options(await session(), [record("a"), record("b")]))).toEqual({
      delivery: "accepted",
      sourceSequence: 0,
    });
    expect(f.calls).toEqual(["create", "init", "source", "finish"]);
    expect(f.requests).toEqual([]);
    expect(await phase()).toBe("complete");
  });
  it.each(["invalid-row", "invalid-key", "duplicate-key"])(
    "does not hide %s behind an empty projection",
    async failure => {
      const f = fixture();
      await runSnapshotMirror(f.options(await session(), [record("old")]));
      f.requests.length = 0;
      f.adapter.projection.project = () => [];
      const bad =
        failure === "invalid-row"
          ? { key: contentHash("bad"), row: { id: 42, value: "excluded" } }
          : failure === "invalid-key"
          ? { ...record("bad"), key: "invalid" }
          : record("a");
      await expect(
        runSnapshotMirror(f.options(await session("second"), [record("a"), record("b"), bad]))
      ).rejects.toThrow("Snapshot mirror stopped");
      expect(f.requests).toEqual([]);
      expect(f.remote.size).toBe(1);
      expect(await membership()).toHaveLength(1);
    }
  );
  it.each(["throw", "invalid-payload"])("does not turn a %s projection failure into exclusion", async failure => {
    const f = fixture();
    await runSnapshotMirror(f.options(await session(), [record("old")]));
    f.requests.length = 0;
    f.adapter.projection.project = row => {
      if (row.id !== "bad") return [];
      if (failure === "throw") throw new Error("Invalid source data");
      return [{ identity: "bad", upsert: { member: 42, value: "v" }, remove: { member: "bad" } }];
    };
    await expect(
      runSnapshotMirror(f.options(await session("second"), [record("a"), record("b"), record("bad")]))
    ).rejects.toThrow("Snapshot mirror stopped");
    expect(f.requests).toEqual([]);
    expect(f.remote.size).toBe(1);
    expect(await membership()).toHaveLength(1);
  });
  it("finishes empty and unchanged snapshots explicitly without synthetic batches", async () => {
    const f = fixture(),
      first = await session();
    await runSnapshotMirror(f.options(first, []));
    expect(f.calls).toEqual(["create", "init", "source", "finish"]);
    const second = await session("second");
    await runSnapshotMirror(f.options(second, [record("a")]));
    const third = await session("third");
    f.calls.length = 0;
    f.requests.length = 0;
    expect(await runSnapshotMirror(f.options(third, [record("a")]))).toEqual({
      delivery: "accepted",
      sourceSequence: 0,
    });
    expect(f.calls).toEqual(["create", "init", "source", "finish"]);
    expect(f.requests).toEqual([]);
  });
  it.each(["invalid", "duplicate", "conflict", "source-throw"])(
    "suppresses all delivery for late %s source failure",
    async failure => {
      const f = fixture(),
        run = await session();
      const rows = [
        record("a"),
        record("b"),
        failure === "duplicate"
          ? record("a")
          : failure === "conflict"
          ? record("c", "a", "different")
          : { key: contentHash("c"), row: { id: 42, value: "v" } },
      ];
      const input = f.options(run, rows);
      if (failure === "source-throw")
        input.source = async function* () {
          yield record("a");
          throw new Error("private-source-data");
        };
      await expect(runSnapshotMirror(input)).rejects.toThrow("Snapshot mirror stopped");
      expect(f.requests).toEqual([]);
      expect(f.calls.at(-1)).toBe("abort");
      expect(await phase()).toBe("aborted");
      expect(await membership()).toEqual([]);
    }
  );
  it.each(["staged", "rejected"] as const)(
    "persists %s results before suppressing removal and promotion",
    async status => {
      const f = fixture(),
        first = await session();
      await runSnapshotMirror(f.options(first, [record("old")]));
      const run = await session("next");
      f.calls.length = 0;
      f.setBatch((_action, batch) =>
        Promise.resolve({
          outcomes: batch.records.map((row, i) =>
            i === 0
              ? { operationId: row.operationId, status: "accepted" }
              : status === "staged"
              ? { operationId: row.operationId, status }
              : { operationId: row.operationId, status, code: "invalid", safeReason: "Invalid row" }
          ),
        })
      );
      await expect(runSnapshotMirror(f.options(run, [record("a"), record("b")]))).rejects.toThrow();
      expect(f.calls).not.toContain("remove");
      expect(f.calls).not.toContain("finish");
      expect(f.calls.at(-1)).toBe("abort");
      expect(await membership()).toHaveLength(2);
      expect(
        (
          await admin.query(
            "SELECT committed_generation FROM newjitsu.reverse_sync_control ORDER BY run_order DESC LIMIT 1"
          )
        ).rows[0].committed_generation
      ).toBe("run");
    }
  );
  it("removes accepted failed-run additions on the next successful full snapshot", async () => {
    const f = fixture(),
      first = await session();
    f.setBatch((_action, batch) =>
      Promise.resolve({
        outcomes: batch.records.map((row, i) =>
          i
            ? { operationId: row.operationId, status: "rejected", code: "invalid", safeReason: "Invalid" }
            : { operationId: row.operationId, status: "accepted" }
        ),
      })
    );
    await expect(runSnapshotMirror(f.options(first, [record("a"), record("b")]))).rejects.toThrow();
    expect(await membership()).toHaveLength(1);
    const next = await session("next");
    f.setBatch(undefined);
    await runSnapshotMirror(f.options(next, []));
    expect(await membership()).toEqual([]);
    expect(f.remote.size).toBe(0);
  });
  it("reconciles an ambiguous batch from its exact stored payload without source replay", async () => {
    const f = fixture(),
      run = await session();
    f.setAfterBatch(() => {
      throw new Error("private-remote-response");
    });
    await expect(runSnapshotMirror(f.options(run, [record("a"), record("b"), record("c")]))).rejects.toThrow();
    expect(await phase()).toBe("running");
    expect(f.calls).not.toContain("abort");
    expect(await membership()).toEqual([]);
    const recovered = await takeover(run);
    f.setAfterBatch(undefined);
    const before = f.projections();
    f.calls.length = 0;
    const options = f.options(recovered, [record("changed-query")]);
    expect(await resumeSnapshotMirror(options, f.recovery)).toEqual({ delivery: "accepted", sourceSequence: 3 });
    expect(f.calls).toEqual(["reconcile", "attach", "upsert", "finish"]);
    expect(f.projections()).toBe(before);
    expect(await membership()).toHaveLength(3);
    expect((await recovered.core.state()).store).toMatchObject({ reconciled: true });
    expect(f.requests[0].batch.records[0].row.member).toMatch(/^[a-f0-9]{64}$/);
  });
  it("keeps unknown work blocked without reconciliation and rejects non-recovery sessions", async () => {
    const f = fixture(),
      run = await session();
    f.setAfterBatch(() => {
      throw new Error("lost response");
    });
    await expect(runSnapshotMirror(f.options(run, [record("a")]))).rejects.toThrow();
    const recovered = await takeover(run);
    const before = f.calls.length;
    await expect(
      resumeSnapshotMirror(f.options(recovered, []), { attachWriter: f.recovery.attachWriter })
    ).rejects.toThrow();
    await expect(resumeSnapshotMirror(f.options(run, []), f.recovery)).rejects.toThrow();
    expect(f.calls).toHaveLength(before);
    expect(await phase()).toBe("running");
  });
  it.each(["pending", "throw"] as const)(
    "reconciles %s finalization without a second finish or writer initialization",
    async outcome => {
      const f = fixture(),
        run = await session();
      f.setFinish(outcome);
      if (outcome === "pending")
        expect((await runSnapshotMirror(f.options(run, [record("a")]))).delivery).toBe("pending");
      else await expect(runSnapshotMirror(f.options(run, [record("a")]))).rejects.toThrow();
      expect(f.calls).not.toContain("abort");
      const recovered = await takeover(run);
      f.calls.length = 0;
      expect(await resumeSnapshotMirror(f.options(recovered, []), f.recovery)).toEqual({
        delivery: "accepted",
        sourceSequence: 1,
      });
      expect(f.calls).toEqual(["reconcileFinish"]);
      expect((await recovered.core.state()).store.final).toBe(true);
      expect(await phase()).toBe("complete");
    }
  );
  it("retries local final-state promotion without repeating accepted finalization", async () => {
    const f = fixture(),
      run = await session();
    await admin.query(
      "ALTER TABLE newjitsu.source_state ADD CONSTRAINT injected_commit_failure CHECK (false) NOT VALID"
    );
    try {
      await expect(runSnapshotMirror(f.options(run, [record("a")]))).rejects.toThrow();
    } finally {
      await admin.query("ALTER TABLE newjitsu.source_state DROP CONSTRAINT injected_commit_failure");
    }
    expect(await phase()).toBe("finish_accepted");
    expect(f.calls).not.toContain("abort");
    expect(
      (
        await admin.query(
          "SELECT committed_generation FROM newjitsu.reverse_sync_control ORDER BY run_order DESC LIMIT 1"
        )
      ).rows[0].committed_generation
    ).toBeNull();
    const recovered = await takeover(run);
    f.calls.length = 0;
    await resumeSnapshotMirror(f.options(recovered, []), f.recovery);
    expect(f.calls).toEqual([]);
    expect(await phase()).toBe("complete");
  });
  it("stops cancellation after acceptance before any removals", async () => {
    const f = fixture(),
      first = await session();
    await runSnapshotMirror(f.options(first, [record("old")]));
    const next = await session("next");
    f.calls.length = 0;
    f.setAfterBatch(() => {
      f.controller.abort();
    });
    await expect(runSnapshotMirror(f.options(next, [record("new")]))).rejects.toThrow();
    expect(f.calls).not.toContain("remove");
    expect(f.calls).not.toContain("finish");
    expect(await membership()).toHaveLength(2);
  });
  it("rejects unsupported adapter capabilities before provider calls", async () => {
    const f = fixture(),
      run = await session();
    f.adapter.batchDelivery = "finish" as any;
    await expect(runSnapshotMirror(f.options(run, []))).rejects.toThrow(/per-batch/);
    expect(f.calls).toEqual([]);
  });
  it("splits provider requests by exact manifest bytes", async () => {
    const f = fixture(),
      run = await session();
    f.adapter.stream.batchSize = 100;
    const input = f.options(
      run,
      Array.from({ length: 5 }, (_, i) => record(String(i), String(i), "x".repeat(400)))
    );
    input.maxBatchBytes = 1400;
    await runSnapshotMirror(input);
    expect(f.requests).toHaveLength(5);
    for (const request of f.requests) expect(Buffer.byteLength(JSON.stringify(request.batch))).toBeLessThan(1400);
  });
  it("stops on a rejected removal while retaining accepted removal progress", async () => {
    const f = fixture(),
      first = await session();
    await runSnapshotMirror(f.options(first, [record("a"), record("b"), record("c")]));
    const next = await session("next");
    f.calls.length = 0;
    f.setBatch((_action, batch) =>
      Promise.resolve({
        outcomes: batch.records.map((row, i) =>
          i
            ? { operationId: row.operationId, status: "rejected", code: "invalid", safeReason: "Rejected removal" }
            : { operationId: row.operationId, status: "accepted" }
        ),
      })
    );
    await expect(runSnapshotMirror(f.options(next, []))).rejects.toThrow();
    expect(f.calls.filter(call => call === "remove")).toHaveLength(1);
    expect(f.calls).not.toContain("finish");
    expect(await membership()).toHaveLength(2);
    expect(
      (
        await admin.query(
          "SELECT committed_generation FROM newjitsu.reverse_sync_control ORDER BY run_order DESC LIMIT 1"
        )
      ).rows[0].committed_generation
    ).toBe("run");
  });
  it("reconciles an ambiguous removal without recreating the writer or replaying the remove", async () => {
    const f = fixture(),
      first = await session();
    await runSnapshotMirror(f.options(first, [record("a")]));
    const next = await session("next");
    f.setAfterBatch(() => {
      throw new Error("lost removal response");
    });
    await expect(runSnapshotMirror(f.options(next, []))).rejects.toThrow();
    expect(await membership()).toHaveLength(1);
    const recovered = await takeover(next);
    f.setAfterBatch(undefined);
    f.calls.length = 0;
    await resumeSnapshotMirror(f.options(recovered, []), f.recovery);
    expect(f.calls).toEqual(["reconcile", "attach", "finish"]);
    expect(await membership()).toEqual([]);
  });
  it("does not reopen incomplete input during recovery", async () => {
    const f = fixture(),
      first = await session();
    await first.delivery.prepareInit({});
    await first.delivery.acknowledgeInit({});
    await first.snapshots.start();
    const recovered = await takeover(first);
    await expect(resumeSnapshotMirror(f.options(recovered, [record("changed")]), f.recovery)).rejects.toThrow();
    expect(f.calls).toEqual([]);
    expect(await phase()).toBe("running");
  });
  it("does not initialize a writer after cancellation during construction", async () => {
    const f = fixture(),
      run = await session();
    const createWriter = f.adapter.stream.createWriter;
    f.adapter.stream.createWriter = async ctx => {
      const writer = await createWriter(ctx);
      f.controller.abort();
      return writer;
    };
    await expect(runSnapshotMirror(f.options(run, [record("a")]))).rejects.toThrow();
    expect(f.controller.signal.aborted).toBe(true);
    expect(f.calls).toEqual(["create"]);
    expect(await phase()).toBe("init_prepared");
    expect(await membership()).toEqual([]);
  });
  it("does not construct a writer or open a source after cancellation", async () => {
    const f = fixture(),
      run = await session();
    f.controller.abort();
    await expect(runSnapshotMirror(f.options(run, []))).rejects.toThrow();
    expect(f.calls).toEqual([]);
    expect(await phase()).toBe("new");
  });
  it("enforces snapshot storage limits before any delivery", async () => {
    const limited = new Database(runtimeConfig, { ...storage, limits: { snapshotEntries: 1 } });
    try {
      const f = fixture(),
        run = await session("run", "run", limited);
      await expect(runSnapshotMirror(f.options(run, [record("a"), record("b")]))).rejects.toThrow();
      expect(f.requests).toEqual([]);
      expect(await phase()).toBe("aborted");
    } finally {
      await limited.close();
    }
  });
  // Optional index/pagination scale check; deliberately seeds SQL rows rather than exercising million-row provider calls.
  // eslint-disable-next-line no-restricted-properties -- opt-in test workload, not application configuration.
  it.skipIf(process.env.RETL_MIRROR_SCALE_TEST !== "1")(
    "bounds diff pages with one million desired identities",
    async () => {
      const run = await session();
      await run.delivery.prepareInit({});
      await run.delivery.acknowledgeInit({});
      await run.snapshots.start();
      run.core.local.sql.exec(
        "WITH RECURSIVE n(x) AS (VALUES(1) UNION ALL SELECT x+1 FROM n WHERE x<999999) INSERT INTO desired SELECT printf('%064x',x),'unchanged','{}' FROM n; INSERT INTO members SELECT identity,payload,value,'2026-01-01T00:00:00.000Z',0 FROM desired;"
      );
      const f = fixture();
      const desired = f.adapter.projection.project({ id: "last", value: "v" })[0];
      const effect = {
        ...desired,
        identityHash: contentHash(desired.identity),
        payloadHash: contentHash(desired.upsert),
      };
      run.core.local.sql
        .prepare("INSERT INTO desired VALUES (?,?,?)")
        .run(effect.identityHash, effect.payloadHash, JSON.stringify(effect));
      run.core.local.sql.exec("ANALYZE");
      await run.snapshots.seal();
      expect(await run.snapshots.page("additions", "", 2)).toEqual([effect]);
      expect(await run.snapshots.page("additions", effect.identityHash, 2)).toEqual([]);
      expect(Number(run.core.local.sql.prepare("SELECT count(*) AS n FROM desired").get()!.n)).toBe(1_000_000);
    },
    120_000
  );
});