diff --git a/src/lineage/lineage-emitter.ts b/src/lineage/lineage-emitter.ts index b1cf9fa..d57c128 100644 --- a/src/lineage/lineage-emitter.ts +++ b/src/lineage/lineage-emitter.ts @@ -1,24 +1,18 @@ /** - * src/lineage/lineage-emitter.ts - * =============================== - * 0pt's OpenLineage emitter (Master-DoD §10 Lineage). One narrow API for - * capability services to record "this run read X and produced Y" against - * AUD0's real OpenLineage ingest (`POST /v1/aud0/openlineage` — see - * src/clients/aud0-client.ts for the wire). + * 0pt OpenLineage emitter. * - * Design points: - * - Deterministic runIds (caller-supplied), so an idempotent retry of a - * capability re-emits the SAME event and AUD0 dedups it instead of - * recording a phantom second run. - * - Canonical SIAX resource refs (`st0re:artifact:...`, `0pt:dataset:...`) - * go in dataset facets so evidence chains can be joined across products - * (§16 "Evidence chains: Link source → transform → model → decision"). - * - Emission is evidence of what already happened — a lineage POST failure - * is surfaced to the caller as a warning value, never an exception that - * rolls back the completed work. Callers that need hard evidence - * guarantees must use an outbox (tracked honestly in STATUS.md). + * 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. */ import type { Aud0Client } from "../clients/aud0-client"; +import { + createAud0OutboxRepositoryFromEnv, + type Aud0OutboxRepository, +} from "../evidence/aud0-outbox"; import { OPT_OPENLINEAGE_PRODUCER, type OpenLineageDatasetRef, @@ -29,35 +23,30 @@ import { export interface LineageEmitInput { tenant_id: string; project_id?: string; - /** The 0pt capability that ran, e.g. "0pt.quality.run.v1". Becomes job.name. */ capability_id: string; - /** Deterministic run id (e.g. the run/version/export id being recorded). */ run_id: string; event_type: OpenLineageEventType; - /** ISO timestamp; defaults to now. */ event_time?: string; - /** Canonical SIAX refs and/or plain names consumed by the run. */ inputs?: LineageDatasetInput[]; - /** Canonical SIAX refs and/or plain names produced by the run. */ outputs?: LineageDatasetInput[]; correlation_id?: string; - /** Extra run facets (namespaced siax*). */ run_facets?: Record; } export interface LineageDatasetInput { - /** Human/system dataset name, e.g. "dataset/ds_x" or "opensearch/opt-companies__g2". */ name: string; - /** Optional canonical SIAX resource ref for cross-product joining. */ ref?: string; namespace?: string; facets?: Record; } export interface LineageEmitResult { + /** True only when AUD0 acknowledged the event synchronously. */ emitted: boolean; + /** True when the exact event is durably persisted for at-least-once delivery. */ + queued: boolean; run_id: string; - /** Present when emission failed (AUD0 unreachable etc.). */ + outbox_id?: string; warning?: string; } @@ -74,8 +63,21 @@ function toDatasetRef(d: LineageDatasetInput): OpenLineageDatasetRef { }; } +function safeWarning(error: unknown): string { + return error instanceof Error ? error.name : "UnknownError"; +} + export class LineageEmitter { - constructor(private readonly aud0: Aud0Client) {} + private readonly outbox?: Aud0OutboxRepository; + + constructor( + private readonly aud0: Aud0Client, + opts?: { outbox?: Aud0OutboxRepository | null }, + ) { + this.outbox = opts && "outbox" in opts + ? opts.outbox ?? undefined + : createAud0OutboxRepositoryFromEnv(); + } buildRunEvent(input: LineageEmitInput): OpenLineageRunEvent { return { @@ -96,17 +98,56 @@ export class LineageEmitter { }; } - /** Emit one run event. Never throws — see module doc comment. */ + /** + * Record one run event without throwing 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. + */ async emit(input: LineageEmitInput): Promise { 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); - return { emitted: true, run_id: input.run_id }; - } catch (err) { + 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: ${err instanceof Error ? err.message : String(err)}`, + warning: `openlineage emission failed: ${safeWarning(error)}`, }; } }