konnosDashboard
db.ts126 lines · main
1import pg from 'pg';
2
3import { env } from './env.js';
4
5/**
6 * One cached `pg` (node-postgres) pool per project, each bound to that
7 * project's DoltGres database. DoltGres speaks the Postgres wire protocol
8 * but — like Postgres — cannot switch database mid-connection (there is no
9 * `USE`), so the database is selected at connect time and the pool is reused
10 * across invocations.
11 *
12 * @README-BRIVEN ADR 0001 — converged onto DoltGres. We use the `pg` driver
13 * (node-postgres), NOT postgres.js: hands-on testing proved postgres.js's
14 * extended-protocol pipelining desyncs with DoltGres (`unhandled message
15 * "&{}"` even on `SELECT 1`), while `pg` works perfectly including the full
16 * git-for-data loop. We parse the base data-plane URL once and build each
17 * per-project pool from explicit fields with the database overridden to
18 * `proj_<id>`.
19 */
20const _pools = new Map<string, pg.Pool>();
21let _base: URL | null = null;
22
23function baseUrl(): URL {
24 if (!env.BRIVEN_DATA_PLANE_URL) {
25 throw new Error('BRIVEN_DATA_PLANE_URL is not configured on the runtime');
26 }
27 if (!_base) _base = new URL(env.BRIVEN_DATA_PLANE_URL);
28 return _base;
29}
30
31function poolFor(projectId: string): pg.Pool {
32 const db = dbNameFor(projectId);
33 let pool = _pools.get(db);
34 if (!pool) {
35 const base = baseUrl();
36 pool = new pg.Pool({
37 host: base.hostname,
38 port: Number(base.port || 5432),
39 user: decodeURIComponent(base.username),
40 password: decodeURIComponent(base.password),
41 database: db,
42 max: 20,
43 idleTimeoutMillis: 30000,
44 connectionTimeoutMillis: 5000,
45 });
46 _pools.set(db, pool);
47 }
48 return pool;
49}
50
51/**
52 * Mirror of `apps/api/src/db/data-plane.ts:dbNameFor` — must stay in sync.
53 * Both sides derive the database name deterministically from the project id
54 * so the api never has to ship the database name in the invoke request
55 * payload.
56 *
57 * Each project gets a dedicated DoltGres database named `proj_<sanitized id>`.
58 */
59export function dbNameFor(projectId: string): string {
60 const safe = projectId.replace(/[^a-zA-Z0-9]/g, '_').toLowerCase();
61 return `proj_${safe}`;
62}
63
64/**
65 * The adapter handed to `fn` inside `withProjectTx`. Preserves the interface
66 * callers (query-builder.ts, runtime-bootstrap.ts) depend on: a single
67 * `unsafe(text, params?)` method that returns a `Promise` of the result rows
68 * directly (mirrors postgres.js's `tx.unsafe`, which is what this replaced).
69 */
70export interface ProjectTx {
71 unsafe(text: string, params?: readonly unknown[]): Promise<unknown[]>;
72}
73
74/**
75 * Open a project-scoped transaction against the project's DoltGres database.
76 * Reuses the cached per-project pool, checks out one client, and runs `fn`
77 * inside a BEGIN/COMMIT; it rolls back automatically on throw.
78 *
79 * Auto-commit-per-write: the first statement turns on
80 * `dolt_transaction_commit`, so the SQL COMMIT that closes this transaction
81 * ALSO creates a Dolt commit — advancing the version log that powers Undo and
82 * realtime change-detection. (Verified against a real DoltGres: a plain INSERT
83 * does NOT advance `DOLT_HASHOF('HEAD')` unless this is set or DOLT_COMMIT is
84 * called explicitly.)
85 *
86 * @README-BRIVEN ADR 0001 — uses `pg` (node-postgres), not postgres.js:
87 * - `sql.begin(fn)` → explicit `client.connect()` + BEGIN / COMMIT / ROLLBACK
88 * - first-statement `SET dolt_transaction_commit = 1` makes COMMIT a Dolt commit
89 * - `fn` receives a `{ unsafe }` adapter so callers are unchanged
90 */
91export async function withProjectTx<T>(
92 projectId: string,
93 fn: (tx: ProjectTx) => Promise<T>,
94): Promise<T> {
95 const pool = poolFor(projectId);
96 const client = await pool.connect();
97 const tx: ProjectTx = {
98 unsafe: (text, params) =>
99 client.query(text, params ? [...params] : undefined).then((r) => r.rows),
100 };
101 try {
102 await client.query('BEGIN');
103 await client.query('SET dolt_transaction_commit = 1');
104 const result = await fn(tx);
105 await client.query('COMMIT');
106 return result;
107 } catch (err) {
108 try {
109 await client.query('ROLLBACK');
110 } catch {
111 // A failed rollback must not mask the original error.
112 }
113 throw err;
114 } finally {
115 client.release();
116 }
117}
118
119/**
120 * End every cached project pool. Test/shutdown hook — after this the next
121 * `withProjectTx` lazily rebuilds the pool it needs.
122 */
123export async function closeAllProjectClients(): Promise<void> {
124 await Promise.all([..._pools.values()].map((p) => p.end()));
125 _pools.clear();
126}