From f0958cf7086a32ddd2d1a44d6f09f39e4e3f2d72 Mon Sep 17 00:00:00 2001
From: Simon Axelsson <simon@siax.io>
Date: Mon, 7 Sep 2026 17:54:55 +0200
Subject: [PATCH] fix: centralize lineage durability in AUD0 client boundary

---
 src/lineage/lineage-emitter.ts | 83 ++++++++++------------------------
 1 file changed, 23 insertions(+), 60 deletions(-)

diff --git a/src/lineage/lineage-emitter.ts b/src/lineage/lineage-emitter.ts
index d57c128..294aa07 100644
--- a/src/lineage/lineage-emitter.ts
+++ b/src/lineage/lineage-emitter.ts
@@ -1,18 +1,13 @@
 /**
  * 0pt OpenLineage emitter.
  *
- * AUD0 is canonical evidence. When 0pt has a repository DATABASE_URL the
- * emitter first persists the exact RunEvent to the local AUD0 outbox and
- * returns `queued:true`; the dispatcher is then the only delivery path. This
- * preserves the historic never-throw-after-domain-work contract while removing
- * AUD0 availability as an evidence-loss window. In env-less/unit runtimes the
- * emitter keeps the direct client behavior.
+ * Durability belongs to the AUD0 client boundary, not here. The default
+ * createAud0Client() persists to the local outbox when DATABASE_URL exists;
+ * the raw dispatcher client (durable:false) performs the actual HTTP delivery.
+ * This emitter therefore builds one canonical RunEvent and records it exactly
+ * once through the injected client.
  */
 import type { Aud0Client } from "../clients/aud0-client";
-import {
-  createAud0OutboxRepositoryFromEnv,
-  type Aud0OutboxRepository,
-} from "../evidence/aud0-outbox";
 import {
   OPT_OPENLINEAGE_PRODUCER,
   type OpenLineageDatasetRef,
@@ -41,9 +36,9 @@ export interface LineageDatasetInput {
 }
 
 export interface LineageEmitResult {
-  /** True only when AUD0 acknowledged the event synchronously. */
+  /** True only when AUD0 itself acknowledged the event synchronously. */
   emitted: boolean;
-  /** True when the exact event is durably persisted for at-least-once delivery. */
+  /** True when the event is durably queued for at-least-once AUD0 delivery. */
   queued: boolean;
   run_id: string;
   outbox_id?: string;
@@ -68,16 +63,7 @@ function safeWarning(error: unknown): string {
 }
 
 export class LineageEmitter {
-  private readonly outbox?: Aud0OutboxRepository;
-
-  constructor(
-    private readonly aud0: Aud0Client,
-    opts?: { outbox?: Aud0OutboxRepository | null },
-  ) {
-    this.outbox = opts && "outbox" in opts
-      ? opts.outbox ?? undefined
-      : createAud0OutboxRepositoryFromEnv();
-  }
+  constructor(private readonly aud0: Aud0Client) {}
 
   buildRunEvent(input: LineageEmitInput): OpenLineageRunEvent {
     return {
@@ -99,55 +85,32 @@ export class LineageEmitter {
   }
 
   /**
-   * Record one run event without throwing after domain work has completed.
+   * Record one run event. Never throws after domain work has completed.
    *
-   * With an outbox: enqueue only; the dispatcher owns delivery and retry.
-   * Without an outbox: direct AUD0 call for local/tests/backwards compatibility.
-   * If durable enqueue itself fails, a direct AUD0 attempt is still made so an
-   * available AUD0 can preserve evidence; the warning makes the durability
-   * degradation explicit rather than silently green.
+   * A DurableAud0Client returns an acknowledgement whose event_id is the
+   * immutable `outbox:<id>` handle. A raw HTTP client returns AUD0's event id.
+   * This lets callers distinguish durable queueing from live acknowledgement
+   * without duplicating the outbox implementation in this layer.
    */
   async emit(input: LineageEmitInput): Promise<LineageEmitResult> {
     const event = this.buildRunEvent(input);
-
-    if (this.outbox) {
-      try {
-        const row = await this.outbox.enqueueOpenLineage(input.tenant_id, APP_ID, event);
-        return {
-          emitted: row.status === "delivered",
-          queued: row.status === "pending",
-          run_id: input.run_id,
-          outbox_id: row.outbox_id,
-        };
-      } catch (queueError) {
-        try {
-          await this.aud0.ingestOpenLineageEvent(input.tenant_id, APP_ID, event);
-          return {
-            emitted: true,
-            queued: false,
-            run_id: input.run_id,
-            warning: `durable outbox enqueue failed (${safeWarning(queueError)}); AUD0 accepted direct fallback`,
-          };
-        } catch (aud0Error) {
-          return {
-            emitted: false,
-            queued: false,
-            run_id: input.run_id,
-            warning: `durable outbox enqueue failed (${safeWarning(queueError)}); direct AUD0 delivery failed (${safeWarning(aud0Error)})`,
-          };
-        }
-      }
-    }
-
     try {
-      await this.aud0.ingestOpenLineageEvent(input.tenant_id, APP_ID, event);
+      const ack = await this.aud0.ingestOpenLineageEvent(input.tenant_id, APP_ID, event);
+      if (ack.event_id.startsWith("outbox:")) {
+        return {
+          emitted: false,
+          queued: true,
+          run_id: input.run_id,
+          outbox_id: ack.event_id.slice("outbox:".length),
+        };
+      }
       return { emitted: true, queued: false, run_id: input.run_id };
     } catch (error) {
       return {
         emitted: false,
         queued: false,
         run_id: input.run_id,
-        warning: `openlineage emission failed: ${safeWarning(error)}`,
+        warning: `openlineage recording failed: ${safeWarning(error)}`,
       };
     }
   }