0pt/tests/p0/external-writeback-service.test.ts

import { describe, it, expect, vi, afterEach } from "vitest";
import { ExternalWritebackService } from "@/src/services/external-writeback-service";
import type { WritebackAdapter, WritebackAdapterContext, WritebackRequest, WritebackResult } from "@/src/adapters/writeback-adapter";
import type { ExternalWriteRequest } from "@/src/types/external-writeback";

function jsonResponse(body: unknown, opts: { status?: number } = {}) {
  return {
    ok: (opts.status ?? 200) < 400,
    status: opts.status ?? 200,
    json: async () => body,
    text: async () => JSON.stringify(body),
  } as unknown as Response;
}

/** A configurable in-memory WritebackAdapter double -- records every `write` call so tests can
 * assert on call count (critical for the idempotency/concurrency tests below), and can be told
 * to fail N times before succeeding, or to always fail/always succeed. */
class TestWritebackAdapter implements WritebackAdapter {
  readonly adapterId = "test_adapter";
  calls: Array<{ request: WritebackRequest; ctx: WritebackAdapterContext }> = [];
  configured = true;
  private failuresRemaining = 0;
  private failureError = "test_adapter_failure";
  externalIdPrefix = "ext";

  isConfigured(): boolean {
    return this.configured;
  }

  failNextCalls(n: number, error = "test_adapter_failure"): void {
    this.failuresRemaining = n;
    this.failureError = error;
  }

  async write(request: WritebackRequest, ctx: WritebackAdapterContext): Promise<WritebackResult> {
    this.calls.push({ request, ctx });
    if (this.failuresRemaining > 0) {
      this.failuresRemaining -= 1;
      return { ok: false, error: this.failureError };
    }
    return { ok: true, externalId: `${this.externalIdPrefix}_${this.calls.length}` };
  }
}

function baseRequest(overrides: Partial<ExternalWriteRequest> = {}): ExternalWriteRequest {
  return {
    adapter_id: "test_adapter",
    idempotency_key: "idem-1",
    record_type: "test_record",
    payload: { foo: "bar" },
    tenant_id: "tenant1",
    project_id: "proj1",
    actor_id: "actor1",
    correlation_id: "corr-1",
    ...overrides,
  };
}

/** Stubs global fetch to answer CL0UD policy evaluation (always "allow" unless `denyOrApproval`
 * is set) and swallow AUD0 event/lineage POSTs, recording them for assertions. */
function stubCl0udAllowAndAud0(opts: { decision?: "allow" | "deny" | "approval_required" } = {}) {
  // eslint-disable-next-line @typescript-eslint/no-explicit-any -- AUD0-ingest-body är fri nyttolast
  const aud0Calls: { url: string; body: any }[] = [];
  const fetchMock = vi.fn(async (url: string, init?: RequestInit) => {
    // REAL CL0UD wire contract (cl0ud main): POST /api/policy/evaluate with a
    // `{ data: { decision, requiresApproval, ... } }` envelope; there is no
    // "approval_required" decision value on the wire — it is allow +
    // requiresApproval, which 0pt's client maps back onto the historical
    // three-value vocabulary.
    if (url.includes("/api/policy/evaluate")) {
      const wanted = opts.decision ?? "allow";
      return jsonResponse({
        data: {
          decision: wanted === "deny" ? "deny" : "allow",
          requiresApproval: wanted === "approval_required",
          decisionId: "pdec_test",
          reason: wanted === "deny" ? "test_deny" : wanted === "approval_required" ? "test_approval_required" : undefined,
        },
        meta: {},
      });
    }
    // REAL AUD0 lineage ingest is the OpenLineage endpoint (/v1/aud0/openlineage);
    // /v1/aud0/lineage/events never existed on AUD0.
    if (url.includes("/v1/aud0/events") || url.includes("/v1/aud0/openlineage")) {
      aud0Calls.push({ url, body: init?.body ? JSON.parse(init.body as string) : null });
      return jsonResponse({});
    }
    throw new Error(`unexpected fetch in test: ${url}`);
  });
  vi.stubGlobal("fetch", fetchMock);
  return { aud0Calls, fetchMock };
}

