src/merge-resume.ts

Functions

Imports

It imports src/env.ts, src/github/client.ts, src/jobs-seat.ts, src/jobs-transition.ts, src/outcome-prs.ts, src/stale-jobs.ts. Imported by src/tick.ts.

Source

1import type { Env } from "./env";
2import { ghFetch, parsePrUrl, resolveRepo } from "./github/client";
3import { autoResumeJob, MERGE_RESUME_ACTOR } from "./jobs-seat";
4import { readJob } from "./jobs-transition";
5import { prUrlsFromJob } from "./outcome-prs";
6import { nextStaleCache, readStaleCache, STALE_PRS_KEY, tallyPrStates, type PrState, type StalePrCache } from "./stale-jobs";
7
8// The stale-jobs step on the five-minute tick.
9// A merge is not completion, so this never closes anything: it returns a BLOCKED job to
10// its holder (or the queue) once EVERY pull request the job names has merged, with a
11// note per pull request. A pull request that is closed without merging, unreadable or
12// still open leaves the job blocked. A job is resumed by this step once: a job that
13// blocks again on the same merged pull requests is waiting on something else.
14//
15// Every state it reads is also written, as counts per job, to the stale:prs cache that
16// the stale view reads (src/stale-jobs.ts, D5), so that view costs no GitHub reads.
17
18const MAX_JOBS = 25;
19const MAX_PRS = 5;
20
21export interface MergeResumeReport {
22  looked: number;
23  resumed: string[];
24  note: string;
25  // Set when the stale:prs cache could not be read or written. The resumes stand.
26  cache_error?: string;
27}
28
29interface PrRead {
30  state: PrState;
31  number: number;
32  sha: string;
33}
34
35async function prState(env: Env, namespace: string, url: string): Promise<PrRead> {
36  const parsed = parsePrUrl(url);
37  if (!parsed) return { state: "unread", number: 0, sha: "" };
38  const unread: PrRead = { state: "unread", number: parsed.number, sha: "" };
39  try {
40    const { owner, repo } = await resolveRepo(env, namespace, `${parsed.owner}/${parsed.repo}`);
41    const resp = await ghFetch(env, owner, repo, `/repos/${owner}/${repo}/pulls/${parsed.number}`);
42    if (!resp.ok) return unread;
43    const pr = (await resp.json()) as { state?: string; merged?: boolean; merge_commit_sha?: string | null };
44    if (pr.merged === true) return { state: "merged", number: parsed.number, sha: (pr.merge_commit_sha ?? "").slice(0, 7) };
45    if (pr.state === "closed") return { state: "closed", number: parsed.number, sha: "" };
46    if (pr.state === "open") return { state: "open", number: parsed.number, sha: "" };
47    return unread;
48  } catch {
49    return unread;
50  }
51}
52
53// Replaces each read job's entry and drops every job no longer blocked, including one
54// this tick just resumed. Returns the problem, or null.
55async function writeStaleCache(env: Env, reads: StalePrCache): Promise<string | null> {
56  try {
57    const prior = await readStaleCache(env.APP_KV);
58    const ids = [...new Set([...Object.keys(prior.cache), ...Object.keys(reads)])];
59    if (ids.length === 0 && prior.problem === null) return null;
60    const { results } = await env.DB.prepare(
61      `SELECT id FROM jobs WHERE status = 'blocked' AND id IN (SELECT value FROM json_each(?1))`
62    )
63      .bind(JSON.stringify(ids))
64      .all<{ id: string }>();
65    const stillBlocked = new Set((results ?? []).map((r) => r.id));
66    await env.APP_KV.put(STALE_PRS_KEY, JSON.stringify(nextStaleCache(prior.cache, reads, stillBlocked)));
67    return prior.problem;
68  } catch (err) {
69    return err instanceof Error ? err.message : String(err);
70  }
71}
72
73export async function mergeResumeTick(env: Env, now: Date): Promise<MergeResumeReport> {
74  const blocked = await env.DB.prepare(
75    `SELECT id FROM jobs WHERE status = 'blocked'
76       AND NOT EXISTS (SELECT 1 FROM job_touches t WHERE t.job_id = jobs.id AND t.actor = ?1)
77     ORDER BY updated_at LIMIT ?2`
78  )
79    .bind(MERGE_RESUME_ACTOR, MAX_JOBS)
80    .all<{ id: string }>();
81  const resumed: string[] = [];
82  const reads: StalePrCache = {};
83  const at = now.toISOString();
84  let looked = 0;
85  for (const { id } of blocked.results ?? []) {
86    const job = await readJob(env.DB, id);
87    if (!job) continue;
88    const urls = prUrlsFromJob(job).slice(0, MAX_PRS + 1);
89    if (urls.length === 0 || urls.length > MAX_PRS) continue;
90    looked++;
91    const facts = await Promise.all(urls.map((u) => prState(env, job.namespace, u)));
92    reads[id] = tallyPrStates(
93      facts.map((f) => f.state),
94      at
95    );
96    if (facts.some((f) => f.state !== "merged")) continue;
97    const note = facts
98      .map((f) => `PR #${f.number} merged${f.sha ? ` at ${f.sha}` : ""}; confirm the deploy and complete with your claim`)
99      .join("\n");
100    const result = await autoResumeJob(env, now, job, note);
101    if (result.resumed) resumed.push(`${id}(${result.to})`);
102  }
103  const cacheError = await writeStaleCache(env, reads);
104  const note =
105    `looked at ${looked} blocked job(s) naming pull requests, resumed ${resumed.length}${resumed.length ? `: ${resumed.join(", ")}` : ""}` +
106    (cacheError ? `; the ${STALE_PRS_KEY} cache: ${cacheError}` : "");
107  return { looked, resumed, note, ...(cacheError ? { cache_error: cacheError } : {}) };
108}