feat(bridge): persist authenticated events before OpenFn CLI execution

services/event-bridge/README.md

@@ -1,11 +1,13 @@
 # BREG event bridge
 
-This pilot service verifies BREG CloudEvents before submitting a work order to a
-fixed OpenFn webhook. It uses Node.js 22.23.2 built-ins and has no package
+This service verifies BREG CloudEvents before either committing a local CLI
+workflow to a durable inbox or submitting a work order to a fixed OpenFn webhook. It uses Node.js 22.23.2 built-ins and has no package
 dependencies. Start with `node services/event-bridge/src/server.js` from the
 repository root.
 
-`POST /events/breg` is the only ingress route. The method, path without query,
+`POST /events/breg` is the default ingress route. Set `BREG_EVENT_PATH` to one
+fixed `/events/<name>` path when an authored BREG destination uses another
+route. The method, exact path without query,
 content type, all signed headers, and bounded exact body bytes are covered by
 the BREG `breg-webhook-signature-v1` HMAC contract: unsigned 64-bit big-endian
 byte lengths precede each field in the order defined in Registry Stack's
@@ -14,25 +16,29 @@ decoded. Signatures are compared in constant time. Duplicate signed headers,
 content encoding, wrong source/type/schema, stale delivery times, and invalid
 payloads are refused before forwarding.
 
-The pilot accepts exactly the common BREG payload members `entity`, `recordId`,
+The receiver accepts exactly the common BREG payload members `entity`, `recordId`,
 `revision`, `trigger`, `packageRevision`, and `values`. Entity, event schema,
 trigger, and projected field names must match configuration. Record/event ids
 must be canonical lowercase UUIDs, revision a positive safe integer, package
 revision a SHA-256 binding, and projected values strings of at most 1024 UTF-8
-bytes. This deliberately implements the pilot's string projection; it is not
-a general JSON Schema validator or lifecycle-event receiver.
+bytes. Reviewed `request_lifecycle` events also carry closed request transition
+metadata. Per-event `entity` and `valueFields` override the common defaults.
+This is a bounded string projection receiver, not a general JSON Schema validator.
 
 Configuration is required unless a default is listed:
 
 | Variable | Meaning |
 | --- | --- |
 | `BREG_HMAC_KEY_FILE` | File containing exact HMAC key bytes, at least 32 bytes. |
 | `BREG_EXPECTED_SOURCE` | Exact CloudEvent source for the registry instance. |
+| `BREG_EVENT_PATH` | Optional exact signed ingress path, default `/events/breg`; one lowercase name under `/events/`, with letters, digits or hyphens. |
 | `BREG_EXPECTED_ENTITY` | Exact payload entity, `farm` for this pilot. |
-| `BREG_EXPECTED_EVENTS_FILE` | JSON object mapping each exact type to `{ "schema": "exact-dataschema", "trigger": "created" }`; `patched` is also supported. |
+| `BREG_EXPECTED_EVENTS_FILE` | JSON object mapping each exact type to `{ "schema": "exact-dataschema", "trigger": "created" }`; `patched` and `request_lifecycle` are also supported. Optional per-event `entity` and `valueFields` override defaults; CLI mode requires `effect`. |
 | `BREG_ALLOWED_VALUE_FIELDS_FILE` | JSON array of exact projected string fields, `["local-identifier"]` for this pilot. |
-| `OPENFN_WEBHOOK_URL` | Fixed webhook URL, with no credentials, query or fragment. |
-| `OPENFN_API_KEY_FILE` | Exact printable API key for `x-api-key`; no trailing newline. |
+| `OPENFN_DELIVERY_MODE` | Default `webhook`; use `cli` for durable local execution. |
+| `OPENFN_INBOX_PATH` | Required in CLI mode. Persistent SQLite file in a private directory. |
+| `OPENFN_WEBHOOK_URL` | Webhook mode only. Fixed URL, with no credentials, query or fragment. |
+| `OPENFN_API_KEY_FILE` | Webhook mode only. Exact printable API key for `x-api-key`; no trailing newline. |
 | `ALLOW_HTTP` | Default false. Set exactly `true` only for an explicitly trusted internal HTTP deployment. HTTPS otherwise required. |
 | `PORT` | Default 8081. |
 | `MAX_BODY_BYTES` | Default 65536; maximum 1048576. |
