1// Insights exports: small aggregates of the lake (lib/lake.ts) for the2// insights pages, as JSON the charts draw from (site/assets/insights.ts),3// plus the rows behind them as Parquet to explore with DuckDB.4//5// Every repo's go to $FORGE_STATE/lake/export/<owner>/<repo>/insights/, the6// forge's to .../export/insights/. The build copies a repo's into that repo's7// pages (lib/forge.ts), so they follow its page directory wherever its pages8// are served: www for a public repo, the private root for a private one,9// whose readers alone see them. publishInsights also puts them straight into10// the live pages after each run, since the pages fetch them and need no11// rebuild. The forge's cover public repos only: a private repo's facts are12// in no public export. A repo made public again publishes its facts from13// while it was private too, as it publishes all of its history then (the14// secret check of ssh public reads every commit first); a removed one's15// leave the lake (forge repo rm, lib/lake.ts forget), so a new repo of the16// same name starts with none. A build's reason is "push" or "other" in the17// exports: the others (private, removed, profile...) would say when a repo18// went private or away.19import fs from 'node:fs';20import path from 'node:path';21import { listRepos } from './lake-sources.ts';22import { lit, type Lake, type LakeConfig } from './lake.ts';23import type { RepoActivityExport } from './activity.ts';2425export const FILES = ['insights.json', 'activity.json', 'events.parquet', 'builds.parquet', 'reviews.parquet', 'ci.parquet', 'ci_metrics.parquet'];26// How far back the activity exports go: a year's grid, and some.27const DAYS = 400;28// The most points a chart gets; older ones are only in the Parquet files.29const CAP = 1000;30// The most points a metric's chart gets: there can be many metrics.31const METRIC_CAP = 400;3233const inList = (col: string, xs: string[]): string => (xs.length ? `${col} IN (${xs.map(lit).join(', ')})` : 'false');34const n = (v: unknown): number | null => (v == null ? null : Number(v));3536interface Scope {37 // A repo's slug, or null for the whole forge.38 slug: string | null;39 dir: string;40 // Rows of events, reviews and CI: on repo.41 where: string;42 // Rows of builds: on scope.43 builds: string;44}4546async function exportScope(lake: Lake, s: Scope, publicRepos: string[]): Promise<void> {47 const tmp = `${s.dir}.tmp`;48 fs.rmSync(tmp, { recursive: true, force: true });49 fs.mkdirSync(tmp, { recursive: true });50 const rows = (sql: string) => lake.rows(sql);51 const one = async (sql: string): Promise<unknown[]> => (await rows(sql))[0] ?? [];5253 const builds = (await rows(`SELECT epoch_ms("at"), ms_total FROM lake.builds WHERE ${s.builds} AND "at" IS NOT NULL AND ms_total IS NOT NULL ORDER BY "at" DESC LIMIT ${CAP}`)).reverse();54 const pushes = await rows(`SELECT strftime(t, '%Y-%m-%d'), count(*) FROM lake.events WHERE type = 'push' AND t IS NOT NULL AND ${s.where} GROUP BY 1 ORDER BY 1`);55 const reviews = (await rows(`56 WITH r AS (57 SELECT repo, review, min(ts) FILTER (kind = 'created') AS created, count(*) FILTER (kind = 'rev') AS revisions,58 count(DISTINCT "user") FILTER (kind = 'approve') AS approvals59 FROM lake.reviews WHERE ${s.where} GROUP BY repo, review)60 SELECT r.repo, r.review, epoch_ms(r.created), epoch_ms(e.ts), e.kind, r.revisions, r.approvals61 FROM r LEFT JOIN lake.review_ends e USING (repo, review)62 ORDER BY coalesce(e.ts, r.created) DESC NULLS LAST LIMIT ${CAP}`)).reverse();63 const ci = (await rows(`SELECT epoch_ms(started), duration_s, result = 'pass' FROM lake.ci WHERE ${s.where} AND started IS NOT NULL ORDER BY started DESC LIMIT ${CAP}`)).reverse();64 const [pushN, firstT, lastT] = await one(`SELECT count(*) FILTER (type = 'push'), epoch_ms(min(t)), epoch_ms(max(t)) FROM lake.events WHERE ${s.where}`);65 const [buildN, buildMed] = await one(`SELECT count(*), median(ms_total) FROM lake.builds WHERE ${s.builds}`);66 const [landed, closed, landH, revMed] = await one(`67 WITH r AS (SELECT repo, review, min(ts) FILTER (kind = 'created') AS created, count(*) FILTER (kind = 'rev') AS revisions FROM lake.reviews WHERE ${s.where} GROUP BY ALL)68 SELECT count(*) FILTER (e.kind = 'landed'), count(*) FILTER (e.kind = 'closed'),69 median(epoch(e.ts) - epoch(r.created)) FILTER (e.kind = 'landed') / 3600, median(r.revisions)70 FROM r LEFT JOIN lake.review_ends e USING (repo, review)`);71 const [reviewN] = await one(`SELECT count(DISTINCT (repo, review)) FROM lake.reviews WHERE ${s.where}`);72 const [ciN, ciPass, ciMed, ciVerified] = await one(`SELECT count(*), count(*) FILTER (result = 'pass'), median(duration_s), count(*) FILTER (verified) FROM lake.ci WHERE ${s.where}`);7374 const out: Record<string, unknown> = {75 scope: s.slug ?? 'forge',76 generated: new Date().toISOString(),77 from: n(firstT), to: n(lastT),78 totals: {79 pushes: n(pushN), builds: n(buildN), build_ms_median: n(buildMed), reviews: n(reviewN), landed: n(landed), closed: n(closed),80 land_hours_median: n(landH), revisions_median: n(revMed), ci: n(ciN), ci_pass: n(ciPass), ci_verified: n(ciVerified), ci_seconds_median: n(ciMed),81 },82 // [time (ms), total (ms)] for each build; [day, pushes]; [repo, id,83 // created, ended, how, revisions, approvals]; [started, seconds, passed].84 builds: builds.map((r) => r.map(n)),85 pushes: pushes.map(([d, c]) => [d, n(c)]),86 reviews: reviews.map(([repo, id, c, e, how, revs, appr]) => [repo, n(id), n(c), n(e), how ?? null, n(revs), n(appr)]),87 ci: ci.map(([t, d, p]) => [n(t), n(d), p ? 1 : 0]),88 };89 if (s.slug) out.metrics = await metrics(lake, s.slug);90 if (!s.slug) {91 const per = await rows(`92 WITH p AS (SELECT repo, count(*) FILTER (type = 'push') AS pushes, max(t) FILTER (type = 'push') AS last FROM lake.events WHERE ${s.where} GROUP BY repo),93 b AS (SELECT scope AS repo, count(*) AS builds, median(ms_total) AS build_ms FROM lake.builds WHERE ${inList('scope', publicRepos)} GROUP BY scope),94 e AS (SELECT repo, count(*) FILTER (kind = 'landed') AS landed FROM lake.review_ends WHERE ${s.where} GROUP BY repo),95 c AS (SELECT repo, count(*) AS ci, count(*) FILTER (result = 'pass') AS pass FROM lake.ci WHERE ${s.where} GROUP BY repo)96 SELECT repo, coalesce(p.pushes, 0), coalesce(b.builds, 0), b.build_ms, coalesce(e.landed, 0), coalesce(c.ci, 0), coalesce(c.pass, 0), epoch_ms(p.last)97 FROM (SELECT unnest([${publicRepos.map(lit).join(', ')}]::VARCHAR[]) AS repo) LEFT JOIN p USING (repo) LEFT JOIN b USING (repo) LEFT JOIN e USING (repo) LEFT JOIN c USING (repo)98 ORDER BY lower(repo)`);99 // [repo, pushes, builds, median build (ms), landed, CI runs, passed, last event]100 out.repos = per.map(([r, ...v]) => [r, ...v.map(n)]);101 }102103 const copy = async (file: string, sql: string): Promise<void> => {104 await lake.run(`COPY (${sql}) TO ${lit(path.join(tmp, file))} (FORMAT parquet, COMPRESSION zstd)`);105 };106 await copy('events.parquet', `SELECT t, repo, type, "user", review, refs, ms, rendered FROM lake.events WHERE ${s.where} ORDER BY t`);107 await copy('builds.parquet', `SELECT id, "at", code, CASE reason WHEN 'push' THEN 'push' ELSE 'other' END AS reason, scope, ms_total, ms_data, ms_render, rss_mb, written, linked FROM lake.builds WHERE ${s.builds} ORDER BY "at"`);108 await copy('reviews.parquet', `109 SELECT repo, review, kind, ts, "user", rev, sha, base, tree, branch, title, NULL::BOOLEAN AS observed FROM lake.reviews WHERE ${s.where}110 UNION ALL SELECT repo, review, kind, ts, "user", NULL, sha, NULL, NULL, NULL, NULL, observed FROM lake.review_ends WHERE ${s.where}111 ORDER BY repo, review, ts`);112 await copy('ci.parquet', `SELECT repo, "commit", tree, result, exit, started, duration_s, system, env, signer, verified FROM lake.ci WHERE ${s.where} ORDER BY started`);113 await copy('ci_metrics.parquet', `114 SELECT m.repo, m."commit", m.tree, m.name, m.value, m.unit, c.started, c.result, c.signer, c.verified115 FROM (SELECT * FROM lake.ci_metrics WHERE ${s.where}) m JOIN lake.ci c USING (id) ORDER BY c.started, m.name`);116 out.files = FILES.slice(2).map((f) => [f, fs.statSync(path.join(tmp, f)).size]);117 if (s.slug) fs.writeFileSync(path.join(tmp, 'activity.json'), JSON.stringify(await repoActivity(lake, s.slug)) + '\n');118 fs.writeFileSync(path.join(tmp, 'insights.json'), JSON.stringify(out) + '\n');119120 // Swapped in whole; a build reading it meanwhile (lib/forge.ts) skips what121 // it doesn't find, and the next run puts it back.122 const old = `${s.dir}.old`;123 fs.rmSync(old, { recursive: true, force: true });124 if (fs.existsSync(s.dir)) fs.renameSync(s.dir, old);125 fs.renameSync(tmp, s.dir);126 fs.rmSync(old, { recursive: true, force: true });127}128129// A repo's metrics on its default branch, for a chart each: [name, unit,130// points], by name, where a point is [commit time (ms), value, commit,131// subject], oldest first. A commit gets the value of an attestation of its132// tree (or of itself, for commits the lake has no tree of), the verified133// and passing one first, then the latest; its unit is the latest point's.134// Commits made the same second go in the order they were attested.135async function metrics(lake: Lake, slug: string): Promise<[string, string | null, [number, number, string, string][]][]> {136 const got = await lake.rows(`137 WITH p AS (138 SELECT m.name, m.value, m.unit, k.sha, k.t, k.subject, c.started139 FROM lake.ci_metrics m JOIN lake.ci c USING (id)140 JOIN lake.commits k ON k.repo = m.repo AND (k.tree = m.tree OR (k.tree IS NULL AND k.sha = m."commit"))141 WHERE m.repo = ${lit(slug)} AND k.t IS NOT NULL142 QUALIFY row_number() OVER (PARTITION BY m.name, k.sha ORDER BY c.verified DESC, c.result = 'pass' DESC, c.started DESC NULLS LAST) = 1),143 q AS (SELECT *, row_number() OVER (PARTITION BY name ORDER BY t DESC, started DESC NULLS LAST, sha) AS i FROM p)144 SELECT name, arg_max(unit, t), list([epoch_ms(t)::VARCHAR, value::VARCHAR, sha, left(subject, 100)] ORDER BY t, started NULLS FIRST, sha)145 FROM q WHERE i <= ${METRIC_CAP} GROUP BY name ORDER BY name`);146 return got.map(([name, unit, pts]) => [147 String(name), unit == null ? null : String(unit),148 (pts as string[][]).map(([t, v, sha, subj]) => [Number(t), Number(v), sha, subj ?? ''] as [number, number, string, string]),149 ]);150}151152const since = (): string => {153 const now = Number(process.env.FORGE_NOW) * 1000 || Date.now();154 return lit(new Date(now - DAYS * 86400000).toISOString().slice(0, 10));155};156const day = (ms: number): string => new Date(ms).toISOString().slice(0, 10);157158// A repo's activity, for the grid on its insights page (ui.activity): how159// many things happened each day (pushes, landed reviews, releases, site160// deploys, CI results), and each one's time (ms), newest first, so the161// page can count them on the reader's days. Days are UTC.162export async function repoActivity(lake: Lake, slug: string): Promise<RepoActivityExport> {163 const where = `repo = ${lit(slug)}`;164 const events: number[] = [];165 for (const [t, refs] of await lake.rows(`SELECT epoch_ms(t), refs FROM lake.events WHERE type = 'push' AND ${where} AND t >= ${since()} AND refs IS NOT NULL`)) {166 let heads = false;167 const named = new Set<string>();168 for (const ref of String(refs).split(/\s+/).filter(Boolean)) {169 const m = /^refs\/(heads|tags|releases|sites)\/(.+)$/.exec(ref);170 if (!m) continue;171 if (m[1] === 'heads') heads = true;172 else if (m[1] === 'sites') events.push(Number(t));173 else if (!named.has(m[2])) {174 // A tag and its release in one push are one release.175 named.add(m[2]);176 events.push(Number(t));177 }178 }179 // Its branches are one push.180 if (heads) events.push(Number(t));181 }182 for (const [t] of await lake.rows(`SELECT epoch_ms(ts) FROM lake.review_ends WHERE kind = 'landed' AND ${where} AND ts >= ${since()}`)) events.push(Number(t));183 for (const [t] of await lake.rows(`SELECT epoch_ms(started) FROM lake.ci WHERE ${where} AND started >= ${since()} AND "commit" IS NOT NULL`)) events.push(Number(t));184 events.sort((a, b) => b - a);185 const days: Record<string, number> = {};186 for (const t of events) days[day(t)] = (days[day(t)] ?? 0) + 1;187 return { days, events };188}189190// Each owner's activity, for the grid on their page: when each commit on191// the default branches of their public repos was made (its author's time),192// by repo, as minutes since 1970, oldest first, each after the first as193// the minutes since the one before (lib/activity.ts reads them), so the page194// can put them on the reader's days; and when they joined: their first195// public event, or the first push to a public repo of theirs. A private196// repo's commits are in no export.197async function exportPeople(lake: Lake, pub: string[]): Promise<number> {198 const dir = path.join(lake.cfg.exportDir, 'people');199 const tmp = `${dir}.tmp`;200 fs.rmSync(tmp, { recursive: true, force: true });201 fs.mkdirSync(tmp, { recursive: true });202 const people = new Map<string, { repos: Record<string, number[]>; joined: number | null }>();203 const of = (o: string) => {204 let p = people.get(o);205 if (!p) people.set(o, (p = { repos: {}, joined: null }));206 return p;207 };208 for (const slug of pub) of(slug.split('/')[0]);209 for (const [repo, mins] of await lake.rows(`SELECT repo, list(floor(epoch(t) / 60)::BIGINT ORDER BY t) FROM lake.commits WHERE ${inList('repo', pub)} AND t >= ${since()} GROUP BY repo ORDER BY repo`)) {210 const [o, name] = String(repo).split('/');211 of(o).repos[name] = encodeMinutes((mins as unknown[]).map(Number));212 }213 for (const [who, t] of await lake.rows(`214 SELECT who, epoch_ms(min(t)) FROM (215 SELECT "user" AS who, t FROM lake.events WHERE ${inList('repo', pub)} AND "user" IS NOT NULL216 UNION ALL SELECT split_part(repo, '/', 1), t FROM lake.events WHERE ${inList('repo', pub)}217 ) WHERE t IS NOT NULL GROUP BY who`)) {218 if (people.has(String(who))) of(String(who)).joined = Number(t);219 }220 for (const [o, p] of people) fs.writeFileSync(path.join(tmp, `${o}.json`), JSON.stringify(p) + '\n');221 const old = `${dir}.old`;222 fs.rmSync(old, { recursive: true, force: true });223 if (fs.existsSync(dir)) fs.renameSync(dir, old);224 fs.renameSync(tmp, dir);225 fs.rmSync(old, { recursive: true, force: true });226 return people.size;227}228229// Ascending minutes as the first, then each one's distance from the one230// before (decodeMinutes in lib/activity.ts undoes it).231export const encodeMinutes = (mins: number[]): number[] => mins.map((m, i) => (i ? m - mins[i - 1] : m));232233const exportFingerprint = (repos: string[]): string => repos.join('\n');234235// Export the forge's insights, and those of each repo in `touched` (and of236// any repo without an export yet). Returns how many scopes it wrote.237export async function exportInsights(lake: Lake, touched: Set<string>): Promise<number> {238 const cfg = lake.cfg;239 const repos = listRepos(cfg.root);240 const pub = repos.filter((r) => r.public).map((r) => r.slug);241 const dir = (slug: string | null): string => path.join(cfg.exportDir, ...(slug ? [...slug.split('/'), 'insights'] : ['insights']));242 // Which repos are public is part of the forge's export: a repo that went243 // private leaves it at the next run, even if nothing else changed.244 const mark = path.join(cfg.exportDir, 'public');245 const was = fs.existsSync(mark) ? fs.readFileSync(mark, 'utf8') : null;246 let count = 0;247 fs.mkdirSync(cfg.exportDir, { recursive: true });248 for (const r of repos) {249 if (!touched.has(r.slug) && fs.existsSync(dir(r.slug))) continue;250 const w = `repo = ${lit(r.slug)}`;251 await exportScope(lake, { slug: r.slug, dir: dir(r.slug), where: w, builds: `scope = ${lit(r.slug)}` }, pub);252 count++;253 }254 if (touched.size || was !== exportFingerprint(pub) || !fs.existsSync(dir(null))) {255 await exportScope(lake, { slug: null, dir: dir(null), where: inList('repo', pub), builds: `(${inList('scope', pub)} OR scope IN ('-', 'all'))` }, pub);256 fs.writeFileSync(mark, exportFingerprint(pub));257 count++;258 }259 await exportPeople(lake, pub);260 // Repos that are gone.261 const live = new Set(repos.map((r) => r.slug));262 for (const o of fs.readdirSync(cfg.exportDir, { withFileTypes: true })) {263 if (!o.isDirectory() || o.name === 'insights' || o.name === 'people' || o.name.endsWith('.tmp') || o.name.endsWith('.old')) continue;264 for (const r of fs.readdirSync(path.join(cfg.exportDir, o.name))) {265 if (!live.has(`${o.name}/${r}`)) fs.rmSync(path.join(cfg.exportDir, o.name, r), { recursive: true, force: true });266 }267 }268 return count;269}270271// Put each export into the live pages: the forge's into www/insights/, a272// public repo's into its page directory in www and a private repo's into273// its page directory in the private root, if it has one there (pages of a274// repo that just flipped are gone, and stay gone until its build). A file275// at a time, each replaced by a rename, so a page never fetches half of one.276export function publishInsights(cfg: LakeConfig): number {277 let count = 0;278 const put = (from: string, to: string): void => {279 fs.mkdirSync(to, { recursive: true });280 for (const f of FILES) {281 const src = path.join(from, f);282 if (!fs.existsSync(src)) continue;283 const tmp = path.join(to, `.${f}.${process.pid}`);284 fs.rmSync(tmp, { force: true });285 try {286 fs.linkSync(src, tmp);287 } catch {288 fs.copyFileSync(src, tmp);289 }290 fs.renameSync(tmp, path.join(to, f));291 }292 count++;293 };294 const forge = path.join(cfg.exportDir, 'insights');295 if (fs.existsSync(cfg.www) && fs.existsSync(forge)) put(forge, path.join(cfg.www, 'insights'));296 // An owner's activity, beside their page (/<owner>/activity.json).297 const people = path.join(cfg.exportDir, 'people');298 for (const f of fs.existsSync(people) ? fs.readdirSync(people) : []) {299 const page = path.join(cfg.www, f.slice(0, -'.json'.length));300 if (!f.endsWith('.json') || !fs.existsSync(path.join(page, 'index.html'))) continue;301 const tmp = path.join(page, `.activity.json.${process.pid}`);302 fs.copyFileSync(path.join(people, f), tmp);303 fs.renameSync(tmp, path.join(page, 'activity.json'));304 }305 for (const r of listRepos(cfg.root)) {306 const from = path.join(cfg.exportDir, r.slug, 'insights');307 const pages = path.join(r.public ? cfg.www : cfg.priv, r.slug);308 if (!fs.existsSync(from) || !fs.existsSync(pages)) continue;309 put(from, path.join(pages, 'insights'));310 // Flipped meanwhile: take them out again; its next build puts them311 // where its pages are now.312 if (fs.existsSync(path.join(r.dir, 'git-daemon-export-ok')) !== r.public) {313 for (const f of FILES) fs.rmSync(path.join(pages, 'insights', f), { force: true });314 }315 }316 return count;317}