import { afterAll, beforeAll, beforeEach, describe, expect, test } from "bun:test";
import { seedRow } from "@cosmicdrift/kumiko-framework/testing";
import { z } from "zod";
import type { TableColumns } from "../../db/dialect";
import { createEventStoreExecutor } from "../../db/event-store-executor";
import { asRawClient, selectMany } from "../../db/query";
import { buildEntityTable } from "../../db/table-builder";
import {
  createEntity,
  createNumberField,
  createTextField,
  defineFeature,
  type EntityId,
  HookPhases,
  type SaveContext,
} from "../../engine";
import { UnprocessableError, writeFailure } from "../../errors";
import { RedisKeys } from "../../pipeline/redis-keys";
import { setupTestStack, type TestStack, TestUsers, unsafeCreateEntityTable } from "../../stack";

// Entity: a simple "item" with name + counter
const itemEntity = createEntity({
  table: "batch_items",
  fields: {
    name: createTextField({ required: true }),
    counter: createNumberField({ default: 0 }),
  },
});

const itemTable = buildEntityTable("item", itemEntity);

// Second entity used by an inTransaction hook to prove that hook DB writes
// roll back with the main transaction.
const auditEntity = createEntity({
  table: "batch_audit",
  fields: {
    action: createTextField({ required: true }),
    itemId: createTextField({ required: true }),
  },
});
// Brand (#742) is compile-time-only; the postSave hook writes this sink via method-form,
// so hold it at the unbranded TableColumns view (identical runtime shape).
const auditTable: TableColumns = buildEntityTable("audit", auditEntity);

// Hook invocation logs — reset per test. Captures which phase each hook saw.
const inTxHookLog: Array<{ id: EntityId; name: string }> = [];
const afterCommitHookLog: Array<{ id: EntityId; name: string }> = [];

// Toggles for afterCommit fault-injection test
let afterCommitShouldThrow = false;
const afterCommitThirdHookRan: string[] = [];

// Delay injected into item:create-slow — reset per test.
let slowHandlerDelayMs = 0;

const itemFeature = defineFeature("batch", (r) => {
  const item = r.entity("item", itemEntity);

  r.writeHandler(
    "item:create",
    z.object({ name: z.string().min(1), counter: z.number().optional() }),
    async (event, ctx) => {
      const crud = createEventStoreExecutor(itemTable, itemEntity, { entityName: "item" });
      return crud.create(event.payload, event.user, ctx.db);
    },
    { access: { roles: ["Admin"] } },
  );

  // Handler that always fails validation — used to trigger rollback mid-batch
  r.writeHandler(
    "item:fail",
    z.object({ name: z.string().min(1) }),
    async () => writeFailure(new UnprocessableError("intentional_failure")),
    { access: { roles: ["Admin"] } },
  );

  // Handler that always throws — used to verify unexpected throws surface as failures
  r.writeHandler(
    "item:throw",
    z.object({ name: z.string().min(1) }),
    async () => {
      throw new Error("handler_crashed");
    },
    { access: { roles: ["Admin"] } },
  );

  // Handler with a controllable delay — used to simulate a slow in-flight
  // handler for the parallel-idempotency race test.
  r.writeHandler(
    "item:create-slow",
    z.object({ name: z.string().min(1) }),
    async (event, ctx) => {
      if (slowHandlerDelayMs > 0) {
        await new Promise((resolve) => setTimeout(resolve, slowHandlerDelayMs));
      }
      const crud = createEventStoreExecutor(itemTable, itemEntity, { entityName: "item" });
      return crud.create(event.payload, event.user, ctx.db);
    },
    { access: { roles: ["Admin"] } },
  );

  // Entity hook: inTransaction — records in memory
  r.hook(
    "postSave",
    { allOf: item },
    async (result: SaveContext) => {
      inTxHookLog.push({ id: result.id, name: (result.data["name"] as string) ?? "" });
    },
    { phase: HookPhases.inTransaction },
  );

  // Entity hook: inTransaction — writes to DB via ctx.db (the tx-scoped TenantDb).
  // Proves that hook DB writes roll back with the main transaction on failure.
  r.hook(
    "postSave",
    { allOf: item },
    async (result, ctx) => {
      if (!ctx.db || !("insertOne" in ctx.db)) return;
      await ctx.db.insertOne(auditTable, { action: "item_saved", itemId: result.id });
    },
    { phase: HookPhases.inTransaction },
  );

  // Entity hook: afterCommit — records in memory (default phase)
  r.hook("postSave", { allOf: item }, async (result: SaveContext) => {
    afterCommitHookLog.push({ id: result.id, name: (result.data["name"] as string) ?? "" });
  });

  // Entity hook: afterCommit — may throw, used to verify error isolation
  r.hook("postSave", { allOf: item }, async () => {
    if (afterCommitShouldThrow) throw new Error("afterCommit_boom");
  });

  // Entity hook: afterCommit — runs AFTER the throwing one. Used to prove the
  // next hooks still fire despite the earlier failure.
  r.hook("postSave", { allOf: item }, async (result: SaveContext) => {
    afterCommitThirdHookRan.push((result.data["name"] as string) ?? "");
  });

  // Two hooks used by the parallelism test. Each records its start+end
  // timestamps so the assertion can compare intervals rather than elapsed
  // wall-clock time (which is timing-flaky on loaded CI boxes).
  r.hook("postSave", { allOf: item }, async (result: SaveContext) => {
    const name = result.data["name"] as string;
    if (!name?.startsWith("slowness-")) return;
    parallelismWindows.push({ hook: "A", start: Date.now() });
    await new Promise((r) => setTimeout(r, 80));
    parallelismWindows.push({ hook: "A", end: Date.now() });
  });
  r.hook("postSave", { allOf: item }, async (result: SaveContext) => {
    const name = result.data["name"] as string;
    if (!name?.startsWith("slowness-")) return;
    parallelismWindows.push({ hook: "B", start: Date.now() });
    await new Promise((r) => setTimeout(r, 80));
    parallelismWindows.push({ hook: "B", end: Date.now() });
  });
});

