diff --git a/CLAUDE.md b/CLAUDE.md index a772e1a..a7b0c2e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -22,6 +22,28 @@ the self-hosting playbook live in the private `peetzweg/devops` repo. - **Ambient `GITHUB_TOKEN` in the shell shadows `.env`** and lacks `read:org` — prefix dev/bun script runs with `env -u GITHUB_TOKEN` when org lookups misbehave. +## Background jobs + +Recipe, proven by `monthly-user-refresh` and `backfill-orgs`: a CLI in `scripts/` (flags, +each mirrored by an env var) → bundled by `build:worker` into `.output/worker/.mjs` → +the Dockerfile already copies `.output`, so shipping a new job needs no Dockerfile change → +a Coolify **scheduled task** runs `node .output/worker/.mjs` (cron + `docker exec` into +the running app container; container TZ is UTC). Logic goes in a pure engine +(`src/lib/.ts`) over an injected store (`-store.ts`) so it unit-tests against fakes. + +Every job must: take its advisory lock from `LOCK_KEYS` in `src/lib/job-runner.ts` (unique key +per job); use `createBudgetGuard` so it can't drain the GitHub token below the floor that live +traffic needs; stop cleanly on a wall-clock cap instead of being killed; resume off a DB +freshness marker (**the missing row is the retry queue** — no job-state table); write only +idempotent upserts; isolate per-item failures to a narrow status allowlist; and emit one +greppable ` done status=… ` summary line. Stagger schedules: the prod box has no swap and +`docker exec` adds a second node process inside the app container. + +Two failure modes worth knowing: cleanup must release the lock **first and defensively**, or a +crash mid-teardown leaves it held by a server-side session that outlives the process and every +later run reports `status=locked` and silently does nothing; and never let pool shutdown throw, +or a completed run shows up as a failed Coolify execution. + ## Domain guardrails - Single-segment paths are GitHub logins (`$user` route); editorial pages live under the diff --git a/package.json b/package.json index 0aef984..99585a2 100644 --- a/package.json +++ b/package.json @@ -10,7 +10,7 @@ "generate-routes": "tsr generate", "og": "tsx scripts/generate-og.ts", "build": "tsx scripts/generate-og.ts && vite build && pnpm run build:worker", - "build:worker": "esbuild scripts/monthly-user-refresh.ts --bundle --platform=node --target=node24 --format=esm --outfile=.output/worker/monthly-user-refresh.mjs", + "build:worker": "esbuild scripts/monthly-user-refresh.ts scripts/backfill-orgs.ts --bundle --platform=node --target=node24 --format=esm --outdir=.output/worker --out-extension:.js=.mjs", "start": "node .output/server/index.mjs", "preview": "vite preview", "test": "vitest run", @@ -23,6 +23,7 @@ "refresh-user": "bun scripts/refresh-user.ts", "monthly:user": "node --env-file-if-exists=.env --import tsx scripts/monthly-user-refresh.ts", "backfill": "bun scripts/backfill-contributions.ts", + "backfill:orgs": "node --env-file-if-exists=.env --import tsx scripts/backfill-orgs.ts", "backup": "bash scripts/backup.sh", "db:generate": "drizzle-kit generate", "db:migrate": "drizzle-kit migrate", diff --git a/scripts/backfill-orgs.ts b/scripts/backfill-orgs.ts index 52a6830..73928c1 100644 --- a/scripts/backfill-orgs.ts +++ b/scripts/backfill-orgs.ts @@ -1,309 +1,233 @@ /** - * Lifetime-totals backfill for organizations the request path won't build on demand. + * Fill organization lifetime totals in the background, a bounded slice at a time. * - * After #103, an org with more than MAX_ORG_MEMBERS (25) public members is *recorded* on lookup - * (an `entities` row, `builtAt` null) but never built live — building a mega-org on a page request - * would burn the shared token's hourly budget. This script fills those recorded-but-empty orgs - * (and any org you name explicitly, e.g. google/microsoft/github) with the SAME numbers the - * request path produces for small orgs: each public member's lifetime contributions *to that org*, - * summed. It writes only the existing tables — `org_members` rows + the `entities` roll-up — so it - * needs NO database migration. (The per-month resolution / company chart is separate future work - * on the refresh-orgs worker; this deliberately does none of that.) + * Bundled by `pnpm build` into `.output/worker/backfill-orgs.mjs`, which is what the production + * image runs as a Coolify scheduled task (daily, 06:00 UTC — deliberately clear of the monthly + * user refresh, which runs on the 1st at 03:00 and takes a couple of hours): * - * Politeness: paced well under GitHub's 5,000 points/hour with a reserved floor for live traffic, - * same model as backfill-contributions.ts. Run with bun (auto-loads .env; beware a shell-exported - * GITHUB_TOKEN overriding it — prefix with `env -u GITHUB_TOKEN` if your shell sets one): + * node .output/worker/backfill-orgs.mjs * - * bun scripts/backfill-orgs.ts # every recorded org not yet filled, SMALLEST first - * bun scripts/backfill-orgs.ts google microsoft # these orgs specifically (records them if new) - * bun scripts/backfill-orgs.ts --force # re-fetch every member (refresh, don't resume) + * Locally, `pnpm backfill:orgs` runs this source through tsx. * - * No-arg runs go smallest-org-first (fewest public members) so small orgs resolve quickly and - * mega-orgs fall to the back instead of blocking everyone behind one 1,000-member org. + * The engine (src/lib/org-backfill.ts) is built to *not* finish: it spends at most + * --max-requests GraphQL requests per run, keeps a floor of budget free for live site traffic, + * stops on a wall-clock cap, and resumes from `org_members.last_fetched` next time. So a + * 2,800-member org gets chipped away over several nights instead of blocking the queue, and every + * org the site has ever recorded eventually gets filled. * - * Safe to re-run and interrupt: every write is an idempotent upsert; a member's `lastFetched` - * marks it done, so a re-run SKIPS members already fetched (in this run or a previous, aborted - * one) and continues from where it stopped. An org's `builtAt` is stamped only after all its - * members are fetched, so an interrupted org stays unfilled and is picked up again. Pass --force - * to re-fetch already-filled members (refresh their numbers) instead of resuming. + * Execute-by-default. `--dry-run` reports what it would do without spending requests or writing. + * + * node .output/worker/backfill-orgs.mjs # drain the queue, smallest org first + * pnpm backfill:orgs --dry-run # preview the slice + * pnpm backfill:orgs google microsoft # these orgs specifically + * pnpm backfill:orgs --force jupyter # re-fetch every member, don't resume + * pnpm backfill:orgs --stale-after-days=90 --max-stale-orgs=5 # refresh the 5 oldest built orgs + * + * Member enumeration needs the read:org token from .env, so beware a shell-exported GITHUB_TOKEN + * shadowing it: prefix local runs with `env -u GITHUB_TOKEN` if your shell sets one. */ -import { and, asc, eq, isNull, sql } from "drizzle-orm"; import { db } from "#/lib/db"; -import { entities, orgMembers } from "#/lib/db/schema"; import { - fetchOrgMembers, fetchOrgMemberContributions, + fetchOrgMembers, fetchOrgProfile, - GitHubError, - type OrgMemberTotals, + fetchRateLimitBudget, yearlyWindows, } from "#/lib/github"; +import { runOrgBackfill } from "#/lib/org-backfill"; +import { createOrgBackfillStore } from "#/lib/org-backfill-store"; + +const VALID_FLAGS = new Set([ + "dry-run", + "force", + "h", + "help", + "max-orgs", + "max-requests", + "max-runtime-minutes", + "max-stale-orgs", + "member-pages", + "poll-every", + "rate-per-hour", + "remaining-floor", + "stale-after-days", +]); + +const { flags, logins } = parseArgv(process.argv.slice(2)); +const env = process.env; + +if (booleanFlag(flags, "help") || booleanFlag(flags, "h")) { + console.log(usage()); + process.exit(0); +} + +const DATABASE_URL = env.DATABASE_URL; +const GITHUB_TOKEN = env.GITHUB_TOKEN; -if (!process.env.DATABASE_URL) - throw new Error("DATABASE_URL is required (add it to .env)"); -const GITHUB_TOKEN = process.env.GITHUB_TOKEN; -if (!GITHUB_TOKEN) throw new Error("GITHUB_TOKEN is required (add it to .env)"); +if (!DATABASE_URL) throw new Error("DATABASE_URL is required."); +if (!GITHUB_TOKEN) throw new Error("GITHUB_TOKEN is required."); if (!db) throw new Error("Database client failed to initialise."); -const token = GITHUB_TOKEN; -const database = db; -const orgEntityId = (login: string) => `org:${login.trim().toLowerCase()}`; -const userEntityId = (login: string) => `user:${login.trim().toLowerCase()}`; +const maxRuntimeMinutes = numberValue( + flags, + "max-runtime-minutes", + env.ORG_BACKFILL_MAX_RUNTIME_MINUTES, + 120, +); + +const result = await runOrgBackfill({ + store: createOrgBackfillStore(db), + token: GITHUB_TOKEN, + now: new Date(), + logins, + dryRun: booleanFlag(flags, "dry-run") || isTrue(env.ORG_BACKFILL_DRY_RUN), + force: booleanFlag(flags, "force") || isTrue(env.ORG_BACKFILL_FORCE), + maxOrgs: numberValue(flags, "max-orgs", env.ORG_BACKFILL_MAX_ORGS, 25), + maxRequests: numberValue( + flags, + "max-requests", + env.ORG_BACKFILL_MAX_REQUESTS, + 1500, + ), + ratePerHour: numberValue( + flags, + "rate-per-hour", + env.ORG_BACKFILL_RATE_PER_HOUR, + 1200, + ), + remainingFloor: numberValue( + flags, + "remaining-floor", + env.ORG_BACKFILL_REMAINING_FLOOR, + 500, + ), + pollEvery: numberValue(flags, "poll-every", env.ORG_BACKFILL_POLL_EVERY, 25), + memberPages: numberValue( + flags, + "member-pages", + env.ORG_BACKFILL_MEMBER_PAGES, + 30, + ), + staleAfterDays: numberValue( + flags, + "stale-after-days", + env.ORG_BACKFILL_STALE_AFTER_DAYS, + 0, + { allowZero: true }, + ), + maxStaleOrgs: numberValue( + flags, + "max-stale-orgs", + env.ORG_BACKFILL_MAX_STALE_ORGS, + 0, + { allowZero: true }, + ), + maxRuntimeMs: maxRuntimeMinutes * 60_000, + // The token is shared with live site traffic, so the run polls its own GraphQL budget and + // refuses to spend below the floor. `rateLimit` queries cost 0 points. + fetchRateLimit: () => fetchRateLimitBudget(GITHUB_TOKEN), + fetchOrgProfile, + fetchOrgMembers, + fetchOrgMemberContributions, + yearlyWindows, + logger: console, +}); -// Requests/hour we aim to spend (≈ GraphQL points), well under the 5,000/hr limit so the live -// site keeps working. Override with REFRESH_RATE=. -const TARGET_RATE = Number(process.env.REFRESH_RATE ?? 2500); -// Never let the remaining budget drop below this — headroom for live traffic. -const REMAINING_FLOOR = 500; -const CHUNK_ROWS = 100; -// fetchOrgMemberContributions batches windows this many at a time; used only to pace, not for -// correctness — an over-estimate just spends the budget more conservatively. -const WINDOWS_PER_REQUEST = 6; -// --force re-fetches every member even if already filled (to refresh numbers). Default: resume, -// i.e. skip members that already have a lastFetched from any prior run. -const force = process.argv.slice(2).includes("--force"); +// Stopping on a cap is normal: the queue is the retry mechanism and tomorrow's run continues. +// A systemic fault is not, and a scheduled task that exits 0 while achieving nothing is invisible +// in Coolify — so those two exit non-zero and show up as a failed execution. +const systemicFailure = + result.stopReason === "consecutive_failures" || + (result.orgsFailed > 0 && result.membersFetched === 0 && !result.dryRun); -const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); +await shutdown(systemicFailure ? 1 : 0); -interface RateLimit { - remaining: number; - resetAt: string; +/** + * Close the pool so Postgres isn't left holding idle backends until TCP keepalive reaps them — + * best-effort only. postgres.js can throw *asynchronously* while tearing sockets down (seen when + * the link to the database blipped mid-run), and an uncaught throw at this point would turn a run + * that already did its work and logged its summary into a failed Coolify execution. Shutdown is + * explicitly not allowed to change the outcome. + */ +async function shutdown(code: number): Promise { + process.on("uncaughtException", (err) => { + console.warn( + `org-backfill status=shutdown_error error=${JSON.stringify(String(err))}`, + ); + process.exit(code); + }); + await db?.$client.end({ timeout: 5 }).catch(() => {}); + process.exit(code); } -/** Query GitHub's current rate-limit budget. `rateLimit` queries themselves cost 0 points. */ -async function rateLimit(): Promise { - try { - const res = await fetch("https://api.github.com/graphql", { - method: "POST", - headers: { - Authorization: `Bearer ${token}`, - "Content-Type": "application/json", - "User-Agent": "commit-history-backfill-orgs", - }, - body: JSON.stringify({ - query: "query { rateLimit { remaining resetAt } }", - }), - }); - const json = (await res.json()) as { data?: { rateLimit: RateLimit } }; - return json.data?.rateLimit ?? null; - } catch { - return null; +function parseArgv(argv: string[]): { + flags: Map; + logins: string[]; +} { + const flags = new Map(); + const logins: string[] = []; + for (const arg of argv) { + if (!arg.startsWith("--")) { + logins.push(arg); + continue; + } + const [rawName, ...rest] = arg.slice(2).split("="); + const name = rawName.trim(); + if (!name) throw new Error(`Invalid flag "${arg}".`); + if (!VALID_FLAGS.has(name)) throw new Error(`Unknown flag "--${name}".`); + flags.set(name, rest.length === 0 ? true : rest.join("=")); } + return { flags, logins }; } -/** If we're near the reserved floor, sleep until GitHub's window resets (plus a small buffer). */ -async function respectFloor(): Promise { - const rl = await rateLimit(); - if (!rl) return; - if (rl.remaining > REMAINING_FLOOR) return; - const waitMs = Math.max(0, new Date(rl.resetAt).getTime() - Date.now()) + 2000; - console.log( - `… budget low (${rl.remaining} left) — pausing ${Math.ceil(waitMs / 1000)}s until reset`, - ); - await sleep(waitMs); +function booleanFlag(flags: Map, name: string): boolean { + return flags.get(name) === true; } -/** - * Fill one org with lifetime totals. Returns the (approximate) number of GitHub requests spent. - * Re-fetches the profile, (re-)enumerates members, fetches each member's org-scoped lifetime - * totals, then rolls the sums onto the org's `entities` row and stamps `builtAt`. - */ -async function fillOrg(login: string): Promise { - const orgId = orgEntityId(login); - const runStart = new Date(); - let requests = 0; - console.log(`${login}: resolving profile…`); - - // Profile first — validates the login, yields nodeId (keys every org-scoped query) + createdAt - // (bounds each member's windows). Upsert so a never-looked-up org (e.g. a fresh google) is - // recorded here rather than requiring a prior page visit. builtAt stays null until the roll-up. - const profile = await fetchOrgProfile(login, token); - requests += 1; - const profileCols = { - login: profile.login, - name: profile.name, - avatarUrl: profile.avatarUrl, - htmlUrl: profile.htmlUrl, - createdAt: new Date(profile.createdAt), - bio: profile.description, - location: profile.location, - websiteUrl: profile.websiteUrl, - twitterUsername: profile.twitterUsername, - publicRepos: profile.publicRepos, - isVerified: profile.isVerified, - githubNodeId: profile.nodeId, - memberCount: profile.memberCount, - lastFetched: runStart, - }; - await database - .insert(entities) - .values({ id: orgId, kind: "org", ...profileCols }) - .onConflictDoUpdate({ target: entities.id, set: profileCols }); - const orgCreated = new Date(profile.createdAt); - - // Enumerate members → a user stub (FK target, never overwrites a real user row) + a pending - // org_members row each. Both idempotent. - const members = await fetchOrgMembers(login, token); - requests += Math.max(1, Math.ceil(members.length / 100)); - for (let i = 0; i < members.length; i += CHUNK_ROWS) { - const chunk = members.slice(i, i + CHUNK_ROWS); - await database - .insert(entities) - .values( - chunk.map((m) => ({ - id: userEntityId(m.login), - kind: "user", - login: m.login, - name: m.name, - avatarUrl: m.avatarUrl, - htmlUrl: `https://github.com/${m.login}`, - createdAt: new Date(m.createdAt), - })), - ) - .onConflictDoNothing(); - await database - .insert(orgMembers) - .values( - chunk.map((m) => ({ - orgId, - memberId: userEntityId(m.login), - role: m.role, - source: "public_member", - })), - ) - .onConflictDoNothing(); - } - - // Resume support: a member's `lastFetched` is set once we've fetched it — in this run OR a - // previous, aborted one. By default we skip anything already fetched, so re-running continues - // from exactly where an abort stopped (e.g. mid-Microsoft) instead of restarting the org. - // --force re-fetches everyone, to refresh an already-filled org's numbers. - const existing = await database - .select({ - memberId: orgMembers.memberId, - lastFetched: orgMembers.lastFetched, - }) - .from(orgMembers) - .where(eq(orgMembers.orgId, orgId)); - const fetchedById = new Map(existing.map((r) => [r.memberId, r.lastFetched])); - const isDone = (login: string) => - !force && fetchedById.get(userEntityId(login)) != null; - const already = members.filter((m) => isDone(m.login)).length; - console.log( - ` ${members.length} public members${already ? ` (${already} already done — resuming)` : ""} — fetching lifetime totals, ~${Math.round((Math.max(1, Math.ceil(20 / WINDOWS_PER_REQUEST)) / TARGET_RATE) * 3600)}s each`, - ); - - let done = 0; - for (const m of members) { - const memberId = userEntityId(m.login); - if (isDone(m.login)) continue; - - // A member can't have contributed to the org before either account existed. - const start = new Date( - Math.max(orgCreated.getTime(), new Date(m.createdAt).getTime()), - ); - const windows = yearlyWindows(start, runStart); - let totals: OrgMemberTotals = { - commits: 0, - issues: 0, - pullRequests: 0, - reviews: 0, - }; - try { - totals = await fetchOrgMemberContributions( - m.login, - profile.nodeId, - token, - windows, - ); - } catch (e) { - // Isolate per-member failures so one bad member can't sink the whole org: - // - 404: deleted/renamed member — gone forever. - // - 400: login our validator can't accept (shouldn't happen now that legacy - // hyphen-ending usernames pass, but stay defensive). - // Record zeros + mark done so resume doesn't wedge on it. Rate limits (403/429) and - // 5xx still propagate → the run aborts and resumes later rather than storing false zeros. - if (!(e instanceof GitHubError && (e.status === 404 || e.status === 400))) { - throw e; - } - console.log(` – ${m.login.padEnd(22)} skipped (${(e as GitHubError).status})`); - } - const req = Math.max(1, Math.ceil(windows.length / WINDOWS_PER_REQUEST)); - requests += req; +function isTrue(value: string | undefined): boolean { + return value?.toLowerCase() === "true"; +} - await database - .update(orgMembers) - .set({ ...totals, lastFetched: new Date() }) - .where( - and(eq(orgMembers.orgId, orgId), eq(orgMembers.memberId, memberId)), - ); - done += 1; - console.log( - ` ✓ ${m.login.padEnd(22)} ${totals.commits.toLocaleString()} commits (${done}/${members.length})`, +function numberValue( + flags: Map, + name: string, + envValue: string | undefined, + fallback: number, + opts: { allowZero?: boolean } = {}, +): number { + const flagValue = flags.get(name); + if (flagValue === true) throw new Error(`--${name} requires a value.`); + const raw = typeof flagValue === "string" ? flagValue : envValue; + if (raw == null || raw === "") return fallback; + const value = Number(raw); + const floor = opts.allowZero ? 0 : 1; + if (!Number.isInteger(value) || value < floor) { + throw new Error( + `--${name} must be an integer >= ${floor}${opts.allowZero ? " (0 disables it)" : ""}.`, ); - // Politeness pacing: spread this member's request cost across the target hourly rate. - await sleep((req / TARGET_RATE) * 3_600_000); } - - // Roll up org totals from org_members (departed rows, if any, stay frozen — same as #97) and - // stamp builtAt so the org now looks finished to the request path. - const [sums] = await database - .select({ - commits: sql`coalesce(sum(${orgMembers.commits}), 0)`, - pullRequests: sql`coalesce(sum(${orgMembers.pullRequests}), 0)`, - reviews: sql`coalesce(sum(${orgMembers.reviews}), 0)`, - issues: sql`coalesce(sum(${orgMembers.issues}), 0)`, - }) - .from(orgMembers) - .where(eq(orgMembers.orgId, orgId)); - await database - .update(entities) - .set({ - totalCommits: Number(sums?.commits ?? 0), - totalPullRequests: Number(sums?.pullRequests ?? 0), - totalReviews: Number(sums?.reviews ?? 0), - totalIssues: Number(sums?.issues ?? 0), - builtAt: new Date(), - lastFetched: new Date(), - }) - .where(eq(entities.id, orgId)); - - console.log( - `✓ ${login.padEnd(24)} ${Number(sums?.commits ?? 0).toLocaleString()} commits · ${done} fetched${already ? ` + ${already} already done` : ""} / ${members.length} members`, - ); - return requests; + return value; } -const args = process.argv.slice(2).filter((a) => !a.startsWith("-")); - -let targets: { login: string }[]; -if (args.length > 0) { - targets = args.map((login) => ({ login })); - console.log(`Backfilling ${targets.length} org(s) by name\n`); -} else { - // Every recorded-but-unfilled org (builtAt null), SMALLEST first (fewest public members) so - // small orgs resolve quickly and mega-orgs (microsoft, google) fall to the back instead of - // blocking everyone. An org interrupted mid-fill still has builtAt null, so it's picked up - // again and resumes from where it stopped (members already fetched are skipped). - const rows = await database - .select({ login: entities.login }) - .from(entities) - .where(and(eq(entities.kind, "org"), isNull(entities.builtAt))) - .orderBy(sql`${entities.memberCount} asc nulls last`, asc(entities.id)); - targets = rows; - console.log( - `${rows.length} recorded org(s) not yet filled, smallest first, ~${TARGET_RATE} req/hr\n`, - ); -} - -let spent = 0; -for (const { login } of targets) { - await respectFloor(); - try { - spent += await fillOrg(login); - } catch (e) { - const msg = e instanceof Error ? e.message : String(e); - console.log(`✗ ${login.padEnd(24)} ${msg} — continuing`); - } +function usage(): string { + return [ + "Usage:", + " node .output/worker/backfill-orgs.mjs [flags] [org…] # built image", + " pnpm backfill:orgs [flags] [org…] # local source via tsx", + "", + "Named orgs bypass the queue. With none, drains recorded-but-unfilled orgs, smallest first.", + "", + "Flags override env vars:", + " --dry-run env ORG_BACKFILL_DRY_RUN", + " --force env ORG_BACKFILL_FORCE", + " --max-orgs=N env ORG_BACKFILL_MAX_ORGS, default 25", + " --max-requests=N env ORG_BACKFILL_MAX_REQUESTS, default 1500", + " --rate-per-hour=N env ORG_BACKFILL_RATE_PER_HOUR, default 1200", + " --remaining-floor=N env ORG_BACKFILL_REMAINING_FLOOR, default 500", + " --poll-every=N env ORG_BACKFILL_POLL_EVERY, default 25", + " --member-pages=N env ORG_BACKFILL_MEMBER_PAGES, default 30 (×100 members)", + " --max-runtime-minutes=N env ORG_BACKFILL_MAX_RUNTIME_MINUTES, default 120", + " --stale-after-days=N env ORG_BACKFILL_STALE_AFTER_DAYS, default 0 (off)", + " --max-stale-orgs=N env ORG_BACKFILL_MAX_STALE_ORGS, default 0 (off)", + ].join("\n"); } -console.log(`\nDone. ~${spent.toLocaleString()} requests spent.`); diff --git a/scripts/monthly-user-refresh.ts b/scripts/monthly-user-refresh.ts index 2e01db5..c276887 100644 --- a/scripts/monthly-user-refresh.ts +++ b/scripts/monthly-user-refresh.ts @@ -72,8 +72,24 @@ await runMonthlyUserRefresh({ logger: console, }); -await db.$client.end(); -process.exit(0); +await shutdown(0); + +/** + * Close the pool so Postgres isn't left holding an idle backend until TCP keepalive reaps it — + * best-effort only. postgres.js can throw *asynchronously* while tearing sockets down (seen when + * the link to the database blipped mid-run), and an uncaught throw at this point would turn a run + * that already did its work and logged its summary into a failed Coolify execution. + */ +async function shutdown(code: number): Promise { + process.on("uncaughtException", (err) => { + console.warn( + `monthly-user-refresh status=shutdown_error error=${JSON.stringify(String(err))}`, + ); + process.exit(code); + }); + await db?.$client.end({ timeout: 5 }).catch(() => {}); + process.exit(code); +} interface Config { help: boolean; diff --git a/src/lib/github.ts b/src/lib/github.ts index dd41788..c49bfdf 100644 --- a/src/lib/github.ts +++ b/src/lib/github.ts @@ -541,21 +541,28 @@ export async function fetchOrgProfile( // Hard stop for the members pagination loop (pages of 100). The org build enforces its own, // lower member cap before fetching contributions — this is just a runaway/mega-org backstop. +// The backfill worker raises it (see `maxPages`), since a 2,800-member org is exactly the case +// the request path refuses and the worker exists to handle. const MAX_MEMBER_PAGES = 10; /** * Enumerate an org's members visible to the token — public members only unless the token itself * belongs to the org. Each node carries `createdAt`, so the caller can window every member's * contribution fetch without per-member profile requests. + * + * `truncated` means the page cap stopped us with more members still to come. Callers that stamp an + * org as built need it: rolling up a truncated membership silently undercounts the org forever. */ export async function fetchOrgMembers( rawLogin: string, token: string, -): Promise { + opts: { maxPages?: number } = {}, +): Promise<{ members: OrgMember[]; truncated: boolean }> { const login = assertLogin(rawLogin); + const maxPages = Math.max(1, opts.maxPages ?? MAX_MEMBER_PAGES); const members: OrgMember[] = []; let cursor: string | null = null; - for (let page = 0; page < MAX_MEMBER_PAGES; page++) { + for (let page = 0; page < maxPages; page++) { const after: string = cursor ? `, after: "${cursor}"` : ""; const data = await graphql<{ organization: { @@ -587,10 +594,10 @@ export async function fetchOrgMembers( if (!edge.node) continue; members.push({ ...edge.node, role: edge.role }); } - if (!conn.pageInfo.hasNextPage) return members; + if (!conn.pageInfo.hasNextPage) return { members, truncated: false }; cursor = conn.pageInfo.endCursor; } - return members; + return { members, truncated: true }; } /** A member's summed lifetime contributions to one org (org-scoped, not their global totals). */ diff --git a/src/lib/job-runner.ts b/src/lib/job-runner.ts new file mode 100644 index 0000000..bd81bb2 --- /dev/null +++ b/src/lib/job-runner.ts @@ -0,0 +1,197 @@ +/** + * Shared plumbing for the background workers that run as Coolify scheduled tasks + * (`.output/worker/*.mjs`, see the Dockerfile). Extracted from monthly-user-refresh once a second + * job needed the same three guarantees: + * + * 1. one run at a time (Postgres advisory lock), + * 2. never drain the shared GitHub token below a reserved floor, + * 3. stop cleanly at a wall-clock cap instead of being killed by the task timeout. + * + * Every job here is expected to be *resumable off its own data* — a missing row is the retry queue — + * so stopping early is normal and cheap, and exiting 0 with work left over is a success. + */ + +import type { DB } from "#/lib/db"; +import type { RateLimitBudget } from "#/lib/github"; + +export interface JobLogger { + info(message: string): void; + warn(message: string): void; + error(message: string): void; +} + +export const quietLogger: JobLogger = { + info: () => {}, + warn: () => {}, + error: () => {}, +}; + +export const defaultSleep = (ms: number) => + new Promise((resolve) => setTimeout(resolve, ms)); + +export function positiveInteger(value: number, name: string): number { + if (!Number.isInteger(value) || value <= 0) { + throw new Error(`${name} must be a positive integer.`); + } + return value; +} + +export function nonNegativeInteger(value: number, name: string): number { + if (!Number.isInteger(value) || value < 0) { + throw new Error(`${name} must be a non-negative integer.`); + } + return value; +} + +export interface JobLock { + tryLock(): Promise; + releaseLock(): Promise; +} + +/** + * Session-scoped advisory lock pinned to one reserved connection. + * + * `database.execute` takes an arbitrary connection out of the postgres.js pool, and unlocking from + * a different connection than the one holding the lock silently returns false. Mutual exclusion + * would work either way (a second process is a second session), but a release that quietly no-ops + * is a trap, so the connection is held for the run's lifetime and released in `releaseLock`. + * + * Keys must be unique per job — see the LOCK_KEYS table below. + */ +export function createJobLock( + database: DB, + key1: number, + key2: number, + label: string, +): JobLock { + let lockConn: Awaited> | null = null; + + return { + async tryLock() { + lockConn = await database.$client.reserve(); + const [row] = await lockConn>` + select pg_try_advisory_lock(${key1}, ${key2}) as acquired`; + const acquired = Boolean(row?.acquired); + if (!acquired) { + lockConn.release(); + lockConn = null; + } + return acquired; + }, + + async releaseLock() { + if (!lockConn) return; + try { + const [row] = await lockConn>` + select pg_advisory_unlock(${key1}, ${key2}) as released`; + if (!row?.released) { + console.warn( + `${label} lock=release_returned_false (was it held by this session?)`, + ); + } + } finally { + lockConn.release(); + lockConn = null; + } + }, + }; +} + +/** + * Advisory-lock keys, one row per background job. Same first key, distinct second — collisions + * would make two unrelated jobs exclude each other, which is much harder to notice than a job + * that never runs. + */ +export const LOCK_KEYS = { + monthlyUserRefresh: [20260701, 1], + orgBackfill: [20260701, 2], +} as const; + +export interface BudgetGuard { + /** True to proceed; false when the run must stop rather than dip below the floor. */ + ensure(force?: boolean): Promise; + /** Account for requests just spent, between polls. */ + spend(requests?: number): void; + /** Total requests this guard has been told about — the run's spend estimate. */ + spent(): number; +} + +/** + * Keeps a run above `remainingFloor` GraphQL points. The token is shared with live site traffic, so + * a batch job that drains the hourly quota takes the site down with it. Polling costs nothing but a + * round-trip, so we poll every `pollEvery` requests and decrement locally in between. + * + * Below the floor the only options are "wait for the window to reset" or "stop". Waiting is allowed + * only if it fits inside the remaining max-runtime budget: a scheduled job that sleeps past its + * window is worse than one that exits 0 and leaves the rest for the next pass. + */ +export function createBudgetGuard(opts: { + /** Log prefix, e.g. `org-backfill` — every line is `