@@ -68,13 +74,13 @@ The authenticated OpenFn webhook receives this JSON envelope:
 }
 ```
 
-A 2xx OpenFn response must contain a UUID `work_order_id` and no `error` field.
+In webhook mode, a 2xx OpenFn response must contain a UUID `work_order_id` and no `error` field.
 Lightning 2.18.2 does not return a `run_id` for asynchronous acceptance.
 Only after this check does the bridge return `202 {"status":"accepted"}` to BREG. This
 acknowledges work-order acceptance, not workflow completion. Upstream rejection,
 malformed/generic success responses, redirects, transport failures, or oversized
 responses produce 502; deadline expiry produces 504. There are no internal
-retries, redirect following, or persistent bridge inbox.
+retries, redirect following, or persistent bridge inbox in webhook mode.
 
 If acceptance occurs but the response is lost, BREG can submit another work
 order. The workflow and destination must tolerate this, including replay under
@@ -86,3 +92,93 @@ administrators because it contains no independent BREG verifier.
 `GET /healthz` returns a static readiness response. Request payloads, headers,
 secrets, and upstream error bodies are never logged or reflected in failures.
 Run focused checks with `npm --prefix services/event-bridge test`.
+
+## Durable CLI mode
+
+Set `OPENFN_DELIVERY_MODE=cli` and `OPENFN_INBOX_PATH` in the receiver. Omit
+webhook URL/API-key settings. Each accepted event type has one stable logical
+`effect`, such as `notify-applicant-v1`. Run one receiver per configured source;
+receivers and workers may share a local inbox. The receiver validates HMAC and
+payload, commits SQLite with WAL and FULL synchronous mode, and returns 202.
+It never waits for a workflow. Full or unavailable storage produces a refusal
+so BREG can retry; a stopped worker does not delay registry decisions.
+
+The worker uses the same `OPENFN_INBOX_PATH` and a reviewed
+`OPENFN_WORKFLOWS_FILE` mapping effects to jobs:
+
+```json
+{
+  "notify-applicant-v1": {
+    "job": "./jobs/notify.js",
+    "adaptor": "@openfn/language-common=/absolute/path/to/installed/common",
+    "configurationFile": "./private/notification.json"
+  }
+}
+```
+
+Job and configuration paths resolve relative to this file. Use an absolute
+installed adaptor path as shown. `configurationFile` is optional. Use
+`adaptors: ["package=/path", "second-package=/path"]` instead of `adaptor`
+when a job composes two or more adaptors; never set both forms. Pin the CLI
+and adaptors at installation; execution disables automatic installation and
+adaptor-name expansion. `OPENFN_BINARY` selects the installed CLI executable
+(default `openfn`). `OPENFN_JOB_TIMEOUT_MS` defaults to 60000, maximum 1800000.
+Use a local filesystem, not a network share, and preserve the inbox directory
+across ordinary restarts. Protect configuration files and the whole state
+volume; SQLite and per-attempt state files are owner-only.
+
+```sh
+npm --prefix services/event-bridge run worker -- run
+npm --prefix services/event-bridge run worker -- status
+npm --prefix services/event-bridge run worker -- replay <effect-id>
+```
+
+`once` executes at most one ready item. Status returns only opaque effect IDs,
+state, attempt counts and generic failures. A workflow receives
+`state.configuration`, the authenticated event envelope as `state.data`, and
+`state.eventEffectId`. The latter is a stable SHA-256 identity over source,
+event ID and effect. Retry generation and delivery attempt do not change it.
+Reusing an identity with different event content is refused. Destination writes
+must use this stable identity or a domain-stable report identity for idempotency.
+
+The CLI runs outside the receiver, with a bounded lease, deadline and private
+temporary working directory. Exit code zero alone does not establish success:
+OpenFn CLI 1.40.1 reports workflow failures in `state.errors`. The worker requires
+a valid bounded result with no errors before marking the item successful.
+It discards stdout/stderr, removes temporary input/output, and drops the logical
+event payload after success while retaining the receipt needed for deduplication.
+SQLite/WAL backups may retain old pages; protect the full volume as event data.
+Tokens, raw Evidence and credentials must never be event projections.
+
+Failures retry after 1, 2, 4 and 8 seconds, then enter `dead` after five attempts.
+An interrupted worker's expired lease also counts toward that bound. Fix the
+underlying job, credentials or destination, inspect status, then replay the dead
+item by its effect ID. Replay keeps the effect identity. Successful items cannot
+be replayed by this command. Preserve receipts for the whole source replay
+window; deleting the inbox also deletes destination deduplication history.
+
+SIGINT/SIGTERM lets the current job finish within its deadline and stops the
+worker. Allow the supervisor at least the configured job timeout before force
+termination. After a forced kill, stop its remaining CLI child process group before
+restarting it, wait for the lease to expire, and remove only its abandoned
+`attempt-*` directories. A destination may already have accepted an effect
+before the worker records success. Delivery is therefore at least once; email
+can be duplicated after such a crash. Destinations that support idempotency
+should enforce it independently. Dead-letter recovery cannot undo an email.
+
+## Verify the local handoff
+
+Run `npm --prefix services/event-bridge run check` and
+`npm --prefix services/event-bridge test`. The explicit CLI journey additionally
+requires installed OpenFn CLI 1.40.1 and common adaptor 3.3.4:
+
+```sh
+OPENFN_BINARY=/absolute/path/to/openfn \
+OPENFN_COMMON_PATH=/absolute/path/to/installed/common \
+  npm --prefix services/event-bridge run test:cli
+```
+
+This exercises real CLI success, restart/replay, a zero-exit workflow error,
+retry and transient-state cleanup. It installs nothing during execution.
+See the maintained [OpenFn CLI usage](https://docs.openfn.org/documentation/cli-usage)
+for installation and adaptor arguments.

services/event-bridge/package.json

@@ -5,8 +5,10 @@
   "type": "module",
   "scripts": {
     "start": "node src/server.js",
-    "check": "node --check src/server.js",
-    "test": "node --test test/*.test.js"
+    "worker": "node src/worker.js",
+    "check": "node --check src/server.js && node --check src/inbox.js && node --check src/worker.js",
+    "test": "node --test test/*.test.js",
+    "test:cli": "node test/cli-journey.mjs"
   },
   "engines": { "node": ">=22.23.2" }
 }

services/event-bridge/src/inbox.js

@@ -0,0 +1,99 @@
+import { createHash, randomUUID } from 'node:crypto';
+import { chmodSync, mkdirSync } from 'node:fs';
+import { dirname, resolve } from 'node:path';
+import { DatabaseSync } from 'node:sqlite';
+
+function canonical(value) {
+  if (Array.isArray(value)) return value.map(canonical);
+  if (value !== null && typeof value === 'object') {
+    return Object.fromEntries(Object.keys(value).sort().map(key => [key, canonical(value[key])]));
+  }
+  return value;
+}
+const digest = value => createHash('sha256').update(JSON.stringify(canonical(value))).digest('hex');
+
+// The receiver commits before acknowledging. Workers hold no transaction while
+// executing a workflow. A crashed worker's lease eventually permits redelivery.
+export class DurableInbox {
+  constructor(path, { maxPending = 10000 } = {}) {
+    if (!Number.isSafeInteger(maxPending) || maxPending < 1) throw new Error('invalid inbox limit');
+    mkdirSync(dirname(resolve(path)), { recursive: true, mode: 0o700 });
+    this.db = new DatabaseSync(path);
+    chmodSync(path, 0o600);
+    this.maxPending = maxPending;
+    this.db.exec(`PRAGMA busy_timeout=1000; PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;
+      CREATE TABLE IF NOT EXISTS inbox (
+        id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, effect TEXT NOT NULL,
+        envelope TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'pending',
+        attempts INTEGER NOT NULL DEFAULT 0, ready_at INTEGER NOT NULL DEFAULT 0,
+        lease TEXT, lease_until INTEGER, failure TEXT,
+        CHECK(status IN ('pending','running','succeeded','dead'))
+      );
+      CREATE INDEX IF NOT EXISTS inbox_ready ON inbox(status, ready_at);`);
+  }
+
+  accept(envelope, effect) {
+    if (typeof effect !== 'string' || !/^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$/.test(effect)) {
+      throw new Error('invalid workflow effect');
+    }
+    const id = digest([envelope.event.source, envelope.event.id, effect]);
+    const fingerprint = digest({ event: envelope.event, data: envelope.data });
+    this.db.exec('BEGIN IMMEDIATE');
+    try {
+      const prior = this.db.prepare('SELECT fingerprint FROM inbox WHERE id=?').get(id);
+      if (prior && prior.fingerprint !== fingerprint) throw new Error('event identity conflict');
+      if (!prior) {
+        const { count } = this.db.prepare("SELECT count(*) AS count FROM inbox WHERE status != 'succeeded'").get();
+        if (count >= this.maxPending) throw new Error('inbox full');
+        this.db.prepare('INSERT INTO inbox(id,fingerprint,effect,envelope) VALUES(?,?,?,?)')
+          .run(id, fingerprint, effect, JSON.stringify(envelope));
+      }
+      this.db.exec('COMMIT');
+      return id;
+    } catch (error) { this.db.exec('ROLLBACK'); throw error; }
+  }
+
+  claim({ now = Date.now(), leaseMs = 65000, maxAttempts = 5 } = {}) {
+    if (!Number.isSafeInteger(leaseMs) || leaseMs < 1000) throw new Error('invalid lease');
+    this.db.exec('BEGIN IMMEDIATE');
+    try {
+      this.db.prepare(`UPDATE inbox SET status=CASE WHEN attempts>=? THEN 'dead' ELSE 'pending' END,
+        lease=NULL, lease_until=NULL, failure='worker_interrupted'
+        WHERE status='running' AND lease_until<=?`).run(maxAttempts, now);
+      const row = this.db.prepare("SELECT * FROM inbox WHERE status='pending' AND ready_at<=? ORDER BY ready_at, rowid LIMIT 1").get(now);
+      if (row) {
+        row.lease = randomUUID();
+        row.attempts += 1;
+        this.db.prepare("UPDATE inbox SET status='running', attempts=?, lease=?, lease_until=? WHERE id=?")
+          .run(row.attempts, row.lease, now + leaseMs, row.id);
+      }
+      this.db.exec('COMMIT');
+      return row ? { id: row.id, effect: row.effect, envelope: JSON.parse(row.envelope),
+        attempts: row.attempts, lease: row.lease } : undefined;
+    } catch (error) { this.db.exec('ROLLBACK'); throw error; }
+  }
+
+  complete(item) {
+    return this.db.prepare(`UPDATE inbox SET status='succeeded', envelope='{}', lease=NULL, lease_until=NULL, failure=NULL
+      WHERE id=? AND status='running' AND lease=?`).run(item.id, item.lease).changes === 1;
+  }
+
+  fail(item, { now = Date.now(), maxAttempts = 5, retryMs = 1000 } = {}) {
+    const status = item.attempts >= maxAttempts ? 'dead' : 'pending';
+    return this.db.prepare(`UPDATE inbox SET status=?, ready_at=?, lease=NULL, lease_until=NULL, failure='workflow_failed'
+      WHERE id=? AND status='running' AND lease=?`)
+      .run(status, now + retryMs * 2 ** Math.min(item.attempts - 1, 10), item.id, item.lease).changes === 1;
+  }
+
+  replay(id) {
+    return this.db.prepare("UPDATE inbox SET status='pending', attempts=0, ready_at=0, failure=NULL WHERE id=? AND status='dead'")
+      .run(id).changes === 1;
+  }
+
+  // Operator status deliberately omits projected data and credential material.
+  status() {
+    return this.db.prepare('SELECT id,effect,status,attempts,ready_at,lease_until,failure FROM inbox ORDER BY rowid').all();
+  }
+
+  close() { this.db.close(); }
+}