// Start + end timestamps recorded by the parallelism hooks above. A pair of
// hooks that ran truly in parallel will show B.start < A.end (and vice-versa),
// regardless of how long the whole request took overall.
//
// Module-level mutable state — safe here because Vitest runs tests inside a
// single file sequentially (the default). If someone flips vitest's
// `sequence.concurrent` on for this file, the test body would need its own
// window collector passed through ctx instead.
type ParallelismEvent = { hook: "A" | "B"; start?: number; end?: number };
const parallelismWindows: ParallelismEvent[] = [];

let stack: TestStack;
const admin = TestUsers.admin;

beforeAll(async () => {
  stack = await setupTestStack({ features: [itemFeature] });
  await unsafeCreateEntityTable(stack.db, itemEntity);
  await unsafeCreateEntityTable(stack.db, auditEntity);
});

afterAll(async () => {
  await stack.cleanup();
});

beforeEach(async () => {
  inTxHookLog.length = 0;
  afterCommitHookLog.length = 0;
  afterCommitThirdHookRan.length = 0;
  afterCommitShouldThrow = false;
  parallelismWindows.length = 0;
  stack.events.reset();
  await asRawClient(stack.db).unsafe(`DELETE FROM "${itemTable.tableName}"`);
  await asRawClient(stack.db).unsafe(`DELETE FROM "${auditTable.tableName}"`);
});

