import { mkdtempSync, readFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { fileURLToPath } from "node:url";
import { spawn } from "node:child_process";
import postgres from "postgres";
import { describe, expect, it } from "vitest";
import { IdempotentOutboxConsumer } from "@awp/application";
import { unsafeOpaqueId, type OutboxMessageId, type ProjectId } from "@awp/contracts";
import { createPostgresRuntime } from "@awp/persistence";

const databaseUrl = process.env.AWP_TEST_DATABASE_URL;
const proofDescribe = databaseUrl ? describe : describe.skip;
const childPath = fileURLToPath(new URL("./durable-seam-child.ts", import.meta.url));
const migrationsFolder = fileURLToPath(
  new URL("../../../packages/persistence/drizzle/", import.meta.url),
);

interface ChildExit {
  readonly code: number | null;
  readonly signal: NodeJS.Signals | null;
  readonly stdout: string;
  readonly stderr: string;
}

function runChild(
  mode: "start" | "recover",
  stateDir: string,
  crashStep?: string,
): Promise<ChildExit> {
  return new Promise((resolve, reject) => {
    const env: NodeJS.ProcessEnv = {
      ...process.env,
      AWP_TEST_DATABASE_URL: databaseUrl!,
      AWP_MIGRATIONS_FOLDER: migrationsFolder,
      AWP_DURABLE_SEAM_STATE_DIR: stateDir,
      AWP_DURABLE_SEAM_MODE: mode,
    };
    delete env.AWP_CRASH_AFTER_STEP_COMMIT;
    if (crashStep) env.AWP_CRASH_AFTER_STEP_COMMIT = crashStep;
    const child = spawn(process.execPath, ["--import", "tsx", childPath], {
      cwd: fileURLToPath(new URL("../../../", import.meta.url)),
      env,
      stdio: ["ignore", "pipe", "pipe"],
    });
    let stdout = "";
    let stderr = "";
    child.stdout.on("data", (chunk) => (stdout += String(chunk)));
    child.stderr.on("data", (chunk) => (stderr += String(chunk)));
    child.on("error", reject);
    child.on("close", (code, signal) => resolve({ code, signal, stdout, stderr }));
  });
}

function entryCount(stateDir: string): number {
  const text = readFileSync(join(stateDir, "step-entries.log"), "utf8");
  return text.split("\n").filter(Boolean).length;
}

proofDescribe("durable seam real DBOS crash window", () => {
  it("crash between domain commit and DBOS checkpoint replays the step without duplicate effects", async () => {
    const stateDir = mkdtempSync(join(tmpdir(), "awp-durable-seam-"));
    const stepName = "awp-durable-seam-proof-domain-commit";
    const first = await runChild("start", stateDir, stepName);
    expect(
      { code: first.code, signal: first.signal },
      `initial child stdout=${first.stdout} stderr=${first.stderr}`,
    ).toEqual({ code: null, signal: "SIGKILL" });
    expect(entryCount(stateDir)).toBe(1);

    const sql = postgres(databaseUrl!, { max: 1 });
    const countsAfterCrash = {
      project: Number(
        (
          await sql`select count(*)::int as count from projects where id = 'project:durable-seam-crash-proof'`
        )[0]!.count,
      ),
      event: Number(
        (
          await sql`select count(*)::int as count from business_events where id = 'event:durable-seam-crash-proof'`
        )[0]!.count,
      ),
      audit: Number(
        (
          await sql`select count(*)::int as count from audit_records where id = 'audit:durable-seam-crash-proof'`
        )[0]!.count,
      ),
      outbox: Number(
        (
          await sql`select count(*)::int as count from outbox_messages where id = 'outbox:durable-seam-crash-proof'`
        )[0]!.count,
      ),
      marker: Number(
        (
          await sql`select count(*)::int as count from workflow_step_outcomes where operation_id = 'operation:durable-seam-crash-proof' and step_name = ${stepName}`
        )[0]!.count,
      ),
    };
    expect(countsAfterCrash).toEqual({ project: 1, event: 1, audit: 1, outbox: 1, marker: 1 });
    const markerRows = await sql`
        select outcome
        from workflow_step_outcomes
        where operation_id = 'operation:durable-seam-crash-proof'
          and step_name = ${stepName}
      `;
    expect(markerRows[0]!.outcome).toEqual({
      projectId: "project:durable-seam-crash-proof",
      disposition: "committed",
      revision: 1,
    });

    const recovered = await runChild("recover", stateDir);
    expect(
      { code: recovered.code, signal: recovered.signal },
      `recovery child stdout=${recovered.stdout} stderr=${recovered.stderr}`,
    ).toEqual({ code: 0, signal: null });
    expect(entryCount(stateDir)).toBe(2);
    const result = JSON.parse(readFileSync(join(stateDir, "workflow-result.json"), "utf8"));
    expect(result).toEqual(markerRows[0]!.outcome);

    const countsAfterRecovery = {
      project: Number(
        (
          await sql`select count(*)::int as count from projects where id = 'project:durable-seam-crash-proof'`
        )[0]!.count,
      ),
      event: Number(
        (
          await sql`select count(*)::int as count from business_events where id = 'event:durable-seam-crash-proof'`
        )[0]!.count,
      ),
      audit: Number(
        (
          await sql`select count(*)::int as count from audit_records where id = 'audit:durable-seam-crash-proof'`
        )[0]!.count,
      ),
      outbox: Number(
        (
          await sql`select count(*)::int as count from outbox_messages where id = 'outbox:durable-seam-crash-proof'`
        )[0]!.count,
      ),
      marker: Number(
        (
          await sql`select count(*)::int as count from workflow_step_outcomes where operation_id = 'operation:durable-seam-crash-proof' and step_name = ${stepName}`
        )[0]!.count,
      ),
    };
    expect(countsAfterRecovery).toEqual(countsAfterCrash);

    const runtime = createPostgresRuntime(databaseUrl!, migrationsFolder);
    const consumer = new IdempotentOutboxConsumer(runtime.uow, {
      now: () => new Date("2026-08-24T02:05:00.000Z"),
    });
    const messageId = unsafeOpaqueId<OutboxMessageId>("outbox:durable-seam-crash-proof");
    const consume = () =>
      consumer.consume("durable-seam-proof-consumer", messageId, async (tx) => {
        const projectedId = unsafeOpaqueId<ProjectId>("project:durable-seam-consumer-effect");
        await tx.projects.insert({
          id: projectedId,
          name: "Durable seam consumer projection",
          repositoryUrl: "https://example.com/durable-seam-consumer.git",
          status: "active",
          revision: 1,
        });
        return projectedId;
      });
    expect(await consume()).toMatchObject({ processed: true });
    expect(await consume()).toEqual({ processed: false });
    expect(
      Number(
        (
          await sql`select count(*)::int as count from projects where id = 'project:durable-seam-consumer-effect'`
        )[0]!.count,
      ),
    ).toBe(1);
    expect(
      Number(
        (
          await sql`select count(*)::int as count from outbox_consumer_receipts where consumer_id = 'durable-seam-proof-consumer' and message_id = ${messageId}`
        )[0]!.count,
      ),
    ).toBe(1);

    await runtime.close();
    await sql.end({ timeout: 5 });
  }, 120_000);
});