External Database

External-database integration for app data — Postgres / MySQL / SQLite / MongoDB connection pooling, retry, circuit breaker, classification routing, residency enforcement, and audit hooks.

Framework state (audit_log, consent_log, _blamejs_*) stays in the local SQLite via b.db. This module is for APP DATA — when an operator keeps domain tables in Postgres / MySQL / MongoDB / libsql, they configure a backend here and use b.externalDb.query() instead of b.db.from() for those tables. The same surface also serves cluster-mode coordination (leader election advisory locks, cross-replica routing) when the cluster provider points at the same backend.

Bring-your-own-client design (per "zero npm runtime deps" rule): the operator supplies the actual DB driver via each backend's connect / query / close hooks. The framework layers connection pooling (lazy-create, idle reaping), transient-error retry, per-backend circuit breaker, classification routing (which backend serves which data class), residency enforcement against db.getDataResidency().region, and audit hooks (system.externaldb.{query,transaction,read}).

Read-replica routing exposes b.externalDb.read.query() and b.externalDb.write.query() — reads weight-round-robin across declared replicas with health tracking and primary fallback; writes always route to primary.

b.externalDb.init(opts) #

{
  backends:        { [name]: BackendConfig },   // required; one or more named backends
  defaultBackend?: string,                      // pool used when no opts.backend / classification / role match (defaults to first)
  dbRoleBackends?: { [sqlRole]: backendName },  // request-time role → backend mapping for the dbRoleFor middleware

  // BackendConfig shape:
  //   connect():            async () → driver client                 (required)
  //   query(client, sql, p): async → { rows, rowCount }              (required)
  //   close(client):        async → void                             (optional; default no-op)
  //   ping(client):         async → void                             (optional; default `SELECT 1`)
  //   beginTx / commit / rollback(client):  async → void             (optional; default `BEGIN`/`COMMIT`/`ROLLBACK`)
  //   batch(client, statements):  async → void                        (optional; atomic multi-statement path for batch-only adapters, e.g. D1 db.batch)
  //   supportsTransactions:  boolean                                  (interactive-tx capability; set false on a stateless/autocommit-per-statement adapter so transaction()/outbox REFUSE rather than silently run non-atomic — default assumes stateful)
  //   dialect:              "postgres" | "mysql" | "sqlite" | "mongodb" | "other"  (default "postgres")
  //   requireTls:           boolean                                  (opt-in TLS posture gate; default off — see below)
  //   tls / ssl / sslmode:  transport-TLS declaration consulted by requireTls (tls:true | ssl: | sslmode:"require"|"verify-ca"|"verify-full")
  //   applicationName:      string ≤ 63 bytes, no CR/LF/NUL          (Postgres pg_stat_activity tag; default null)
  //   pool:                 { min, max, idleTimeoutMs }              (defaults: 1 / 10 / C.TIME.minutes(1))
  //   classifications:      string[]                                 (defaults to ["*"])
  //   residencyTag:         "EU" | "US" | "unrestricted" | ...       (defaults to "unrestricted")
  //   retry, breaker:       passthrough to b.retry / CircuitBreaker
  //   replicas:             [{ connect, query, weight?, residencyTag?, allowCrossBorder? }]
  //   replicaFallbackToPrimary: boolean                              (default true)
}

Register one or more app-data backends. Each backend declares its connect / query driver hooks plus optional pooling, classification, residency, retry, and replica configuration. Throws synchronously on malformed input (missing hooks, unknown dialect, residency mismatch against db.getDataResidency(), dotted GUC names that fail identifier validation).

Boot-time residency check: when db.getDataResidency().region is set, any backend serving personal (or *) data must carry a residencyTag in the allowed-region list — refused with RESIDENCY_VIOLATION when not.

Opt-in transport posture: set requireTls: true on a backend to refuse it at config time (TLS_REQUIRED) unless its declared transport is encrypted (tls: true, an ssl object, or sslmode: "require" | "verify-ca" | "verify-full"). sslmode values that permit a plaintext fallback (prefer / allow / disable) are refused. The gate is OFF by default — a backend that omits requireTls is used exactly as supplied, with no transport check. Mandated for cardholder data by PCI-DSS v4.0 Req 4 and for ePHI by HIPAA §164.312(e).

