0pt/src/services/external-writeback-service.ts

/**
 * src/services/external-writeback-service.ts
 * =============================================
 * `0pt.write_external_record` -- "POST a validated, already-approved
 * structured record to a registered external-system adapter." Generic in its
 * core: this service has no Odoo/invoice/Protosell-specific logic anywhere
 * in it -- it resolves an `adapter_id` to a registered `WritebackAdapter`
 * (../adapters/writeback-adapter.ts) and delegates the actual write to it.
 * `../adapters/odoo-bills-adapter.ts`'s OdooBillsAdapter is ONE such
 * adapter, registered by a caller the same way any other adapter would be --
 * this service never imports it directly.
 *
 * WHY THIS EXISTS / WHERE IT SITS IN THE CHAIN: `act0/src/
 * approval-writeback-templates.ts`'s generic `0pt_document_approval_then_
 * external_write` Template chains a captured+extracted document
 * (0pt.document_ocr's real terminal event, `0pt.document_extracted`) through
 * an Approval Engine gate, THEN calls this capability. By the time
 * `writeExternalRecord` runs, a human has already approved the write -- this
 * service still re-checks CL0UD policy (per docs/PRODUCT_PLAN.md section
 * 4.1's hard rule that CL0UD is always re-consulted, the same rule
 * document-extraction-service.ts already follows -- confirmed by grepping
 * this repo's src/services/ for evaluatePolicy()/Cl0udClient usage; the
 * only other real service that does the same is iceberg-write-service.ts.
 * An earlier version of this comment also cited `quality-check-service.ts`
 * here; that file does not exist anywhere in this repo),
 * but does not itself re-verify the upstream Approval record (ACT0 owns
 * approvals; 0pt's own capability boundary is "did CL0UD say yes for THIS
 * call", same separation of concerns every other 0pt service in this repo
 * already draws).
 *
 * IDEMPOTENCY (per this task's own instruction not to regress the proven
 * pattern ported from v0-datahub's `lib/odoo/invoice-bridge.ts`
 * `processInboxToOdoo`): that original code CLAIMS a row (an atomic
 * Postgres `UPDATE ... WHERE odoo_move_id IS NULL AND odoo_state != 'posting'`)
 * BEFORE ever calling the external system, so two concurrent triggers (a
 * manual approve racing a cron, or a retried ACT0 step racing itself) cannot
 * both post the same record. This service ports that SAME "claim before you
 * call out" shape forward as a generic, adapter-agnostic in-memory claim,
 * keyed on `(adapter_id, idempotency_key)`:
 *   - the claim check-and-set happens SYNCHRONOUSLY, before this function's
 *     first `await` -- JavaScript's single-threaded run-to-first-await
 *     semantics make this exactly as race-free, within one process, as the
 *     original's atomic SQL UPDATE was across concurrent DB clients (see
 *     `writeExternalRecord`'s own inline comment for exactly where the line
 *     is drawn, and tests/p0/external-writeback-service.test.ts's own
 *     concurrency test, which calls `writeExternalRecord` twice with the
 *     same idempotency key via `Promise.all` and asserts the underlying
 *     adapter's `write` is invoked exactly once).
 *   - HONEST SCOPE: this in-memory claim is process-local -- it does not
 *     survive a restart and does not protect a multi-instance deployment,
 *     unlike the original's real Postgres row lock. An adapter with its own
 *     persistent storage (OdooBillsAdapter) additionally refuses to
 *     re-post a record whose payload already carries an `existing_external_id`
 *     -- see WritebackAdapter.write's own doc comment for why this
 *     second layer matters and odoo-bills-adapter.ts for exactly how it is
 *     ported. A production deployment backing this service with real
 *     persistent storage (same "pure core, thin adapter" doctrine every
 *     other 0pt service in this repo already documents) should replace this
 *     in-memory Map with a real claim row, at which point the SAME
 *     synchronous-claim-before-any-await shape still applies, now backed by
 *     a real transaction instead of a JS Map.
 */