describe("ExternalWritebackService — adapter resolution", () => {
  afterEach(() => vi.restoreAllMocks());

  it("throws when no adapter is registered for the requested adapter_id", async () => {
    stubCl0udAllowAndAud0();
    const service = new ExternalWritebackService({ cl0udBaseUrl: "https://cl0ud.example.com" });
    await expect(service.writeExternalRecord(baseRequest())).rejects.toThrow(/no adapter registered/);
  });

  it("throws when the registered adapter reports isConfigured() === false, without calling write()", async () => {
    stubCl0udAllowAndAud0();
    const adapter = new TestWritebackAdapter();
    adapter.configured = false;
    const service = new ExternalWritebackService({ cl0udBaseUrl: "https://cl0ud.example.com", adapters: [adapter] });
    await expect(service.writeExternalRecord(baseRequest())).rejects.toThrow(/not configured/);
    expect(adapter.calls).toHaveLength(0);
  });
});

describe("ExternalWritebackService — CL0UD policy gate", () => {
  afterEach(() => vi.restoreAllMocks());

  it("denies the write and marks the record failed when CL0UD denies", async () => {
    stubCl0udAllowAndAud0({ decision: "deny" });
    const adapter = new TestWritebackAdapter();
    const service = new ExternalWritebackService({ cl0udBaseUrl: "https://cl0ud.example.com", adapters: [adapter] });
    await expect(service.writeExternalRecord(baseRequest())).rejects.toThrow(/Policy denied/);
    expect(adapter.calls).toHaveLength(0);
    const record = await service.findByIdempotencyKey("test_adapter", "idem-1");
    expect(record?.status).toBe("failed");
  });

  it("does not silently bypass a CL0UD approval_required decision even though ACT0's own approval step already ran upstream", async () => {
    stubCl0udAllowAndAud0({ decision: "approval_required" });
    const adapter = new TestWritebackAdapter();
    const service = new ExternalWritebackService({ cl0udBaseUrl: "https://cl0ud.example.com", adapters: [adapter] });
    await expect(service.writeExternalRecord(baseRequest())).rejects.toThrow(/Policy requires approval/);
    expect(adapter.calls).toHaveLength(0);
  });
});

describe("ExternalWritebackService — successful write", () => {
  afterEach(() => vi.restoreAllMocks());

  it("writes via the adapter, returns status 'written' with the adapter's external id, and emits AUD0 event + lineage", async () => {
    const { aud0Calls } = stubCl0udAllowAndAud0();
    const adapter = new TestWritebackAdapter();
    const service = new ExternalWritebackService({
      cl0udBaseUrl: "https://cl0ud.example.com",
      aud0BaseUrl: "https://aud0.example.com",
      adapters: [adapter],
    });

    const record = await service.writeExternalRecord(baseRequest());

    expect(record.status).toBe("written");
    expect(record.external_id).toBe("ext_1");
    expect(adapter.calls).toHaveLength(1);
    expect(adapter.calls[0]?.request.recordType).toBe("test_record");
    expect(adapter.calls[0]?.request.payload).toEqual({ foo: "bar" });

    const writtenEvent = aud0Calls.find((c) => c.body?.event_type === "0pt.external_record_written");
    expect(writtenEvent).toBeTruthy();
    expect(writtenEvent!.body.capability_id).toBe("0pt.write_external_record");
    expect(writtenEvent!.body.risk_level).toBe("L4");
    expect(writtenEvent!.body.payload.external_id).toBe("ext_1");

    // Lineage now goes to AUD0's REAL ingest contract: an OpenLineage
    // RunEvent on /v1/aud0/openlineage (the legacy source/target shape is
    // converted by aud0-client's legacyLineageToOpenLineage).
    const lineageEvent = aud0Calls.find((c) => c.url.includes("/openlineage"));
    expect(lineageEvent).toBeTruthy();
    expect(lineageEvent!.body.eventType).toBe("COMPLETE");
    expect(lineageEvent!.body.inputs[0].name).toContain("idem-1");
    expect(lineageEvent!.body.outputs[0].name).toContain("ext_1");
    expect(lineageEvent!.body.run.facets.siaxLegacyLineage.lineage_type).toBe("produced");
  });
});