var pg = require("pg");
var pool = new pg.Pool({ connectionString: "postgres://app:pw@db.example.com/app" });

b.externalDb.init({
  backends: {
    main: {
      dialect:         "postgres",
      applicationName: "blamejs-app",
      connect:         function () { return pool.connect(); },
      query:           function (client, sql, params) { return client.query(sql, params); },
      close:           function (client) { return client.release(); },
      classifications: ["personal", "operational"],
      residencyTag:    "EU",
      pool:            { min: 2, max: 20, idleTimeoutMs: 60000 },
    },
  },
  defaultBackend: "main",
});

b.externalDb.supportsTransactions(opts?) #

{
  backend?:        string,   // explicit backend name
  classification?: string,   // route by data class
}

Report whether the picked (or default) backend can provide an interactive transaction. Returns false only when the backend declares supportsTransactions: false at init() — a stateless / autocommit-per-statement adapter on which transaction() would run BEGIN / the body / COMMIT on different sessions (no isolation, no rollback). Consumers built on the dual-write guarantee (b.outbox) call this at construction so a non-atomic backend is refused up front rather than at the first transaction.

Same backend-selection opts as b.externalDb.query (backend / classification).

if (!b.externalDb.supportsTransactions()) {
  throw new Error("this backend cannot run atomic transactions");
}

b.externalDb.query(sql, params, opts) #

{
  backend?:           string,   // explicit backend name; bypasses classification + role pick
  classification?:    string,   // route to first backend whose classifications include this value
  includeSqlInAudit?: boolean,  // emit SQL text in audit metadata (off by default — may carry literal PII)
  rowResidencyTag?:   string,   // the row's residency region tag; required for a write (DML, CALL/EXECUTE/DO, COPY ... FROM, REPLACE, or a WITH/EXPLAIN-ANALYZE wrapping one) to a residency-tagged backend under a cross-border regulated posture (pass "global"/"unrestricted" for region-neutral rows)
}

Execute a single statement against the picked backend. Returns the driver-shaped { rows, rowCount } from the backend's query hook. Wraps the call in b.retry.withRetry for transient driver errors and the per-backend circuit breaker; emits system.externaldb.query audit events plus duration / slow-query metrics; surfaces Postgres SQLSTATE 28000 / 28P01 / 42501 as db.auth.failed audit rows for SOC2 forensic walks.

Backend selection precedence: opts.backend (explicit) → opts.classification (first backend serving the class) → ALS-bound dbRole + dbRoleBackends map (set by b.middleware.dbRoleFor or b.externalDb.runAs) → the configured defaultBackend.

var res = await b.externalDb.query(
  "SELECT id, email FROM users WHERE tenant_id = $1",
  ["acme"],
  { classification: "personal" }
);
res.rowCount;   // → 42
res.rows[0];    // → { id: 1, email: "ada@example.com" }

b.externalDb.transaction(fn, opts) #

{
  backend?:                    string,                       // explicit backend name
  classification?:             string,                       // route by data class
  sessionGucs?:                { [name]: string|number|boolean },  // SET LOCAL bindings (e.g. { "app.tenant_id": "acme" })
  statementTimeoutMs?:         number,                       // SET LOCAL statement_timeout
  idleInTransactionTimeoutMs?: number,                       // SET LOCAL idle_in_transaction_session_timeout
  deadlockRetries?:            number,                       // retries for 40P01 / 40001 (default 3)
  rowResidencyTag?:            string,                       // residency tag applied to every statement; a per-call tx.query(sql, params, { rowResidencyTag }) overrides it for that statement
}

Run fn(tx) inside a transaction on the picked backend. Wraps the body in BEGIN / COMMIT / ROLLBACK via the backend's hooks; commits on resolve, rolls back on throw. Transient deadlock / serialization failures (Postgres SQLSTATE 40P01 / 40001) retry automatically with a small jittered backoff (default 3 attempts; tune via opts.deadlockRetries).