describe("POST /api/batch", () => {
  test("empty commands array returns success with empty results", async () => {
    const res = await stack.http.batch([], admin);
    expect(res.status).toBe(200);
    const body = await res.json();
    expect(body.isSuccess).toBe(true);
    expect(body.results).toEqual([]);
  });

  test("rejects non-array commands with 400", async () => {
    const res = await stack.http.raw(
      "POST",
      "/api/batch",
      // biome-ignore lint/suspicious/noExplicitAny: intentional bad body
      { commands: "not-an-array" as any },
      { Authorization: `Bearer ${await stack.jwt.sign(admin)}` },
    );
    expect(res.status).toBe(400);
  });

  test("all-succeed: writes persist, both phases fire per command", async () => {
    const res = await stack.http.batch(
      [
        { type: "batch:write:item:create", payload: { name: "alpha" } },
        { type: "batch:write:item:create", payload: { name: "beta" } },
        { type: "batch:write:item:create", payload: { name: "gamma" } },
      ],
      admin,
    );

    const body = await res.json();
    expect(res.status).toBe(200);
    expect(body.isSuccess).toBe(true);
    expect(body.results).toHaveLength(3);
    for (const r of body.results) expect(r.isSuccess).toBe(true);

    // Both phases fired once per command, same ids, same order
    expect(inTxHookLog).toHaveLength(3);
    expect(afterCommitHookLog).toHaveLength(3);
    expect(inTxHookLog.map((h) => h.name)).toEqual(["alpha", "beta", "gamma"]);
    expect(afterCommitHookLog.map((h) => h.name)).toEqual(["alpha", "beta", "gamma"]);

    // Rows actually persisted
    const rows = await selectMany(stack.db, itemTable);
    expect(rows).toHaveLength(3);
  });

  test("mid-batch failure: all writes roll back, afterCommit hooks do NOT fire", async () => {
    // Seed with one existing item so we can verify the batch didn't persist anything
    await seedRow(stack.db, itemTable, {
      name: "seed",
      counter: 0,
      tenantId: "00000000-0000-4000-8000-000000000001",
    });
    const seedCount = (await selectMany(stack.db, itemTable)).length;

    const res = await stack.http.batch(
      [
        { type: "batch:write:item:create", payload: { name: "will-rollback-1" } },
        { type: "batch:write:item:fail", payload: { name: "fails" } },
        { type: "batch:write:item:create", payload: { name: "never-runs" } },
      ],
      admin,
    );

    const body = await res.json();
    // UnprocessableError → 422 (business-rule violation), which is the
    // "expected failure" HTTP status. The batch envelope keeps `failedIndex`
    // + `results` alongside the error payload so callers know which command
    // tripped the rollback.
    expect(res.status).toBe(422);
    expect(body.isSuccess).toBe(false);
    expect(body.failedIndex).toBe(1);
    expect(body.error.code).toBe("unprocessable");
    expect(body.error.details.reason).toBe("intentional_failure");

    // inTransaction hook fired for the first successful command (then rolled back
    // — but the hook log is in-memory, it persists)
    expect(inTxHookLog.map((h) => h.name)).toEqual(["will-rollback-1"]);

    // afterCommit hook must NOT have fired (transaction rolled back)
    expect(afterCommitHookLog).toEqual([]);

    // DB: only the seed row remains, the batch's first successful write rolled back
    const rows = await selectMany(stack.db, itemTable);
    expect(rows).toHaveLength(seedCount);
    expect((rows[0] as { name: string }).name).toBe("seed");
  });

  test("inTransaction hook DB writes roll back with the batch", async () => {
    // Successful batch: audit rows should be written
    const okRes = await stack.http.batch(
      [{ type: "batch:write:item:create", payload: { name: "alpha" } }],
      admin,
    );
    expect((await okRes.json()).isSuccess).toBe(true);
    const auditAfterOk = await selectMany(stack.db, auditTable);
    expect(auditAfterOk).toHaveLength(1);
    expect((auditAfterOk[0] as { action: string }).action).toBe("item_saved");

    // Reset — new batch fails mid-way. Both entity rows AND audit rows must roll back.
    await asRawClient(stack.db).unsafe(`DELETE FROM "${itemTable.tableName}"`);
    await asRawClient(stack.db).unsafe(`DELETE FROM "${auditTable.tableName}"`);

    const failRes = await stack.http.batch(
      [
        { type: "batch:write:item:create", payload: { name: "beta" } },
        { type: "batch:write:item:fail", payload: { name: "stop" } },
      ],
      admin,
    );
    expect((await failRes.json()).isSuccess).toBe(false);

    // Both tables are empty — the inTransaction audit hook's write rolled back
    // together with the item row.
    const itemsAfterFail = await selectMany(stack.db, itemTable);
    const auditAfterFail = await selectMany(stack.db, auditTable);
    expect(itemsAfterFail).toHaveLength(0);
    expect(auditAfterFail).toHaveLength(0);
  });

  test("afterCommit hooks run in parallel (B starts before A finishes)", async () => {
    const res = await stack.http.batch(
      [{ type: "batch:write:item:create", payload: { name: "slowness-parallel" } }],
      admin,
    );
    expect(res.status).toBe(200);

    // Extract each hook's interval independently — checks overlap of
    // intervals, not total elapsed time. Robust against CI noise.
    const aStart = parallelismWindows.find((e) => e.hook === "A" && e.start !== undefined)?.start;
    const aEnd = parallelismWindows.find((e) => e.hook === "A" && e.end !== undefined)?.end;
    const bStart = parallelismWindows.find((e) => e.hook === "B" && e.start !== undefined)?.start;
    const bEnd = parallelismWindows.find((e) => e.hook === "B" && e.end !== undefined)?.end;

    expect(aStart).toBeDefined();
    expect(aEnd).toBeDefined();
    expect(bStart).toBeDefined();
    expect(bEnd).toBeDefined();

    // Parallel iff the two intervals overlap: one starts before the other
    // ends. Sequential execution would produce disjoint intervals.
    const overlap = (aStart as number) < (bEnd as number) && (bStart as number) < (aEnd as number);
    expect(overlap).toBe(true);
  });

  test("afterCommit hook error is isolated: batch succeeds, other hooks still fire", async () => {
    afterCommitShouldThrow = true;

    const res = await stack.http.batch(
      [{ type: "batch:write:item:create", payload: { name: "omega" } }],
      admin,
    );

    // Batch is reported successful despite the afterCommit hook throwing
    const body = await res.json();
    expect(res.status).toBe(200);
    expect(body.isSuccess).toBe(true);

    // DB row persisted (tx committed)
    const rows = await selectMany(stack.db, itemTable);
    expect(rows).toHaveLength(1);

    // The hook AFTER the throwing one still ran — errors don't cascade
    expect(afterCommitThirdHookRan).toEqual(["omega"]);
  });

  test("idempotency: repeated batch with same requestId returns cached result, no re-exec", async () => {
    const requestId = "batch-rid-123";
    const commands = [{ type: "batch:write:item:create", payload: { name: "once" } }];

    const first = await stack.http.batch(commands, admin, requestId);
    const firstBody = await first.json();
    expect(firstBody.isSuccess).toBe(true);
    expect(firstBody.results).toHaveLength(1);

    const rowsAfterFirst = await selectMany(stack.db, itemTable);
    expect(rowsAfterFirst).toHaveLength(1);

    // Hook logs reflect one execution
    expect(inTxHookLog).toHaveLength(1);
    expect(afterCommitHookLog).toHaveLength(1);

    // Retry with the same requestId — same response, but commands did NOT run again
    const second = await stack.http.batch(commands, admin, requestId);
    const secondBody = await second.json();

    expect(secondBody.isSuccess).toBe(true);
    expect(secondBody.results).toEqual(firstBody.results);

    // DB still has only one row (no double-insert)
    const rowsAfterSecond = await selectMany(stack.db, itemTable);
    expect(rowsAfterSecond).toHaveLength(1);

    // Hooks didn't fire a second time
    expect(inTxHookLog).toHaveLength(1);
    expect(afterCommitHookLog).toHaveLength(1);
  });

  test("idempotency: failed batch is cached — retry returns same error without re-exec", async () => {
    const requestId = "batch-rid-fail-cache";
    const commands = [
      { type: "batch:write:item:create", payload: { name: "should-rollback" } },
      { type: "batch:write:item:fail", payload: { name: "stop" } },
    ];

    const first = await stack.http.batch(commands, admin, requestId);
    const firstBody = await first.json();
    expect(first.status).toBe(422);
    expect(firstBody.isSuccess).toBe(false);
    expect(firstBody.failedIndex).toBe(1);
    expect(inTxHookLog.map((h) => h.name)).toEqual(["should-rollback"]);
    expect(await selectMany(stack.db, itemTable)).toHaveLength(0);

    inTxHookLog.length = 0;
    afterCommitHookLog.length = 0;

    const second = await stack.http.batch(commands, admin, requestId);
    const secondBody = await second.json();
    expect(second.status).toBe(422);
    expect(secondBody.isSuccess).toBe(false);
    expect(secondBody.failedIndex).toBe(firstBody.failedIndex);
    // HTTP re-serializes the cached failure with a fresh requestId/timestamp —
    // pin the stable business fields, not the transport envelope.
    expect(secondBody.error.code).toBe(firstBody.error.code);
    expect(secondBody.error.details).toEqual(firstBody.error.details);
    // Cached path must not re-run commands (hooks stay empty, DB empty)
    expect(inTxHookLog).toHaveLength(0);
    expect(afterCommitHookLog).toHaveLength(0);
    expect(await selectMany(stack.db, itemTable)).toHaveLength(0);
  });

  test("idempotency: parallel writes with the same requestId — second waits for the first instead of re-executing", async () => {
    slowHandlerDelayMs = 250;
    const requestId = "batch-rid-parallel-slow";
    const commands = [{ type: "batch:write:item:create-slow", payload: { name: "parallel-once" } }];

    try {
      const [first, second] = await Promise.all([
        stack.http.batch(commands, admin, requestId),
        (async () => {
          // Give request #1 a head start so it wins the lock acquisition —
          // the race being tested is "#2 waits", not "who acquires first".
          await new Promise((resolve) => setTimeout(resolve, 50));
          return stack.http.batch(commands, admin, requestId);
        })(),
      ]);

      const firstBody = await first.json();
      const secondBody = await second.json();

      expect(firstBody.isSuccess).toBe(true);
      expect(secondBody.isSuccess).toBe(true);
      // Same cached response, not a fresh execution.
      expect(secondBody.results).toEqual(firstBody.results);

      // Exactly one row and one hook run — the second call did not re-run
      // the handler and create a duplicate side effect.
      const rows = await selectMany(stack.db, itemTable);
      expect(rows).toHaveLength(1);
      expect(inTxHookLog).toHaveLength(1);
      expect(afterCommitHookLog).toHaveLength(1);
    } finally {
      slowHandlerDelayMs = 0;
    }
  });

  test("idempotency: corrupted cache entry is treated as miss and re-runs", async () => {
    const requestId = "batch-rid-corrupt";
    const cacheKey = `${RedisKeys.idempotency}${admin.tenantId}:${admin.id}:${requestId}`;

    // Prove key-coupling first: seed a well-formed cached entry under the
    // exact manually-built key and confirm the batch short-circuits on it
    // (no handler run, no DB write) — if the manual key ever drifts from
    // what the guard actually reads, this half fails loud instead of the
    // corrupt-JSON half below silently becoming a no-op cache-miss test.
    const seeded = { isSuccess: true, results: [{ seeded: true }] };
    await stack.redis.redis.set(cacheKey, JSON.stringify(seeded), "EX", 60);
    const cachedRes = await stack.http.batch(
      [{ type: "batch:write:item:create", payload: { name: "should-not-run" } }],
      admin,
      requestId,
    );
    expect(await cachedRes.json()).toEqual(seeded);
    expect(await selectMany(stack.db, itemTable)).toHaveLength(0);
    expect(inTxHookLog).toHaveLength(0);

    // Now corrupt the same key and confirm it's treated as a miss.
    await stack.redis.redis.set(cacheKey, "{not-json", "EX", 60);

    const res = await stack.http.batch(
      [{ type: "batch:write:item:create", payload: { name: "after-corrupt" } }],
      admin,
      requestId,
    );
    const body = await res.json();
    expect(res.status).toBe(200);
    expect(body.isSuccess).toBe(true);
    expect(await selectMany(stack.db, itemTable)).toHaveLength(1);
    expect(inTxHookLog.map((h) => h.name)).toEqual(["after-corrupt"]);
  });

  test("mid-batch handler throw: rolls back prior writes, afterCommit does not fire", async () => {
    const res = await stack.http.batch(
      [
        { type: "batch:write:item:create", payload: { name: "before-throw" } },
        { type: "batch:write:item:throw", payload: { name: "boom" } },
        { type: "batch:write:item:create", payload: { name: "never" } },
      ],
      admin,
    );
    const body = await res.json();
    expect(body.isSuccess).toBe(false);
    expect(body.failedIndex).toBe(1);
    expect(inTxHookLog.map((h) => h.name)).toEqual(["before-throw"]);
    expect(afterCommitHookLog).toEqual([]);
    expect(await selectMany(stack.db, itemTable)).toHaveLength(0);
    expect(await selectMany(stack.db, auditTable)).toHaveLength(0);
  });
});

