1// What `forge lake` reads (lib/lake.ts): the forge's append-only facts, as2// rows, and where it got to in each source so the next run starts there.3//4// events.jsonl and metrics.jsonl are read from a byte offset. cmd_build cuts5// events.jsonl to its last 1000 lines once it passes 1 MB, which makes a new6// file (tail into a temporary file, mv over it), so an offset alone would7// skip or repeat lines. Each source also remembers the file's inode, a hash8// of its first line, and hashes of the last lines it read. If the inode, the9// first line or the line before the offset changed, or the file is shorter,10// it was cut or replaced: the run reads it whole and starts after the last11// line it had read, which is still there unless the cut went past it. Then12// the lines between are gone, and so is everything before them.13import fs from 'node:fs';14import path from 'node:path';15import { createHash } from 'node:crypto';16import { catFileBatch, isAncestor, tryRun } from './git.ts';17import { parseNote, type Verifier } from './attest.ts';18import { readReviewLogs } from './reviews.ts';19import type { MetricsLine } from './metrics.ts';20import type { ReviewLog } from './types.ts';2122// Where a jsonl source was read up to.23export interface Offset {24 ino: number;25 pos: number;26 head: string;27 // Hashes of the last lines read, oldest first (at most TAIL).28 tail: string[];29}3031export interface Read {32 lines: string[];33 next: Offset;34 // continue: appended to; found: cut or replaced, and the run found where it35 // had got to; lost: cut past that, so lines in between are gone; new: first read.36 how: 'continue' | 'found' | 'lost' | 'new';37}3839const TAIL = 16;40const NL = 0x0a;41export const lineHash = (b: Buffer | string): string => createHash('sha1').update(b).digest('hex').slice(0, 16);4243function readAt(fd: number, pos: number, len: number): Buffer {44 const buf = Buffer.alloc(len);45 let got = 0;46 while (got < len) {47 const n = fs.readSync(fd, buf, got, len - got, pos + got);48 if (!n) break;49 got += n;50 }51 return buf.subarray(0, got);52}5354// The complete lines of buf (each ends in a newline), with the byte offset55// just past each.56function split(buf: Buffer, base = 0): { line: Buffer; end: number }[] {57 const out: { line: Buffer; end: number }[] = [];58 let at = 0;59 for (let nl = buf.indexOf(NL); nl >= 0; nl = buf.indexOf(NL, at)) {60 out.push({ line: buf.subarray(at, nl), end: base + nl + 1 });61 at = nl + 1;62 }63 return out;64}6566function firstLine(fd: number, size: number): string {67 for (let len = 4096; ; len *= 4) {68 const buf = readAt(fd, 0, Math.min(len, size));69 const nl = buf.indexOf(NL);70 if (nl >= 0) return lineHash(buf.subarray(0, nl));71 if (buf.length >= size) return '';72 }73}7475// Does the line that ends at pos hash to want? (Reads back at most 1 MB; a76// longer line counts as changed, and the file is read whole.)77function endsWith(fd: number, pos: number, want: string | undefined): boolean {78 if (!want) return pos === 0;79 if (pos === 0) return false;80 const start = Math.max(0, pos - (1 << 20));81 const buf = readAt(fd, start, pos - start);82 if (buf.length !== pos - start || buf[buf.length - 1] !== NL) return false;83 const prev = buf.lastIndexOf(NL, buf.length - 2);84 if (prev < 0 && start > 0) return false;85 return lineHash(buf.subarray(prev + 1, buf.length - 1)) === want;86}8788// Where to start in a file that was cut or replaced: after the last place89// whose preceding lines are the tail that was read. A match may run back to90// the start of the file, when the cut fell inside the tail.91function resume(hashes: string[], tail: string[]): number | null {92 for (let p = hashes.length; p >= 1; p--) {93 const m = Math.min(tail.length, p);94 if (m < tail.length && p !== m) continue;95 let ok = m > 0;96 for (let i = 1; ok && i <= m; i++) ok = hashes[p - i] === tail[tail.length - i];97 if (ok) return p;98 }99 return null;100}101102// The lines appended to file since prev, and where the next read starts.103// Only complete lines count: one being written is read next time.104export function readNew(file: string, prev: Offset | null): Read | null {105 let fd: number;106 try {107 fd = fs.openSync(file, 'r');108 } catch {109 return null;110 }111 try {112 const st = fs.fstatSync(fd);113 const head = firstLine(fd, st.size);114 if (prev && prev.ino === st.ino && st.size >= prev.pos && prev.head === head && endsWith(fd, prev.pos, prev.tail.at(-1))) {115 const got = split(readAt(fd, prev.pos, st.size - prev.pos), prev.pos);116 const tail = [...prev.tail, ...got.map((l) => lineHash(l.line))].slice(-TAIL);117 return { lines: got.map((l) => l.line.toString('utf8')), next: { ino: st.ino, pos: got.at(-1)?.end ?? prev.pos, head, tail }, how: 'continue' };118 }119 const all = split(readAt(fd, 0, st.size));120 const hashes = all.map((l) => lineHash(l.line));121 const at = prev ? resume(hashes, prev.tail) : null;122 const from = at ?? 0;123 return {124 lines: all.slice(from).map((l) => l.line.toString('utf8')),125 next: { ino: st.ino, pos: all.at(-1)?.end ?? 0, head, tail: hashes.slice(-TAIL) },126 how: !prev ? 'new' : at == null ? 'lost' : 'found',127 };128 } finally {129 fs.closeSync(fd);130 }131}132133// ---------------------------------------------------------------- rows134135// Times are UTC strings DuckDB reads as TIMESTAMP.136export const ts = (ms: number | null | undefined): string | null =>137 ms == null || !Number.isFinite(ms) ? null : new Date(ms).toISOString().replace('T', ' ').replace('Z', '');138139const int = (v: unknown): number | null => (typeof v === 'number' && Number.isFinite(v) ? Math.round(v) : null);140const num = (v: unknown): number | null => (typeof v === 'number' && Number.isFinite(v) ? v : null);141const str = (v: unknown): string | null => (typeof v === 'string' ? v : null);142143export interface EventRow {144 t: string | null;145 repo: string | null;146 type: string | null;147 user: string | null;148 review: number | null;149 refs: string | null;150 ms: number | null;151 rendered: number | null;152 private: boolean;153}154155// One line of events.jsonl (hooks and builds append them; lib/events-server.ts).156export function eventRow(line: string): EventRow | null {157 let e: Record<string, unknown>;158 try {159 e = JSON.parse(line);160 } catch {161 return null;162 }163 if (!e || typeof e !== 'object') return null;164 return {165 t: ts(num(e.t)), repo: str(e.repo), type: str(e.type), user: str(e.user), review: int(e.review),166 refs: str(e.refs)?.trim() || null, ms: num(e.ms), rendered: int(e.rendered), private: e.private === true,167 };168}169170export interface BuildRow {171 id: string;172 at: string | null;173 code: string | null;174 reason: string | null;175 scope: string;176 ms_total: number | null;177 ms_data: number | null;178 ms_render: number | null;179 rss_mb: number | null;180 written: number | null;181 linked: number | null;182 line: string;183}184185export interface BuildRepoRow {186 build: string;187 at: string | null;188 repo: string;189 ms_load: number | null;190 pages: number | null;191 rendered: number | null;192 skipped: number | null;193 stages: string;194}195196// One line of metrics.jsonl (lib/metrics.ts): the build, and a row for each197// repo it loaded. The id is a hash of the line, which forge metrics uses to198// merge the lake's history with lines it hasn't read yet.199export function buildRows(line: string): { build: BuildRow; repos: BuildRepoRow[] } | null {200 let m: MetricsLine;201 try {202 m = JSON.parse(line);203 } catch {204 return null;205 }206 if (!m || typeof m !== 'object' || typeof m.at !== 'string') return null;207 const id = lineHash(line);208 const at = ts(Date.parse(m.at));209 return {210 build: {211 id, at, code: str(m.code), reason: str(m.reason), scope: m.only || 'all',212 ms_total: num(m.ms?.total), ms_data: num(m.ms?.data), ms_render: num(m.ms?.render), rss_mb: int(m.rss_mb),213 written: int(m.written), linked: int(m.linked), line,214 },215 repos: (Array.isArray(m.repos) ? m.repos : []).map((r) => ({216 build: id, at, repo: r.slug, ms_load: num(r.ms?.load), pages: int(r.counts?.pages), rendered: int(r.counts?.rendered),217 skipped: int(r.counts?.skipped), stages: JSON.stringify(r.ms ?? {}),218 })),219 };220}221222export interface RepoRef {223 slug: string;224 dir: string;225 public: boolean;226}227228// Every repo under the root, as bin/forge sees them: <owner>/<name>.git with229// a HEAD; public while it has git-daemon-export-ok.230export function listRepos(root: string): RepoRef[] {231 const out: RepoRef[] = [];232 let owners: string[];233 try {234 owners = fs.readdirSync(root).filter((o) => !o.startsWith('.')).sort();235 } catch {236 return out;237 }238 for (const o of owners) {239 let names: string[];240 try {241 names = fs.readdirSync(path.join(root, o)).filter((d) => d.endsWith('.git')).sort();242 } catch {243 continue;244 }245 for (const d of names) {246 const dir = path.join(root, o, d);247 if (!fs.existsSync(path.join(dir, 'HEAD'))) continue;248 out.push({ slug: `${o}/${d.slice(0, -4)}`, dir, public: fs.existsSync(path.join(dir, 'git-daemon-export-ok')) });249 }250 }251 return out;252}253254export interface ReviewRow {255 repo: string;256 review: number;257 kind: 'created' | 'rev' | 'approve' | 'deleted';258 ts: string | null;259 user: string | null;260 rev: number | null;261 sha: string | null;262 base: string | null;263 tree: string | null;264 branch: string | null;265 title: string | null;266}267268export interface ReviewEnd {269 repo: string;270 review: number;271 kind: 'landed' | 'closed';272 ts: string | null;273 user: string | null;274 sha: string | null;275 // The time is when the lake first saw it landed, not a logged one: the276 // branch is still there, so no deletion was logged.277 observed: boolean;278}279280export interface ReviewLogFile {281 log: ReviewLog;282 hash: string;283}284285// A repo's review logs, each with a hash of its file, so a run only redoes286// the reviews whose logs grew.287export function reviewLogs(dir: string): ReviewLogFile[] {288 const logs = readReviewLogs(dir);289 return logs.map((log) => {290 let hash = '';291 try {292 hash = lineHash(fs.readFileSync(path.join(dir, 'forge', 'reviews', String(log.id))));293 } catch {}294 return { log, hash };295 });296}297298// Every line of one review log, as rows: created, each revision, each299// approval (with the revision whose tip it approved), deleted.300export function reviewRows(slug: string, log: ReviewLog): ReviewRow[] {301 const row = (kind: ReviewRow['kind'], t: number | null, user: string | null, extra: Partial<ReviewRow> = {}): ReviewRow =>302 ({ repo: slug, review: log.id, kind, ts: t ? ts(t * 1000) : null, user, rev: null, sha: null, base: null, tree: null, branch: null, title: null, ...extra });303 const rows: ReviewRow[] = [];304 if (log.created) rows.push(row('created', log.created.ts, log.created.user, { branch: log.branch, title: log.opts.title || null }));305 for (const r of log.revs) rows.push(row('rev', r.ts, r.user, { rev: r.n, sha: r.sha, base: r.base }));306 for (const a of log.approvals) {307 const rev = [...log.revs].reverse().find((r) => r.sha === a.sha)?.n ?? null;308 rows.push(row('approve', a.ts, a.user, { rev, sha: a.sha, tree: a.tree }));309 }310 if (log.deleted) rows.push(row('deleted', log.deleted.ts, log.deleted.user));311 return rows;312}313314// Has this review ended, and how? Deleted with its tip in the default branch315// is landed (forge land and ssh land delete what they land), deleted316// otherwise is closed, and a branch still there whose tip is in the default317// branch landed without the deletion (observed at `now`). A tip that is its318// own merge base had no commits of its own, so it isn't counted as landed.319export function reviewEnd(slug: string, dir: string, log: ReviewLog, now: number): ReviewEnd | null {320 const last = log.revs.at(-1);321 if (!last) return null;322 const def = tryRun(dir, ['symbolic-ref', '--short', 'HEAD'])?.trim() || 'main';323 const defTip = tryRun(dir, ['rev-parse', '-q', '--verify', `refs/heads/${def}^{commit}`])?.trim();324 const live = !log.deleted ? tryRun(dir, ['rev-parse', '-q', '--verify', `refs/heads/${log.branch}^{commit}`])?.trim() : null;325 const tip = live || last.sha;326 const own = !(tip === last.sha && tip === last.base);327 const landed = !!defTip && own && isAncestor(dir, tip, defTip);328 const end = (kind: ReviewEnd['kind'], t: number, user: string | null, observed: boolean): ReviewEnd =>329 ({ repo: slug, review: log.id, kind, ts: ts(t * 1000), user, sha: tip, observed });330 if (log.deleted) return end(landed ? 'landed' : 'closed', log.deleted.ts, log.deleted.user, false);331 if (live && landed) return end('landed', now, null, true);332 return null;333}334335export interface CiRow {336 id: string;337 repo: string;338 commit: string | null;339 tree: string | null;340 result: string | null;341 exit: number | null;342 started: string | null;343 duration_s: number | null;344 runner: string | null;345 system: string | null;346 env: string | null;347 signer: string | null;348 verified: boolean;349}350351export interface CiMetricRow {352 id: string;353 repo: string;354 commit: string | null;355 tree: string | null;356 name: string;357 value: number;358 unit: string | null;359}360361// The tip of a repo's CI notes, so a run skips repos whose notes didn't move.362export const notesTip = (dir: string): string | null => tryRun(dir, ['rev-parse', '-q', '--verify', 'refs/notes/ci^{commit}'])?.trim() || null;363364// The attestations in a repo's refs/notes/ci that aren't in `known` (ids are365// a hash of the signed statement and its signature, so a note that gained a366// block only adds that block). Signatures are checked as they are read.367// Each one's metrics come with it, keyed by the same id.368export function ciRows(slug: string, dir: string, known: Set<string>, verifier: Verifier): { rows: CiRow[]; metrics: CiMetricRow[] } {369 const list = tryRun(dir, ['ls-tree', '-r', 'refs/notes/ci']);370 if (!list) return { rows: [], metrics: [] };371 const blobs = list.split('\n').filter(Boolean).map((l) => l.split(/\s+/)).filter((f) => f[1] === 'blob').map((f) => f[2]);372 const out: CiRow[] = [];373 const metrics: CiMetricRow[] = [];374 const seen = new Set(known);375 for (const b of catFileBatch(dir, blobs).values()) {376 for (const a of parseNote(b.data.toString('utf8'))) {377 const id = createHash('sha256').update(a.payload + a.signature).digest('hex').slice(0, 32);378 if (seen.has(id)) continue;379 seen.add(id);380 const v = verifier.verify(a);381 const dur = /^(\d+)s$/.exec(a.duration ?? '');382 out.push({383 id, repo: slug, commit: a.commit ?? null, tree: a.tree ?? null, result: a.result ?? null,384 exit: a.exit != null && /^-?\d+$/.test(a.exit) ? Number(a.exit) : null,385 started: a.started ? ts(Date.parse(a.started)) : null, duration_s: dur ? Number(dur[1]) : null,386 runner: a.runner ?? null, system: a.system ?? null, env: a.env ?? null, signer: v.signer, verified: v.ok,387 });388 for (const m of a.metrics) metrics.push({ id, repo: slug, commit: a.commit ?? null, tree: a.tree ?? null, name: m.name, value: m.value, unit: m.unit });389 }390 }391 return { rows: out, metrics };392}393394export interface CommitRow {395 repo: string;396 sha: string;397 t: string | null;398 author: string | null;399 subject: string | null;400 tree: string | null;401}402403// The commits reachable from tip and not from since (all of them without404// it), first parent or not: every commit the default branch took in.405export function commitRows(slug: string, dir: string, tip: string, since: string | null): CommitRow[] {406 const out = tryRun(dir, ['log', '-z', '--no-color', '--format=%H%x1f%at%x1f%an%x1f%T%x1f%s', tip, ...(since ? [`^${since}`] : []), '--']);407 if (!out) return [];408 return out.split('\0').filter(Boolean).map((rec) => {409 const [sha, at, author, tree, subject] = rec.replace(/^\n/, '').split('\x1f');410 return { repo: slug, sha, t: ts(Number(at) * 1000), author: author || null, subject: subject ?? null, tree: tree || null };411 });412}