tx.query(sql, params) runs against the same client used by BEGIN, so RLS state set by sessionGucs (SET LOCAL) applies for the duration of the transaction and resets at COMMIT/ROLLBACK.

Refuses (NON_ATOMIC_BACKEND) when the picked backend declares supportsTransactions: false — a stateless / autocommit-per-statement adapter on which BEGIN / the body / COMMIT would run on different sessions (no isolation, no rollback). Supply interactive beginTx/commit/rollback hooks or a batch adapter on the backend instead of shipping a silently non-atomic block.

var summary = await b.externalDb.transaction(async function (tx) {
  await tx.query("INSERT INTO orders(id, total) VALUES ($1, $2)", ["o-1", 4200]);
  await tx.query("UPDATE inventory SET qty = qty - 1 WHERE sku = $1", ["sku-7"]);
  var res = await tx.query("SELECT count(*) AS n FROM orders WHERE id = $1", ["o-1"]);
  return res.rows[0];
}, {
  classification: "operational",
  sessionGucs:    { "app.tenant_id": "acme" },
  statementTimeoutMs: 5000,
});
summary.n;   // → 1

b.externalDb.healthCheck(backendName) #

Ping a backend by acquiring a client and running its ping hook (or SELECT 1 when none is supplied). Returns { ok, breakerState, pool } for a single backend, or a { [name]: result } map when called with no argument. Connection-shape errors destroy the client; the breaker state is reflected in the returned record so health endpoints can surface circuit-open conditions.

var all = await b.externalDb.healthCheck();
all.main.ok;             // → true
all.main.breakerState;   // → "closed"
all.main.pool;           // → { idle: 1, active: 0, waiters: 0 }

var one = await b.externalDb.healthCheck("main");
one.ok;                  // → true

b.externalDb.listBackends() #

Snapshot every registered backend's name, dialect, classifications, residency tag, breaker state, and live pool stats. Returns [] when init() has not run. Cheap — does not open any new connections.

var rows = b.externalDb.listBackends();
rows[0].name;             // → "main"
rows[0].dialect;          // → "postgres"
rows[0].classifications;  // → ["personal", "operational"]
rows[0].residencyTag;     // → "EU"
rows[0].breakerState;     // → "closed"
rows[0].pool;             // → { idle: 2, active: 0, waiters: 0 }

b.externalDb.shutdown() #

Drain every backend pool (and replica pool), close idle clients, then clear all registry state so a subsequent init() starts from scratch. Idempotent — calling before init() is a no-op. Wire to b.appShutdown so process exit waits for in-flight queries to release their clients.

process.on("SIGTERM", async function () {
  await b.externalDb.shutdown();
  process.exit(0);
});

b.externalDb.read.query(sql, params, opts) #

{
  backend?:        string,   // explicit backend name
  classification?: string,   // route by data class
}

Route a read against the backend's declared replicas using weighted round-robin. A failed replica is sidelined for 30 seconds and the call falls back to primary when replicaFallbackToPrimary is true (the default). Backends without replicas transparently route to primary. Same opts selection rules as b.externalDb.query (backend / classification / ALS-bound role).

var res = await b.externalDb.read.query(
  "SELECT id, total FROM orders WHERE tenant_id = $1",
  ["acme"],
  { classification: "operational" }
);
res.rowCount;   // → 7
res.rows[0];    // → { id: "o-1", total: 4200 }

b.externalDb.write.query(sql, params, opts) #

{
  backend?:           string,   // explicit backend name
  classification?:    string,   // route by data class
  includeSqlInAudit?: boolean,  // emit SQL text in audit metadata
}

Symmetric alias for b.externalDb.query — always routes to primary. Pair with b.externalDb.read.query when an operator wants the call site to express read/write intent without a magic-comment hint. Same opts selection rules as b.externalDb.query.

