@intent-driven/bridge-postgres

@intent-driven/bridge-postgres

Bridge между Postgres БД и IDF Φ event-log: CDC mirror (read-only, БД primary) и gateway-mode (Fold принимает agent calls, валидирует, при confirm пишет в БД).

Status

Feature	Status
pgoutput decoder + createMirror skeleton	✅ PR 5a
createPgLogicalMirror + idempotent setup helpers	✅ PR 5b
Schema-drift detector (detectDrift)	✅ PR 5c
Gateway mode + transactional outbox	✅ PR 5d
Reference deployment + migration guide	планируется PR 5e

Gateway mode (PR 5d)

import {
  ensureOutboxTable,
  createGateway,
  createOutboxProcessor,
} from "@intent-driven/bridge-postgres";
import pg from "pg";

const pool = new pg.Pool({ connectionString });
const setupClient = await pool.connect();
await ensureOutboxTable(setupClient);
setupClient.release();

const gateway = createGateway({ pool, ontology });

// Agent call:
const result = await gateway.execute({
  intentName: "create_order",
  params: { amount: 100, status: "new" },
  ctx: { idempotencyKey: "uuid-...", role: { base: "agent" } },
});
// → { ok: true, result: <row>, outboxId: 42 }
//   ИЛИ
// → { ok: false, code: "preapproval_denied", message, details }

// Outbox processor (отдельный worker):
const processor = createOutboxProcessor({
  pool,
  publish: async (row) => { await foldRuntime.publishEffect(row); },
});
await processor.start();

Atomicity: business write + outbox row в одной TX. Validator reject (preapproval / invariants / irreversibility) → ничего не пишется. Idempotency через idempotency_key UNIQUE constraint (replay safe).

Mirror API (PR 5a)

import { createMirror } from "@intent-driven/bridge-postgres";

const mirror = createMirror({
  stream: someAsyncIterableOfPgoutputBuffers,
  async onEffect(effect, ctx) {
    // effect: { alpha: "insert"|"replace"|"remove", target, value?, key?, ... }
    // ctx: { transaction: { xid, finalLSN, commitTimestamp }, relation }
    await foldRuntime.proposeEffect(effect);
  },
  async onCommit(info) {
    // commit batch boundary — flush к Φ event-log
  },
});

await mirror.start();
console.log(mirror.stats);
// { decoded: 1234, insert: 100, update: 80, delete: 5, relation: 3, begin: 50, commit: 50 }

В PR 5b будет high-level helper:

const mirror = createPgLogicalMirror({
  connectionString: "postgres://repluser:pass@host/db",
  slotName: "idf_mirror",
  publicationName: "idf_pub",
  onEffect, onCommit,
});

License

MIT.