describe("POST /api/write (single write runs in its own transaction)", () => {
  test("inTransaction hook DB write persists with the entity write", async () => {
    const res = await stack.http.write("batch:write:item:create", { name: "single" }, admin);
    const body = await res.json();
    expect(body.isSuccess).toBe(true);

    // Both the item row AND the audit row exist — proves the single write
    // went through a transaction and the inTx hook shared it.
    const items = await selectMany(stack.db, itemTable);
    const audits = await selectMany(stack.db, auditTable);
    expect(items).toHaveLength(1);
    expect(audits).toHaveLength(1);
  });

  test("handler throw rolls back inTransaction hook writes too", async () => {
    // First a successful write so there's something to compare against
    await stack.http.write("batch:write:item:create", { name: "survivor" }, admin);
    const beforeItems = await selectMany(stack.db, itemTable);
    const beforeAudits = await selectMany(stack.db, auditTable);
    expect(beforeItems).toHaveLength(1);
    expect(beforeAudits).toHaveLength(1);

    // Now a write whose handler throws — nothing new should be committed
    const res = await stack.http.write("batch:write:item:throw", { name: "crash" }, admin);
    const body = await res.json();
    expect(body.isSuccess).toBe(false);

    const afterItems = await selectMany(stack.db, itemTable);
    const afterAudits = await selectMany(stack.db, auditTable);
    // Counts unchanged — no partial commit
    expect(afterItems).toHaveLength(beforeItems.length);
    expect(afterAudits).toHaveLength(beforeAudits.length);
  });
});