var res = await b.externalDb.write.query(
  "INSERT INTO orders(id, tenant_id, total) VALUES ($1, $2, $3)",
  ["o-2", "acme", 1500],
  { classification: "operational" }
);
res.rowCount;   // → 1

b.externalDb.write.transaction(fn, opts) #

{
  backend?:                    string,
  classification?:             string,
  sessionGucs?:                { [name]: string|number|boolean },
  statementTimeoutMs?:         number,
  idleInTransactionTimeoutMs?: number,
  deadlockRetries?:            number,
}

Symmetric alias for b.externalDb.transaction — always runs against primary. Same opts shape (sessionGucs / statementTimeoutMs / idleInTransactionTimeoutMs / deadlockRetries) as the canonical form.

var n = await b.externalDb.write.transaction(async function (tx) {
  await tx.query("UPDATE counters SET n = n + 1 WHERE k = $1", ["hits"]);
  var res = await tx.query("SELECT n FROM counters WHERE k = $1", ["hits"]);
  return res.rows[0].n;
}, { sessionGucs: { "app.tenant_id": "acme" } });
typeof n;   // → "number"

b.externalDb.configurePool(backendName, opts) #

{
  min?:           number,   // positive integer; floor on idle clients
  max?:           number,   // positive integer; ceiling on total clients (must be >= min)
  idleTimeoutMs?: number,   // positive integer; reap idle clients after this many ms
}

Resize a registered backend's pool at runtime. New max takes effect on the next acquire; existing idle clients are kept; min is honored when the pool next refills; idleTimeoutMs applies on the next reaper tick. Throws on unknown options or non-positive integers so a config typo surfaces at the call site.

b.externalDb.configurePool("main", {
  min:           4,
  max:           50,
  idleTimeoutMs: 120000,
});

b.externalDb.adapters.connectAs(connect, opts) #

{
  query:               function,    // required — the backend's query function (used to issue SET statements)
  role?:               string,      // SQL identifier; runs SET ROLE ""
  searchPath?:         string[],    // SQL identifiers; runs SET search_path TO "", "", ...
  applicationName?:    string,      // appears in pg_stat_activity
  statementTimeoutMs?: number,      // positive integer; SET statement_timeout TO 
  gucs?:               { [name]: string|number },   // raw GUC bindings; finite numbers required for numeric values
}

var pg = require("pg");
var pool = new pg.Pool({ connectionString: "postgres://app:pw@db.example.com/app" });
var rawConnect = function () { return pool.connect(); };
var rawQuery   = function (client, sql, params) { return client.query(sql, params); };

b.externalDb.init({
  backends: {
    analytics: {
      dialect: "postgres",
      connect: b.externalDb.adapters.connectAs(rawConnect, {
        query:               rawQuery,
        role:                "analytics_user",
        searchPath:          ["analytics", "public"],
        applicationName:     "blamejs:analytics",
        statementTimeoutMs:  30000,
        gucs:                { idle_in_transaction_session_timeout: "60s" },
      }),
      query: rawQuery,
    },
  },
});

await b.externalDb.runAs("analytics_user", async function () {
  var res = await b.externalDb.read.query(
    "SELECT count(*) AS n FROM events WHERE day = $1",
    ["2026-05-09"]
  );
  return res.rows[0].n;
});

await b.externalDb.runAs("analytics_user", async function () {
  b.externalDb.currentRole();   // → "analytics_user"
});
b.externalDb.currentRole();     // → null

{
  declaredRoles: string[],            // required; allowlist of expected role names
  backend?:      string,              // explicit backend name (defaults to defaultBackend)
  mode?:         "audit" | "throw",   // default "audit"
  ignoreSystem?: boolean,             // skip postgres / pg_* / rds_* / azure_* / cloudsqlsuperuser (default true)
}

var report = await b.externalDb.assertRoleHardening({
  backend:       "main",
  declaredRoles: ["app_user", "analytics_user", "admin"],
  mode:          "audit",
  ignoreSystem:  true,
});
report.unrecognized;   // → []
report.missing;        // → []
report.observed;       // → ["admin", "analytics_user", "app_user"]