import crypto from "crypto";
import { createCl0udClient, type Cl0udClient } from "../clients/cl0ud-client";
import { createAud0Client, type Aud0Client } from "../clients/aud0-client";
import type { WritebackAdapter } from "../adapters/writeback-adapter";
import type { ExternalWriteRecord, ExternalWriteRequest } from "../types/external-writeback";

export interface ExternalWritebackServiceOptions {
  cl0udBaseUrl?: string;
  aud0BaseUrl?: string;
  adapters?: readonly WritebackAdapter[];
}

export class ExternalWritebackService {
  private writes: Map<string, ExternalWriteRecord> = new Map();
  /** `${adapter_id}::${idempotency_key}` -> write_id. See this module's own doc comment's
   * "IDEMPOTENCY" section for exactly what this does and does not guarantee. */
  private claimedByIdempotencyKey: Map<string, string> = new Map();
  private adapters: Map<string, WritebackAdapter> = new Map();
  private cl0ud: Cl0udClient;
  private aud0: Aud0Client;

  constructor(opts?: ExternalWritebackServiceOptions) {
    this.cl0ud = createCl0udClient(opts?.cl0udBaseUrl);
    this.aud0 = createAud0Client(opts?.aud0BaseUrl);
    for (const adapter of opts?.adapters ?? []) {
      this.registerAdapter(adapter);
    }
  }

  registerAdapter(adapter: WritebackAdapter): void {
    this.adapters.set(adapter.adapterId, adapter);
  }

  getAdapter(adapterId: string): WritebackAdapter {
    const adapter = this.adapters.get(adapterId);
    if (!adapter) throw new Error(`ExternalWritebackService: no adapter registered for '${adapterId}'`);
    return adapter;
  }

