diff --git a/cloudflare/GITHUB.md b/cloudflare/GITHUB.md index f1ca7cfc..af7f49e6 100644 --- a/cloudflare/GITHUB.md +++ b/cloudflare/GITHUB.md @@ -313,8 +313,8 @@ The jobs/funding quests own live eligibility checks and contribution execution. ## Project state and activity An enrolled project's setup page links to `/projects/github///state`. -A maintainer explicitly refreshes observations with its same-origin form. The -service rereads the default branch at an exact commit using the preview's shared +Webhooks reread observations in the background (see below), and a maintainer +can refresh them explicitly with its same-origin form. Each read rereads the default branch at an exact commit using the preview's shared snapshot reader and validates the Quest tree with the pure core. The authenticated board and nested quest pages use the existing `Project` seam. Source links retain the exact commit; live project state never replaces GitHub as authority. @@ -333,6 +333,31 @@ This deliberately avoids GitHub's bounded, delayed activity feed; GitHub does [not automatically redeliver failed webhooks](https://docs.github.com/en/webhooks/using-webhooks/handling-failed-webhook-deliveries). There is no scheduled background poll in this stage. +Each webhook that names an issue or PR (a comment on a PR names the PR by its +number), or that invalidates candidates (below), asks for a background read of its enabled project once its delivery commits. +The read runs after the response (`waitUntil`) with the App's read-only, +repository-scoped installation token, and is the same read as a refresh: one +page of each pass, plus every entity a webhook named since it was last read, +fetched by number in the same GraphQL query. A new issue or PR therefore shows +its title and state after one read however far the history pass is from it, +and a closed one stops showing as open. A lease in `github_project_state` lets +one read run at a time per project: it answers every request made before it +started, requests made during it get one more read, and duplicate deliveries +never reach it, so a burst costs one or two reads rather than one each. A +per-entity webhook count lets a read clear only the updates it saw. One +invocation reads at most three rounds, to stay within a Worker's subrequest +budget; a request left after them waits for the next webhook or a refresh. +A read overtaken by another read or by a webhook's revision bump reads again. +Paused and disconnected projects are not read. A failed or rate-limited read is +recorded and shown next to the refresh fallback, and is not retried: only a +request made during it gets another round, and otherwise the next webhook or a +refresh reads again. Checks, statuses and other pushes name no entity and +change no candidate, so they cost no read. + +The inbox also treats a PR whose row still says open as closed once a +complete, current occupancy pass no longer lists it, so a missed close webhook +corrects itself on the next read. + A single GraphQL activity query reads up to 50 issues, 50 PRs, 100 open PRs and 100 branch claims, keeping external request count independent of entity/file counts. Bulk D1 writes use `json_each`, avoiding a query per entity or SQLite's parameter @@ -340,10 +365,10 @@ limit. A revision comparison commits each page and cursor atomically; concurrent refreshes or an intervening webhook cannot publish over a newer observation. Only a PR opening, closing, reopening or receiving new commits, a `quest/` branch being created or deleted, or a push to the default branch invalidates candidates -and restarts the occupancy pass; new work waits for the next refresh, so a job +and restarts the occupancy pass; new work waits for the next read, so a job binds the PR head and tree the maintainer saw. Checks, statuses, comments, -reviews and other pushes wait for the history pass, so a busy repository neither -withholds candidates nor races a refresh. Installation tokens remain read-only, +reviews and other pushes do not invalidate candidates, so a busy repository +neither withholds candidates nor races a refresh. Installation tokens remain read-only, repository-scoped, and revoked after use. Durable activity items use the GitHub repository ID, entity kind and full database @@ -365,7 +390,10 @@ Claimed on the board. Runner/run associations belong to job coordination. Every cached read and refresh proves current maintainer authority, live repository identity/public visibility, installation access, and enrollment for that -installation. Refresh rechecks those boundaries after provider reads. Paused +installation. Refresh rechecks those boundaries after provider reads. A +background read has no user, so it checks the same boundaries except maintainer +authority; it publishes observations only, and grants or caches no authority. +Every page and action still rechecks the maintainer acting on it. Paused projects remain readable but have no candidates. Disconnect cascades cached snapshots and activity; explicitly enrolling a replacement installation clears prior observations. No positive authority cache is created. Candidate data is an diff --git a/cloudflare/README.md b/cloudflare/README.md index d08682c3..6370227b 100644 --- a/cloudflare/README.md +++ b/cloudflare/README.md @@ -172,9 +172,11 @@ progress, and their projects. Each item is one job target (`issues/1`, `pulls/2`, a quest path), so an issue, its planning job, and that job's questions stay one item, and webhook updates and reconnects never add another prompt. Agents waiting on an answer come first, then failed jobs, new pull -requests, new issues, and ready quests. An item offers only actions that -would work: none on a paused project, and only a refresh while availability -is pending. Every action is a plain form post to the job routes, which +requests, new issues, and ready quests. Each webhook rereads its project in the +background, so a new issue or PR arrives with its title, ready to act on. An +item offers only actions that would work: none on a paused project, and only a +refresh while availability is pending or a background read failed, with the +failure's reason. Every action is a plain form post to the job routes, which recheck authority and move on to the next active decision (or the empty state) with a short notice such as "Answer recorded"; the item acted on stays reachable from its list with its earlier answers. A refused action says why diff --git a/cloudflare/dev/github.ts b/cloudflare/dev/github.ts index 857705c4..de23f825 100644 --- a/cloudflare/dev/github.ts +++ b/cloudflare/dev/github.ts @@ -101,7 +101,17 @@ function graphql(repo: Repo, query: string) { })), ); } - return { data: { repository: result } }; + // Entities a webhook named, read by number as GitHub does: an issue number + // that is a pull request, or a missing one, is NOT_FOUND at its alias. + const errors: unknown[] = []; + for (const [, alias, field, number] of query.matchAll(/(s\d+):(issue|pullRequest)\(number:(\d+)\)/g)) { + const pull = repo.pulls.find((p) => p.number === Number(number)); + const issue = repo.issues.find((i) => i.number === Number(number)); + const found = field === "pullRequest" ? pull && pullNode(repo, pull) : issue && entity(repo, issue); + result[alias!] = found ?? null; + if (!found) errors.push({ type: "NOT_FOUND", path: ["repository", alias] }); + } + return { data: { repository: result }, ...(errors.length ? { errors } : {}) }; } function tree(repo: Repo) { diff --git a/cloudflare/dev/seed.ts b/cloudflare/dev/seed.ts index ea880316..6de077f8 100644 --- a/cloudflare/dev/seed.ts +++ b/cloudflare/dev/seed.ts @@ -213,7 +213,7 @@ export async function seed(options: { fetch: Fetch; db: D1Database; world: World "pause archive", ); - // An issue opened since the refresh: its webhook leaves a pointer with no data yet. + // An issue opened since the refresh: its webhook rereads the project in the background. const pointer = { number: 9, title: "Sync stops after the laptop sleeps", author: "reporter" }; rocket.issues.push(pointer); const body = JSON.stringify({ diff --git a/cloudflare/migrations/0013_reconcile.sql b/cloudflare/migrations/0013_reconcile.sql new file mode 100644 index 00000000..4fc8bd05 --- /dev/null +++ b/cloudflare/migrations/0013_reconcile.sql @@ -0,0 +1,11 @@ +-- A webhook asks for a background read of its project. `requested` stays set +-- until a read starts, and `lease` (an expiry in ms) lets only one read run at +-- a time, so a burst of deliveries coalesces into one read. `failure` keeps the +-- last background read's error until a read or refresh succeeds. +ALTER TABLE github_project_state ADD COLUMN requested INTEGER NOT NULL DEFAULT 0 CHECK(requested IN (0,1)); +ALTER TABLE github_project_state ADD COLUMN lease INTEGER; +ALTER TABLE github_project_state ADD COLUMN failure TEXT; +-- Counts the webhooks that named an entity since it was last read by number, so +-- a read clears only what it saw. +ALTER TABLE github_project_activity ADD COLUMN stale INTEGER NOT NULL DEFAULT 0; +UPDATE github_project_activity SET stale=1 WHERE data IS NULL AND missing=0; diff --git a/cloudflare/src/github/service.ts b/cloudflare/src/github/service.ts index b4ac07b2..2871cfa0 100644 --- a/cloudflare/src/github/service.ts +++ b/cloudflare/src/github/service.ts @@ -2,7 +2,15 @@ import { createHash, createPrivateKey, sign } from "node:crypto"; import { check, isQuest, parse, ready } from "quest/core"; import { osiApproved } from "../licenses"; import { positiveId } from "./webhook"; -import { activityQuery, type ActivityPage, type ProjectState, type Scan, reconcile, readState } from "./state"; +import { + activityQuery, + type ActivityPage, + type ProjectState, + type Scan, + reconcile, + readState, + staleEntities, +} from "./state"; export interface GitHubRepository { owner: string; @@ -11,6 +19,17 @@ export interface GitHubRepository { } type Permission = "metadata" | "contents" | "issues" | "pull_requests" | "checks" | "statuses"; type Permissions = Partial>; +/** What reading a project's observations needs. */ +const observing: Permissions = { + contents: "read", + issues: "read", + pull_requests: "read", + checks: "read", + statuses: "read", +}; +// Longer than a Worker keeps background work after its response, so a read +// that died with its Worker frees the project for the next webhook. +const observeLease = 60_000; interface Installation { id: number; app_id: number; @@ -54,6 +73,13 @@ export class GitHubError extends Error { super(message); } } + +/** Another read published, or a webhook invalidated, while this one read, so it must read again. */ +export class ProjectChanged extends GitHubError { + constructor() { + super("Project changed during refresh; refresh again"); + } +} /** A publication the bot cannot make, which needs a maintainer's decision rather than a retry. */ export class PublicationBlocked extends GitHubError {} /** A maintainer check that was answered with a denial, rather than one that could not complete. */ @@ -521,80 +547,150 @@ export class GitHubService { /** Observations only. Neither a refresh nor an arriving entity launches work. */ projectState(repo: GitHubRepository, userId: string, refresh = false): Promise { - return this.#authorized( - repo, - userId, - refresh - ? { contents: "read", issues: "read", pull_requests: "read", checks: "read", statuses: "read" } - : {}, - async (token, current, installation) => { - this.#public(current); - if ((await this.#state(repo, installation)) === null) - throw new GitHubError("Project is not enrolled for this installation"); - if (!refresh) return this.#observations(repo, installation); - await this.#env.DB.prepare("INSERT OR IGNORE INTO github_project_state(repository,epoch) VALUES(?,?)") - .bind(repo.id, crypto.randomUUID()) - .run(); - const row = await this.#env.DB.prepare( - "SELECT revision,scan,invalidated,epoch FROM github_project_state WHERE repository=?", - ) - .bind(repo.id) - .first<{ revision: number; scan: string | null; invalidated: number; epoch: string }>(); - const previous: Scan | null = row?.scan ? JSON.parse(row.scan) : null; - // History resumes through any webhook; only occupancy, which is - // small, restarts to observe a possibly opened, closed, or claimed quest. - const scan: Scan = { - history: - previous && !previous.history.complete - ? previous.history - : { id: crypto.randomUUID(), issues: null, pulls: null, complete: false }, - occupancy: - previous && !previous.occupancy.complete && row!.invalidated !== 1 - ? previous.occupancy - : { open: null, refs: null, pulls: [], claims: [], complete: false }, - }; - const snapshot = await this.#snapshot(token, repo, current); - const response = await this.#api<{ data?: { repository: ActivityPage | null }; errors?: unknown[] }>( - token, - "/graphql", - "POST", - { - query: activityQuery(scan), - variables: { - owner: repo.owner, - name: repo.name, - issues: scan.history.issues || null, - pulls: scan.history.pulls || null, - open: scan.occupancy.open || null, - refs: scan.occupancy.refs || null, - }, - }, - ); - if (response.errors?.length || !response.data?.repository) - throw new GitHubError("Project activity unavailable"); - // A rename, transfer, visibility change, revocation, or lost - // maintainer role during the scan must not publish fresh observations. - const latest = await this.#api(token, repositoryPath(repo)); - this.#repository(repo, latest); - this.#public(latest); - if (latest.default_branch !== current.default_branch) - throw new GitHubError("Default branch changed; refresh again"); - await this.#maintainer(token, repo, userId); - await this.#gate(installation); - if ((await this.#state(repo, installation)) === null) - throw new GitHubError("Project enrollment changed"); - await reconcile( - this.#env.DB, - repo, - snapshot, - scan, - response.data.repository, - row!.revision, - row!.epoch, - ); - return this.#observations(repo, installation); + return this.#authorized(repo, userId, refresh ? observing : {}, async (token, current, installation) => { + this.#public(current); + if ((await this.#state(repo, installation)) === null) + throw new GitHubError("Project is not enrolled for this installation"); + // A lost maintainer role during the read must not publish fresh observations. + if (refresh) + await this.#read(token, repo, current, installation, () => this.#maintainer(token, repo, userId)); + return this.#observations(repo, installation); + }); + } + + /** + * Read one page of a project's observations from GitHub and publish them, + * unless `recheck` fails or the repository, installation, or enrollment + * changed during the read. + */ + async #read( + token: string, + repo: GitHubRepository, + current: Repository, + installation: number, + recheck: () => Promise, + ) { + await this.#env.DB.prepare("INSERT OR IGNORE INTO github_project_state(repository,epoch) VALUES(?,?)") + .bind(repo.id, crypto.randomUUID()) + .run(); + const row = await this.#env.DB.prepare( + "SELECT revision,scan,invalidated,epoch FROM github_project_state WHERE repository=?", + ) + .bind(repo.id) + .first<{ revision: number; scan: string | null; invalidated: number; epoch: string }>(); + const previous: Scan | null = row?.scan ? JSON.parse(row.scan) : null; + // History resumes through any webhook; only occupancy, which is + // small, restarts to observe a possibly opened, closed, or claimed quest. + const scan: Scan = { + history: + previous && !previous.history.complete + ? previous.history + : { id: crypto.randomUUID(), issues: null, pulls: null, complete: false }, + occupancy: + previous && !previous.occupancy.complete && row!.invalidated !== 1 + ? previous.occupancy + : { open: null, refs: null, pulls: [], claims: [], complete: false }, + }; + // What webhooks named is read by number, so it is current after one read + // however far the history pass is from it. + const stale = await staleEntities(this.#env.DB, repo.id); + const snapshot = await this.#snapshot(token, repo, current); + const response = await this.#api<{ + data?: { repository: ActivityPage | null }; + errors?: { path?: unknown[] }[]; + }>(token, "/graphql", "POST", { + query: activityQuery(scan, stale), + variables: { + owner: repo.owner, + name: repo.name, + issues: scan.history.issues || null, + pulls: scan.history.pulls || null, + open: scan.occupancy.open || null, + refs: scan.occupancy.refs || null, }, + }); + // An entity read by number may be gone, such as a deleted issue (GitHub + // reports NOT_FOUND at its alias). Any other error is a partial result. + // Accepting any error at the alias keeps one unreadable entity from + // blocking every later read. + const partial = response.errors?.some( + (error) => + error.path?.length !== 2 || error.path[0] !== "repository" || !/^s\d+$/.test(String(error.path[1])), ); + if (partial || !response.data?.repository) throw new GitHubError("Project activity unavailable"); + // A rename, transfer, visibility change, or revocation during the read + // must not publish fresh observations. + const latest = await this.#api(token, repositoryPath(repo)); + this.#repository(repo, latest); + this.#public(latest); + if (latest.default_branch !== current.default_branch) + throw new GitHubError("Default branch changed; refresh again"); + await recheck(); + await this.#gate(installation); + if ((await this.#state(repo, installation)) === null) throw new GitHubError("Project enrollment changed"); + await reconcile(this.#env.DB, repo, snapshot, scan, response.data.repository, stale, row!.revision, row!.epoch); + } + + /** + * Reread an enabled project after a webhook asked for it, with the App's + * own token. One read runs at a time: it answers every request made before + * it started, and requests made during it get one more read, so a burst of + * deliveries costs one or two reads rather than one each. This publishes + * observations only. It checks no maintainer, so it neither grants nor + * caches authority; every page and action still rechecks whoever acts. A + * read that another read or a webhook overtook reads again. A failure is + * recorded for the inbox and not retried: only a request made during the + * failed read gets another round. Each invocation reads at most three + * rounds to stay within one Worker invocation's subrequest budget; a + * request left after them waits for the next webhook or a refresh. + */ + async observe(id: number) { + const db = this.#env.DB; + for (let round = 0; round < 3; round++) { + const now = Date.now(); + const lease = now + observeLease; + const claimed = await db + .prepare( + `UPDATE github_project_state SET requested=0,lease=? WHERE repository=? AND requested=1 AND (lease IS NULL OR lease(); + if (repo) + await this.#withRepository(repo, observing, async (token, current, installation) => { + this.#public(current); + if ((await this.#state(repo, installation)) !== "enabled") + throw new GitHubError("Project enrollment changed"); + await this.#read(token, repo, current, installation, async () => {}); + }); + } catch (error) { + // A manual refresh that started before the webhook, or a webhook's + // revision bump, can publish first and leave this read's entities stale. + if (error instanceof ProjectChanged) overtaken = true; + else { + // Provider errors carry no response bodies or secrets; anything else is a bug. + failure = error instanceof GitHubError ? error.message : "Unexpected error"; + if (!(error instanceof GitHubError)) + console.error({ event: "github_observe_failed", name: (error as Error)?.name }); + } + } + // An overtaken read, or more named entities than one read covers, asks for another round. + const more = overtaken || (!failure && (await staleEntities(db, id)).length > 0); + await db + .prepare( + "UPDATE github_project_state SET lease=NULL,failure=COALESCE(?,failure),requested=MAX(requested,?) WHERE repository=? AND lease=?", + ) + .bind(failure, more ? 1 : 0, id, lease) + .run(); + } } /** diff --git a/cloudflare/src/github/state.ts b/cloudflare/src/github/state.ts index 548fda0d..a0e0e41b 100644 --- a/cloudflare/src/github/state.ts +++ b/cloudflare/src/github/state.ts @@ -1,6 +1,6 @@ import { parse, ready } from "quest/core"; import type { Project, Snapshot } from "../board/project"; -import { GitHubError, type GitHubRepository } from "./service"; +import { GitHubError, type GitHubRepository, ProjectChanged } from "./service"; import { positiveId } from "./webhook"; type Cursor = string | null | false; @@ -39,7 +39,11 @@ export type ActivityPage = { pullRequests?: Connection; open?: Connection; refs?: Connection<{ name: string; prefix: string }>; + /** Stale entities read by number, in the order requested; null when GitHub no longer has one. */ + [stale: `s${number}`]: Entity | Pull | null | undefined; }; +/** An entity a webhook named since it was last read, and how many times. */ +export type Stale = { kind: "issue" | "pull"; id: string; number: number; stale: number }; export type Activity = { kind: "issue" | "pull"; id: string; @@ -56,13 +60,17 @@ export type ProjectState = { complete: boolean; observedAt: number; revision: number; + /** PRs the latest occupancy pass found open, or null until it completes after the last invalidating webhook. */ + open: string[] | null; + /** Why the last background read failed, until a read or refresh succeeds. */ + failure: string | null; }; const entityFields = "databaseId:fullDatabaseId number title state updatedAt author { login } comments { totalCount }"; const pullFields = `${entityFields} headRefName headRefOid isDraft reviewDecision mergeStateStatus reviews { totalCount } latestReviews(first:20) { nodes { state author {login} commit {oid} } pageInfo {hasNextPage} } commits(last:1) { nodes { commit { oid statusCheckRollup { state } } } }`; // Cursor variables are declared only for unfinished connections (GraphQL rejects // unused variables). Fixed page sizes bound subrequests, independent of tree size. -export function activityQuery({ history, occupancy }: Scan) { +export function activityQuery({ history, occupancy }: Scan, stale: Stale[] = []) { const fields: string[] = []; const variables = ["$owner: String!", "$name: String!"]; if (history.issues !== false) { @@ -89,6 +97,15 @@ export function activityQuery({ history, occupancy }: Scan) { 'refs(refPrefix:"refs/heads/quest/",first:100,after:$refs) { nodes { name prefix } pageInfo { hasNextPage endCursor } }', ); } + // Numbers come from validated webhook pointers and are interpolated as integers. + stale.forEach((entity, i) => { + if (!positiveId(entity.number)) throw new GitHubError("Invalid entity observation"); + fields.push( + entity.kind === "pull" + ? `s${i}:pullRequest(number:${entity.number}) { ${pullFields} }` + : `s${i}:issue(number:${entity.number}) { ${entityFields} }`, + ); + }); return `query(${variables.join(",")}) { repository(owner:$owner,name:$name) { ${fields.join("\n")} } }`; } @@ -109,12 +126,41 @@ export function associatedQuest(branch: string, paths: Set) { return paths.has(exact) ? exact : paths.has(epic) ? epic : null; } +function validate(kind: "issue" | "pull", item: Entity) { + if ( + typeof item.databaseId !== "string" || + !/^[1-9]\d*$/.test(item.databaseId) || + !positiveId(item.number) || + typeof item.title !== "string" || + !Number.isFinite(Date.parse(item.updatedAt)) + ) + throw new GitHubError("Invalid entity observation"); + if (kind === "pull") { + const pull = item as Pull; + if (!/^[a-f0-9]{40}$/.test(pull.headRefOid) || pull.commits.nodes.at(-1)?.commit.oid !== pull.headRefOid) + throw new GitHubError("Pull request head changed; refresh again"); + } +} + +/** Entities that webhooks named since they were last read by number, oldest first. */ +export async function staleEntities(db: D1Database, repository: number) { + return ( + await db + .prepare( + "SELECT kind,id,number,stale FROM github_project_activity WHERE repository=? AND stale>0 ORDER BY changedAt,id LIMIT 50", + ) + .bind(repository) + .all() + ).results; +} + export async function reconcile( db: D1Database, repo: GitHubRepository, snapshot: Snapshot, scan: Scan, page: ActivityPage, + stale: Stale[], revision: number, epoch: string, ) { @@ -145,7 +191,8 @@ export async function reconcile( updated.history.complete = updated.history.issues === false && updated.history.pulls === false; updated.occupancy.complete = updated.occupancy.open === false && updated.occupancy.refs === false; // Keyed by identity: an open PR can arrive from both PR connections at once. - const entities = new Map(); + // `stale` is the webhook count a read by number saw, so only it is cleared. + const entities = new Map(); for (const [kind, connection, cursor, open] of [ ["issue", page.issues, history.issues, false], ["pull", page.pullRequests, history.pulls, false], @@ -153,52 +200,54 @@ export async function reconcile( ] as const) { if (cursor === false) continue; for (const item of connection!.nodes) { - if ( - typeof item.databaseId !== "string" || - !/^[1-9]\d*$/.test(item.databaseId) || - !positiveId(item.number) || - typeof item.title !== "string" || - !Number.isFinite(Date.parse(item.updatedAt)) - ) - throw new GitHubError("Invalid entity observation"); - if (kind === "pull") { - const pull = item as Pull; - if ( - !/^[a-f0-9]{40}$/.test(pull.headRefOid) || - pull.commits.nodes.at(-1)?.commit.oid !== pull.headRefOid - ) - throw new GitHubError("Pull request head changed; refresh again"); - } - entities.set(`${kind}/${item.databaseId}`, { kind, item }); + validate(kind, item); + entities.set(`${kind}/${item.databaseId}`, { kind, item, stale: 0 }); if (open) updated.occupancy.pulls.push(item.databaseId); } } + // A read by number is the latest, so it replaces a page's copy. + const gone: Stale[] = []; + stale.forEach((entity, i) => { + const item = page[`s${i}`]; + if (item === undefined) throw new GitHubError("Incomplete project activity"); + if (item) validate(entity.kind, item); + // A number GitHub no longer resolves to this entity, such as a deleted issue. + if (!item || item.databaseId !== entity.id) gone.push(entity); + else entities.set(`${entity.kind}/${entity.id}`, { kind: entity.kind, item, stale: entity.stale }); + }); updated.occupancy.claims = [...new Set(updated.occupancy.claims)]; updated.occupancy.pulls = [...new Set(updated.occupancy.pulls)]; - // json_each keeps the transaction to three statements and a few bindings, + const current = "EXISTS(SELECT 1 FROM github_project_state WHERE repository=? AND revision=? AND epoch=?)"; + // json_each keeps the transaction to four statements and a few bindings, // within D1's free query budget and SQLite's bound-parameter limit. const statements = [ db .prepare( - `INSERT INTO github_project_activity(repository,kind,id,number,firstSeen,changedAt,data,seenIn) - SELECT ?,json_extract(value,'$.kind'),json_extract(value,'$.item.databaseId'),json_extract(value,'$.item.number'),?,?,json_extract(value,'$.item'),? - FROM json_each(?) WHERE EXISTS(SELECT 1 FROM github_project_state WHERE repository=? AND revision=? AND epoch=?) - ON CONFLICT(repository,kind,id) DO UPDATE SET number=excluded.number,changedAt=CASE WHEN github_project_activity.data IS NOT excluded.data THEN excluded.changedAt ELSE github_project_activity.changedAt END,data=excluded.data,seenIn=excluded.seenIn,missing=0`, + `INSERT INTO github_project_activity(repository,kind,id,number,firstSeen,changedAt,data,seenIn,stale) + SELECT ?,json_extract(value,'$.kind'),json_extract(value,'$.item.databaseId'),json_extract(value,'$.item.number'),?,?,json_extract(value,'$.item'),?,json_extract(value,'$.stale') + FROM json_each(?) WHERE ${current} + ON CONFLICT(repository,kind,id) DO UPDATE SET number=excluded.number,changedAt=CASE WHEN github_project_activity.data IS NOT excluded.data THEN excluded.changedAt ELSE github_project_activity.changedAt END,data=excluded.data,seenIn=excluded.seenIn,missing=0,stale=CASE WHEN excluded.stale>0 AND github_project_activity.stale=excluded.stale THEN 0 ELSE github_project_activity.stale END`, ) .bind(repo.id, now, now, history.id, JSON.stringify([...entities.values()]), repo.id, revision, epoch), db .prepare( - "UPDATE github_project_activity SET missing=1 WHERE repository=? AND (seenIn IS NULL OR seenIn<>?) AND ?=1 AND EXISTS(SELECT 1 FROM github_project_state WHERE repository=? AND revision=? AND epoch=?)", + `UPDATE github_project_activity SET missing=1,stale=CASE WHEN stale=json_extract(value,'$.stale') THEN 0 ELSE stale END + FROM json_each(?) WHERE repository=? AND kind=json_extract(value,'$.kind') AND github_project_activity.id=json_extract(value,'$.id') AND ${current}`, + ) + .bind(JSON.stringify(gone), repo.id, repo.id, revision, epoch), + db + .prepare( + `UPDATE github_project_activity SET missing=1 WHERE repository=? AND (seenIn IS NULL OR seenIn<>?) AND ?=1 AND ${current}`, ) .bind(repo.id, history.id, updated.history.complete ? 1 : 0, repo.id, revision, epoch), db .prepare( - "UPDATE github_project_state SET snapshot=?,scan=?,observedAt=?,invalidated=?,revision=revision+1 WHERE repository=? AND revision=? AND epoch=?", + "UPDATE github_project_state SET snapshot=?,scan=?,observedAt=?,invalidated=0,failure=NULL,revision=revision+1 WHERE repository=? AND revision=? AND epoch=?", ) - .bind(JSON.stringify(snapshot), JSON.stringify(updated), now, 0, repo.id, revision, epoch), + .bind(JSON.stringify(snapshot), JSON.stringify(updated), now, repo.id, revision, epoch), ]; const result = await db.batch(statements); - if (result.at(-1)!.meta.changes !== 1) throw new GitHubError("Project changed during refresh; refresh again"); + if (result.at(-1)!.meta.changes !== 1) throw new ProjectChanged(); } /** @@ -222,7 +271,7 @@ export async function readState(db: D1Database, repo: GitHubRepository): Promise const records = await db.batch([ db .prepare( - "SELECT snapshot,scan,observedAt,revision,invalidated FROM github_project_state WHERE repository=?", + "SELECT snapshot,scan,observedAt,revision,invalidated,failure FROM github_project_state WHERE repository=?", ) .bind(repo.id), db @@ -232,7 +281,14 @@ export async function readState(db: D1Database, repo: GitHubRepository): Promise .bind(repo.id), ]); const row = records[0]!.results[0] as - | { snapshot: string | null; scan: string | null; observedAt: number; revision: number; invalidated: number } + | { + snapshot: string | null; + scan: string | null; + observedAt: number; + revision: number; + invalidated: number; + failure: string | null; + } | undefined; if (!row?.snapshot || !row.scan) return null; const snapshot: Snapshot = JSON.parse(row.snapshot); @@ -265,10 +321,10 @@ export async function readState(db: D1Database, repo: GitHubRepository): Promise const occupied = new Set([...claims, ...changes.map((c) => c.quest)]); // Unknown until every open PR and quest branch is read since the last // webhook that could have opened, closed, or claimed a quest. - const available = - scan.occupancy.complete && row.invalidated === 0 - ? ready(snapshot.documents.map((d) => parse(d.path, d.content))).filter((path) => !occupied.has(path)) - : null; + const settled = scan.occupancy.complete && row.invalidated === 0; + const available = settled + ? ready(snapshot.documents.map((d) => parse(d.path, d.content))).filter((path) => !occupied.has(path)) + : null; const project: Project = { ...snapshot, name: `github/${repo.owner}/${repo.name}`, @@ -289,5 +345,7 @@ export async function readState(db: D1Database, repo: GitHubRepository): Promise complete: scan.history.complete, observedAt: row.observedAt, revision: row.revision, + open: settled ? scan.occupancy.pulls : null, + failure: row.failure, }; } diff --git a/cloudflare/src/github/webhook.ts b/cloudflare/src/github/webhook.ts index ac307382..adb4f181 100644 --- a/cloudflare/src/github/webhook.ts +++ b/cloudflare/src/github/webhook.ts @@ -1,5 +1,6 @@ import { Hono } from "hono"; import { bodyLimit } from "hono/body-limit"; +import { GitHubService } from "./service"; export function positiveId(value: unknown): value is number { return typeof value === "number" && Number.isSafeInteger(value) && value > 0; @@ -120,6 +121,8 @@ githubWebhooks.post("/webhooks/github", async (c) => { ).bind(`repositories_${action}`, installation, delivery, digest, event), ); } + // The statement that asks for a background read, if this delivery does. + let reads: number | null = null; if (observed.has(event)) { const pending = "EXISTS(SELECT 1 FROM github_deliveries WHERE id=? AND digest=? AND event=? AND processed=0)"; const enrolled = "EXISTS(SELECT 1 FROM github_projects WHERE id=? AND installation=?)"; @@ -129,12 +132,13 @@ githubWebhooks.post("/webhooks/github", async (c) => { const issue = payload.issue; const kind = pull ? "pull" : issue && !issue.pull_request ? "issue" : null; const entity = pull ?? issue; - if (kind && positiveId(entity.id) && positiveId(entity.number)) { + const named = kind && positiveId(entity.id) && positiveId(entity.number); + if (named) { statements.push( c.env.DB.prepare( - `INSERT INTO github_project_activity(repository,kind,id,number,firstSeen,changedAt) - SELECT ?,?,?,?,?,? WHERE ${pending} AND ${enrolled} - ON CONFLICT(repository,kind,id) DO UPDATE SET changedAt=excluded.changedAt`, + `INSERT INTO github_project_activity(repository,kind,id,number,firstSeen,changedAt,stale) + SELECT ?,?,?,?,?,?,1 WHERE ${pending} AND ${enrolled} + ON CONFLICT(repository,kind,id) DO UPDATE SET changedAt=excluded.changedAt,stale=github_project_activity.stale+1`, ).bind( repository, kind, @@ -150,12 +154,34 @@ githubWebhooks.post("/webhooks/github", async (c) => { ), ); } - if (invalidates(event, payload)) + // A comment on a PR carries the PR's number, but under its issue ID, so + // it marks the PR's row by number. An untracked PR waits for history. + const discussed = !pull && issue?.pull_request && positiveId(issue.number); + if (discussed) + statements.push( + c.env.DB.prepare( + `UPDATE github_project_activity SET changedAt=?,stale=stale+1 + WHERE repository=? AND kind='pull' AND number=? AND ${pending} AND ${enrolled}`, + ).bind(now, repository, issue.number, delivery, digest, event, repository, installation), + ); + const invalidating = invalidates(event, payload); + if (invalidating) statements.push( c.env.DB.prepare( `UPDATE github_project_state SET invalidated=1,revision=revision+1 WHERE repository=? AND ${pending} AND ${enrolled}`, ).bind(repository, delivery, digest, event, repository, installation), ); + // What the inbox shows changed: ask for a background read. CI and other + // pushes change neither, so they cost no read. + if (named || discussed || invalidating) { + reads = statements.length; + statements.push( + c.env.DB.prepare( + `INSERT INTO github_project_state(repository,epoch,requested) SELECT ?,?,1 WHERE ${pending} AND ${enrolled} + ON CONFLICT(repository) DO UPDATE SET requested=1`, + ).bind(repository, crypto.randomUUID(), delivery, digest, event, repository, installation), + ); + } } statements.push( @@ -177,7 +203,9 @@ githubWebhooks.post("/webhooks/github", async (c) => { if (winner?.digest !== digest || winner.event !== event) return c.json({ error: "Delivery conflict" }, 409); return c.json({ status: "duplicate" }); } - // Observations never approve or launch work. Trusted consumers reconcile + // Observations never approve or launch work. The background read reconciles // current GitHub state rather than treating delayed webhook bodies as facts. + if (reads !== null && committed[reads]!.meta.changes > 0) + c.executionCtx.waitUntil(new GitHubService(c.env).observe(repository)); return c.json({ status: "accepted" }, 202); }); diff --git a/cloudflare/src/inbox/model.ts b/cloudflare/src/inbox/model.ts index c19b9ef4..efad07e6 100644 --- a/cloudflare/src/inbox/model.ts +++ b/cloudflare/src/inbox/model.ts @@ -80,6 +80,9 @@ export function entities(project: Enrollment, state: ProjectState | null) { // A webhook's pointer has no data until a refresh reads it, but it is already news. if (activity.missing || (activity.data && activity.data.state !== "OPEN")) continue; const pull = activity.kind === "pull"; + // A PR the latest occupancy pass did not find open has closed since its + // row was read, even if no webhook said so. + if (pull && activity.data && state.open && !state.open.includes(activity.id)) continue; found.set(`${pull ? "pulls" : "issues"}/${activity.number}`, { kind: activity.kind, title: activity.data?.title ?? `${pull ? "Pull request" : "Issue"} #${activity.number}`, diff --git a/cloudflare/src/inbox/page.tsx b/cloudflare/src/inbox/page.tsx index 9c4633be..6e78500c 100644 --- a/cloudflare/src/inbox/page.tsx +++ b/cloudflare/src/inbox/page.tsx @@ -266,13 +266,16 @@ function AccountForm(props: { item: Item; runners: Runners }) { function blocked(item: Item, view: ProjectView | undefined) { if (item.project.state !== "enabled") return "Contributions are paused for this project."; const state = view?.state; - if (!state?.project.available) return "GitHub reported changes. Refresh to start new work from current state."; + // Webhooks reread the project in the background; a failed read says why. + const failed = state?.failure ? `The background read failed: ${state.failure}. ` : ""; + if (!state?.project.available) + return `${failed || "GitHub reported changes. "}Refresh to start new work from current state.`; // A retry continues its failed job's pull request, so that claim does not block it. const retry = item.need === "retry" ? (item.job?.pull ?? null) : null; if (item.kind === "quest" && !startable(state, item.target, retry)) return "This quest is no longer ready to start, such as when it is claimed or removed."; if ((item.kind === "issue" || item.kind === "pull") && !item.activity?.data) - return "Refresh to read this from GitHub."; + return `${failed}Refresh to read this from GitHub.`; return null; } @@ -455,6 +458,7 @@ function Project(props: { view: ProjectView }) { <> {project.state === "enabled" ? "Contributions enabled" : "Contributions paused"} {!state?.project.available && Refresh needed} + {state?.failure && Background read failed: {state.failure}} Quest board Activity diff --git a/cloudflare/src/projects.tsx b/cloudflare/src/projects.tsx index bcfc03ed..a6dce431 100644 --- a/cloudflare/src/projects.tsx +++ b/cloudflare/src/projects.tsx @@ -230,9 +230,10 @@ export function ProjectStatePage(props: { owner: string; repository: string; sta

