import { readdirSync, readFileSync } from "node:fs";
import { fileURLToPath } from "node:url";
import { sql } from "drizzle-orm";
import { createPgliteClient } from "@platform-modules/db/postgres/pglite";
import { describe, expect, it } from "vitest";
import {
  WorkflowStepTransactionRunner,
  type ApplicationTransaction,
  type Clock,
} from "@awp/application";
import {
  unsafeOpaqueId,
  type AuditRecordId,
  type CorrelationId,
  type EventId,
  type OperationId,
  type OutboxMessageId,
  type PrincipalId,
  type ProjectId,
} from "@awp/contracts";
import { PostgresUnitOfWork, schema } from "@awp/persistence";

const migrationDirectory = fileURLToPath(
  new URL("../../packages/persistence/drizzle/", import.meta.url),
);

async function database() {
  const db = createPgliteClient({ schema });
  for (const name of readdirSync(migrationDirectory)
    .filter((name) => name.endsWith(".sql"))
    .sort()) {
    const migration = readFileSync(`${migrationDirectory}/${name}`, "utf8");
    for (const statement of migration.split("--> statement-breakpoint")) {
      const sqlText = statement.trim();
      if (sqlText.length > 0) await db.execute(sql.raw(sqlText));
    }
  }
  return db;
}

const clock: Clock = { now: () => new Date("2026-08-24T00:00:00.000Z") };

function identity(suffix: string) {
  return {
    operationId: unsafeOpaqueId<OperationId>(`operation:${suffix}`),
    stepName: "persist-project",
    stepKey: `project:${suffix}`,
  };
}

async function projectMutation(tx: ApplicationTransaction, suffix: string) {
  const projectId = unsafeOpaqueId<ProjectId>(`project:${suffix}`);
  const principalId = unsafeOpaqueId<PrincipalId>(`principal:${suffix}`);
  const correlationId = unsafeOpaqueId<CorrelationId>(`correlation:${suffix}`);
  const occurredAt = "2026-08-24T00:00:00.000Z";
  await tx.projects.insert({
    id: projectId,
    name: `Project ${suffix}`,
    repositoryUrl: "https://example.com/repository.git",
    status: "active",
    revision: 1,
  });
  await tx.events.append({
    id: unsafeOpaqueId<EventId>(`event:${suffix}`),
    type: "ProjectPersisted",
    schemaVersion: 1,
    occurredAt,
    aggregateType: "Project",
    aggregateId: projectId,
    aggregateRevision: 1,
    projectId,
    principalId,
    correlationId,
    payload: { projectId },
  });
  await tx.audit.append({
    id: unsafeOpaqueId<AuditRecordId>(`audit:${suffix}`),
    occurredAt,
    principalId,
    action: "ProjectPersisted",
    targetType: "Project",
    targetId: projectId,
    projectId,
    disposition: "allowed",
    correlationId,
    safeMetadata: {},
  });
  await tx.outbox.append({
    id: unsafeOpaqueId<OutboxMessageId>(`outbox:${suffix}`),
    topic: "ProjectPersisted",
    payload: { projectId },
    occurredAt,
  });
  return { projectId, disposition: "persisted" as const };
}

describe("durable workflow step transaction seam", () => {
  it("commits domain/event/audit/outbox/marker once and returns the recorded outcome on replay", async () => {
    const db = await database();
    const uow = new PostgresUnitOfWork(db);
    const runner = new WorkflowStepTransactionRunner(uow, clock);
    let executions = 0;

    const first = await runner.run(identity("once"), async (tx) => {
      executions += 1;
      return projectMutation(tx, "once");
    });
    const replay = await runner.run(identity("once"), async (tx) => {
      executions += 1;
      return projectMutation(tx, "once");
    });

    expect(first).toEqual({
      outcome: { projectId: "project:once", disposition: "persisted" },
      replayed: false,
    });
    expect(replay).toEqual({ ...first, replayed: true });
    expect(executions).toBe(1);
    expect(await db.select().from(schema.projects)).toHaveLength(1);
    expect(await db.select().from(schema.businessEvents)).toHaveLength(1);
    expect(await db.select().from(schema.auditRecords)).toHaveLength(1);
    expect(await db.select().from(schema.outboxMessages)).toHaveLength(1);
    expect(await db.select().from(schema.workflowStepOutcomes)).toEqual([
      expect.objectContaining({
        operationId: "operation:once",
        stepName: "persist-project",
        stepKey: "project:once",
        outcome: { projectId: "project:once", disposition: "persisted" },
      }),
    ]);
    await db.$client.close();
  }, 15_000);

  it("rolls domain effects back when marker insertion loses the primary-key race", async () => {
    const db = await database();
    const uow = new PostgresUnitOfWork(db);
    const marker = {
      ...identity("winner"),
      outcome: { projectId: "project:winner", disposition: "persisted" },
      recordedAt: "2026-08-24T00:00:00.000Z",
    };
    await uow.transaction((tx) => tx.workflowStepOutcomes.insert(marker));

    await expect(
      uow.transaction(async (tx) => {
        await tx.projects.insert({
          id: unsafeOpaqueId<ProjectId>("project:loser"),
          name: "Must roll back",
          repositoryUrl: "https://example.com/loser.git",
          status: "active",
          revision: 1,
        });
        await tx.workflowStepOutcomes.insert(marker);
      }),
    ).rejects.toMatchObject({ name: "WorkflowStepOutcomeConflictError" });

    expect(await db.select().from(schema.projects)).toHaveLength(0);
    expect(await db.select().from(schema.workflowStepOutcomes)).toHaveLength(1);
    await db.$client.close();
  }, 15_000);

  it("invokes the after-commit boundary only for the winning mutation, never on replay", async () => {
    const db = await database();
    const uow = new PostgresUnitOfWork(db);
    const commits: unknown[] = [];
    const runner = new WorkflowStepTransactionRunner(uow, clock, (commit) => {
      commits.push(commit);
    });

    await runner.run(identity("hook"), (tx) => projectMutation(tx, "hook"));
    await runner.run(identity("hook"), (tx) => projectMutation(tx, "hook"));

    expect(commits).toHaveLength(1);
    expect(commits[0]).toMatchObject({
      identity: identity("hook"),
      outcome: { projectId: "project:hook", disposition: "persisted" },
      recordedAt: "2026-08-24T00:00:00.000Z",
    });
    await db.$client.close();
  }, 15_000);
});