  async writeExternalRecord(request: ExternalWriteRequest): Promise<ExternalWriteRecord> {
    const adapter = this.adapters.get(request.adapter_id);
    if (!adapter) {
      throw new Error(`ExternalWritebackService: no adapter registered for '${request.adapter_id}'`);
    }
    if (!adapter.isConfigured()) {
      throw new Error(
        `ExternalWritebackService: adapter '${request.adapter_id}' is not configured — refusing to fabricate a write`,
      );
    }

    // --- SYNCHRONOUS claim section: no `await` above this line, and none until the claim is
    // set below. This is what makes the claim race-free under this module's own doc comment's
    // "IDEMPOTENCY" doctrine -- see that comment for the full explanation and its own scope
    // caveat (process-local only).
    const claimKey = `${request.adapter_id}::${request.idempotency_key}`;
    const existingWriteId = this.claimedByIdempotencyKey.get(claimKey);
    if (existingWriteId) {
      const existing = this.writes.get(existingWriteId);
      if (existing && existing.status === "written") {
        return existing; // already posted -- idempotent no-op, matches invoice-bridge.ts's own
        // "Redan bokförd i Odoo... hoppar över" early return.
      }
      if (existing && existing.status === "claimed") {
        // A write for this exact idempotency key is already in flight (or, in this in-memory
        // reference implementation, was left 'claimed' by a caller that never awaited the
        // result) -- do not race a second POST to the external system. Matches
        // invoice-bridge.ts's own "Redan under postning eller bokförd — hoppar över" claim-
        // conflict response.
        return { ...existing, status: "skipped_duplicate" };
      }
      // existing.status === 'failed' (or the record is missing for any other reason): fall
      // through and re-claim -- a failed write is safe and expected to retry, matching
      // invoice-bridge.ts's own "Release the 'posting' claim... can be retried later" behavior.
    }

    const writeId = `ewr_${crypto.randomUUID()}`;
    const now = new Date().toISOString();
    let record: ExternalWriteRecord = {
      write_id: writeId,
      adapter_id: request.adapter_id,
      idempotency_key: request.idempotency_key,
      record_type: request.record_type,
      status: "claimed",
      tenant_id: request.tenant_id,
      project_id: request.project_id,
      actor_id: request.actor_id,
      correlation_id: request.correlation_id,
      created_at: now,
    };
    this.writes.set(writeId, record);
    this.claimedByIdempotencyKey.set(claimKey, writeId);
    // --- End synchronous claim section. Every branch below is async and safe to interleave. ---

    try {
      const policy = await this.cl0ud.evaluatePolicy({
        tenant_id: request.tenant_id,
        project_id: request.project_id,
        actor_id: request.actor_id,
        capability_id: "0pt.write_external_record",
        resource_type: "external_write",
        resource_id: writeId,
        risk_level: "L4",
        correlation_id: request.correlation_id,
      });
      if (policy.decision === "deny") {
        throw new Error(`Policy denied: ${policy.reason}`);
      }
      if (policy.decision === "approval_required") {
        // Per this service's own module doc comment: the ACT0 Approval Engine gate that
        // precedes this capability in the approval-writeback-templates.ts chain is a SEPARATE
        // mechanism from CL0UD's own policy decision here -- if CL0UD independently still says
        // approval_required for this specific call, this service does not silently bypass it
        // (same "never lie about what happened" doctrine every other 0pt service in this repo
        // follows for a denied/approval_required decision).
        throw new Error(`Policy requires approval: ${policy.reason}`);
      }

      const result = await adapter.write(
        { recordType: request.record_type, payload: request.payload, idempotencyKey: request.idempotency_key },
        {
          tenant_id: request.tenant_id,
          project_id: request.project_id,
          actor_id: request.actor_id,
          correlation_id: request.correlation_id,
        },
      );
      if (!result.ok) {
        throw new Error(result.error ?? "adapter_write_failed");
      }

      record = { ...record, status: "written", external_id: result.externalId, updated_at: new Date().toISOString() };
      this.writes.set(writeId, record);

      await this.aud0.ingestEvent({
        event_id: `evt_0pt_external_record_written_${writeId}`,
        event_type: "0pt.external_record_written",
        tenant_id: request.tenant_id,
        project_id: request.project_id,
        actor_id: request.actor_id,
        app_id: "0pt",
        capability_id: "0pt.write_external_record",
        resource_type: "external_write",
        resource_id: writeId,
        risk_level: "L4",
        payload: {
          adapter_id: request.adapter_id,
          record_type: request.record_type,
          idempotency_key: request.idempotency_key,
          external_id: result.externalId,
        },
        created_at: new Date().toISOString(),
        correlation_id: request.correlation_id,
      });

      await this.aud0.ingestLineageEvent({
        lineage_event_id: `lineage_0pt_external_write_${writeId}`,
        tenant_id: request.tenant_id,
        project_id: request.project_id,
        app_id: "0pt",
        capability_id: "0pt.write_external_record",
        // `idempotency_key` is, by convention (not enforced by this generic service), the
        // upstream record's stable id -- e.g. a document_extraction_id when this write was
        // triggered by act0's document_approval_then_external_write template.
        source_type: "document_extraction",
        source_id: request.idempotency_key,
        target_type: "external_record",
        target_id: result.externalId ?? writeId,
        lineage_type: "produced",
        metadata: { adapter_id: request.adapter_id, record_type: request.record_type },
        created_at: new Date().toISOString(),
        correlation_id: request.correlation_id,
      });

      return record;
    } catch (e: unknown) {
      const message = e instanceof Error ? e.message : String(e);
      record = { ...record, status: "failed", error: message, updated_at: new Date().toISOString() };
      this.writes.set(writeId, record);
      throw e;
    }
  }

  async getWrite(writeId: string): Promise<ExternalWriteRecord> {
    const record = this.writes.get(writeId);
    if (!record) throw new Error(`External write ${writeId} not found`);
    return record;
  }

  async findByIdempotencyKey(adapterId: string, idempotencyKey: string): Promise<ExternalWriteRecord | undefined> {
    const writeId = this.claimedByIdempotencyKey.get(`${adapterId}::${idempotencyKey}`);
    return writeId ? this.writes.get(writeId) : undefined;
  }
}

/** Shared singleton, same "cross-route in-memory state" convention document-capture-service.ts's
 * own header documents (this service's writes/claims must be visible across whatever API routes
 * call writeExternalRecord vs getWrite) -- no adapters pre-registered; a caller (an API route,
 * a test, or a future 0pt/api/documents/writeback route) registers whichever adapters it needs
 * via registerAdapter before calling writeExternalRecord. */
export const externalWritebackService = new ExternalWritebackService();