- GitHub issues and pull requests are source material. Refreshing observes activity and does not start - work or approve a merge. + GitHub issues and pull requests are source material. Webhooks reread the project in the background; + refreshing does the same by hand. Neither starts work or approves a merge.

+ {state?.failure &&

The last background read failed: {state.failure}

} {state ? ( <>

diff --git a/cloudflare/test/inbox.test.ts b/cloudflare/test/inbox.test.ts index 32f82e68..94f2c255 100644 --- a/cloudflare/test/inbox.test.ts +++ b/cloudflare/test/inbox.test.ts @@ -148,10 +148,21 @@ describe("decision inbox", () => { expect(html).toMatch(/Fix<\/a>New/); // A webhook only points at the new issue; refreshing reads it, and no work starts before. expect(html).toContain("Issue #3"); - expect(thread(html)).toContain("GitHub reported changes"); + expect(thread(html)).toContain("Refresh to start new work"); expect(thread(html)).not.toContain("Plan this issue"); }); + it("drops a pull request the latest occupancy pass no longer found open", async () => { + expect(thread(await inbox({ project: "10", item: "pulls/2" }))).toContain("Iterate on this PR"); + // History has not reread it, so its row still says open. + await db + .prepare( + "UPDATE github_project_state SET scan=json_set(scan,'$.occupancy.pulls',json('[]')) WHERE repository=10", + ) + .run(); + expect(await inbox({ project: "10", item: "pulls/2" })).toContain("no longer open"); + }); + it("starts work bound to the revision on screen and shows its progress plainly", async () => { const stale = await submit(jobs, [ ["workflow", "iterate"], diff --git a/cloudflare/test/project-fixtures.ts b/cloudflare/test/project-fixtures.ts index af644c1f..aea41440 100644 --- a/cloudflare/test/project-fixtures.ts +++ b/cloudflare/test/project-fixtures.ts @@ -1,5 +1,6 @@ // A signed-in maintainer of an enrolled project with complete observations, // and a mocked GitHub that answers the requests authority checks make. +import { createExecutionContext, waitOnExecutionContext } from "cloudflare:test"; import { env } from "cloudflare:workers"; import { vi } from "vitest"; import { app } from "../src/index"; @@ -166,12 +167,13 @@ export async function maintainerOf() { return { user, cookie, github, mock }; } -/** Deliver a signed GitHub webhook. */ +/** Deliver a signed GitHub webhook, and finish the background read it asks for. */ export async function delivery( event: string, payload: unknown, id = crypto.randomUUID(), changes: Record = {}, + bindings: Env = env, ) { const body = typeof payload === "string" ? payload : JSON.stringify(payload); const key = await crypto.subtle.importKey( @@ -183,13 +185,17 @@ export async function delivery( ); const digest = await crypto.subtle.sign("HMAC", key, new TextEncoder().encode(body)); const signature = "sha256=" + Array.from(new Uint8Array(digest), (b) => b.toString(16).padStart(2, "0")).join(""); - return app.request( + const ctx = createExecutionContext(); + const response = await app.request( "http://localhost:8787/webhooks/github", { method: "POST", headers: { "X-GitHub-Event": event, "X-GitHub-Delivery": id, "X-Hub-Signature-256": signature, ...changes }, body, }, - env, + bindings, + ctx, ); + await waitOnExecutionContext(ctx); + return response; } diff --git a/cloudflare/test/projects.test.ts b/cloudflare/test/projects.test.ts index 86690e68..eb43b281 100644 --- a/cloudflare/test/projects.test.ts +++ b/cloudflare/test/projects.test.ts @@ -34,6 +34,8 @@ function github( options: { role?: string; activity?: ActivityPage; + /** What a read by number finds, by number; otherwise the activity page's entity with it. */ + entities?: Record; onActivity?: () => Promise; activityErrors?: boolean; repositoryName?: string; @@ -182,10 +184,21 @@ function github( const query = JSON.parse(String(init?.body)).query as string; if (!query.includes("object(oid:")) { await options.onActivity?.(); - return Response.json({ - data: { repository: options.activity ?? activity() }, - errors: options.activityErrors ? [{ message: "partial" }] : undefined, - }); + const page = options.activity ?? activity(); + const repository: Record = { ...page }; + const errors: unknown[] = options.activityErrors ? [{ message: "partial" }] : []; + for (const [, alias, field, number] of query.matchAll(/(s\d+):(issue|pullRequest)\(number:(\d+)\)/g)) { + const nodes = (field === "issue" ? [page.issues] : [page.pullRequests, page.open]).flatMap( + (c) => c?.nodes ?? [], + ); + const found = + options.entities && Number(number) in options.entities + ? options.entities[Number(number)] + : (nodes.find((node) => node.number === Number(number)) ?? null); + repository[alias!] = found; + if (!found) errors.push({ type: "NOT_FOUND", path: ["repository", alias] }); + } + return Response.json({ data: { repository }, errors: errors.length ? errors : undefined }); } const body = calls.at(-1)!.body; @@ -692,7 +705,7 @@ function issue(id = 100, number = 1) { } function pull(branch = "quest/first", id = 200) { return { - ...issue(id, 2), + ...issue(id, id - 198), headRefName: branch, headRefOid: commit, isDraft: false, @@ -813,8 +826,8 @@ describe("enrolled project observations", () => { ["status", {}], ["issues", { action: "opened", issue: { id: 102, number: 5 } }], ["issue_comment", { action: "created", issue: { id: 100, number: 1 } }], - ["pull_request", { action: "edited", pull_request: { id: 201, number: 2 } }], - ["pull_request_review", { action: "submitted", pull_request: { id: 201, number: 2 } }], + ["pull_request", { action: "edited", pull_request: { id: 201, number: 3 } }], + ["pull_request_review", { action: "submitted", pull_request: { id: 201, number: 3 } }], ["push", { ref: "refs/heads/human-branch" }], ["push", { ref: "refs/heads/quest/first", created: false, deleted: false }], ] as const; @@ -956,6 +969,249 @@ describe("enrolled project observations", () => { }); }); +describe("webhook reconciliation", () => { + // The webhook route reads GitHub through the global fetch. + function mocked(options: Parameters[0] = {}) { + const mock = github(options); + vi.spyOn(globalThis, "fetch").mockImplementation(mock.request as typeof fetch); + return mock; + } + const reads = (calls: ReturnType["calls"]) => + calls.filter((call) => call.path === "/graphql" && !call.body.query.includes("object(oid:")); + const opened = (id: number, number: number, id_ = crypto.randomUUID()) => + delivery("issues", { ...event, action: "opened", issue: { id, number } }, id_, {}, testEnv); + async function row(id: string) { + return env.DB.prepare("SELECT data,stale,missing FROM github_project_activity WHERE repository=10 AND id=?") + .bind(id) + .first<{ data: string | null; stale: number; missing: number }>(); + } + async function enrolled() { + await github().service.setContributions(repo, "enrollment-test", "enabled"); + await github().service.projectState(repo, "enrollment-test", true); + } + + it("reads a new issue by number with the App's read-only token and no maintainer check", async () => { + await enrolled(); + const { calls } = mocked({ entities: { 5: { ...issue(102, 5), title: "New idea" } } }); + expect((await opened(102, 5)).status).toBe(202); + expect(reads(calls)).toHaveLength(1); + expect(reads(calls)[0]!.body.query).toContain("s0:issue(number:5)"); + expect(calls.some((call) => call.path.endsWith("/permission"))).toBe(false); + expect(calls.find((call) => call.path.endsWith("/access_tokens"))!.body.permissions).toEqual({ + metadata: "read", + contents: "read", + issues: "read", + pull_requests: "read", + checks: "read", + statuses: "read", + }); + expect(calls.at(-1)?.method).toBe("DELETE"); + expect(await row("102")).toMatchObject({ stale: 0, missing: 0 }); + const state = await github().service.projectState(repo, "enrollment-test"); + expect(state!.project.issues).toContainEqual({ path: "issues/5", title: "New idea" }); + expect(state!.failure).toBeNull(); + }); + + it("reads once for a duplicate delivery and skips webhooks that change nothing shown", async () => { + await enrolled(); + const { calls } = mocked({ entities: { 5: issue(102, 5) } }); + const id = crypto.randomUUID(); + await opened(102, 5, id); + expect(await (await opened(102, 5, id)).json()).toEqual({ status: "duplicate" }); + for (const [name, payload] of [ + ["check_run", {}], + ["status", {}], + ["push", { ref: "refs/heads/human-branch" }], + ] as const) + await delivery(name, { ...event, ...payload }, undefined, {}, testEnv); + expect(reads(calls)).toHaveLength(1); + }); + + it("coalesces a burst of deliveries into one more read", async () => { + await enrolled(); + let started!: () => void; + const reading = new Promise((resolve) => (started = resolve)); + let release!: () => void; + const gate = new Promise((resolve) => (release = resolve)); + const { calls } = mocked({ + entities: { 5: issue(102, 5), 6: issue(103, 6), 7: issue(104, 7) }, + onActivity: async () => { + started(); + await gate; + }, + }); + const first = opened(102, 5); + await reading; + // Another update to the entity being read, and new ones, while the read is in progress. + for (const [id, number] of [ + [102, 5], + [103, 6], + [104, 7], + [103, 6], + ]) + expect((await opened(id!, number!)).status).toBe(202); + release(); + await first; + expect(reads(calls)).toHaveLength(2); + expect(reads(calls)[1]!.body.query).toContain("issue(number:5)"); + for (const id of ["102", "103", "104"]) expect(await row(id)).toMatchObject({ stale: 0 }); + }); + + it("reads again when a webhook during a read fails its revision check", async () => { + await enrolled(); + let started!: () => void; + const reading = new Promise((resolve) => (started = resolve)); + let release!: () => void; + const gate = new Promise((resolve) => (release = resolve)); + let first = true; + const { calls } = mocked({ + entities: { 5: issue(102, 5) }, + onActivity: async () => { + if (!first) return; + first = false; + started(); + await gate; + }, + }); + const pending = opened(102, 5); + await reading; + // A new PR head bumps the revision the read in progress publishes against. + const synchronized = { ...event, action: "synchronize", pull_request: { id: 200, number: 2 } }; + expect((await delivery("pull_request", synchronized, undefined, {}, testEnv)).status).toBe(202); + release(); + await pending; + expect(reads(calls)).toHaveLength(2); + expect(await row("102")).toMatchObject({ stale: 0 }); + expect((await github().service.projectState(repo, "enrollment-test"))!.failure).toBeNull(); + }); + + it("reads again when a refresh that started first publishes first", async () => { + await enrolled(); + const gate = () => { + let started!: () => void; + let release!: () => void; + const reading = new Promise((resolve) => (started = resolve)); + const held = new Promise((resolve) => (release = resolve)); + return { reading, release, hold: async () => (started(), await held) }; + }; + // The refresh lists stale entities before the webhook arrives. + const refresh = gate(); + const refreshing = github({ onActivity: refresh.hold }).service.projectState(repo, "enrollment-test", true); + await refresh.reading; + const observer = gate(); + let first = true; + const { calls } = mocked({ + entities: { 5: issue(102, 5) }, + onActivity: async () => { + if (!first) return; + first = false; + await observer.hold(); + }, + }); + const pending = opened(102, 5); + await observer.reading; + refresh.release(); + await refreshing; + observer.release(); + await pending; + expect(reads(calls)).toHaveLength(2); + expect(await row("102")).toMatchObject({ stale: 0 }); + expect((await github().service.projectState(repo, "enrollment-test"))!.failure).toBeNull(); + }); + + it("rereads a PR by number after a comment on it", async () => { + await enrolled(); + const { calls } = mocked({ entities: { 2: { ...pull(), comments: { totalCount: 3 } } } }); + const comment = { ...event, action: "created", issue: { id: 999, number: 2, pull_request: {} } }; + expect((await delivery("issue_comment", comment, undefined, {}, testEnv)).status).toBe(202); + expect(reads(calls)).toHaveLength(1); + expect(reads(calls)[0]!.body.query).toContain("s0:pullRequest(number:2)"); + const read = await row("200"); + expect(read).toMatchObject({ stale: 0 }); + expect(JSON.parse(read!.data!).comments).toEqual({ totalCount: 3 }); + expect(await row("999")).toBeNull(); + }); + + it("records a failed read without retrying, and a later webhook reads again", async () => { + await enrolled(); + const failed = mocked({ activityErrors: true }); + await opened(102, 5); + expect(reads(failed.calls)).toHaveLength(1); + expect(await row("102")).toMatchObject({ data: null, stale: 1 }); + const state = await github().service.projectState(repo, "enrollment-test"); + expect(state!.failure).toBe("Project activity unavailable"); + const { calls } = mocked({ entities: { 5: issue(102, 5) } }); + await opened(102, 5); + expect(reads(calls)).toHaveLength(1); + expect(await row("102")).toMatchObject({ stale: 0 }); + expect((await github().service.projectState(repo, "enrollment-test"))!.failure).toBeNull(); + }); + + it("shows why a background read failed next to the refresh fallback", async () => { + const who = identity("github"); + const login = await oauth(who); + const user = await env.DB.prepare("SELECT userId FROM account WHERE providerId='github' AND accountId=?") + .bind(who.id) + .first<{ userId: string }>(); + const { service } = github({ account: Number(who.id) }); + await service.setContributions(repo, user!.userId, "enabled"); + await service.projectState(repo, user!.userId, true); + mocked({ account: Number(who.id), activityErrors: true }); + await opened(102, 5); + const page = await app.request( + `${origin}/projects/github/upstream/project/state`, + { headers: { Cookie: login.session! } }, + testEnv, + ); + expect(await page.text()).toContain("The last background read failed: Project activity unavailable"); + }); + + it("rereads a closed issue and marks a deleted one missing", async () => { + await enrolled(); + mocked({ entities: { 1: { ...issue(), state: "CLOSED" } } }); + await delivery("issues", { ...event, action: "closed", issue: { id: 100, number: 1 } }, undefined, {}, testEnv); + expect((await github().service.projectState(repo, "enrollment-test"))!.project.issues).toEqual([]); + mocked({ entities: { 1: null } }); + await delivery( + "issues", + { ...event, action: "deleted", issue: { id: 100, number: 1 } }, + undefined, + {}, + testEnv, + ); + expect(await row("100")).toMatchObject({ stale: 0, missing: 1 }); + }); + + it("rereads availability after a webhook that can claim a quest", async () => { + await enrolled(); + const quiet = { + issues: connection([]), + pullRequests: connection([]), + open: connection([]), + refs: connection([]), + }; + await github({ activity: quiet }).service.projectState(repo, "enrollment-test", true); + mocked({ activity: { ...quiet, refs: connection([{ name: "first", prefix: "refs/heads/quest/" }]) } }); + await delivery("push", { ...event, ref: "refs/heads/quest/first", created: true }, undefined, {}, testEnv); + const state = await github().service.projectState(repo, "enrollment-test"); + expect(state!.claims).toEqual(["quest/first.md"]); + expect(state!.project.available).toEqual([]); + }); + + it("does not read a paused or disconnected project", async () => { + await enrolled(); + await github().service.setContributions(repo, "enrollment-test", "paused"); + const paused = mocked(); + expect((await opened(102, 5)).status).toBe(202); + expect(paused.calls).toEqual([]); + await github().service.setContributions(repo, "enrollment-test", "disconnected"); + const disconnected = mocked(); + expect((await opened(103, 6)).status).toBe(202); + expect(disconnected.calls).toEqual([]); + expect(await env.DB.prepare("SELECT * FROM github_project_state").first()).toBeNull(); + }); +}); + describe("project activity routes", () => { const base = `${origin}/projects/github/upstream/project`; let githubId = 0; diff --git a/cloudflare/test/seed.test.ts b/cloudflare/test/seed.test.ts index 16dd1163..59a85b42 100644 --- a/cloudflare/test/seed.test.ts +++ b/cloudflare/test/seed.test.ts @@ -1,5 +1,6 @@ // The seeded local server's data must load on the current schema and routes, // and show every inbox state it exists to check. +import { createExecutionContext, waitOnExecutionContext } from "cloudflare:test"; import { env } from "cloudflare:workers"; import { afterEach, expect, it, vi } from "vitest"; import { fakeGitHub } from "../dev/github"; @@ -21,7 +22,13 @@ it("seeds a representative inbox through the Worker's own routes", async () => { ]); const data = world(); vi.spyOn(globalThis, "fetch").mockImplementation(fakeGitHub(data)); - const request = (path: string, init?: RequestInit) => app.request(`${origin}${path}`, init, testEnv); + // Webhooks read in the background, so each request finishes its `waitUntil` work. + const request = async (path: string, init?: RequestInit) => { + const ctx = createExecutionContext(); + const response = await app.request(`${origin}${path}`, init, testEnv, ctx); + await waitOnExecutionContext(ctx); + return response; + }; const { cookie } = await seed({ fetch: request, db, @@ -45,7 +52,8 @@ it("seeds a representative inbox through the Worker's own routes", async () => { "Build the landing page hero", "Bump dependencies", "Add a dark mode toggle", - "Issue #9", + // Read by number after its webhook. + "Sync stops after the laptop sleeps", "Export a project as a ZIP archive", "Settings page loses unsaved changes when the session expires", "Keep working offline", diff --git a/quest/m0/kix/README.md b/quest/m0/kix/README.md index 13ef6375..f141577a 100644 --- a/quest/m0/kix/README.md +++ b/quest/m0/kix/README.md @@ -101,7 +101,6 @@ operator-paid-hosting decisions: - [Runner](/quest/m0/kix/runner.md) - `quest runner` on Linux and macOS with Nix devShells, accounts, and the Claude Code adapter - [Job sandbox](/quest/m0/kix/sandbox.md) - OS-sandboxed commands, with untrusted PR devShells built in a VM on Linux runners - [opencode adapter](/quest/m0/kix/opencode.md) - OpenCode Go, ChatGPT plans, and API-key providers through opencode -- [Webhook reconcile](/quest/m0/kix/reconcile.md) - new issues and PRs arrive in the inbox ready to act on, without a manual refresh - [Merge jobs](/quest/m0/kix/merge-job.md) - a merge pick fixes conflicts, CI, and reviews on a runner, then lands the PR - [Decision paper trail](/quest/m0/kix/paper-trail.md) - planning PRs carry the recorded questions and answers, rendered by the service - [Agent activity](/quest/m0/kix/activity.md) - agent cards and map highlights built from checkpoint summaries diff --git a/quest/m0/kix/reconcile.md b/quest/m0/kix/reconcile.md deleted file mode 100644 index bba5f2c9..00000000 --- a/quest/m0/kix/reconcile.md +++ /dev/null @@ -1,29 +0,0 @@ -# [S] Reconcile projects after each webhook - -## Goal - -A new GitHub issue, PR, or quest change arrives in the decision inbox with its -title and context, ready to act on, without anyone clicking "Refresh from -GitHub". The manual refresh stays as a fallback. Webhooks still never approve -or launch work. - -## Plan - -Decided 2026-10-18 as a follow-up of the decision inbox (#144), which shows a -webhook's item as a bare pointer ("Issue #N") and offers only a refresh until -someone reads the project from GitHub: - -- After a webhook records an observation, reconcile that project in the - background with the App's installation token. Coalesce bursts so repeated - or duplicate deliveries for one project trigger one read, not one each. -- Background reads publish observations only. Authority stays per request: - every action still rechecks the acting maintainer, as decided for the - inbox, so a background read never grants or caches authority. -- A failed or rate-limited reconcile leaves the item as it is today, with - the refresh fallback and its reason shown, and a later webhook or refresh - retries. No retry loops. -- The per-load GitHub call count (about five per project) stays as is until - it becomes a measured problem; caching authority checks would weaken the - per-request recheck. -- Test with duplicate deliveries, a burst of deliveries, a failed read, and a - webhook for a paused or disconnected project.