konnosDashboard
snapshots.ts610 lines · main
1import { brivenError, ValidationError } from '@briven/shared';
2import pg from 'pg';
3
4import { dbNameFor } from '../db/data-plane.js';
5import { env } from '../env.js';
6import { log } from '../lib/logger.js';
7
8/**
9 * Snapshots — the non-coder "undo button", now built on DoltGres-native
10 * version control instead of the old schema-clone hack.
11 *
12 * Each project is its own DoltGres DATABASE (`proj_<id>`) with an independent
13 * commit history. A snapshot is a **Dolt tag** — a stable, named pointer at a
14 * commit — created over a `DOLT_COMMIT` of the project's current state:
15 *
16 * - create → ensure a commit of the working set, then `DOLT_TAG` it.
17 * - list → read `dolt_tags` (one row per snapshot).
18 * - restore → `DOLT_RESET('--hard', tag)` rolls the working data back to the
19 * snapshot. Tags survive resets, so restoring to a *different*
20 * snapshot afterwards ("undo of the undo") always works.
21 * - diff → `DOLT_DIFF_SUMMARY` / `DOLT_DIFF` between the snapshot and the
22 * current (WORKING) data.
23 * - delete → `DOLT_TAG('-d', tag)`.
24 *
25 * Why tags and not a registry table: a `_briven_snapshots` table would itself
26 * be versioned data inside the project's branch, so `DOLT_RESET('--hard', …)`
27 * would roll the registry back too and lose snapshot rows. `dolt_tags` is Dolt
28 * system metadata — it is NOT affected by a hard reset (verified against the
29 * live DoltGres build) — so all snapshot bookkeeping lives there. Per-snapshot
30 * metadata (the human label, the auto/manual flag, the table count) is stored
31 * as a small JSON blob in the tag's message.
32 *
33 * Connection model: DOLT_* procedures are run in autocommit on a dedicated
34 * per-project `pg` pool (NOT `runInProjectDatabase`, which wraps everything in
35 * one BEGIN/COMMIT). The reason is `DOLT_COMMIT` raises "nothing to commit"
36 * when the working set is clean; in Postgres a raised statement aborts the
37 * whole surrounding transaction, so we must pre-check `dolt_status` and run
38 * each procedure as its own implicitly-committed statement. The `pg` driver is
39 * required here (postgres.js desyncs against DoltGres — see ADR 0001).
40 */
41
42export const SNAP_ID_RE = /^s[0-9a-f]{24}$/;
43
44function newSnapId(): string {
45 const buf = new Uint8Array(12);
46 crypto.getRandomValues(buf);
47 return 's' + Array.from(buf, (b) => b.toString(16).padStart(2, '0')).join('');
48}
49
50/** Per-snapshot metadata we encode into the Dolt tag message as JSON. */
51interface TagMeta {
52 /** Human label. */
53 readonly l: string;
54 /** Auto (worker-created) vs manual. */
55 readonly a: boolean;
56 /** User-table count captured at create time. */
57 readonly t: number;
58}
59
60export interface SnapshotSummary {
61 readonly id: string;
62 readonly name: string;
63 readonly tableCount: number;
64 readonly createdAt: string;
65 /**
66 * True for snapshots taken automatically by the scheduled auto-snapshot
67 * worker; false for ones a person saved by hand. Only `auto` snapshots
68 * are ever pruned by retention — manual ones are kept until deleted.
69 */
70 readonly auto: boolean;
71 /** The Dolt commit hash this snapshot's tag points at. */
72 readonly commitHash: string;
73}
74
75// ---------------------------------------------------------------------------
76// Per-project DoltGres connection (autocommit) — see file header for why this
77// is a dedicated pool rather than runInProjectDatabase.
78// ---------------------------------------------------------------------------
79
80const _pools = new Map<string, pg.Pool>();
81
82function poolForDb(dbName: string): pg.Pool {
83 const url = env.BRIVEN_DATA_PLANE_URL;
84 if (!url) {
85 throw new Error('BRIVEN_DATA_PLANE_URL is not configured');
86 }
87 let pool = _pools.get(dbName);
88 if (!pool) {
89 const base = new URL(url);
90 pool = new pg.Pool({
91 host: base.hostname,
92 port: Number(base.port || 5432),
93 user: decodeURIComponent(base.username),
94 password: decodeURIComponent(base.password),
95 database: dbName,
96 max: 5,
97 idleTimeoutMillis: 30000,
98 connectionTimeoutMillis: 5000,
99 });
100 _pools.set(dbName, pool);
101 log.info('snapshot_db_pool_opened', { dbName });
102 }
103 return pool;
104}
105
106/** Run one autocommit statement against a project's DoltGres database. */
107async function q(
108 dbName: string,
109 text: string,
110 params?: readonly unknown[],
111): Promise<Record<string, unknown>[]> {
112 const pool = poolForDb(dbName);
113 const res = await pool.query(text, params ? [...params] : undefined);
114 return res.rows as Record<string, unknown>[];
115}
116
117/** Close every cached snapshot pool (call on shutdown if wired). */
118export async function closeSnapshotPools(): Promise<void> {
119 const pools = Array.from(_pools.values());
120 _pools.clear();
121 await Promise.all(pools.map((p) => p.end()));
122}
123
124/** The first column value of the first row (DOLT_* funcs return one scalar). */
125function scalar(rows: Record<string, unknown>[]): string {
126 const row = rows[0];
127 if (!row) return '';
128 const v = Object.values(row)[0];
129 return v == null ? '' : String(v);
130}
131
132/** Some DOLT_* funcs wrap their result like `{hash}`; strip the braces. */
133function unwrap(v: string): string {
134 return v.startsWith('{') && v.endsWith('}') ? v.slice(1, -1) : v;
135}
136
137/** Count the project's user tables (public schema, excluding `_briven_*`). */
138async function countUserTables(dbName: string): Promise<number> {
139 const rows = await q(
140 dbName,
141 `SELECT count(*)::int AS n
142 FROM information_schema.tables
143 WHERE table_schema = 'public'
144 AND table_type = 'BASE TABLE'
145 AND left(table_name, 8) <> '_briven_'`,
146 );
147 return Number(rows[0]?.n) || 0;
148}
149
150/** True when the working set has uncommitted changes (`dolt_status` rows). */
151async function isDirty(dbName: string): Promise<boolean> {
152 const rows = await q(dbName, `SELECT count(*)::int AS n FROM dolt_status`);
153 return (Number(rows[0]?.n) || 0) > 0;
154}
155
156function parseMeta(message: string, fallbackLabel: string): TagMeta {
157 try {
158 const m = JSON.parse(message) as Partial<TagMeta>;
159 if (m && typeof m === 'object') {
160 return {
161 l: typeof m.l === 'string' ? m.l : fallbackLabel,
162 a: m.a === true,
163 t: Number(m.t) || 0,
164 };
165 }
166 } catch {
167 // Not our JSON (e.g. a hand-made tag) — fall through to defaults.
168 }
169 return { l: fallbackLabel, a: false, t: 0 };
170}
171
172function toIso(v: unknown): string {
173 if (v instanceof Date) return v.toISOString();
174 // DoltGres returns tag dates as 'YYYY-MM-DD HH:MM:SS.mmm' (UTC).
175 const s = String(v ?? '');
176 const d = new Date(s.includes('T') ? s : s.replace(' ', 'T') + 'Z');
177 return Number.isNaN(d.getTime()) ? s : d.toISOString();
178}
179
180/**
181 * Take a snapshot of the project's current data: commit the working set (if
182 * dirty), then create a Dolt tag pointing at HEAD. The tag name is the
183 * snapshot id (`s` + 24 hex, matching SNAP_ID_RE); the label/auto/tableCount
184 * are stored in the tag's message.
185 */
186export async function createSnapshot(
187 projectId: string,
188 name: string,
189 options: { auto?: boolean } = {},
190): Promise<SnapshotSummary> {
191 const auto = options.auto ?? false;
192 const cleanName = (name ?? '').trim().slice(0, 80) || 'snapshot';
193 const dbName = dbNameFor(projectId);
194
195 const tableCount = await countUserTables(dbName);
196
197 // Only commit when there is something to commit — DOLT_COMMIT raises
198 // "nothing to commit" on a clean working set, which would otherwise surface
199 // as an error to the caller.
200 if (await isDirty(dbName)) {
201 await q(dbName, `SELECT DOLT_COMMIT('-A', '-m', $1)`, [`snapshot: ${cleanName}`]);
202 }
203
204 const commitHash = unwrap(scalar(await q(dbName, `SELECT DOLT_HASHOF('HEAD')`)));
205
206 const snapId = newSnapId();
207 const meta: TagMeta = { l: cleanName, a: auto, t: tableCount };
208 await q(dbName, `SELECT DOLT_TAG($1, '-m', $2, 'HEAD')`, [snapId, JSON.stringify(meta)]);
209
210 return {
211 id: snapId,
212 name: cleanName,
213 tableCount,
214 createdAt: new Date().toISOString(),
215 auto,
216 commitHash,
217 };
218}
219
220/** Read a project's snapshot tags (our `s…` ids only), newest first. */
221async function readSnapshotTags(dbName: string): Promise<SnapshotSummary[]> {
222 const rows = await q(
223 dbName,
224 `SELECT tag_name, tag_hash, message, date FROM dolt_tags ORDER BY date DESC`,
225 );
226 const out: SnapshotSummary[] = [];
227 for (const r of rows) {
228 const id = String(r.tag_name ?? '');
229 if (!SNAP_ID_RE.test(id)) continue; // ignore non-snapshot tags
230 const meta = parseMeta(String(r.message ?? ''), id);
231 out.push({
232 id,
233 name: meta.l,
234 tableCount: meta.t,
235 createdAt: toIso(r.date),
236 auto: meta.a,
237 commitHash: String(r.tag_hash ?? ''),
238 });
239 }
240 return out;
241}
242
243/** List a project's snapshots, newest first. */
244export async function listSnapshots(projectId: string): Promise<SnapshotSummary[]> {
245 return readSnapshotTags(dbNameFor(projectId));
246}
247
248/** A snapshot row tagged with the project (database) it belongs to. */
249export interface CrossProjectSnapshot extends SnapshotSummary {
250 /** The data-plane database (`proj_<id>`) the snapshot was found in. */
251 readonly schema: string;
252}
253
254/**
255 * Open an admin connection to the data plane's default database so we can
256 * enumerate `proj_*` databases. Reuses the per-db pool cache keyed by the
257 * URL's own database name.
258 */
259function adminDbName(): string {
260 const url = env.BRIVEN_DATA_PLANE_URL;
261 if (!url) throw new Error('BRIVEN_DATA_PLANE_URL is not configured');
262 const path = new URL(url).pathname.replace(/^\//, '');
263 return path || 'postgres';
264}
265
266/**
267 * Operator-facing read of recent snapshots ACROSS every project, newest
268 * first. Under database-per-project each project keeps its own `dolt_tags`,
269 * so this enumerates the `proj_*` databases (one catalog read) then reads
270 * each project's snapshot tags. Per-database failures are skipped so one bad
271 * project can't sink the whole list. When the data plane isn't configured
272 * (local dev), returns an empty list rather than throwing.
273 */
274export async function listRecentSnapshotsAcrossProjects(
275 limit = 200,
276): Promise<CrossProjectSnapshot[]> {
277 const cap = Math.min(Math.max(limit, 1), 1000);
278 if (!env.BRIVEN_DATA_PLANE_URL) return [];
279
280 let dbRows: Record<string, unknown>[];
281 try {
282 dbRows = await q(
283 adminDbName(),
284 `SELECT datname FROM pg_database WHERE left(datname, 5) = 'proj_' ORDER BY datname`,
285 );
286 } catch (err) {
287 log.warn('snapshot_cross_project_enumerate_failed', {
288 message: err instanceof Error ? err.message : String(err),
289 });
290 return [];
291 }
292
293 const all: CrossProjectSnapshot[] = [];
294 for (const row of dbRows) {
295 const dbName = String(row.datname ?? '');
296 if (!dbName) continue;
297 try {
298 const snaps = await readSnapshotTags(dbName);
299 for (const s of snaps) all.push({ ...s, schema: dbName });
300 } catch (err) {
301 // Skip projects whose DB can't be read (e.g. mid-teardown).
302 log.warn('snapshot_cross_project_read_failed', {
303 dbName,
304 message: err instanceof Error ? err.message : String(err),
305 });
306 }
307 }
308
309 all.sort((a, b) => (a.createdAt < b.createdAt ? 1 : a.createdAt > b.createdAt ? -1 : 0));
310 return all.slice(0, cap);
311}
312
313/**
314 * Prune automatic snapshots beyond a retention count, oldest first. Only
315 * snapshots flagged `auto` are ever considered — manual ones are never
316 * counted or deleted. Returns the ids of the snapshots that were pruned.
317 */
318export async function pruneAutoSnapshots(
319 projectId: string,
320 keepCount: number,
321): Promise<{ pruned: string[] }> {
322 const keep = Math.max(0, Math.floor(keepCount));
323 const dbName = dbNameFor(projectId);
324 const snaps = (await readSnapshotTags(dbName)).filter((s) => s.auto);
325 // readSnapshotTags is newest-first; keep the first `keep`, drop the rest.
326 const doomed = snaps.slice(keep).map((s) => s.id);
327 for (const id of doomed) {
328 await q(dbName, `SELECT DOLT_TAG('-d', $1)`, [id]);
329 }
330 return { pruned: doomed };
331}
332
333/**
334 * Restore the project's data to a snapshot via `DOLT_RESET('--hard', tag)`.
335 * Because every write to the project auto-commits, the next change after a
336 * restore commits forward from here; and since the snapshot tags themselves
337 * survive the reset, restoring to a different snapshot afterwards ("undo of
338 * the undo") always works. Returns the user-table count after the restore.
339 */
340export async function restoreSnapshot(
341 projectId: string,
342 snapId: string,
343): Promise<{ restored: number }> {
344 if (!SNAP_ID_RE.test(snapId)) {
345 throw new ValidationError('invalid snapshot id', { snapId });
346 }
347 const dbName = dbNameFor(projectId);
348
349 const exists = await q(dbName, `SELECT 1 AS ok FROM dolt_tags WHERE tag_name = $1`, [snapId]);
350 if (!exists[0]) {
351 throw new brivenError('not_found', `snapshot not found: ${snapId}`, { status: 404 });
352 }
353
354 await q(dbName, `SELECT DOLT_RESET('--hard', $1)`, [snapId]);
355 const restored = await countUserTables(dbName);
356 return { restored };
357}
358
359/**
360 * Maximum rows examined per table when diffing — retained for API shape and
361 * reported back so the UI can phrase "showing the first N rows". DoltGres
362 * computes the row delta server-side, so this is an examination ceiling, not a
363 * hard cut.
364 */
365const DIFF_ROW_CAP = 1000;
366
367/** A column present on one side of the diff but not the other. */
368export interface ColumnDiff {
369 readonly name: string;
370 readonly dataType: string;
371}
372
373export interface TableRowDiff {
374 /** Rows present live but not in the snapshot. */
375 readonly added: number;
376 /** Rows present in the snapshot but not live. */
377 readonly removed: number;
378 /** Rows present on both sides whose contents differ. */
379 readonly changed: number;
380 /** Row count on the live side. */
381 readonly liveRowCount: number;
382 /** Row count on the snapshot side. */
383 readonly snapshotRowCount: number;
384 /** True when either side exceeded DIFF_ROW_CAP and the diff is partial. */
385 readonly truncated: boolean;
386}
387
388/** Per-table diff entry: schema delta + (optional) row delta. */
389export interface TableDiff {
390 readonly name: string;
391 /** Columns present live but absent from the snapshot. */
392 readonly columnsAdded: readonly ColumnDiff[];
393 /** Columns present in the snapshot but absent live. */
394 readonly columnsRemoved: readonly ColumnDiff[];
395 /** Row-level delta. Null when it couldn't be computed. */
396 readonly rows: TableRowDiff | null;
397 /** True when row diffing was skipped (kept for API shape). */
398 readonly noPrimaryKey: boolean;
399}
400
401export interface SnapshotDiff {
402 readonly snapshotId: string;
403 readonly snapshotName: string;
404 readonly snapshotCreatedAt: string;
405 /** Tables that exist live now but were not in the snapshot. */
406 readonly tablesAdded: readonly string[];
407 /** Tables captured in the snapshot but since dropped live. */
408 readonly tablesRemoved: readonly string[];
409 /** Per-table diffs for tables present on BOTH sides but changed. */
410 readonly tables: readonly TableDiff[];
411 /** The per-table row examination cap (so the UI can phrase truncation). */
412 readonly rowCap: number;
413}
414
415/** Strip a leading `public.` from a Dolt table name. */
416function bareTable(name: string): string {
417 return name.startsWith('public.') ? name.slice('public.'.length) : name;
418}
419
420/** A function-arg string literal, single-quotes doubled. */
421function lit(s: string): string {
422 return `'${s.replace(/'/g, "''")}'`;
423}
424
425/**
426 * Best-effort parse of column name → type from a Dolt `CREATE TABLE` body.
427 * `AS OF` is unsupported on information_schema in this DoltGres build, so the
428 * snapshot-side schema is read from DOLT_SCHEMA_DIFF's create statements.
429 */
430function parseCreateColumns(stmt: string): Map<string, string> {
431 const cols = new Map<string, string>();
432 if (!stmt) return cols;
433 const open = stmt.indexOf('(');
434 const close = stmt.lastIndexOf(')');
435 if (open < 0 || close <= open) return cols;
436 const body = stmt.slice(open + 1, close);
437 const segs: string[] = [];
438 let depth = 0;
439 let cur = '';
440 for (const ch of body) {
441 if (ch === '(') depth++;
442 else if (ch === ')') depth--;
443 if (ch === ',' && depth === 0) {
444 segs.push(cur);
445 cur = '';
446 } else {
447 cur += ch;
448 }
449 }
450 if (cur.trim()) segs.push(cur);
451 const constraintKw = /^(primary|key|unique|constraint|foreign|index|check)\b/i;
452 for (const seg of segs) {
453 const t = seg.trim();
454 if (!t || constraintKw.test(t)) continue;
455 const m = t.match(/^["`]?([A-Za-z_][A-Za-z0-9_]*)["`]?\s+(.+)$/s);
456 const colName = m?.[1];
457 const colType = m?.[2];
458 if (colName && colType) cols.set(colName, colType.trim().replace(/\s+/g, ' '));
459 }
460 return cols;
461}
462
463/**
464 * Compare a snapshot against the project's CURRENT (WORKING) data and return a
465 * structured diff: tables added/removed, per-table columns added/removed, and
466 * per-table row delta (added/removed/changed). Strictly read-only. Built on
467 * `DOLT_DIFF_SUMMARY` (table/row presence) and `DOLT_DIFF` (per-row delta).
468 */
469export async function diffSnapshot(projectId: string, snapId: string): Promise<SnapshotDiff> {
470 if (!SNAP_ID_RE.test(snapId)) {
471 throw new ValidationError('invalid snapshot id', { snapId });
472 }
473 const dbName = dbNameFor(projectId);
474
475 const tag = (await q(dbName, `SELECT tag_hash, message, date FROM dolt_tags WHERE tag_name = $1`, [
476 snapId,
477 ])) as Array<{ tag_hash?: unknown; message?: unknown; date?: unknown }>;
478 if (!tag[0]) {
479 throw new brivenError('not_found', `snapshot not found: ${snapId}`, { status: 404 });
480 }
481 const meta = parseMeta(String(tag[0].message ?? ''), snapId);
482 const snapshotCreatedAt = toIso(tag[0].date);
483
484 // Table-level presence: from = snapshot tag, to = WORKING (live). The tag id
485 // is SNAP_ID_RE-validated, so interpolating it into the function call is
486 // safe.
487 const summary = await q(
488 dbName,
489 `SELECT from_table_name, to_table_name, diff_type, schema_change
490 FROM DOLT_DIFF_SUMMARY(${lit(snapId)}, 'WORKING')`,
491 );
492
493 const tablesAdded: string[] = [];
494 const tablesRemoved: string[] = [];
495 const changedTables: { name: string; schemaChange: boolean }[] = [];
496 for (const r of summary) {
497 const type = String(r.diff_type ?? '');
498 const to = bareTable(String(r.to_table_name ?? ''));
499 const from = bareTable(String(r.from_table_name ?? ''));
500 if (type === 'added') tablesAdded.push(to || from);
501 else if (type === 'removed') tablesRemoved.push(from || to);
502 else changedTables.push({ name: to || from, schemaChange: String(r.schema_change) === '1' });
503 }
504
505 const tables: TableDiff[] = [];
506 for (const { name, schemaChange } of changedTables) {
507 // Row delta, grouped by Dolt's diff_type.
508 let added = 0;
509 let removed = 0;
510 let changed = 0;
511 try {
512 const rows = await q(
513 dbName,
514 `SELECT diff_type, count(*)::int AS n
515 FROM DOLT_DIFF(${lit(snapId)}, 'WORKING', ${lit(name)})
516 GROUP BY diff_type`,
517 );
518 for (const r of rows) {
519 const n = Number(r.n) || 0;
520 const t = String(r.diff_type ?? '');
521 if (t === 'added') added = n;
522 else if (t === 'removed') removed = n;
523 else if (t === 'modified') changed = n;
524 }
525 } catch (err) {
526 log.warn('snapshot_diff_rows_failed', {
527 dbName,
528 table: name,
529 message: err instanceof Error ? err.message : String(err),
530 });
531 }
532
533 const liveRowCount = await safeCount(dbName, `SELECT count(*)::int AS n FROM "${name}"`);
534 const snapshotRowCount = await safeCount(
535 dbName,
536 `SELECT count(*)::int AS n FROM "${name}" AS OF ${lit(snapId)}`,
537 );
538
539 let columnsAdded: ColumnDiff[] = [];
540 let columnsRemoved: ColumnDiff[] = [];
541 if (schemaChange) {
542 try {
543 const sd = await q(
544 dbName,
545 `SELECT from_create_statement, to_create_statement
546 FROM DOLT_SCHEMA_DIFF(${lit(snapId)}, 'WORKING', ${lit(name)})`,
547 );
548 const fromCols = parseCreateColumns(String(sd[0]?.from_create_statement ?? ''));
549 const toCols = parseCreateColumns(String(sd[0]?.to_create_statement ?? ''));
550 columnsAdded = [...toCols]
551 .filter(([c]) => !fromCols.has(c))
552 .map(([name2, dataType]) => ({ name: name2, dataType }));
553 columnsRemoved = [...fromCols]
554 .filter(([c]) => !toCols.has(c))
555 .map(([name2, dataType]) => ({ name: name2, dataType }));
556 } catch (err) {
557 log.warn('snapshot_diff_schema_failed', {
558 dbName,
559 table: name,
560 message: err instanceof Error ? err.message : String(err),
561 });
562 }
563 }
564
565 tables.push({
566 name,
567 columnsAdded,
568 columnsRemoved,
569 rows: {
570 added,
571 removed,
572 changed,
573 liveRowCount,
574 snapshotRowCount,
575 truncated: liveRowCount > DIFF_ROW_CAP || snapshotRowCount > DIFF_ROW_CAP,
576 },
577 noPrimaryKey: false,
578 });
579 }
580
581 return {
582 snapshotId: snapId,
583 snapshotName: meta.l,
584 snapshotCreatedAt,
585 tablesAdded,
586 tablesRemoved,
587 tables,
588 rowCap: DIFF_ROW_CAP,
589 };
590}
591
592async function safeCount(dbName: string, sql: string): Promise<number> {
593 try {
594 const rows = await q(dbName, sql);
595 return Number(rows[0]?.n) || 0;
596 } catch {
597 return 0;
598 }
599}
600
601/** Delete a snapshot (drops its Dolt tag). Idempotent. */
602export async function deleteSnapshot(projectId: string, snapId: string): Promise<void> {
603 if (!SNAP_ID_RE.test(snapId)) {
604 throw new ValidationError('invalid snapshot id', { snapId });
605 }
606 const dbName = dbNameFor(projectId);
607 const exists = await q(dbName, `SELECT 1 AS ok FROM dolt_tags WHERE tag_name = $1`, [snapId]);
608 if (!exists[0]) return;
609 await q(dbName, `SELECT DOLT_TAG('-d', $1)`, [snapId]);
610}