1// forge lake: the forge's facts in a DuckLake, so their history outlives the2// files they come from. Events, builds, review lifecycles and CI3// attestations are append-only facts spread over files that get cut4// (events.jsonl), aren't backed up (metrics.jsonl) or live in each repo5// (review logs, refs/notes/ci). Each run reads what is new since the last6// one (lib/lake-sources.ts) and adds it in one transaction, together with7// where it got to, so a run that dies adds nothing and the next one starts8// from the same place. Then it writes the insights exports (lib/insights.ts).9//10// One writer: bin/forge runs this under $FORGE_STATE/lake.lock. DuckDB11// can't share a database file between processes, and DuckLake over SQLite12// takes writes from several but turns many away with "database is locked"13// under contention. Readers (forge lake query, forge metrics) attach it14// read-only and can run alongside.15//16// node lib/lake.ts ingest [--dry-run] add what is new, then export insights17// node lib/lake.ts query <sql> [--json|--csv] read-only, no file access18// node lib/lake.ts install install the DuckDB extensions (the Dockerfile runs it)19// node lib/lake.ts stage <dir> a consistent copy of the lake for a backup20// node lib/lake.ts forget <owner>/<repo> take a removed repo's rows out (forge repo rm)21import fs from 'node:fs';22import os from 'node:os';23import path from 'node:path';24import { createRequire } from 'node:module';25import { fileURLToPath } from 'node:url';26import { DuckDBInstance, StatementType, type DuckDBConnection } from '@duckdb/node-api';27import { Verifier } from './attest.ts';28import { isAncestor, tryRun } from './git.ts';29import {30 buildRows, ciRows, commitRows, eventRow, listRepos, notesTip, readNew, reviewEnd, reviewLogs, reviewRows,31 type BuildRepoRow, type BuildRow, type CiMetricRow, type CiRow, type CommitRow, type EventRow, type Offset, type Read, type ReviewEnd, type ReviewRow,32} from './lake-sources.ts';33import { exportInsights, publishInsights } from './insights.ts';3435const home = path.resolve(path.dirname(fileURLToPath(import.meta.url)), '..');3637// The extensions DuckDB 1.5.6 installs from its core repository, and the38// builds of them this code was tested with. DuckDB checks their signatures;39// `install` also checks these versions, so an image never ships different40// ones unnoticed.41export const EXTENSIONS: Record<string, string> = { ducklake: 'ac7595b0', sqlite_scanner: '5274128', httpfs: '4bc690d' };4243export interface LakeConfig {44 root: string;45 state: string;46 www: string;47 // The private root, where private repos' pages are (bin/forge build_private).48 priv: string;49 dir: string;50 catalog: string;51 // A directory, or an s3:// URL (FORGE_LAKE_DATA).52 data: string;53 local: boolean;54 exportDir: string;55 extensions: string;56}5758export function lakeConfig(env: NodeJS.ProcessEnv = process.env): LakeConfig {59 const root = path.resolve(env.FORGE_ROOT || 'demo/root');60 const state = path.join(root, '.forge');61 const dir = path.join(state, 'lake');62 const data = env.FORGE_LAKE_DATA || path.join(dir, 'data');63 const local = !/^[a-z0-9]+:\/\//i.test(data);64 // Extensions: the image's (installed at build time, so the container never65 // downloads one), else a cache outside FORGE_ROOT's backups.66 const bundled = path.join(home, '.duckdb', 'extensions');67 const extensions = env.FORGE_DUCKDB_EXTENSIONS ||68 (fs.existsSync(bundled) ? bundled : path.join(env.XDG_CACHE_HOME || path.join(os.homedir(), '.cache'), 'duckdb', 'extensions'));69 return {70 root, state, www: env.FORGE_WWW || path.join(state, 'www'), priv: env.FORGE_PRIVATE_WWW || path.join(state, 'private', 'www'), dir, catalog: path.join(dir, 'catalog.sqlite'),71 data: local ? path.resolve(data) + '/' : data.replace(/\/?$/, '/'), local, exportDir: path.join(dir, 'export'), extensions,72 };73}7475export const lit = (s: string): string => `'${s.replaceAll("'", "''")}'`;76export const ident = (s: string): string => `"${s.replaceAll('"', '""')}"`;7778// Tables and their columns. Times are UTC.79const TABLES: Record<string, [string, string][]> = {80 // events.jsonl: pushes, builds, comments, approvals, as appended.81 events: [['t', 'TIMESTAMP'], ['repo', 'VARCHAR'], ['type', 'VARCHAR'], ['user', 'VARCHAR'], ['review', 'INTEGER'], ['refs', 'VARCHAR'], ['ms', 'DOUBLE'], ['rendered', 'INTEGER'], ['private', 'BOOLEAN']],82 // metrics.jsonl: one row a build (scope is the repo, "-" or "all"), the83 // line itself for forge metrics, and a row for each repo it loaded.84 builds: [['id', 'VARCHAR'], ['at', 'TIMESTAMP'], ['code', 'VARCHAR'], ['reason', 'VARCHAR'], ['scope', 'VARCHAR'], ['ms_total', 'DOUBLE'], ['ms_data', 'DOUBLE'], ['ms_render', 'DOUBLE'], ['rss_mb', 'INTEGER'], ['written', 'INTEGER'], ['linked', 'INTEGER'], ['line', 'VARCHAR']],85 build_repos: [['build', 'VARCHAR'], ['at', 'TIMESTAMP'], ['repo', 'VARCHAR'], ['ms_load', 'DOUBLE'], ['pages', 'INTEGER'], ['rendered', 'INTEGER'], ['skipped', 'INTEGER'], ['stages', 'VARCHAR']],86 // Review logs: created, rev, approve, deleted; and how each review ended.87 reviews: [['repo', 'VARCHAR'], ['review', 'INTEGER'], ['kind', 'VARCHAR'], ['ts', 'TIMESTAMP'], ['user', 'VARCHAR'], ['rev', 'INTEGER'], ['sha', 'VARCHAR'], ['base', 'VARCHAR'], ['tree', 'VARCHAR'], ['branch', 'VARCHAR'], ['title', 'VARCHAR']],88 review_ends: [['repo', 'VARCHAR'], ['review', 'INTEGER'], ['kind', 'VARCHAR'], ['ts', 'TIMESTAMP'], ['user', 'VARCHAR'], ['sha', 'VARCHAR'], ['observed', 'BOOLEAN']],89 // Commits on each repo's default branch, as it moved: its author's time90 // (the day it counts for on activity grids), subject and tree (which CI91 // results and metrics are keyed by; commits added before it was kept92 // have none).93 commits: [['repo', 'VARCHAR'], ['sha', 'VARCHAR'], ['t', 'TIMESTAMP'], ['author', 'VARCHAR'], ['subject', 'VARCHAR'], ['tree', 'VARCHAR']],94 // refs/notes/ci: one row an attestation.95 ci: [['id', 'VARCHAR'], ['repo', 'VARCHAR'], ['commit', 'VARCHAR'], ['tree', 'VARCHAR'], ['result', 'VARCHAR'], ['exit', 'INTEGER'], ['started', 'TIMESTAMP'], ['duration_s', 'INTEGER'], ['runner', 'VARCHAR'], ['system', 'VARCHAR'], ['env', 'VARCHAR'], ['signer', 'VARCHAR'], ['verified', 'BOOLEAN']],96 // The metrics in each attestation (.forge/ci writes them to97 // $FORGE_METRICS): one row a metric, id the attestation's in ci.98 ci_metrics: [['id', 'VARCHAR'], ['repo', 'VARCHAR'], ['commit', 'VARCHAR'], ['tree', 'VARCHAR'], ['name', 'VARCHAR'], ['value', 'DOUBLE'], ['unit', 'VARCHAR']],99 // Where each source was read up to (lib/lake-sources.ts), written in the100 // same transaction as its rows.101 sources: [['source', 'VARCHAR'], ['ino', 'BIGINT'], ['pos', 'BIGINT'], ['head', 'VARCHAR'], ['tail', 'VARCHAR'], ['at', 'TIMESTAMP']],102};103104export class Lake {105 cfg: LakeConfig;106 conn: DuckDBConnection;107 private db: DuckDBInstance;108109 private constructor(cfg: LakeConfig, db: DuckDBInstance, conn: DuckDBConnection) {110 this.cfg = cfg;111 this.db = db;112 this.conn = conn;113 }114115 // Rows as arrays of JSON values (BIGINTs as strings).116 async rows(sql: string): Promise<unknown[][]> {117 return (await this.conn.runAndReadAll(sql)).getRowsJson() as unknown[][];118 }119120 async run(sql: string): Promise<void> {121 await this.conn.run(sql);122 }123124 close(): void {125 this.conn.closeSync();126 this.db.closeSync();127 }128129 // Attach the lake: to write (creating it the first time), or read-only.130 static async open(cfg: LakeConfig, { write }: { write: boolean }): Promise<Lake> {131 if (!write && !fs.existsSync(cfg.catalog)) throw new LakeError(`no lake yet (${cfg.catalog}); forge lake makes it`);132 const db = await DuckDBInstance.create(':memory:', {133 extension_directory: cfg.extensions, autoinstall_known_extensions: 'false', autoload_known_extensions: 'false',134 });135 const conn = await db.connect();136 const lake = new Lake(cfg, db, conn);137 try {138 await lake.load(cfg.local ? ['ducklake', 'sqlite'] : ['ducklake', 'sqlite', 'httpfs']);139 if (!cfg.local) await lake.secret();140 const fresh = !fs.existsSync(cfg.catalog);141 if (write) {142 fs.mkdirSync(cfg.dir, { recursive: true });143 if (cfg.local) fs.mkdirSync(cfg.data, { recursive: true });144 }145 // The data path is stored in the catalog; this host's wins, so a lake146 // restored under another root still finds its files.147 await lake.run(`ATTACH ${lit('ducklake:sqlite:' + cfg.catalog)} AS lake (DATA_PATH ${lit(cfg.data)}, OVERRIDE_DATA_PATH true${write ? '' : ', READ_ONLY'})`);148 if (write) {149 await lake.run('BEGIN TRANSACTION');150 if (fresh) {151 // Snapshots older than 40 days are dropped at a checkpoint, and the152 // files only they used are deleted a week later, so a backup of the153 // catalog from the last five weeks still finds its files.154 for (const [k, v] of [['expire_older_than', '40 days'], ['delete_older_than', '7 days'], ['parquet_compression', 'zstd']]) {155 await lake.run(`CALL lake.set_option(${lit(k)}, ${lit(v)})`);156 }157 }158 for (const [t, cols] of Object.entries(TABLES)) {159 await lake.run(`CREATE TABLE IF NOT EXISTS lake.${t} (${cols.map(([c, ty]) => `${ident(c)} ${ty}`).join(', ')})`);160 }161 // Columns added since a lake was made: at the end of their table,162 // empty in the rows it already has.163 const have = new Set((await lake.rows(`SELECT table_name || ' ' || column_name FROM information_schema.columns WHERE table_catalog = 'lake'`)).map((r) => String(r[0])));164 for (const [t, cols] of Object.entries(TABLES)) {165 for (const [c, ty] of cols) {166 if (!have.has(`${t} ${c}`)) await lake.run(`ALTER TABLE lake.${t} ADD COLUMN ${ident(c)} ${ty}`);167 }168 }169 await lake.run('COMMIT');170 }171 } catch (e) {172 lake.close();173 throw e;174 }175 return lake;176 }177178 // Load extensions, installing missing ones (only outside the image, which179 // has them; DuckDB downloads them from its repository).180 private async load(names: string[]): Promise<void> {181 const have = new Set((await this.rows(`SELECT extension_name FROM duckdb_extensions() WHERE installed`)).map((r) => String(r[0])));182 for (const n of names) {183 const key = n === 'sqlite' ? 'sqlite_scanner' : n;184 if (!have.has(key)) {185 try {186 await this.run(`INSTALL ${n}`);187 } catch (e) {188 throw new LakeError(`the DuckDB extension ${n} isn't installed in ${this.cfg.extensions}, and installing it failed: ${firstLine(e)}`);189 }190 }191 await this.run(`LOAD ${n}`);192 }193 }194195 // Credentials for an s3:// data path (bin/forge's lake_env passes them).196 private async secret(): Promise<void> {197 const env = process.env;198 const key = env.FORGE_LAKE_KEY_ID || '';199 const secret = env.FORGE_LAKE_SECRET || '';200 const endpoint = (env.FORGE_LAKE_ENDPOINT || '').replace(/^https?:\/\//, '').replace(/\/$/, '');201 const parts = [`TYPE s3`, `KEY_ID ${lit(key)}`, `SECRET ${lit(secret)}`, `REGION ${lit(env.FORGE_LAKE_REGION || 'auto')}`];202 if (endpoint) parts.push(`ENDPOINT ${lit(endpoint)}`, `URL_STYLE 'path'`);203 // A plain-HTTP endpoint (a local S3 server, say).204 if (/^http:\/\//.test(env.FORGE_LAKE_ENDPOINT || '')) parts.push('USE_SSL false');205 await this.run(`CREATE SECRET lake_data (${parts.join(', ')})`);206 }207208 // NDJSON rows into a table, through a scratch file DuckDB reads typed.209 async insert(table: string, rows: object[], scratch: string): Promise<void> {210 if (!rows.length) return;211 const cols = TABLES[table];212 const file = path.join(scratch, `${table}.jsonl`);213 fs.writeFileSync(file, rows.map((r) => JSON.stringify(r)).join('\n') + '\n');214 const spec = cols.map(([c, ty]) => `${lit(c)}: ${lit(ty)}`).join(', ');215 await this.run(`INSERT INTO lake.${table} SELECT ${cols.map(([c]) => ident(c)).join(', ')} FROM read_json(${lit(file)}, format = 'newline_delimited', columns = {${spec}})`);216 fs.rmSync(file, { force: true });217 }218}219220export class LakeError extends Error {}221222const firstLine = (e: unknown): string => String((e as Error)?.message ?? e).split('\n')[0];223224// ---------------------------------------------------------------- ingest225226export interface Summary {227 events: number;228 builds: number;229 reviews: number;230 ended: number;231 ci: number;232 commits: number;233 bad: number;234 notes: string[];235 // Repos whose facts changed, for the exports.236 touched: Set<string>;237}238239interface SourceState extends Offset {240 source: string;241}242243const offsetOf = (s: SourceState | undefined): Offset | null => (s ? { ino: s.ino, pos: s.pos, head: s.head, tail: s.tail } : null);244245// A removed repo (forge repo rm): its rows in every table, and where the246// lake read its sources up to, so a new repo under the same name starts247// with none of them, and its exports with none of the removed one's248// history. events.jsonl and metrics.jsonl were read past its lines249// already, so they don't come back. Returns how many rows went.250export async function forget(lake: Lake, slug: string): Promise<number> {251 const s = lit(slug);252 const where: Record<string, string> = {253 events: `repo = ${s}`, builds: `scope = ${s}`, build_repos: `repo = ${s}`, reviews: `repo = ${s}`,254 review_ends: `repo = ${s}`, commits: `repo = ${s}`, ci: `repo = ${s}`, ci_metrics: `repo = ${s}`,255 sources: `source IN (${['main', 'commits', 'notes'].map((k) => lit(`${k} ${slug}`)).join(', ')}) OR starts_with(source, ${lit(`review ${slug} `)})`,256 };257 let n = 0;258 await lake.run('BEGIN TRANSACTION');259 try {260 for (const [t, w] of Object.entries(where)) {261 if (t !== 'sources') n += Number((await lake.rows(`SELECT count(*) FROM lake.${t} WHERE ${w}`))[0]?.[0] ?? 0);262 await lake.run(`DELETE FROM lake.${t} WHERE ${w}`);263 }264 await lake.run('COMMIT');265 } catch (e) {266 await lake.run('ROLLBACK').catch(() => {});267 throw e;268 }269 return n;270}271272// Add what is new in every source. With dry, the same reads and inserts run273// and are rolled back, so it says what a run would add.274export async function ingest(lake: Lake, { dry = false, now = Date.now() }: { dry?: boolean; now?: number } = {}): Promise<Summary> {275 const cfg = lake.cfg;276 const sources = new Map<string, SourceState>();277 for (const [source, ino, pos, head, tail] of await lake.rows(`SELECT source, ino, pos, head, tail FROM lake.sources`)) {278 sources.set(String(source), { source: String(source), ino: Number(ino), pos: Number(pos), head: String(head ?? ''), tail: JSON.parse(String(tail ?? '[]')) });279 }280 const sum: Summary = { events: 0, builds: 0, reviews: 0, ended: 0, ci: 0, commits: 0, bad: 0, notes: [], touched: new Set() };281 const next = new Map<string, Offset>();282 const events: EventRow[] = [];283 const builds: BuildRow[] = [];284 const buildRepos: BuildRepoRow[] = [];285 const reviews: ReviewRow[] = [];286 const redo: [string, number][] = [];287 const ends: ReviewEnd[] = [];288 const ci: CiRow[] = [];289 const ciMetrics: CiMetricRow[] = [];290 const commits: CommitRow[] = [];291292 const jsonl = <T>(name: string, file: string, parse: (line: string) => T | null, take: (row: T) => void): void => {293 const r: Read | null = readNew(file, offsetOf(sources.get(name)));294 if (!r) return;295 if (r.how === 'found') sum.notes.push(`${path.basename(file)} was cut or replaced; picked up where the last run left off`);296 if (r.how === 'lost') sum.notes.push(`${path.basename(file)} was cut past where the last run left off; what was in between is gone`);297 for (const line of r.lines) {298 if (!line.trim()) continue;299 const row = parse(line);300 if (row) take(row);301 else sum.bad++;302 }303 const was = sources.get(name);304 if (!was || was.ino !== r.next.ino || was.pos !== r.next.pos || was.head !== r.next.head) next.set(name, r.next);305 };306 jsonl('events', path.join(cfg.state, 'events.jsonl'), eventRow, (e) => {307 events.push(e);308 if (e.repo) sum.touched.add(e.repo);309 });310 jsonl('metrics', path.join(cfg.state, 'metrics.jsonl'), buildRows, (b) => {311 builds.push(b.build);312 buildRepos.push(...b.repos);313 sum.touched.add(b.build.scope);314 });315316 // Review logs: a log that changed is read again whole (they are small).317 // Whether a review ended is checked when its log or the default branch moved.318 const ended = new Set((await lake.rows(`SELECT repo, review FROM lake.review_ends`)).map(([r, n]) => `${r} ${n}`));319 let known: Map<string, Set<string>> | null = null;320 // FORGE_NOW pins the clock in the demo and tests, as for review logs.321 const nowS = Number(process.env.FORGE_NOW) || Math.floor(now / 1000);322 const verifier = new Verifier({ allowedSigners: path.join(cfg.state, 'allowed_signers'), cacheFile: null });323 for (const repo of listRepos(cfg.root)) {324 const main = `main ${repo.slug}`;325 const tip = tryRun(repo.dir, ['rev-parse', '-q', '--verify', 'HEAD^{commit}'])?.trim() ?? '';326 const moved = sources.get(main)?.head !== tip;327 if (moved) next.set(main, { ino: 0, pos: 0, head: tip, tail: [] });328 for (const { log, hash } of reviewLogs(repo.dir)) {329 const key = `review ${repo.slug} ${log.id}`;330 const changed = sources.get(key)?.head !== hash;331 if (changed) {332 const rows = reviewRows(repo.slug, log);333 reviews.push(...rows);334 redo.push([repo.slug, log.id]);335 next.set(key, { ino: 0, pos: 0, head: hash, tail: [] });336 sum.reviews += rows.length;337 sum.touched.add(repo.slug);338 }339 if ((changed || moved) && !ended.has(`${repo.slug} ${log.id}`)) {340 const end = reviewEnd(repo.slug, repo.dir, log, nowS);341 if (end) {342 ends.push(end);343 sum.touched.add(repo.slug);344 }345 }346 }347 // Commits: what the default branch gained since the last run (all of348 // it, less what the lake has, when it was rewritten or is new).349 const ckey = `commits ${repo.slug}`;350 const was = sources.get(ckey)?.head ?? '';351 if (tip && was !== tip) {352 const have = was && isAncestor(repo.dir, was, tip) ? null : new Set((await lake.rows(`SELECT sha FROM lake.commits WHERE repo = ${lit(repo.slug)}`)).map((r) => String(r[0])));353 const rows = commitRows(repo.slug, repo.dir, tip, have ? null : was).filter((c) => !have?.has(c.sha));354 commits.push(...rows);355 if (rows.length) sum.touched.add(repo.slug);356 next.set(ckey, { ino: 0, pos: 0, head: tip, tail: [] });357 }358 const notes = notesTip(repo.dir);359 const nkey = `notes ${repo.slug}`;360 if (notes && sources.get(nkey)?.head !== notes) {361 if (!known) {362 known = new Map();363 for (const [r, id] of await lake.rows(`SELECT repo, id FROM lake.ci`)) {364 if (!known.has(String(r))) known.set(String(r), new Set());365 known.get(String(r))!.add(String(id));366 }367 }368 const { rows, metrics } = ciRows(repo.slug, repo.dir, known.get(repo.slug) ?? new Set(), verifier);369 ci.push(...rows);370 ciMetrics.push(...metrics);371 if (rows.length) sum.touched.add(repo.slug);372 next.set(nkey, { ino: 0, pos: 0, head: notes, tail: [] });373 }374 }375 sum.events = events.length;376 sum.builds = builds.length;377 sum.ended = ends.length;378 sum.ci = ci.length;379 sum.commits = commits.length;380 if (!next.size) return sum;381382 const scratch = fs.mkdtempSync(path.join(cfg.dir, 'tmp.'));383 const at = new Date(now).toISOString().replace('T', ' ').replace('Z', '');384 try {385 await lake.run('BEGIN TRANSACTION');386 for (let i = 0; i < redo.length; i += 500) {387 const keys = redo.slice(i, i + 500).map(([r, n]) => `(${lit(r)}, ${n})`).join(', ');388 await lake.run(`DELETE FROM lake.reviews WHERE (repo, review) IN (${keys})`);389 }390 await lake.insert('events', events, scratch);391 await lake.insert('builds', builds, scratch);392 await lake.insert('build_repos', buildRepos, scratch);393 await lake.insert('reviews', reviews, scratch);394 await lake.insert('review_ends', ends, scratch);395 await lake.insert('ci', ci, scratch);396 await lake.insert('ci_metrics', ciMetrics, scratch);397 await lake.insert('commits', commits, scratch);398 const names = [...next.keys()];399 for (let i = 0; i < names.length; i += 500) {400 await lake.run(`DELETE FROM lake.sources WHERE source IN (${names.slice(i, i + 500).map(lit).join(', ')})`);401 }402 await lake.insert('sources', [...next].map(([source, o]) => ({ source, ino: o.ino, pos: o.pos, head: o.head, tail: JSON.stringify(o.tail), at })), scratch);403 await lake.run(dry ? 'ROLLBACK' : 'COMMIT');404 } catch (e) {405 await lake.run('ROLLBACK').catch(() => {});406 throw e;407 } finally {408 fs.rmSync(scratch, { recursive: true, force: true });409 }410 return sum;411}412413// Inlined rows (DuckLake keeps inserts under 10 rows in the catalog) go to414// Parquet, small files are merged and old snapshots dropped, about once a415// day (FORGE_LAKE_CHECKPOINT hours).416async function checkpoint(lake: Lake, now: number): Promise<boolean> {417 const hours = Number(process.env.FORGE_LAKE_CHECKPOINT ?? 24);418 const last = (await lake.rows(`SELECT pos FROM lake.sources WHERE source = 'checkpoint'`))[0]?.[0];419 if (last != null && now / 1000 - Number(last) < hours * 3600) return false;420 // A new lake starts the clock; its first rows are new files already.421 if (last != null) await lake.run('CHECKPOINT lake');422 await lake.run('BEGIN TRANSACTION');423 await lake.run(`DELETE FROM lake.sources WHERE source = 'checkpoint'`);424 await lake.run(`INSERT INTO lake.sources VALUES ('checkpoint', 0, ${Math.floor(now / 1000)}, '', '[]', now()::TIMESTAMP)`);425 await lake.run('COMMIT');426 return last != null;427}428429// ---------------------------------------------------------------- query430431// A table, as the DuckDB CLI's plain mode does it.432function table(cols: string[], rows: unknown[][]): string {433 const cell = (v: unknown): string => (v == null ? 'NULL' : typeof v === 'object' ? JSON.stringify(v) : String(v)).replace(/\s+/g, ' ');434 const text = [cols, ...rows.map((r) => r.map(cell))];435 const w = cols.map((_, i) => Math.min(60, Math.max(...text.map((r) => r[i].length))));436 const fmt = (r: string[]): string => r.map((c, i) => (c.length > w[i] ? c.slice(0, w[i] - 1) + '…' : c.padEnd(w[i]))).join(' ').trimEnd();437 return [fmt(cols), w.map((n) => '-'.repeat(n)).join(' '), ...text.slice(1).map(fmt)].join('\n') + `\n(${rows.length} ${rows.length === 1 ? 'row' : 'rows'})\n`;438}439440// Statements a query may run: reads.441const READS = new Set([StatementType.SELECT, StatementType.EXPLAIN, StatementType.PRAGMA]);442443// Read-only, three ways: the lake is attached READ_ONLY; only SELECT,444// EXPLAIN and PRAGMA statements run, so no COPY or EXPORT writes a file;445// and file access stops at the lake's data, with settings locked, so a446// query can't read other files, attach a database or undo either.447export async function query(cfg: LakeConfig, sql: string, format: 'table' | 'json' | 'csv' = 'table'): Promise<string> {448 const lake = await Lake.open(cfg, { write: false });449 try {450 await lake.run('USE lake');451 await lake.run(`SET allowed_directories = [${lit(cfg.data)}]`);452 await lake.run('SET enable_external_access = false');453 await lake.run('SET lock_configuration = true');454 const stmts = await lake.conn.extractStatements(sql);455 if (!stmts.count) throw new LakeError('no SQL to run');456 let res = null;457 for (let i = 0; i < stmts.count; i++) {458 const p = await stmts.prepare(i);459 if (!READS.has(p.statementType)) throw new LakeError(`forge lake query only reads: ${StatementType[p.statementType]} statements don't run`);460 res = await p.runAndReadAll();461 }462 if (!res) throw new LakeError('no SQL to run');463 const cols = res.columnNames();464 const rows = res.getRowsJson() as unknown[][];465 if (format === 'json') return rows.map((r) => JSON.stringify(Object.fromEntries(cols.map((c, i) => [c, r[i]])))).join('\n') + (rows.length ? '\n' : '');466 if (format === 'csv') {467 const q = (v: unknown): string => {468 const s = v == null ? '' : typeof v === 'object' ? JSON.stringify(v) : String(v);469 return /[",\n]/.test(s) ? `"${s.replaceAll('"', '""')}"` : s;470 };471 return [cols, ...rows].map((r) => r.map(q).join(',')).join('\n') + '\n';472 }473 return table(cols, rows);474 } finally {475 lake.close();476 }477}478479// Every metrics.jsonl line the lake has, oldest first, for forge metrics;480// null without a lake.481export async function buildLines(cfg: LakeConfig): Promise<{ id: string; line: string }[] | null> {482 if (!fs.existsSync(cfg.catalog)) return null;483 const lake = await Lake.open(cfg, { write: false });484 try {485 return (await lake.rows(`SELECT id, line FROM lake.builds ORDER BY "at", id`)).map(([id, line]) => ({ id: String(id), line: String(line) }));486 } finally {487 lake.close();488 }489}490491// ---------------------------------------------------------------- install, stage492493// Install the extensions into the extension directory and check their494// versions (the Dockerfile runs this, so the container never downloads one).495async function install(cfg: LakeConfig): Promise<void> {496 fs.mkdirSync(cfg.extensions, { recursive: true });497 const db = await DuckDBInstance.create(':memory:', { extension_directory: cfg.extensions, autoinstall_known_extensions: 'false', autoload_known_extensions: 'false' });498 const conn = await db.connect();499 for (const n of ['ducklake', 'sqlite', 'httpfs']) await conn.run(`INSTALL ${n}`);500 const got = (await conn.runAndReadAll(`SELECT extension_name, extension_version, install_path FROM duckdb_extensions() WHERE installed AND install_path NOT LIKE '(%'`)).getRowsJson() as string[][];501 let bad = 0;502 for (const [name, version, file] of got) {503 if (!(name in EXTENSIONS)) continue;504 const ok = EXTENSIONS[name] === version;505 if (!ok) bad++;506 console.log(`${name} ${version} ${(fs.statSync(file).size / (1 << 20)).toFixed(1)}M${ok ? '' : ` (expected ${EXTENSIONS[name]}; check it, then update EXTENSIONS in lib/lake.ts)`}`);507 }508 conn.closeSync();509 db.closeSync();510 if (bad) process.exit(1);511}512513// A copy of the lake for a backup, into <dir>/lake: the catalog with514// SQLite's VACUUM INTO, which reads one consistent state of it, never the515// live file, then checked; and a local data directory as hard links516// (Parquet files are written once and never changed). bin/forge holds517// lake.lock meanwhile, so no file the copy names goes away.518function stage(cfg: LakeConfig, dest: string): void {519 if (!fs.existsSync(cfg.catalog)) return;520 const out = path.join(dest, 'lake');521 const copy = path.join(out, 'catalog.sqlite');522 fs.mkdirSync(out, { recursive: true });523 fs.rmSync(copy, { force: true });524 const { DatabaseSync } = sqlite();525 const db = new DatabaseSync(cfg.catalog, { readOnly: true });526 try {527 db.exec(`VACUUM INTO ${lit(copy)}`);528 } finally {529 db.close();530 }531 const check = new DatabaseSync(copy, { readOnly: true });532 try {533 const r = check.prepare('PRAGMA integrity_check').get() as Record<string, unknown> | undefined;534 if (Object.values(r ?? {})[0] !== 'ok') throw new LakeError(`the copy of the catalog fails SQLite's integrity check: ${JSON.stringify(r)}`);535 } finally {536 check.close();537 }538 if (!cfg.local || !fs.existsSync(cfg.data)) return;539 const walk = (from: string, to: string): void => {540 fs.mkdirSync(to, { recursive: true });541 for (const e of fs.readdirSync(from, { withFileTypes: true })) {542 const a = path.join(from, e.name);543 const b = path.join(to, e.name);544 if (e.isDirectory()) walk(a, b);545 else if (e.isFile()) {546 try {547 fs.linkSync(a, b);548 } catch {549 fs.copyFileSync(a, b);550 }551 }552 }553 };554 walk(cfg.data, path.join(out, 'data'));555}556557// node:sqlite, without its experimental warning.558function sqlite(): typeof import('node:sqlite') {559 const emit = process.emitWarning;560 process.emitWarning = ((w: string | Error, ...rest: never[]) => (/SQLite/.test(String(w)) ? undefined : emit.call(process, w, ...rest))) as typeof emit;561 try {562 return createRequire(import.meta.url)('node:sqlite') as typeof import('node:sqlite');563 } finally {564 process.emitWarning = emit;565 }566}567568// ---------------------------------------------------------------- main569570const plural = (n: number, w: string): string => `${n} ${w}${n === 1 ? '' : 's'}`;571572async function main(argv: string[]): Promise<void> {573 const cfg = lakeConfig();574 const [cmd, ...rest] = argv;575 if (cmd === 'ingest') {576 const dry = rest.includes('--dry-run') || rest.includes('-n');577 const t0 = performance.now();578 const now = Date.now();579 // A dry run on a forge without a lake yet reads into a scratch one.580 const scratch = dry && !fs.existsSync(cfg.catalog) ? fs.mkdtempSync(path.join(os.tmpdir(), 'forge-lake.')) : null;581 const lake = await Lake.open(scratch ? { ...cfg, dir: scratch, catalog: path.join(scratch, 'catalog.sqlite'), data: scratch + '/data/', local: true } : cfg, { write: true });582 try {583 const s = await ingest(lake, { dry, now });584 const what = [plural(s.events, 'event'), plural(s.builds, 'build'), plural(s.reviews, 'review log line'), `${s.ended} ended`, plural(s.ci, 'attestation'), plural(s.commits, 'commit')].join(', ');585 for (const n of s.notes) console.log(`lake: ${n}`);586 if (s.bad) console.log(`lake: skipped ${plural(s.bad, 'line')} that aren't JSON`);587 if (dry) {588 console.log(`lake: would add ${what} (dry run: nothing written)`);589 return;590 }591 const ck = await checkpoint(lake, now);592 const ex = await exportInsights(lake, s.touched);593 const pub = publishInsights(cfg);594 const size = du(cfg.dir);595 console.log(`lake: added ${what}${ck ? '; checkpointed' : ''}; exported ${plural(ex, 'scope')}, published ${plural(pub, 'scope')}; ${((performance.now() - t0) / 1000).toFixed(2)}s, ${(size / 1024).toFixed(0)}K on disk`);596 } finally {597 lake.close();598 if (scratch) fs.rmSync(scratch, { recursive: true, force: true });599 }600 } else if (cmd === 'query') {601 const fmt = rest.includes('--json') ? 'json' : rest.includes('--csv') ? 'csv' : 'table';602 const sql = rest.filter((a) => a !== '--json' && a !== '--csv').join(' ');603 if (!sql.trim()) throw new LakeError('usage: forge lake query "<sql>" [--json|--csv]');604 process.stdout.write(await query(cfg, sql, fmt));605 } else if (cmd === 'forget') {606 if (rest.length !== 1 || !/^[^/]+\/[^/]+$/.test(rest[0])) throw new LakeError('usage: node lib/lake.ts forget <owner>/<repo>');607 if (!fs.existsSync(cfg.catalog)) return;608 const lake = await Lake.open(cfg, { write: true });609 try {610 console.log(`lake: forgot ${plural(await forget(lake, rest[0]), 'row')} of ${rest[0]}`);611 } finally {612 lake.close();613 }614 } else if (cmd === 'install') {615 await install(cfg);616 } else if (cmd === 'stage') {617 if (!rest[0]) throw new LakeError('usage: node lib/lake.ts stage <dir>');618 stage(cfg, rest[0]);619 } else {620 throw new LakeError('usage: forge lake [--dry-run] | forge lake query "<sql>" [--json|--csv]');621 }622}623624function du(p: string): number {625 let n = 0;626 try {627 const st = fs.lstatSync(p);628 if (!st.isDirectory()) return st.size;629 for (const e of fs.readdirSync(p)) n += du(path.join(p, e));630 } catch {}631 return n;632}633634if (process.argv[1] && fs.realpathSync(process.argv[1]) === fileURLToPath(import.meta.url)) {635 main(process.argv.slice(2)).catch((e) => {636 console.error(`forge: lake: ${e instanceof LakeError ? e.message : firstLine(e)}`);637 process.exit(1);638 });639}