diff --git a/src/db/repositories.ts b/src/db/repositories.ts index 4016764ec9..3b5c8dfadd 100644 --- a/src/db/repositories.ts +++ b/src/db/repositories.ts @@ -273,6 +273,20 @@ export async function markRepositoriesRemovedFromInstallation(env: Env, installa .where(and(eq(repositories.installationId, installationId), inArray(repositories.fullName, names))); } +/** Every repo full name currently marked `isInstalled` under this installation, so a caller can diff it against + * a freshly-fetched live list and hand the leftovers to {@link markRepositoriesRemovedFromInstallation} (#5028). + * Unlike listRepoFullNamesForInstallation (used for cross-repo aggregation with a truncation-audit concern), + * this is a plain currently-installed set for a single maintainer's own installation, which is never large + * enough to need a cap. */ +export async function listInstalledRepoFullNamesForInstallation(env: Env, installationId: number): Promise { + const db = getDb(env.DB); + const rows = await db + .select({ fullName: repositories.fullName }) + .from(repositories) + .where(and(eq(repositories.installationId, installationId), eq(repositories.isInstalled, true))); + return rows.map((row) => row.fullName); +} + export async function getInstallation(env: Env, installationId: number): Promise { const db = getDb(env.DB); const [row] = await db.select().from(installations).where(eq(installations.id, installationId)).limit(1); diff --git a/src/index.ts b/src/index.ts index fb87e4dab2..d944c9c3b9 100644 --- a/src/index.ts +++ b/src/index.ts @@ -4,6 +4,7 @@ import { delayUntil, shouldWaitForGitHubRateLimit, LOW_REST_RATE_LIMIT_REMAINING import { processDlqBatch } from "./queue/dlq"; import { processJob } from "./queue/processors"; import { isOrbBrokerEnabled } from "./orb/broker"; +import { isOrbBrokerMode } from "./orb/broker-client"; import { isOpsEnabled } from "./review/ops-wire"; import { isRecapEnabled, resolveMaintainerRecapManifestOverride, shouldFireMaintainerRecap } from "./review/maintainer-recap-wire"; import { isSweepWatchdogEnabled } from "./review/sweep-watchdog"; @@ -202,6 +203,11 @@ async function enqueueScheduledJobs(env: Env, controller: ScheduledController): if (selfHostedReviews && isReconciliationWindow && isPrReconciliationEnabled(env)) jobs.push({ type: "reconcile-open-prs", requestedBy: "schedule" }); if (isHourly) { jobs.push({ type: "refresh-registry", requestedBy: "schedule" }); + // Brokered self-host installed-repo sync (#5028): the central Orb relay deliberately does not forward + // installation/installation_repositories events to brokered containers, so a brokered self-host has no + // other way to learn its own repo list beyond the first forwarded PR/issue event per repo. Self-host + + // broker-mode only (isOrbBrokerMode reads ORB_ENROLLMENT_SECRET) — a no-op everywhere else, byte-identical. + if (selfHostedReviews && isOrbBrokerMode(env)) jobs.push({ type: "sync-brokered-installed-repos", requestedBy: "schedule" }); jobs.push({ type: "refresh-scoring-model", requestedBy: "schedule" }); jobs.push({ type: "refresh-upstream-drift", requestedBy: "schedule" }); jobs.push({ type: "rollup-product-usage", requestedBy: "schedule", days: 7 }); diff --git a/src/orb/installed-repos-sync.ts b/src/orb/installed-repos-sync.ts new file mode 100644 index 0000000000..497ef242e9 --- /dev/null +++ b/src/orb/installed-repos-sync.ts @@ -0,0 +1,80 @@ +// Brokered self-host installed-repo sync (#5028, part of the isRegistered/isInstalled untangling epic #5016). A +// brokered self-host learns its own repo list from GitHub directly, via its broker token -- the same list a +// non-brokered self-host or cloud installation already gets eagerly through real `installation`/ +// `installation_repositories` webhooks. The central Orb relay deliberately does NOT forward those two events to +// brokered containers (src/orb/relay.ts's RELAY_FORWARD_EVENTS -- the container runs under the CENTRAL Orb App, +// not its own, so it must not treat those as its own installation state). Without this sync, a brokered +// self-host only learns about a repo the FIRST time a forwarded PR/issue event arrives for it: a freshly +// enrolled, quiet repo has no local `repositories` row at all, and every core feature gated on `isInstalled` +// silently skips it. + +import { listInstalledRepoFullNamesForInstallation, markRepositoriesRemovedFromInstallation, upsertRepositoryFromGitHub } from "../db/repositories"; +import { githubHeaders } from "../github/client"; +import type { GitHubRepositoryPayload } from "../types"; +import { fetchBrokeredInstallationToken, isOrbBrokerMode } from "./broker-client"; + +const GITHUB_INSTALLATION_REPOS_PAGE_SIZE = 100; +// Bounds worst-case pagination for a single sync tick. A real maintainer's installation has a handful of repos; +// this caps runaway pagination against a misbehaving response to a sane worst case (5,000 repos) rather than +// looping unbounded. +const MAX_INSTALLATION_REPOS_PAGES = 50; + +export type InstalledReposSyncResult = + | { status: "skipped" } + | { status: "synced"; installationId: number; repoCount: number; removedCount: number } + | { status: "failed"; reason: string }; + +/** Fetch every repo currently accessible to this brokered installation via GitHub's own + * `GET /installation/repositories`, paginated. Uses the broker token directly -- its response already carries + * the bound installationId, so no separate token mint is needed for this call. */ +async function fetchAllInstallationRepos(token: string, fetchImpl: typeof fetch): Promise { + const repos: GitHubRepositoryPayload[] = []; + for (let page = 1; page <= MAX_INSTALLATION_REPOS_PAGES; page += 1) { + const res = await fetchImpl(`https://api.github.com/installation/repositories?per_page=${GITHUB_INSTALLATION_REPOS_PAGE_SIZE}&page=${page}`, { + headers: githubHeaders({ token }), + signal: AbortSignal.timeout(20_000), + }); + if (!res.ok) throw new Error(`installation_repositories_http_${res.status}`); + const body = (await res.json()) as { repositories?: GitHubRepositoryPayload[] }; + const batch = body.repositories ?? []; + repos.push(...batch); + if (batch.length < GITHUB_INSTALLATION_REPOS_PAGE_SIZE) break; + } + return repos; +} + +/** + * Sync this brokered self-host's `repositories.isInstalled` rows against GitHub's live installation-repos list: + * every returned repo is upserted with `isInstalled: true` (mirrors what the webhook handler already does on + * every forwarded event); every LOCAL repo previously marked installed under this installationId that is no + * longer in the fresh list is flipped to `isInstalled: false` (a repo removed from the installation, or moved + * out of a "selected" install's scope). + * + * No-op (`status: "skipped"`) outside broker mode -- a non-brokered self-host or cloud already gets this + * eagerly via real installation webhooks, and this must never run there (isOrbBrokerMode's signal, the + * enrollment secret's presence, is the same guard every other broker-only self-host path uses). + * + * Best-effort: any failure (broker down, GitHub throttled/erroring) returns `status: "failed"` rather than + * throwing, matching the fail-safe convention of every other cron sync in this codebase (e.g. + * registerOrbRelayTargetWithRetry) -- a sync miss self-heals on the next scheduled tick, never blocks the cron. + */ +export async function syncBrokeredInstalledRepos( + env: { ORB_ENROLLMENT_SECRET?: string | undefined; ORB_BROKER_URL?: string | undefined } & Env, + fetchImpl: typeof fetch = fetch, +): Promise { + if (!isOrbBrokerMode(env)) return { status: "skipped" }; + try { + const { token, installationId } = await fetchBrokeredInstallationToken(env, fetchImpl); + const repos = await fetchAllInstallationRepos(token, fetchImpl); + for (const repo of repos) { + await upsertRepositoryFromGitHub(env, repo, installationId); + } + const freshFullNames = new Set(repos.map((repo) => repo.full_name)); + const previouslyInstalled = await listInstalledRepoFullNamesForInstallation(env, installationId); + const staleFullNames = previouslyInstalled.filter((fullName) => !freshFullNames.has(fullName)); + await markRepositoriesRemovedFromInstallation(env, installationId, staleFullNames); + return { status: "synced", installationId, repoCount: repos.length, removedCount: staleFullNames.length }; + } catch (error) { + return { status: "failed", reason: error instanceof Error ? error.message : "sync_failed" }; + } +} diff --git a/src/queue/job-dispatch.ts b/src/queue/job-dispatch.ts index 35d4bfd096..6ca3b49051 100644 --- a/src/queue/job-dispatch.ts +++ b/src/queue/job-dispatch.ts @@ -32,6 +32,7 @@ import { runSelfTuneBreaker } from "../review/outcomes-wire"; import { isRagEnabled } from "../review/rag-wire"; import { processSubmitDraft } from "../services/draft"; import { retryFailedRelays } from "../orb/relay"; +import { syncBrokeredInstalledRepos } from "../orb/installed-repos-sync"; import { generateSignalSnapshots } from "./signal-snapshot"; import { runRetentionPrune } from "./retention"; // The 15 handlers below have no reason to move -- each is only reachable via this dispatcher (or, for @@ -69,6 +70,9 @@ export async function processJob(env: Env, message: JobMessage): Promise { case "refresh-registry": await refreshRegistry(env); return; + case "sync-brokered-installed-repos": + await syncBrokeredInstalledRepos(env); + return; case "backfill-registered-repos": if (!message.repoFullName && message.requestedBy !== "test") { const repositories = (await listRepositories(env)).filter( diff --git a/src/selfhost/maintenance-admission.ts b/src/selfhost/maintenance-admission.ts index f71d088843..a151730cea 100644 --- a/src/selfhost/maintenance-admission.ts +++ b/src/selfhost/maintenance-admission.ts @@ -43,6 +43,7 @@ import { deterministicJitterMs, parsePositiveIntEnv } from "./queue-common"; export const MAINTENANCE_JOB_TYPES: ReadonlySet = new Set([ "backfill-registered-repos", "refresh-registry", + "sync-brokered-installed-repos", "refresh-installation-health", "refresh-scoring-model", "refresh-upstream-drift", diff --git a/src/selfhost/queue-common.ts b/src/selfhost/queue-common.ts index f6c6a24959..1da60c11c4 100644 --- a/src/selfhost/queue-common.ts +++ b/src/selfhost/queue-common.ts @@ -118,6 +118,11 @@ const GITHUB_BUDGET_BACKGROUND_TYPES = new Set([ // runReviewRecapJob calls loadRepoFocusManifest directly for its one repo. Not yet cron-enqueued (manual/API // trigger only today, per its own doc comment), but still worth gating against a rapid repeated manual trigger. "generate-review-recap", + // syncBrokeredInstalledRepos (#5028) makes a real, paginated, authenticated `GET /installation/repositories` + // REST call using the brokered installation token -- unlike refresh-registry (an unauthenticated/raw-file + // fetch to entrius/gittensor, not the GitHub REST API), this genuinely draws down the shared installation's + // REST budget and must yield alongside every other budget consumer here. + "sync-brokered-installed-repos", ]); const PRIORITY_BY_TYPE = new Map([ ["agent-regate-pr", AGENT_REGATE_PRIORITY], @@ -926,6 +931,7 @@ export function jobCoalesceKey(payload: string): string | null { } switch (type) { case "refresh-registry": + case "sync-brokered-installed-repos": case "refresh-installation-health": case "refresh-scoring-model": case "refresh-upstream-drift": diff --git a/src/types.ts b/src/types.ts index 541bed4b73..9a24fcd9b2 100644 --- a/src/types.ts +++ b/src/types.ts @@ -56,6 +56,10 @@ export type JobMessage = type: "refresh-registry"; requestedBy: "schedule" | "api" | "test"; } + | { + type: "sync-brokered-installed-repos"; + requestedBy: "schedule" | "api" | "test"; + } | { type: "backfill-registered-repos"; requestedBy: "schedule" | "api" | "test"; diff --git a/test/unit/index.test.ts b/test/unit/index.test.ts index 91e1f20942..7c0a82d0ab 100644 --- a/test/unit/index.test.ts +++ b/test/unit/index.test.ts @@ -713,6 +713,46 @@ describe("worker entrypoint", () => { expect(sent.some((m) => m.type === "ops-alerts")).toBe(false); }); + it("enqueues sync-brokered-installed-repos hourly ONLY in broker mode (ORB_ENROLLMENT_SECRET set, flag-OFF is byte-identical)", async () => { + const sentFor = async (enrollmentSecret?: string): Promise> => { + const sent: Array = []; + const env = createTestEnv({ + ...(enrollmentSecret === undefined ? {} : { ORB_ENROLLMENT_SECRET: enrollmentSecret }), + JOBS: { + async send(message: import("../../src/types").JobMessage) { + sent.push(message); + }, + } as unknown as Queue, + }); + const waitUntil: Promise[] = []; + await worker.scheduled(controllerFor("2026-05-25T05:00:00.000Z"), env, executionContext(waitUntil)); + await Promise.all(waitUntil); + return sent; + }; + + // Non-brokered (default) → no sync job; the enqueued set is unchanged from today. + expect((await sentFor()).some((m) => m.type === "sync-brokered-installed-repos")).toBe(false); + // Brokered (an enrollment secret is configured) → exactly one sync job, enqueued in the hourly window. + const brokered = await sentFor("orbsec_x"); + expect(brokered.filter((m) => m.type === "sync-brokered-installed-repos")).toEqual([{ type: "sync-brokered-installed-repos", requestedBy: "schedule" }]); + }); + + it("does NOT enqueue sync-brokered-installed-repos outside the hourly window even in broker mode", async () => { + const sent: Array = []; + const env = createTestEnv({ + ORB_ENROLLMENT_SECRET: "orbsec_x", + JOBS: { + async send(message: import("../../src/types").JobMessage) { + sent.push(message); + }, + } as unknown as Queue, + }); + const waitUntil: Promise[] = []; + await worker.scheduled(controllerFor("2026-05-25T05:15:00.000Z"), env, executionContext(waitUntil)); // non-hourly + await Promise.all(waitUntil); + expect(sent.some((m) => m.type === "sync-brokered-installed-repos")).toBe(false); + }); + it("enqueues the sweep-liveness-watchdog job hourly ONLY when GITTENSORY_SWEEP_WATCHDOG is ON (flag-OFF is byte-identical)", async () => { const sentFor = async (watchdogFlag?: string): Promise> => { const sent: Array = []; diff --git a/test/unit/orb-installed-repos-sync.test.ts b/test/unit/orb-installed-repos-sync.test.ts new file mode 100644 index 0000000000..5a78a2d9b5 --- /dev/null +++ b/test/unit/orb-installed-repos-sync.test.ts @@ -0,0 +1,108 @@ +import { describe, expect, it } from "vitest"; +import { getRepository } from "../../src/db/repositories"; +import { syncBrokeredInstalledRepos } from "../../src/orb/installed-repos-sync"; +import { createTestEnv } from "../helpers/d1"; + +type Call = { url: string; init?: RequestInit | undefined }; + +/** Router-style fetch stub: the broker token exchange goes to `.../v1/orb/token`, everything else is treated + * as a GitHub `GET /installation/repositories` page request and answered from `pages` in order. */ +function routedFetch(args: { tokenResponse: Response; pages: Response[] }): { fetchImpl: typeof fetch; calls: Call[] } { + const calls: Call[] = []; + let pageIndex = 0; + const fetchImpl = (async (url: RequestInfo | URL, init?: RequestInit) => { + calls.push({ url: String(url), init }); + if (String(url).includes("/v1/orb/token")) return args.tokenResponse; + const page = args.pages[pageIndex]; + pageIndex += 1; + return page ?? Response.json({ repositories: [] }); + }) as typeof fetch; + return { fetchImpl, calls }; +} + +function tokenResponse(installationId = 42): Response { + return Response.json({ token: "ghs_x", installationId, expiresAt: "2026-06-25T09:00:00Z", permissions: { contents: "write" } }); +} + +function repoPayload(fullName: string) { + const [owner, name] = fullName.split("/"); + return { full_name: fullName, name, owner: { login: owner }, private: false, html_url: `https://github.com/${fullName}`, default_branch: "main" }; +} + +describe("syncBrokeredInstalledRepos", () => { + it("is a no-op outside broker mode (no enrollment secret)", async () => { + const env = createTestEnv(); + const { fetchImpl, calls } = routedFetch({ tokenResponse: tokenResponse(), pages: [Response.json({ repositories: [repoPayload("owner/repo")] })] }); + const result = await syncBrokeredInstalledRepos(env, fetchImpl); + expect(result).toEqual({ status: "skipped" }); + expect(calls).toHaveLength(0); + }); + + it("upserts every repo returned by GitHub as isInstalled, single page", async () => { + const env = createTestEnv({ ORB_ENROLLMENT_SECRET: "orbsec_x" }); + const { fetchImpl } = routedFetch({ + tokenResponse: tokenResponse(42), + pages: [Response.json({ repositories: [repoPayload("owner/repo-a"), repoPayload("owner/repo-b")] })], + }); + const result = await syncBrokeredInstalledRepos(env, fetchImpl); + expect(result).toEqual({ status: "synced", installationId: 42, repoCount: 2, removedCount: 0 }); + await expect(getRepository(env, "owner/repo-a")).resolves.toMatchObject({ isInstalled: true, installationId: 42 }); + await expect(getRepository(env, "owner/repo-b")).resolves.toMatchObject({ isInstalled: true, installationId: 42 }); + }); + + it("paginates until a short page, following GitHub's own page-size convention", async () => { + const env = createTestEnv({ ORB_ENROLLMENT_SECRET: "orbsec_x" }); + const fullPage = Response.json({ repositories: Array.from({ length: 100 }, (_, i) => repoPayload(`owner/repo-${i}`)) }); + const shortPage = Response.json({ repositories: [repoPayload("owner/repo-last")] }); + const { fetchImpl, calls } = routedFetch({ tokenResponse: tokenResponse(42), pages: [fullPage, shortPage] }); + const result = await syncBrokeredInstalledRepos(env, fetchImpl); + expect(result).toEqual({ status: "synced", installationId: 42, repoCount: 101, removedCount: 0 }); + // token exchange + 2 pages + expect(calls).toHaveLength(3); + await expect(getRepository(env, "owner/repo-last")).resolves.toMatchObject({ isInstalled: true }); + }); + + it("marks a previously-installed repo as no longer installed once GitHub stops returning it", async () => { + const env = createTestEnv({ ORB_ENROLLMENT_SECRET: "orbsec_x" }); + // First sync: two repos installed. + const first = routedFetch({ tokenResponse: tokenResponse(42), pages: [Response.json({ repositories: [repoPayload("owner/kept"), repoPayload("owner/removed")] })] }); + await syncBrokeredInstalledRepos(env, first.fetchImpl); + await expect(getRepository(env, "owner/removed")).resolves.toMatchObject({ isInstalled: true }); + + // Second sync: GitHub now only returns the kept repo. + const second = routedFetch({ tokenResponse: tokenResponse(42), pages: [Response.json({ repositories: [repoPayload("owner/kept")] })] }); + const result = await syncBrokeredInstalledRepos(env, second.fetchImpl); + expect(result).toEqual({ status: "synced", installationId: 42, repoCount: 1, removedCount: 1 }); + await expect(getRepository(env, "owner/kept")).resolves.toMatchObject({ isInstalled: true }); + await expect(getRepository(env, "owner/removed")).resolves.toMatchObject({ isInstalled: false, installationId: null }); + }); + + it("fails safe (never throws) when the broker token exchange fails", async () => { + const env = createTestEnv({ ORB_ENROLLMENT_SECRET: "orbsec_x" }); + const { fetchImpl } = routedFetch({ tokenResponse: new Response("nope", { status: 403 }), pages: [] }); + const result = await syncBrokeredInstalledRepos(env, fetchImpl); + expect(result).toMatchObject({ status: "failed" }); + expect((result as { status: "failed"; reason: string }).reason).toMatch(/403/); + }); + + it("fails safe (never throws) when the GitHub installation-repos call errors", async () => { + const env = createTestEnv({ ORB_ENROLLMENT_SECRET: "orbsec_x" }); + const { fetchImpl } = routedFetch({ tokenResponse: tokenResponse(42), pages: [new Response("rate limited", { status: 429 })] }); + const result = await syncBrokeredInstalledRepos(env, fetchImpl); + expect(result).toMatchObject({ status: "failed", reason: "installation_repositories_http_429" }); + }); + + it("falls back to a generic reason when a non-Error value rejects the fetch", async () => { + const env = createTestEnv({ ORB_ENROLLMENT_SECRET: "orbsec_x" }); + const fetchImpl = (() => Promise.reject("not an Error instance")) as typeof fetch; + const result = await syncBrokeredInstalledRepos(env, fetchImpl); + expect(result).toEqual({ status: "failed", reason: "sync_failed" }); + }); + + it("treats a missing repositories field on a page as an empty batch (ends pagination)", async () => { + const env = createTestEnv({ ORB_ENROLLMENT_SECRET: "orbsec_x" }); + const { fetchImpl } = routedFetch({ tokenResponse: tokenResponse(42), pages: [Response.json({})] }); + const result = await syncBrokeredInstalledRepos(env, fetchImpl); + expect(result).toEqual({ status: "synced", installationId: 42, repoCount: 0, removedCount: 0 }); + }); +}); diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index 1c7265cb7c..f5bde33cad 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -389,6 +389,7 @@ describe("queue processors", () => { }); await processJob(env, { type: "refresh-registry", requestedBy: "test" }); + await processJob(env, { type: "sync-brokered-installed-repos", requestedBy: "test" }); await processJob(env, { type: "refresh-scoring-model", requestedBy: "test" }); await processJob(env, { type: "backfill-registered-repos", requestedBy: "test", repoFullName: "JSONbored/gittensory", force: true }); await processJob(env, { type: "generate-signal-snapshots", requestedBy: "test", repoFullName: "JSONbored/gittensory" });