import fs from "node:fs/promises";
import { afterEach, describe, expect, test, vi } from "vitest";
import { DeterministicBatcher, acquireBatchOwnerLocks, batchDeclarationFingerprint, batchProgressId } from "../src/batches/runtime.js";
import type { EventCandidate, ThoughtEvent } from "../src/events/types.js";
import type { JazzThoughtStore } from "../src/jazz/store.js";
import type { BatchDeclarationManifest } from "../src/runtime/manifest.js";
import { temporaryProject, testStore } from "./helpers.js";

const stores: JazzThoughtStore[] = [];
const roots: string[] = [];

afterEach(async () => {
  vi.restoreAllMocks();
  await Promise.all(stores.splice(0).map((store) => store.close()));
  await Promise.all(roots.splice(0).map((root) => fs.rm(root, { recursive: true, force: true })));
});

describe("deterministic derived-event batching", () => {
  test("flushes one isolated commit after restart-stable quiet time", async () => {
    const { store } = await setup();
    await appendCommit(store, 1, "app.bsky.feed.post", "2026-07-21T00:00:00.000Z");
    const declaration = batchDeclaration({ quietWindowMs: 1_000 });

    expect(await new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:00.500Z")).cycle(declaration))
      .toMatchObject({ waiting: true, emitted: 0 });
    const restarted = new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:02.000Z"));
    expect(await restarted.cycle(declaration)).toMatchObject({ flushReason: "quiet-window", memberCount: 1 });
    expect(await restarted.cycle(declaration)).toMatchObject({ emitted: 0, examined: 0 });
  });

  test("cursor and lifecycle source events do not postpone quiet flush", async () => {
    const { store } = await setup();
    await appendCommit(store, 100, "app.bsky.feed.post", "2026-07-21T00:00:00.000Z");
    await appendNonmatching(store, "cursor-101", "2026-07-21T00:00:00.100Z");
    await appendNonmatching(store, "lifecycle-102", "2026-07-21T00:00:01.500Z");
    const result = await new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:02.000Z"))
      .cycle(batchDeclaration({ quietWindowMs: 1_000 }));
    expect(result).toMatchObject({ flushReason: "quiet-window", memberCount: 1 });
  });

  test("sparse isolated matching commits each flush without raw-source head blocking", async () => {
    const { store } = await setup();
    const declaration = batchDeclaration({ quietWindowMs: 100 });
    await appendCommit(store, 1, "app.bsky.feed.post", "2026-07-21T00:00:00.000Z");
    await appendNonmatching(store, "cursor-after-first", "2026-07-21T00:00:00.010Z");
    const batcher = new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:01.000Z"));
    expect(await batcher.cycle(declaration)).toMatchObject({ memberCount: 1 });
    await appendCommit(store, 2, "app.bsky.feed.post", "2026-07-21T00:00:10.000Z");
    await appendNonmatching(store, "cursor-after-second", "2026-07-21T00:00:10.010Z");
    expect(await new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:11.000Z")).cycle(declaration))
      .toMatchObject({ memberCount: 1 });
  });

  test("max-items forces a burst batch and replay is idempotent", async () => {
    const { store } = await setup();
    for (let index = 1; index <= 3; index += 1) await appendCommit(store, index, "app.bsky.feed.like", `2026-07-21T00:00:0${index}.000Z`);
    const declaration = batchDeclaration({ maxItems: 3, quietWindowMs: 60_000, maxAgeMs: 120_000 });
    const batcher = new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:04.000Z"));
    const first = await batcher.cycle(declaration);
    expect(first).toMatchObject({ flushReason: "max-items", memberCount: 3, emitted: 1 });
    const batch = await store.getEvent(first.batchEventId!);
    expect((batch!.payload.members as unknown[])).toHaveLength(3);
    expect(JSON.stringify(batch!.payload)).not.toContain("LIKE-BODY");
    expect(await batcher.cycle(declaration)).toMatchObject({ emitted: 0 });
  });

  test("max-age prevents starvation under a continuously advancing source", async () => {
    const { store } = await setup();
    for (let index = 1; index <= 4; index += 1) await appendCommit(store, index, "app.bsky.feed.like", `2026-07-21T00:00:0${index}.000Z`);
    const declaration = batchDeclaration({ maxItems: 50, quietWindowMs: 10_000, maxAgeMs: 3_000 });
    expect(await new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:04.500Z")).cycle(declaration))
      .toMatchObject({ flushReason: "max-age", memberCount: 4 });
  });

  test("crash before atomic settlement leaves progress unchanged and replay succeeds", async () => {
    const { store } = await setup();
    await appendCommit(store, 1, "app.bsky.feed.like", "2026-07-21T00:00:00.000Z");
    const declaration = batchDeclaration({ quietWindowMs: 100 });
    const original = store.settleDerivedBatch.bind(store);
    vi.spyOn(store, "settleDerivedBatch").mockRejectedValueOnce(new Error("synthetic crash boundary"));
    const batcher = new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:01.000Z"));
    await expect(batcher.cycle(declaration)).rejects.toThrow("synthetic crash boundary");
    expect(await store.getConsumerProgress(batchProgressId(declaration, RAW_SOURCE))).toBeUndefined();
    vi.mocked(store.settleDerivedBatch).mockImplementation(original);
    expect(await batcher.cycle(declaration)).toMatchObject({ emitted: 1, memberCount: 1 });
  });

  test("filtered progress crosses nonmatching rows but rejects a hidden matching interval member", async () => {
    const { store } = await setup();
    const declaration = batchDeclaration({ quietWindowMs: 100 });
    const commit10 = await appendCommit(store, 10, "app.bsky.feed.post", "2026-07-21T00:00:00.000Z");
    await appendNonmatching(store, "cursor-11", "2026-07-21T00:00:00.010Z");
    const commit12 = await appendCommit(store, 12, "app.bsky.feed.like", "2026-07-21T00:00:00.020Z");
    expect(await new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:01.000Z")).cycle(declaration))
      .toMatchObject({ memberCount: 2 });

    const { store: hiddenStore } = await setup();
    const hidden10 = await appendCommit(hiddenStore, 10, "app.bsky.feed.post", "2026-07-21T00:00:00.000Z");
    const hidden11 = await appendCommit(hiddenStore, 11, "app.bsky.feed.like", "2026-07-21T00:00:00.010Z");
    const hidden12 = await appendCommit(hiddenStore, 12, "app.bsky.feed.like", "2026-07-21T00:00:00.020Z");
    const candidate = batchCandidate(declaration, [hidden10, hidden12]);
    await expect(hiddenStore.settleDerivedBatch({
      candidate,
      progress: {
        id: batchProgressId(declaration, RAW_SOURCE), consumerId: declaration.id,
        consumerVersion: declaration.version, source: RAW_SOURCE, lastSequence: hidden12.sourceSequence,
        lastEventId: hidden12.id, updatedAt: "2026-07-21T00:00:01.000Z",
      },
      members: [hidden10, hidden12],
      priorFilteredSequence: 0,
      declaration: {
        inputEventTypes: declaration.input.eventTypes,
        acceptedPrivacy: ["public-source", "private", "sensitive"],
        source: RAW_SOURCE,
        fingerprint: String((candidate.payload.declaration as Record<string, unknown>).fingerprint),
      },
    })).rejects.toThrow("complete matching filtered interval");
    expect(await hiddenStore.getConsumerProgress(batchProgressId(declaration, RAW_SOURCE))).toBeUndefined();
    expect(await hiddenStore.listEvents({ source: declaration.output.sourceId })).toHaveLength(0);
    expect(hidden11.sourceSequence).toBeGreaterThan(hidden10.sourceSequence);
    expect(commit12.sourceSequence).toBeGreaterThan(commit10.sourceSequence);
  });

  test("producer append after replay-now captured head remains pending", async () => {
    const { store } = await setup();
    const capturedHead = await appendNonmatching(store, "captured-head", "2026-07-21T00:00:00.000Z");
    const declaration = batchDeclaration({ replay: "now", quietWindowMs: 100 });
    const batcher = new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:00.050Z"));
    expect(await batcher.cycle(declaration)).toMatchObject({ emitted: 0 });
    expect(await store.getConsumerProgress(batchProgressId(declaration, RAW_SOURCE)))
      .toMatchObject({ lastSequence: capturedHead.sourceSequence, lastEventId: capturedHead.id });
    await appendCommit(store, 2, "app.bsky.feed.post", "2026-07-21T00:00:00.060Z");
    expect(await new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:01.000Z")).cycle(declaration))
      .toMatchObject({ emitted: 1, memberCount: 1 });
  });

  test("fails closed rather than declassifying mixed privacy", async () => {
    const { store } = await setup();
    await appendCommit(store, 1, "app.bsky.feed.post", "2026-07-21T00:00:00.000Z");
    const candidate = commitCandidate(2, "app.bsky.feed.like", "2026-07-21T00:00:01.000Z");
    await store.appendEvent({ ...candidate, privacy: "private" });
    const declaration = batchDeclaration({ quietWindowMs: 100, privacy: "preserve" });
    await expect(new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:02.000Z")).cycle(declaration))
      .rejects.toThrow("different privacy classes");
    expect(await store.getConsumerProgress(batchProgressId(declaration, RAW_SOURCE))).toBeUndefined();
  });

  test("atomically merges complete prefixes from multiple sources into one most-private window", async () => {
    const { store } = await setup();
    const secondSource = "x:fixture";
    const first = await appendCommit(store, 1, "app.bsky.feed.post", "2026-07-21T00:00:00.000Z");
    const second = (await store.appendEvent({
      ...commitCandidate(2, "app.bsky.feed.like", "2026-07-21T00:00:00.100Z"),
      source: secondSource,
      sourceKind: "x-webhook",
      externalId: "x-event-2",
      idempotencyKey: "x-event-2",
      privacy: "sensitive",
    })).event;
    const third = await appendCommit(store, 3, "app.bsky.feed.like", "2026-07-21T00:00:00.200Z");
    const declaration = batchDeclaration({
      id: "fixture-activity-window",
      input: { eventTypes: ["stream.thought.source.atproto.commit"], sourceIds: [RAW_SOURCE, secondSource] },
      output: { sourceId: "batch:fixture-activity", eventType: "stream.thought.derived.event.batch" },
      quietWindowMs: 100,
      privacy: "most-private",
    });
    const batcher = new DeterministicBatcher(store, () => new Date("2026-07-21T00:00:01.000Z"));
    const result = await batcher.cycle(declaration);
    expect(result).toMatchObject({ emitted: 1, memberCount: 3, flushReason: "quiet-window" });
    const batch = await store.getEvent(result.batchEventId!);
    expect(batch).toMatchObject({ privacy: "sensitive", rootEventId: result.batchEventId });
    expect((batch!.payload.members as Array<{ eventId: string }>).map((member) => member.eventId)).toEqual([
      first.id,
      second.id,
      third.id,
    ]);
    expect(await store.getConsumerProgress(batchProgressId(declaration, RAW_SOURCE)))
      .toMatchObject({ lastSequence: third.sourceSequence, lastEventId: third.id });
    expect(await store.getConsumerProgress(batchProgressId(declaration, secondSource)))
      .toMatchObject({ lastSequence: second.sourceSequence, lastEventId: second.id });
    expect(await batcher.cycle(declaration)).toMatchObject({ emitted: 0, examined: 0 });
  });
});

describe("batch declaration owner locks", () => {
  test("rejects a live owner and releases cleanly", async () => {
    const root = await temporaryProject("thoughtstream-batch-lock-"); roots.push(root);
    const declaration = batchDeclaration();
    const identity = { bootId: async () => "boot-a", processStart: async () => "start-a", pid: 123 };
    const first = await acquireBatchOwnerLocks(root, [declaration], identity);
    await expect(acquireBatchOwnerLocks(root, [declaration], identity)).rejects.toThrow("live batch owner");
    await first.release();
    const second = await acquireBatchOwnerLocks(root, [declaration], identity);
    await second.release();
  });

  test("recovers dead pid and boot or process-start mismatches", async () => {
    const root = await temporaryProject("thoughtstream-batch-lock-"); roots.push(root);
    const declaration = batchDeclaration();
    const first = await acquireBatchOwnerLocks(root, [declaration], {
      bootId: async () => "boot-old", processStart: async () => "start-old", pid: 111,
    });
    const rebooted = await acquireBatchOwnerLocks(root, [declaration], {
      bootId: async () => "boot-new", processStart: async () => "start-new", pid: 222,
    });
    await rebooted.release();
    await first.release();

    const stale = await acquireBatchOwnerLocks(root, [declaration], {
      bootId: async () => "boot-same", processStart: async () => "old-start", pid: 333,
    });
    const reused = await acquireBatchOwnerLocks(root, [declaration], {
      bootId: async () => "boot-same",
      processStart: async (pid) => pid === 333 ? "reused-start" : "new-start",
      pid: 444,
    });
    await reused.release();
    await stale.release();

    const dead = await acquireBatchOwnerLocks(root, [declaration], {
      bootId: async () => "boot-dead", processStart: async () => "dead-start", pid: 555,
    });
    const recovered = await acquireBatchOwnerLocks(root, [declaration], {
      bootId: async () => "boot-dead",
      processStart: async (pid) => pid === 555 ? undefined : "live-start",
      pid: 666,
    });
    await recovered.release();
    await dead.release();
  });
});

const RAW_SOURCE = "jetstream:fixture";

function batchDeclaration(overrides: Partial<BatchDeclarationManifest> = {}): BatchDeclarationManifest {
  return {
    id: "fixture-atproto-batch",
    version: 1,
    enabled: true,
    input: { eventTypes: ["stream.thought.source.atproto.commit"], sourceIds: [RAW_SOURCE] },
    output: { sourceId: "batch:fixture-atproto", eventType: "stream.thought.derived.event.batch" },
    quietWindowMs: 1_000,
    maxAgeMs: 10_000,
    maxItems: 10,
    privacy: "preserve",
    replay: "beginning",
    pollIntervalMs: 100,
    ...overrides,
  };
}

async function appendCommit(store: JazzThoughtStore, index: number, collection: string, occurredAt: string): Promise<ThoughtEvent> {
  return (await store.appendEvent(commitCandidate(index, collection, occurredAt))).event;
}

async function appendNonmatching(store: JazzThoughtStore, key: string, occurredAt: string): Promise<ThoughtEvent> {
  return (await store.appendEvent({
    ...commitCandidate(999, "app.bsky.feed.post", occurredAt),
    type: "stream.thought.connector.cursor.advanced",
    externalId: key,
    idempotencyKey: key,
    payload: { cursor: key },
  })).event;
}

function batchCandidate(declaration: BatchDeclarationManifest, members: ThoughtEvent[]): EventCandidate {
  const fingerprint = sha256ForTest(declaration);
  return {
    type: declaration.output.eventType, schemaVersion: 1, source: declaration.output.sourceId,
    sourceKind: "system", externalId: `batch-${members.map((event) => event.id).join("-")}`,
    idempotencyKey: `batch-${members.map((event) => event.id).join("-")}`,
    occurredAt: members.at(-1)!.occurredAt, actor: `batch:${declaration.id}`,
    correlationId: "batch-test", privacy: "public-source",
    payload: {
      declaration: { id: declaration.id, version: declaration.version, fingerprint },
      flushReason: "quiet-window", firstOccurredAt: members[0]!.occurredAt,
      lastOccurredAt: members.at(-1)!.occurredAt,
      members: members.map((event) => ({
        eventId: event.id, source: event.source, sourceSequence: event.sourceSequence, type: event.type,
        schemaVersion: event.schemaVersion, privacy: event.privacy, occurredAt: event.occurredAt,
        observedAt: event.observedAt, payloadHash: event.payloadHash,
      })),
    },
  };
}

function sha256ForTest(declaration: BatchDeclarationManifest): string {
  return batchDeclarationFingerprint(declaration);
}

function commitCandidate(index: number, collection: string, occurredAt: string): EventCandidate {
  return {
    type: "stream.thought.source.atproto.commit",
    schemaVersion: 1,
    source: RAW_SOURCE,
    sourceKind: "jetstream",
    externalId: `at://did:plc:fixture/${collection}/${index}`,
    idempotencyKey: `commit-${index}`,
    occurredAt,
    observedAt: occurredAt,
    actor: "did:plc:fixture",
    correlationId: "synthetic-burst",
    privacy: "public-source",
    payload: {
      atUri: `at://did:plc:fixture/${collection}/${index}`,
      collection,
      operation: "create",
      record: collection === "app.bsky.feed.like"
        ? { subject: { uri: `at://did:plc:subject/app.bsky.feed.post/${index}`, cid: `bafy-${index}` }, note: "LIKE-BODY" }
        : { text: "POST-BODY" },
    },
  };
}

async function setup(): Promise<{ store: JazzThoughtStore }> {
  const root = await temporaryProject("thoughtstream-batches-");
  roots.push(root);
  const store = await testStore(root);
  stores.push(store);
  return { store };
}