src/trends.ts

Functions

Imports

It imports src/approval-executor.ts, src/env.ts, src/github/client.ts, src/log.ts, src/ops-types/views.ts, src/roster.ts, src/time.ts. Imported by src/merge-safety.ts, src/portal-trends.ts.

Source

1import { COMPLETE_MARKER, DEFAULT_RETENTION } from "@dustinedwards/d1-dump";
2import { AUTO_MERGED_APPROVAL_REASON } from "./approval-executor";
3import type { Env } from "./env";
4import { getDefaultBranch, ghFetch, resolveRepo } from "./github/client";
5import { APP_ROSTER } from "./roster";
6import { logEvent } from "./log";
7import type { TrendBackupSlot, TrendCiDay, TrendCiRepo, TrendCostWeek, TrendMergeWeek, TrendMinutesWeek, TrendQueueDay, PortalTrends } from "./ops-types/views";
8import { MINUTE_MS, DAY_MS } from "./time";
9
10// The series behind the Portal's charts (docs/design/design-portal-evaluation.md section 3,
11// ruled 2026-10-09: DECIDE 5, the trends route, and DECIDE 6, all eleven visuals). The feed
12// carries no series longer than 7 days, so these are read on demand by GET
13// /portal/api/trends (src/portal-trends.ts), read-only:
14//
15//   minutes    visual 1: minutes of Dustin per week, an estimate, and his decisions
16//   queue      visual 3: per day, jobs posted, jobs ended, and jobs open at the day's end
17//   merges     visual 7: per week, merged by the policy, by an approval, by the seat, closed;
18//              and each rollback as an event
19//   cost       visual 6: per week, model spend of the jobs that ended, against the cap in the feed
20//   backups    visual 10: the retention slots, 14 daily, 8 weekly, 6 monthly, filled or not
21//   ci         visual 11: per roster repo and day, the default branch green, red or not run
22//
23// Visuals 2 and 5 (waits, claim agreement) come from GET /portal/api/claims, which already
24// serves them; 4, 8 and 9 from the plan and the feed. Every time here is UTC.
25
26const TRENDS_QUEUE_DAYS = 14;
27const TRENDS_CI_DAYS = 14;
28// A human's activity rows read for the minutes, at most. Far above a real week's count;
29// past it the series says it is cut rather than undercounting quietly. The newest rows
30// are the ones kept, so a cut costs the oldest week, not this one.
31const MINUTES_ROW_CAP = 20_000;
32// A gap longer than this starts a new session (the experience review's section 8).
33const SESSION_GAP_MS = 5 * MINUTE_MS;
34const CI_CACHE_KEY = "portal:trends-ci:v1";
35const CI_CACHE_SECONDS = 60 * 60;
36const BACKUP_PREFIX = "backups/json/";
37
38const dayOf = (ms: number) => new Date(ms).toISOString().slice(0, 10);
39
40/** The Monday (UTC) of the week a day falls in, as YYYY-MM-DD. */
41export function mondayOf(day: string): string {
42  const t = Date.parse(`${day}T00:00:00Z`);
43  const dow = (new Date(t).getUTCDay() + 6) % 7;
44  return dayOf(t - dow * DAY_MS);
45}
46
47/** The ISO 8601 week label of a Monday, as "2026-W41". */
48export function isoWeek(monday: string): string {
49  const t = Date.parse(`${monday}T00:00:00Z`);
50  // The ISO year is the year of the week's Thursday.
51  const thursday = new Date(t + 3 * DAY_MS);
52  const year = thursday.getUTCFullYear();
53  const jan4 = Date.UTC(year, 0, 4);
54  const week1 = jan4 - ((new Date(jan4).getUTCDay() + 6) % 7) * DAY_MS;
55  const n = Math.round((t - week1) / (7 * DAY_MS)) + 1;
56  return `${year}-W${String(n).padStart(2, "0")}`;
57}
58
59/** The Mondays of the last `weeks` weeks, oldest first, ending with this week's. */
60export function weekStarts(now: Date, weeks: number): string[] {
61  const thisMonday = Date.parse(`${mondayOf(dayOf(now.getTime()))}T00:00:00Z`);
62  return Array.from({ length: weeks }, (_, i) => dayOf(thisMonday - (weeks - 1 - i) * 7 * DAY_MS));
63}
64
65/** The last `days` days, oldest first, ending today. */
66export function lastDays(now: Date, days: number): string[] {
67  const today = Date.parse(`${dayOf(now.getTime())}T00:00:00Z`);
68  return Array.from({ length: days }, (_, i) => dayOf(today - (days - 1 - i) * DAY_MS));
69}
70
71/** An audit or touch time to milliseconds. Both "YYYY-MM-DD HH:MM:SS" (SQLite's
72 *  datetime('now'), no zone) and ISO are written; both are UTC. */
73export function timeMs(at: string): number {
74  return Date.parse(at.includes("T") ? (/[zZ]|[+-]\d\d:?\d\d$/.test(at) ? at : `${at}Z`) : `${at.replace(" ", "T")}Z`);
75}
76
77/** Minutes and sessions per week from activity times: a session ends after a gap over
78 *  five minutes and counts its span plus one minute, so a single click is a minute. A
79 *  session is credited to the week it starts in. */
80export function sessionMinutes(times: number[], mondays: string[]): Map<string, { minutes: number; sessions: number }> {
81  const out = new Map(mondays.map((m) => [m, { minutes: 0, sessions: 0 }]));
82  const sorted = [...times].filter(Number.isFinite).sort((a, b) => a - b);
83  let i = 0;
84  while (i < sorted.length) {
85    const start = sorted[i]!;
86    let end = start;
87    while (i + 1 < sorted.length && sorted[i + 1]! - end <= SESSION_GAP_MS) end = sorted[++i]!;
88    i++;
89    const week = out.get(mondayOf(dayOf(start)));
90    if (!week) continue;
91    week.sessions += 1;
92    week.minutes += (end - start) / MINUTE_MS + 1;
93  }
94  for (const w of out.values()) w.minutes = Math.round(w.minutes);
95  return out;
96}
97
98async function readMinutes(db: D1Database, mondays: string[]): Promise<{ weeks: TrendMinutesWeek[]; truncated: boolean }> {
99  const from = `${mondays[0]} 00:00:00`;
100  const [audit, touches, decisions, approvals] = await Promise.all([
101    // A write by the administrator's own login. The Portal's reads are not audited, so
102    // looking without acting is not counted: the figure is a floor, and says so.
103    db
104      .prepare("SELECT at FROM audit_log WHERE at >= ?1 AND actor LIKE 'access:%' ORDER BY at DESC LIMIT ?2")
105      .bind(from, MINUTES_ROW_CAP + 1)
106      .all<{ at: string }>(),
107    db
108      .prepare("SELECT at FROM job_touches WHERE at >= ?1 AND actor_kind = 'human' ORDER BY at DESC LIMIT ?2")
109      .bind(mondays[0], MINUTES_ROW_CAP + 1)
110      .all<{ at: string }>(),
111    // A decision is a human touch that ended a wait or closed a job, not a note.
112    db
113      .prepare(
114        "SELECT substr(at, 1, 10) AS day, COUNT(*) AS n FROM job_touches WHERE at >= ?1 AND actor_kind = 'human' AND kind <> 'note' GROUP BY day"
115      )
116      .bind(mondays[0])
117      .all<{ day: string; n: number }>(),
118    db
119      .prepare("SELECT substr(approved_at, 1, 10) AS day, COUNT(*) AS n FROM pr_approvals WHERE approved_at >= ?1 AND approved_by LIKE 'access:%' GROUP BY day")
120      .bind(mondays[0])
121      .all<{ day: string; n: number }>(),
122  ]);
123  const truncated = audit.results.length > MINUTES_ROW_CAP || touches.results.length > MINUTES_ROW_CAP;
124  const times = [...audit.results.slice(0, MINUTES_ROW_CAP), ...touches.results.slice(0, MINUTES_ROW_CAP)].map((r) => timeMs(r.at));
125  const minutes = sessionMinutes(times, mondays);
126  const perWeek = new Map(mondays.map((m) => [m, 0]));
127  for (const r of [...decisions.results, ...approvals.results]) {
128    const m = mondayOf(r.day);
129    if (perWeek.has(m)) perWeek.set(m, perWeek.get(m)! + r.n);
130  }
131  return {
132    weeks: mondays.map((m) => ({ week: isoWeek(m), start: m, minutes: minutes.get(m)!.minutes, sessions: minutes.get(m)!.sessions, decisions: perWeek.get(m)! })),
133    truncated,
134  };
135}
136
137async function readQueue(db: D1Database, days: string[]): Promise<TrendQueueDay[]> {
138  const from = days[0]!;
139  const [before, posted, ended] = await Promise.all([
140    // Open when the window starts: posted before it and not yet ended by then. A job
141    // that ended (done, failed, superseded) never moves again, so its updated_at is when.
142    db
143      .prepare("SELECT COUNT(*) AS n FROM jobs WHERE created_at < ?1 AND NOT (status IN ('done', 'failed', 'superseded') AND updated_at < ?1)")
144      .bind(from)
145      .first<{ n: number }>(),
146    db.prepare("SELECT substr(created_at, 1, 10) AS day, COUNT(*) AS n FROM jobs WHERE created_at >= ?1 GROUP BY day").bind(from).all<{ day: string; n: number }>(),
147    db
148      .prepare("SELECT substr(updated_at, 1, 10) AS day, COUNT(*) AS n FROM jobs WHERE status IN ('done', 'failed', 'superseded') AND updated_at >= ?1 GROUP BY day")
149      .bind(from)
150      .all<{ day: string; n: number }>(),
151  ]);
152  const p = new Map(posted.results.map((r) => [r.day, r.n]));
153  const e = new Map(ended.results.map((r) => [r.day, r.n]));
154  let open = before?.n ?? 0;
155  return days.map((day) => {
156    const row = { day, posted: p.get(day) ?? 0, ended: e.get(day) ?? 0, open: 0 };
157    open += row.posted - row.ended;
158    row.open = open;
159    return row;
160  });
161}
162
163async function readMerges(db: D1Database, mondays: string[]): Promise<{ weeks: TrendMergeWeek[]; rollbacks: PortalTrends["merges"]["rollbacks"] }> {
164  const from = `${mondays[0]} 00:00:00`;
165  const [rows, rollbacks] = await Promise.all([
166    // auto-merged: the policy merged on its own (src/auto-merge-tick.ts). approved-merged:
167    // the executor merged an approval (src/approval-executor.ts). manage_pr: the seat's
168    // own merge or close through capsid_close_pr. pr-closed: a close from the Portal. A
169    // policy merge of a pull request that held an approval also closes the approval with an
170    // approved-merged row (closeAsAutoMerged); that row is the same merge, so it is skipped.
171    db
172      .prepare(
173        `SELECT substr(at, 1, 10) AS day,
174                CASE WHEN action = 'auto-merged' THEN 'auto'
175                     WHEN action = 'approved-merged' THEN 'approved'
176                     WHEN action = 'pr-closed' THEN 'closed'
177                     WHEN params LIKE '%"action":"close"%' THEN 'closed'
178                     ELSE 'seat' END AS kind,
179                COUNT(*) AS n
180         FROM audit_log
181         WHERE at >= ?1
182           AND (action IN ('auto-merged', 'approved-merged', 'pr-closed')
183                OR (action = 'manage_pr' AND (params LIKE '%"merged":true%' OR params LIKE '%"action":"close"%')))
184           AND NOT (action = 'approved-merged' AND instr(COALESCE(params, ''), ?2) > 0)
185         GROUP BY day, kind`
186      )
187      // instr, not LIKE: D1 refuses a LIKE pattern this long.
188      .bind(from, `"reason":${JSON.stringify(AUTO_MERGED_APPROVAL_REASON)}`)
189      .all<{ day: string; kind: "auto" | "approved" | "seat" | "closed"; n: number }>(),
190    db
191      .prepare("SELECT repo, pr, run_at, outcome FROM deploy_outcomes WHERE run_at >= ?1 AND outcome IN ('rolled_back', 'rollback_failed') ORDER BY run_at")
192      .bind(mondays[0])
193      .all<{ repo: string; pr: number | null; run_at: string; outcome: "rolled_back" | "rollback_failed" }>(),
194  ]);
195  const weeks = new Map<string, TrendMergeWeek>(mondays.map((m) => [m, { week: isoWeek(m), start: m, auto: 0, approved: 0, seat: 0, closed: 0 }]));
196  for (const r of rows.results) {
197    const w = weeks.get(mondayOf(r.day));
198    if (w) w[r.kind] += r.n;
199  }
200  return { weeks: [...weeks.values()], rollbacks: rollbacks.results.map((r) => ({ at: r.run_at, repo: r.repo, pr: r.pr, outcome: r.outcome })) };
201}
202
203/** Model spend per week from outcome rows summed by day (cost_usd, rows that carried one).
204 *  A week with no costed outcome is null, never 0: no telemetry is not a free week. */
205export function costWeeks(rows: Array<{ day: string; usd: number; n: number }>, mondays: string[]): TrendCostWeek[] {
206  const weeks = new Map<string, { usd: number; jobs: number }>(mondays.map((m) => [m, { usd: 0, jobs: 0 }]));
207  for (const r of rows) {
208    const w = weeks.get(mondayOf(r.day));
209    if (!w) continue;
210    w.usd += r.usd;
211    w.jobs += r.n;
212  }
213  return mondays.map((m) => {
214    const w = weeks.get(m)!;
215    return { week: isoWeek(m), start: m, usd: w.jobs ? Math.round(w.usd * 100) / 100 : null, jobs: w.jobs };
216  });
217}
218
219async function readCost(db: D1Database, mondays: string[]): Promise<TrendCostWeek[]> {
220  const rows = await db
221    .prepare("SELECT substr(recorded_at, 1, 10) AS day, SUM(cost_usd) AS usd, COUNT(cost_usd) AS n FROM job_outcomes WHERE recorded_at >= ?1 AND cost_usd IS NOT NULL GROUP BY day")
222    .bind(mondays[0])
223    .all<{ day: string; usd: number; n: number }>();
224  return costWeeks(rows.results, mondays);
225}
226
227/** The retention slots d1-dump keeps (14 daily, 8 weekly, 6 monthly), each filled when a
228 *  complete run falls in it. Run ids begin with their ISO day. */
229export function backupSlots(completeRunIds: string[], now: Date): NonNullable<PortalTrends["backups"]["slots"]> {
230  const days = new Set(completeRunIds.map((id) => id.slice(0, 10)));
231  const weeks = new Set([...days].map(mondayOf));
232  const months = new Set([...days].map((d) => d.slice(0, 7)));
233  const { daily: nDaily, weekly: nWeekly, monthly: nMonthly } = DEFAULT_RETENTION;
234  const daily: TrendBackupSlot[] = lastDays(now, nDaily).map((key) => ({ key, filled: days.has(key) }));
235  const weekly: TrendBackupSlot[] = weekStarts(now, nWeekly).map((key) => ({ key, filled: weeks.has(key) }));
236  const y = now.getUTCFullYear();
237  const m = now.getUTCMonth();
238  const monthly: TrendBackupSlot[] = Array.from({ length: nMonthly }, (_, i) => {
239    const d = new Date(Date.UTC(y, m - (nMonthly - 1 - i), 1));
240    const key = d.toISOString().slice(0, 7);
241    return { key, filled: months.has(key) };
242  });
243  return { daily, weekly, monthly };
244}
245
246async function readBackups(env: Env, now: Date): Promise<PortalTrends["backups"]> {
247  try {
248    const complete: string[] = [];
249    let cursor: string | undefined;
250    do {
251      const page = await env.MEDIA.list({ prefix: BACKUP_PREFIX, cursor });
252      for (const o of page.objects) {
253        if (o.key.endsWith(`/${COMPLETE_MARKER}`)) complete.push(o.key.slice(BACKUP_PREFIX.length).split("/")[0]!);
254      }
255      cursor = page.truncated ? page.cursor : undefined;
256    } while (cursor);
257    return { slots: backupSlots(complete, now), runs: complete.length, error: null };
258  } catch (err) {
259    return { slots: null, runs: null, error: `The backup bucket could not be listed: ${err instanceof Error ? err.message : String(err)}` };
260  }
261}
262
263interface RunLite {
264  conclusion: string | null;
265  status: string;
266  created_at: string;
267  // Each workflow keeps its own verdict, so a green run of one workflow does not hide
268  // another that is red. Absent, every run counts as one workflow.
269  workflow_id?: number;
270}
271
272/** A repo's default branch per day: "red" when any workflow's last completed run up to
273 *  the day's end failed, "green" when every one passed, carried forward over a day with
274 *  none, and "none" before the first run seen. */
275export function ciDays(runs: RunLite[], days: string[]): TrendCiDay[] {
276  const done = runs.filter((r) => r.status === "completed" && r.conclusion !== null && !["cancelled", "skipped", "neutral"].includes(r.conclusion));
277  const byDay = new Map<string, RunLite[]>();
278  for (const r of done) {
279    const d = r.created_at.slice(0, 10);
280    byDay.set(d, [...(byDay.get(d) ?? []), r]);
281  }
282  const latest = new Map<number | undefined, boolean>();
283  return days.map((day) => {
284    const list = (byDay.get(day) ?? []).sort((a, b) => a.created_at.localeCompare(b.created_at));
285    const failed = list.filter((r) => r.conclusion !== "success").length;
286    for (const r of list) latest.set(r.workflow_id, r.conclusion === "success");
287    const state: TrendCiDay["state"] = latest.size === 0 ? "none" : [...latest.values()].every(Boolean) ? "green" : "red";
288    return { day, runs: list.length, failed, state };
289  });
290}
291
292async function readCiRepo(env: Env, namespace: string, days: string[]): Promise<TrendCiRepo> {
293  const { owner, repo, full } = await resolveRepo(env, namespace);
294  const branch = await getDefaultBranch(env, owner, repo);
295  const query = `/repos/${owner}/${repo}/actions/runs?branch=${encodeURIComponent(branch)}&created=${encodeURIComponent(`>=${days[0]}`)}&per_page=100&exclude_pull_requests=true`;
296  const resp = await ghFetch(env, owner, repo, query);
297  if (!resp.ok) throw new Error(`GitHub answered ${resp.status} for ${full}'s runs`);
298  const data = (await resp.json()) as { workflow_runs: RunLite[]; total_count: number };
299  return { namespace, repo: full, days: ciDays(data.workflow_runs, days), truncated: data.total_count > data.workflow_runs.length };
300}
301
302async function readCi(env: Env, now: Date): Promise<PortalTrends["ci"]> {
303  const days = lastDays(now, TRENDS_CI_DAYS);
304  try {
305    const cached = await env.APP_KV.get(CI_CACHE_KEY);
306    if (cached) {
307      const parsed = JSON.parse(cached) as PortalTrends["ci"];
308      if (parsed.repos.every((r) => r.days.length === days.length && r.days[r.days.length - 1]?.day === days[days.length - 1])) return parsed;
309    }
310  } catch (err) {
311    // An unreadable cache entry is read again from GitHub below.
312    logEvent("warn", "TRENDS_CI_CACHE_UNREADABLE", { message: `TRENDS_CI_CACHE_UNREADABLE ${CI_CACHE_KEY}: ${err instanceof Error ? err.message : String(err)}` });
313  }
314  const repos: TrendCiRepo[] = [];
315  const notRead: Array<{ namespace: string; error: string }> = [];
316  const settled = await Promise.allSettled(APP_ROSTER.map((namespace) => readCiRepo(env, namespace, days)));
317  settled.forEach((r, i) => {
318    if (r.status === "fulfilled") repos.push(r.value);
319    else notRead.push({ namespace: APP_ROSTER[i]!, error: r.reason instanceof Error ? r.reason.message : String(r.reason) });
320  });
321  const out: PortalTrends["ci"] = { read_at: now.toISOString(), repos, not_read: notRead };
322  // Only a full read is kept, so a GitHub outage is retried on the next look, not for an hour.
323  // A failed cache write costs only the next look a fresh read, so it does not fail this one.
324  if (!notRead.length) {
325    try {
326      await env.APP_KV.put(CI_CACHE_KEY, JSON.stringify(out), { expirationTtl: CI_CACHE_SECONDS });
327    } catch (err) {
328      logEvent("warn", "TRENDS_CI_CACHE_WRITE_FAILED", { message: `TRENDS_CI_CACHE_WRITE_FAILED ${CI_CACHE_KEY}: ${err instanceof Error ? err.message : String(err)}` });
329    }
330  }
331  return out;
332}
333
334/** Every series, for `weeks` weeks of the weekly ones. */
335export async function readTrends(env: Env, now: Date, weeks: number): Promise<PortalTrends> {
336  const mondays = weekStarts(now, weeks);
337  const [minutes, queue, merges, cost, backups, ci] = await Promise.all([
338    readMinutes(env.DB, mondays),
339    readQueue(env.DB, lastDays(now, TRENDS_QUEUE_DAYS)),
340    readMerges(env.DB, mondays),
341    readCost(env.DB, mondays),
342    readBackups(env, now),
343    readCi(env, now),
344  ]);
345  return {
346    generated: now.toISOString(),
347    weeks,
348    minutes: {
349      weeks: minutes.weeks,
350      truncated: minutes.truncated,
351      note: "An estimate: the administrator's own writes and job touches, sessions split at a 5-minute gap, plus 1 minute each. Reading the Portal without acting is not counted.",
352    },
353    queue,
354    merges,
355    cost: { weeks: cost },
356    backups,
357    ci,
358  };
359}