describe("ExternalWritebackService — idempotency (sequential re-calls)", () => {
  afterEach(() => vi.restoreAllMocks());

  it("a second sequential call with the same idempotency_key after a successful write is an idempotent no-op — the adapter is never called again", async () => {
    stubCl0udAllowAndAud0();
    const adapter = new TestWritebackAdapter();
    const service = new ExternalWritebackService({ cl0udBaseUrl: "https://cl0ud.example.com", adapters: [adapter] });

    const first = await service.writeExternalRecord(baseRequest());
    const second = await service.writeExternalRecord(baseRequest());

    expect(adapter.calls).toHaveLength(1);
    expect(second.write_id).toBe(first.write_id);
    expect(second.status).toBe("written");
    expect(second.external_id).toBe(first.external_id);
  });

  it("a failed write CAN be retried with the same idempotency_key — this is a deliberate 'claim survives failure for retry' behavior, not a bug", async () => {
    stubCl0udAllowAndAud0();
    const adapter = new TestWritebackAdapter();
    adapter.failNextCalls(1, "transient_error");
    const service = new ExternalWritebackService({ cl0udBaseUrl: "https://cl0ud.example.com", adapters: [adapter] });

    await expect(service.writeExternalRecord(baseRequest())).rejects.toThrow(/transient_error/);
    const failed = await service.findByIdempotencyKey("test_adapter", "idem-1");
    expect(failed?.status).toBe("failed");

    const retried = await service.writeExternalRecord(baseRequest());
    expect(retried.status).toBe("written");
    expect(adapter.calls).toHaveLength(2); // one failed attempt, one successful retry
  });

  it("different idempotency_key values against the same adapter never collide", async () => {
    stubCl0udAllowAndAud0();
    const adapter = new TestWritebackAdapter();
    const service = new ExternalWritebackService({ cl0udBaseUrl: "https://cl0ud.example.com", adapters: [adapter] });

    const a = await service.writeExternalRecord(baseRequest({ idempotency_key: "idem-a" }));
    const b = await service.writeExternalRecord(baseRequest({ idempotency_key: "idem-b" }));

    expect(adapter.calls).toHaveLength(2);
    expect(a.write_id).not.toBe(b.write_id);
    expect(a.external_id).not.toBe(b.external_id);
  });
});

describe("ExternalWritebackService — idempotency under concurrency (the regression-critical case)", () => {
  afterEach(() => vi.restoreAllMocks());

  it("two concurrent writeExternalRecord calls with the SAME idempotency_key result in exactly ONE adapter.write call — the claim is synchronous and race-free within one process", async () => {
    stubCl0udAllowAndAud0();
    const adapter = new TestWritebackAdapter();
    const service = new ExternalWritebackService({ cl0udBaseUrl: "https://cl0ud.example.com", adapters: [adapter] });

    const [resultA, resultB] = await Promise.all([
      service.writeExternalRecord(baseRequest()),
      service.writeExternalRecord(baseRequest()),
    ]);

    // The critical assertion: no matter how the two promises interleave afterwards, the
    // external system is only ever actually written to once for this idempotency_key. This is
    // the exact property v0-datahub's invoice-bridge.ts's own atomic Postgres UPDATE claim
    // protected in production (two concurrent triggers can't both create a Bill) -- see
    // external-writeback-service.ts's own module doc comment for how this port achieves the
    // same guarantee via a synchronous, pre-first-await, in-memory claim.
    expect(adapter.calls).toHaveLength(1);

    // One call wins the claim and completes the real write; the other observes the claim
    // in-flight and returns 'skipped_duplicate' without ever touching the adapter.
    expect([resultA.status, resultB.status].sort()).toEqual(["skipped_duplicate", "written"]);
  });

  it("three concurrent calls with the same idempotency_key still only ever produce one adapter.write call", async () => {
    stubCl0udAllowAndAud0();
    const adapter = new TestWritebackAdapter();
    const service = new ExternalWritebackService({ cl0udBaseUrl: "https://cl0ud.example.com", adapters: [adapter] });

    const results = await Promise.all([
      service.writeExternalRecord(baseRequest()),
      service.writeExternalRecord(baseRequest()),
      service.writeExternalRecord(baseRequest()),
    ]);

    expect(adapter.calls).toHaveLength(1);
    const writtenCount = results.filter((r) => r.status === "written").length;
    const skippedCount = results.filter((r) => r.status === "skipped_duplicate").length;
    expect(writtenCount).toBe(1);
    expect(skippedCount).toBe(2);
  });
});