src/push.ts

Functions

Imports

It imports src/auto-merge-tick.ts, src/digest.ts, src/env.ts, src/http.ts, src/jobs-schema.ts, src/merge-safety.ts, src/ops-feed.ts, src/ops-types/views.ts, src/portal-actions.ts, src/portal-auth.ts, src/portal-host.ts, src/store-guards.ts, src/time.ts, src/web-push.ts. Imported by src/routes.ts, src/tick.ts.

Source

1import type { Env } from "./env";
2import { AWAITING_SEAT_KEY, type AwaitingSeat } from "./auto-merge-tick";
3import { RETRY_CAP_REASON } from "./jobs-schema";
4import { PORTAL_HOST } from "./portal-host";
5import { gateActionRequest } from "./portal-actions";
6import { portalGate } from "./portal-auth";
7import { OPS_RETURN_TO } from "./ops-feed";
8import type { PortalPush } from "./ops-types/views";
9import { readPaused } from "./merge-safety";
10import { auditStatement } from "./store-guards";
11import { digestState } from "./digest";
12import { readVapidKey, sendPush, subscriptionProblem, type PushSubscriptionKeys } from "./web-push";
13import { jsonResponse as json, textResponse as text } from "./http";
14import { HOUR_MS, DAY_MS } from "./time";
15
16// PHONE ALERTS (docs/design/design-merge-pipeline.md section 5, PR 7; Dustin's DECIDE 5
17// and 6 of 2026-10-09). Web Push to the devices the administrator turned on in the
18// Portal's Settings, for exactly six things:
19//   1. a `typed` pull request (a money path) waiting more than 2 hours;
20//   2. a rollback (deploy_outcomes: rolled_back or rollback_failed);
21//   3. a job at the correction cap (blocked with the retry-cap reason);
22//   4. a question over 4 hours old on a job with priority 70 or more;
23//   5. auto-merge paused for a namespace by the safety shutoff (src/merge-safety.ts);
24//   6. a pull request labelled visual waiting for Dustin to look at it (src/visual-pr.ts),
25//      once per head, so a new push asks again (job_0368576da167).
26// Everything else waits for the digest.
27//
28// No alert carries an Approve button. The button was for `approve`-class pull requests,
29// and no trigger carries one: a typed PR is approved only by typing its number (design
30// section 8), a visual PR is approved after looking at its preview in the Portal, and the
31// other four are not pull requests. So there is no
32// /ops/approve route: a tap opens the Portal, where every answer is given. An alert names
33// the repo, the pull request or job id and the class, never a title.
34//
35// Each item is sent once: what was sent is kept in KV with the time, for 30 days. A typed
36// PR's 2 hours are counted from the first tick that saw it typed, kept there too, since
37// the awaiting-seat set is rewritten whole on every tick. At most MAX_PER_TICK items go
38// out per tick, so turning a device on does not empty a backlog onto it at once.
39
40const STATE_KEY = "push:state";
41const SENT_TTL_MS = 30 * DAY_MS;
42const TYPED_WAIT_MS = 2 * HOUR_MS;
43const QUESTION_AGE_MS = 4 * HOUR_MS;
44const QUESTION_PRIORITY = 70;
45// A rollback or a cap older than this when first seen is history, not an alert.
46const RECENT_MS = DAY_MS;
47const MAX_PER_TICK = 3;
48const MAX_DEVICES = 10;
49const PORTAL_URL = `https://${PORTAL_HOST}/`;
50
51// Where a tap lands: the item itself on the Portal, not its home page (the UX audit's M2).
52// A pull request's card on Pull requests carries the anchor the dashboard builds
53// (dashboard/src/lib/pr-actions.ts, prAnchor); a job opens in its drawer on Jobs.
54const portalPath = (path: string) => `https://${PORTAL_HOST}${path}`;
55const prUrl = (repo: string, number: number) => portalPath(`/prs#pr-${repo.toLowerCase().replace(/[^a-z0-9]+/g, "-")}-${number}`);
56const jobUrl = (id: string) => portalPath(`/queue/job/${encodeURIComponent(id)}`);
57const DEPLOYS_URL = portalPath("/health?tab=deploys");
58const AUTO_MERGE_URL = portalPath("/prs#auto-merge");
59
60type UrgentKind = "typed-pr" | "rollback" | "correction-cap" | "question" | "automerge-paused" | "visual-pr";
61
62interface Urgent {
63  key: string;
64  kind: UrgentKind;
65  namespace: string;
66  title: string;
67  body: string;
68  /** The Portal page that shows this item. */
69  url: string;
70}
71
72interface PushState {
73  typed_seen: Record<string, string>;
74  sent: Record<string, string>;
75}
76
77/** What a device receives, decrypted by the Portal's service worker. No actions: a tap
78 *  opens the url. */
79export interface PushPayload {
80  title: string;
81  body: string;
82  url: string;
83  tag: string;
84}
85
86async function readState(env: Env): Promise<PushState> {
87  const raw = await env.APP_KV.get(STATE_KEY);
88  const parsed = raw ? (JSON.parse(raw) as Partial<PushState>) : {};
89  return { typed_seen: parsed.typed_seen ?? {}, sent: parsed.sent ?? {} };
90}
91
92const hours = (ms: number) => `${Math.floor(ms / HOUR_MS)}h`;
93
94/** The six triggers, from what the tick, the watcher and the queue already keep. Reads
95 *  only; `typedSeen` is updated in place with the typed PRs seen now. */
96async function gatherUrgent(env: Env, now: Date, typedSeen: Record<string, string>): Promise<Urgent[]> {
97  const out: Urgent[] = [];
98  const t = now.getTime();
99
100  // 1. Typed pull requests, from the awaiting-seat set the auto-merge tick writes.
101  const raw = await env.APP_KV.get(AWAITING_SEAT_KEY);
102  const awaiting = raw ? (JSON.parse(raw) as AwaitingSeat[]) : [];
103  const typedNow = new Set<string>();
104  for (const a of Array.isArray(awaiting) ? awaiting : []) {
105    if (a.class !== "typed") continue;
106    const id = `${a.repo}#${a.number}`;
107    typedNow.add(id);
108    const since = typedSeen[id] ?? now.toISOString();
109    typedSeen[id] = since;
110    const waited = t - Date.parse(since);
111    if (waited > TYPED_WAIT_MS) {
112      out.push({ key: `typed:${id}`, kind: "typed-pr", namespace: a.namespace, title: "Typed approval waiting", body: `${id}, class typed (money path), waiting ${hours(waited)}. Approve it by typing its number.`, url: prUrl(a.repo, a.number) });
113    }
114  }
115  for (const id of Object.keys(typedSeen)) if (!typedNow.has(id)) delete typedSeen[id];
116
117  // 6. Visual pull requests, from the same set, sent at once: a look is quick, and the
118  // key carries the head, so each new push asks again.
119  for (const a of Array.isArray(awaiting) ? awaiting : []) {
120    if (!a.visual) continue;
121    const id = `${a.repo}#${a.number}`;
122    out.push({
123      key: `visual:${id}:${a.head_sha ?? a.at}`,
124      kind: "visual-pr",
125      namespace: a.namespace,
126      title: "Visual change to look at",
127      body: `${id} changes what people see. Open it for its preview and screenshots, then approve it there.`,
128      url: prUrl(a.repo, a.number),
129    });
130  }
131
132  // 2. Rollbacks the watcher recorded.
133  const since = new Date(t - RECENT_MS).toISOString().replace("T", " ").slice(0, 19);
134  const rolled = await env.DB.prepare(
135    `SELECT repo, namespace, run_id, head_sha, outcome FROM deploy_outcomes
136     WHERE outcome IN ('rolled_back', 'rollback_failed') AND recorded_at >= ?1 ORDER BY recorded_at LIMIT 20`
137  )
138    .bind(since)
139    .all<{ repo: string; namespace: string; run_id: number; head_sha: string; outcome: string }>();
140  for (const r of rolled.results ?? []) {
141    const failed = r.outcome === "rollback_failed";
142    out.push({
143      key: `rollback:${r.repo}:${r.run_id}`,
144      kind: "rollback",
145      namespace: r.namespace,
146      title: failed ? "Rollback failed" : "Deploy rolled back",
147      body: `${r.repo} at ${r.head_sha.slice(0, 7)}: ${failed ? "the live gate failed and the rollback did not take effect" : "the live gate failed and the deploy was rolled back"}.`,
148      url: DEPLOYS_URL,
149    });
150  }
151
152  // 3 and 4. Blocked jobs: at the correction cap, or a question on a priority job.
153  const blocked = await env.DB.prepare(
154    `SELECT id, namespace, priority, result_summary, updated_at FROM jobs WHERE status = 'blocked' ORDER BY updated_at LIMIT 200`
155  ).all<{ id: string; namespace: string; priority: number; result_summary: string | null; updated_at: string }>();
156  for (const j of blocked.results ?? []) {
157    const summary = j.result_summary ?? "";
158    const age = t - Date.parse(j.updated_at);
159    if (summary.startsWith(RETRY_CAP_REASON) && age <= RECENT_MS) {
160      out.push({ key: `cap:${j.id}:${j.updated_at}`, kind: "correction-cap", namespace: j.namespace, title: "Correction cap reached", body: `${j.namespace} job ${j.id}: the reviewer and the driver did not converge. It needs your call.`, url: jobUrl(j.id) });
161    } else if (summary.startsWith("QUESTION:") && j.priority >= QUESTION_PRIORITY && age > QUESTION_AGE_MS) {
162      out.push({ key: `question:${j.id}:${j.updated_at}`, kind: "question", namespace: j.namespace, title: "Question waiting", body: `${j.namespace} job ${j.id}, priority ${j.priority}, asked ${hours(age)} ago.`, url: jobUrl(j.id) });
163    }
164  }
165
166  // 5. Namespaces the safety shutoff paused. An unreadable value alerts nothing: the
167  // tick already holds every namespace then (src/auto-merge-tick.ts) and logs it.
168  try {
169    for (const [namespace, p] of Object.entries(await readPaused(env.APP_KV))) {
170      out.push({ key: `paused:${namespace}:${p.since}`, kind: "automerge-paused", namespace, title: "Auto-merge paused", body: `${namespace}: ${p.reason}. Unattended merges stop until you resume it with the automerge_resume control (chat, capsid_admin_run_controls) and a reason.`, url: AUTO_MERGE_URL });
171    }
172  } catch {
173    // See above.
174  }
175  return out;
176}
177
178interface DeviceRow extends PushSubscriptionKeys {
179  id: number;
180}
181
182export interface PushReport {
183  ran: boolean;
184  note: string;
185  sent: string[];
186}
187
188/** One pass: what is urgent and not yet sent goes to every active device. */
189export async function pushTick(env: Env, now: Date, fetcher: typeof fetch = fetch): Promise<PushReport> {
190  const vapid = readVapidKey(env.VAPID_PRIVATE_KEY);
191  if (typeof vapid === "string") return { ran: false, note: vapid, sent: [] };
192  const devices = await env.DB.prepare(`SELECT id, endpoint, p256dh, auth FROM push_subscriptions WHERE status = 'active' ORDER BY id LIMIT ?1`)
193    .bind(MAX_DEVICES)
194    .all<DeviceRow>();
195  const live = devices.results ?? [];
196  if (live.length === 0) return { ran: false, note: "no device has phone alerts on", sent: [] };
197
198  const state = await readState(env);
199  const before = JSON.stringify(state);
200  const urgent = (await gatherUrgent(env, now, state.typed_seen)).filter((u) => !state.sent[u.key]);
201  const due = urgent.slice(0, MAX_PER_TICK);
202  const subject = env.ADMIN_EMAIL ? `mailto:${env.ADMIN_EMAIL}` : `https://${PORTAL_HOST}`;
203  const at = now.toISOString();
204  const sent: string[] = [];
205  const statements: D1PreparedStatement[] = [];
206  const gone = new Set<number>();
207
208  for (const u of due) {
209    const payload: PushPayload = { title: u.title, body: u.body, url: u.url, tag: u.key };
210    let delivered = 0;
211    for (const d of live) {
212      if (gone.has(d.id)) continue;
213      const res = await sendPush(vapid, subject, d, payload, now, fetcher);
214      if (res.gone) {
215        gone.add(d.id);
216        statements.push(
217          env.DB.prepare(`UPDATE push_subscriptions SET status = 'gone', closed_at = ?2, last_status = ?3 WHERE id = ?1 AND status = 'active' RETURNING id`).bind(d.id, at, res.status),
218          auditStatement(env.DB, "capsid:push", "push-device-gone", null, null, { id: d.id, status: res.status })
219        );
220        continue;
221      }
222      statements.push(env.DB.prepare(`UPDATE push_subscriptions SET last_sent_at = ?2, last_status = ?3 WHERE id = ?1 AND status = 'active'`).bind(d.id, at, res.status));
223      if (res.status >= 200 && res.status < 300) delivered++;
224    }
225    if (delivered > 0) {
226      state.sent[u.key] = at;
227      sent.push(u.key);
228      statements.push(auditStatement(env.DB, "capsid:push", "push-sent", u.namespace, null, { kind: u.kind, key: u.key, devices: delivered }));
229    }
230  }
231
232  for (const [k, v] of Object.entries(state.sent)) if (now.getTime() - Date.parse(v) > SENT_TTL_MS) delete state.sent[k];
233  if (statements.length) await env.DB.batch(statements);
234  if (JSON.stringify(state) !== before) await env.APP_KV.put(STATE_KEY, JSON.stringify(state));
235
236  if (due.length === 0) return { ran: false, note: "nothing urgent", sent };
237  const failed = due.length - sent.length;
238  const later = urgent.length - due.length;
239  return {
240    ran: true,
241    note: `sent ${sent.length} alert(s): ${sent.join(", ") || "none"}${failed ? `; ${failed} reached no device and will be tried again` : ""}${later ? `; ${later} more next tick` : ""}${gone.size ? `; ${gone.size} device(s) gone` : ""}`,
242    sent,
243  };
244}
245
246// ---------------------------------------------------------------------------
247// The Portal's route: GET lists, POST turns a device on or off or sends a test.
248
249export const PORTAL_PUSH_PATH = "/portal/api/push";
250
251async function listDevices(env: Env, now: Date): Promise<PortalPush> {
252  const vapid = readVapidKey(env.VAPID_PRIVATE_KEY);
253  const rows = await env.DB.prepare(
254    `SELECT id, label, created_at, last_sent_at, last_status FROM push_subscriptions WHERE status = 'active' ORDER BY id LIMIT ?1`
255  )
256    .bind(MAX_DEVICES)
257    .all<PortalPush["devices"][number]>();
258  return {
259    generated: now.toISOString(),
260    public_key: typeof vapid === "string" ? null : vapid.publicKey,
261    problem: typeof vapid === "string" ? vapid : null,
262    devices: rows.results ?? [],
263    digest: await digestState(env),
264  };
265}
266
267/** GET /portal/api/push: the VAPID public key and the devices with alerts on. Never an
268 *  endpoint: it is a URL anyone could send to. */
269export async function handlePortalPushGet(request: Request, env: Env, now: Date = new Date()): Promise<Response> {
270  const gate = await portalGate(request, env, now, OPS_RETURN_TO);
271  if (!gate.ok) return gate.response;
272  return json(await listDevices(env, now));
273}
274
275/** POST /portal/api/push: { op: "on", subscription, label } | { op: "off", id } | { op: "test" }.
276 *  Behind the actions' checks: the session, same-origin, the body cap and CSRF. */
277export async function handlePortalPushPost(request: Request, env: Env, now: Date = new Date(), fetcher: typeof fetch = fetch): Promise<Response> {
278  const gated = await gateActionRequest(request, env, now);
279  if (!gated.ok) return gated.response;
280  const actor = `access:${gated.email}`;
281  const body = gated.body;
282  const at = now.toISOString();
283
284  if (body.op === "on") {
285    const vapid = readVapidKey(env.VAPID_PRIVATE_KEY);
286    if (typeof vapid === "string") return text(vapid, 503);
287    const s = body.subscription as { endpoint?: unknown; keys?: { p256dh?: unknown; auth?: unknown } } | undefined;
288    const sub = { endpoint: s?.endpoint, p256dh: s?.keys?.p256dh, auth: s?.keys?.auth } as Partial<PushSubscriptionKeys>;
289    const problem = subscriptionProblem(sub);
290    if (problem) return text(`not turned on: ${problem}`, 400);
291    const label = typeof body.label === "string" ? body.label.slice(0, 80) : null;
292    const count = await env.DB.prepare(`SELECT COUNT(*) AS n FROM push_subscriptions WHERE status = 'active'`).first<{ n: number }>();
293    const existing = await env.DB.prepare(`SELECT id FROM push_subscriptions WHERE endpoint = ?1 AND status = 'active'`).bind(sub.endpoint).first<{ id: number }>();
294    if (!existing && (count?.n ?? 0) >= MAX_DEVICES) return text(`not turned on: ${MAX_DEVICES} devices already have alerts on. Turn one off first.`, 409);
295    const host = new URL(sub.endpoint!).hostname;
296    if (!existing) {
297      await env.DB.batch([
298        env.DB.prepare(
299          `INSERT INTO push_subscriptions (endpoint, p256dh, auth, label, status, created_by, created_at) VALUES (?1, ?2, ?3, ?4, 'active', ?5, ?6)`
300        ).bind(sub.endpoint, sub.p256dh, sub.auth, label, actor, at),
301        auditStatement(env.DB, actor, "push-device-on", null, null, { label, push_service: host }),
302      ]);
303    }
304    return json(await listDevices(env, now));
305  }
306
307  if (body.op === "off") {
308    const id = Number(body.id);
309    if (!Number.isInteger(id) || id < 1) return text("op off needs the device's id.", 400);
310    const [moved] = await env.DB.batch([
311      env.DB.prepare(`UPDATE push_subscriptions SET status = 'off', closed_at = ?2 WHERE id = ?1 AND status = 'active' RETURNING id`).bind(id, at),
312    ]);
313    if (!moved?.results?.length) return text(`device ${id} has no alerts on: it was turned off already or is gone.`, 409);
314    await env.DB.batch([auditStatement(env.DB, actor, "push-device-off", null, null, { id })]);
315    return json(await listDevices(env, now));
316  }
317
318  if (body.op === "test") {
319    const vapid = readVapidKey(env.VAPID_PRIVATE_KEY);
320    if (typeof vapid === "string") return text(vapid, 503);
321    const rows = await env.DB.prepare(`SELECT id, endpoint, p256dh, auth FROM push_subscriptions WHERE status = 'active' ORDER BY id LIMIT ?1`).bind(MAX_DEVICES).all<DeviceRow>();
322    const subject = env.ADMIN_EMAIL ? `mailto:${env.ADMIN_EMAIL}` : `https://${PORTAL_HOST}`;
323    const payload: PushPayload = { title: "Capsid test alert", body: "Phone alerts reach this device.", url: PORTAL_URL, tag: "test" };
324    const results = [];
325    for (const d of rows.results ?? []) results.push({ id: d.id, status: (await sendPush(vapid, subject, d, payload, now, fetcher)).status });
326    await env.DB.batch([auditStatement(env.DB, actor, "push-test", null, null, { results })]);
327    return json({ ...(await listDevices(env, now)), results });
328  }
329
330  return text("op must be on, off or test.", 400);
331}