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 { IdempotentOutboxConsumer, OutboxDispatcher } from "@awp/application";
import { unsafeOpaqueId, type OutboxMessageId, 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 = { now: () => new Date("2026-08-24T01:00:00.000Z") };

describe("outbox at-least-once delivery with consumer dedupe", () => {
  it("redelivers after publish-before-mark failure while consumer effect remains exactly once", async () => {
    const db = await database();
    const uow = new PostgresUnitOfWork(db);
    const messageId = unsafeOpaqueId<OutboxMessageId>("outbox-redelivery");
    await uow.transaction((tx) =>
      tx.outbox.append({
        id: messageId,
        topic: "ProjectMaterialized",
        payload: { projectId: "project-outbox-effect" },
        occurredAt: "2026-08-24T00:00:00.000Z",
      }),
    );

    const consumer = new IdempotentOutboxConsumer(uow, clock);
    let deliveries = 0;
    let failAfterFirstConsumerCommit = true;
    const dispatcher = new OutboxDispatcher(
      uow,
      {
        async publish(message) {
          deliveries += 1;
          await consumer.consume("projector:test", message.id, async (tx) => {
            const projectId = unsafeOpaqueId<ProjectId>("project-outbox-effect");
            await tx.projects.insert({
              id: projectId,
              name: "Projected exactly once",
              repositoryUrl: "https://example.com/outbox.git",
              status: "active",
              revision: 1,
            });
            return projectId;
          });
          if (failAfterFirstConsumerCommit) {
            failAfterFirstConsumerCommit = false;
            throw new Error("simulated crash after publish before published=true");
          }
        },
      },
      clock,
    );

    expect(await dispatcher.dispatchPending()).toEqual({ attempted: 1, published: 0, failed: 1 });
    let outboxRows = await db.select().from(schema.outboxMessages);
    expect(outboxRows).toEqual([
      expect.objectContaining({ id: messageId, published: false, attempts: 1 }),
    ]);
    expect(await db.select().from(schema.projects)).toHaveLength(1);
    expect(await db.select().from(schema.outboxConsumerReceipts)).toHaveLength(1);

    expect(await dispatcher.dispatchPending()).toEqual({ attempted: 1, published: 1, failed: 0 });
    outboxRows = await db.select().from(schema.outboxMessages);
    expect(outboxRows).toEqual([
      expect.objectContaining({
        id: messageId,
        published: true,
        attempts: 2,
      }),
    ]);
    expect(Date.parse(outboxRows[0]!.publishedAt!)).toBe(Date.parse("2026-08-24T01:00:00.000Z"));
    expect(deliveries).toBe(2);
    expect(await db.select().from(schema.projects)).toHaveLength(1);
    expect(await db.select().from(schema.outboxConsumerReceipts)).toEqual([
      expect.objectContaining({ consumerId: "projector:test", messageId }),
    ]);

    expect(await dispatcher.dispatchPending()).toEqual({ attempted: 0, published: 0, failed: 0 });
    expect(deliveries).toBe(2);
    await db.$client.close();
  }, 20_000);
});