import { describe, expect, it } from "vitest";
import { unsafeOpaqueId, type OperationId } from "@awp/contracts";
import {
  DbosWorkflowProvider,
  runIdempotentSideEffect,
  workflowIdFor,
  type DbosRuntime,
  type DbosStartRequest,
  type DbosWorkflowRecord,
  type SideEffectJournal,
  type SideEffectJournalEntry,
} from "../../../packages/providers/workflow-dbos/src/index.js";
import { providerContext } from "../../security/execution/provider-context.js";

class PersistentFakeDbos implements DbosRuntime {
  readonly workflows = new Map<string, DbosWorkflowRecord>();
  starts = 0;
  throwAfterDurableStart = false;

  async start(request: DbosStartRequest): Promise<DbosWorkflowRecord> {
    this.starts += 1;
    const existing = this.workflows.get(request.workflowId);
    if (existing) return existing;
    const record: DbosWorkflowRecord = {
      workflowId: request.workflowId,
      workflowKind: request.workflowKind,
      status: "PENDING",
      updatedAt: "2026-08-20T00:00:00.000Z",
    };
    this.workflows.set(request.workflowId, record);
    if (this.throwAfterDurableStart) {
      this.throwAfterDurableStart = false;
      throw new Error("simulated process/network interruption after durable start");
    }
    return record;
  }

  async get(workflowId: string): Promise<DbosWorkflowRecord | undefined> {
    return this.workflows.get(workflowId);
  }

  async cancel(workflowId: string): Promise<void> {
    const existing = this.workflows.get(workflowId);
    if (!existing) return;
    this.workflows.set(workflowId, { ...existing, status: "CANCELLED" });
  }
}

class MemoryJournal<T> implements SideEffectJournal<T> {
  readonly entries = new Map<string, SideEffectJournalEntry<T>>();

  async get(key: string): Promise<SideEffectJournalEntry<T> | undefined> {
    return this.entries.get(key);
  }
  async markStarted(key: string): Promise<void> {
    this.entries.set(key, { key, state: "started" });
  }
  async markCompleted(key: string, result: T): Promise<void> {
    this.entries.set(key, { key, state: "completed", result });
  }
}

describe("DBOS durable workflow provider", () => {
  it("maps an AWP operation to one deterministic DBOS workflow ID and suppresses duplicate starts", async () => {
    const runtime = new PersistentFakeDbos();
    const provider = new DbosWorkflowProvider(runtime);
    const operationId = unsafeOpaqueId<OperationId>("operation-42");

    const first = await provider.start(providerContext(), operationId, "agent-attempt", { n: 1 });
    const second = await provider.start(providerContext(), operationId, "agent-attempt", { n: 2 });

    expect(workflowIdFor(operationId)).toBe("awp:operation-42");
    expect(runtime.starts).toBe(1);
    expect(first.references[0]?.nativeId).toBe("awp:operation-42");
    expect(second.references[0]?.nativeId).toBe("awp:operation-42");
  });

  it("reconciles an ambiguous durable start instead of issuing the side effect twice", async () => {
    const runtime = new PersistentFakeDbos();
    runtime.throwAfterDurableStart = true;
    const provider = new DbosWorkflowProvider(runtime);
    const operationId = unsafeOpaqueId<OperationId>("operation-restart");

    const result = await provider.start(providerContext(), operationId, "factory-run", {});
    expect(runtime.starts).toBe(1);
    expect(result.value.state).toBe("running");

    const afterRestart = new DbosWorkflowProvider(runtime);
    await afterRestart.start(providerContext(), operationId, "factory-run", {});
    expect(runtime.starts).toBe(1);
  });

  it("persists and reconciles cancellation", async () => {
    const runtime = new PersistentFakeDbos();
    const provider = new DbosWorkflowProvider(runtime);
    const operationId = unsafeOpaqueId<OperationId>("operation-cancel");
    await provider.start(providerContext(), operationId, "agent-attempt", {});

    const cancelled = await provider.cancel(providerContext(), operationId);
    expect(cancelled.value.state).toBe("cancelled");

    const reconciled = await provider.reconcile({
      context: providerContext(),
      reference: cancelled.references[0]!,
    });
    expect(reconciled.value.state).toBe("cancelled");
  });

  it("reconciles a started external side effect after a crash before retrying execution", async () => {
    const journal = new MemoryJournal<string>();
    const providerState = new Map<string, string>();
    let executions = 0;
    let simulateCrashAfterProviderMutation = true;

    const effect = {
      key: "publish:attempt-1",
      execute: async () => {
        executions += 1;
        providerState.set("publish:attempt-1", "provider-result");
        if (simulateCrashAfterProviderMutation) {
          simulateCrashAfterProviderMutation = false;
          throw new Error("process killed after provider mutation");
        }
        return "provider-result";
      },
      reconcile: async () => providerState.get("publish:attempt-1"),
    };

    await expect(runIdempotentSideEffect(journal, effect)).rejects.toThrow("process killed");
    expect(journal.entries.get(effect.key)?.state).toBe("started");

    const result = await runIdempotentSideEffect(journal, effect);
    expect(result).toBe("provider-result");
    expect(executions).toBe(1);
    expect(journal.entries.get(effect.key)?.state).toBe("completed");
  });
});