diff --git a/src/github/backfill.ts b/src/github/backfill.ts index 9544ca50fa..5b231c62a5 100644 --- a/src/github/backfill.ts +++ b/src/github/backfill.ts @@ -115,6 +115,11 @@ type BackfillLimits = { type BackfillMode = "light" | "full" | "resume"; type BackfillSegmentName = "labels" | "open_issues" | "open_pull_requests" | "recent_merged_pull_requests"; +type GitHubConditionalValidators = { etag?: string | null | undefined; lastModified?: string | null | undefined }; +type GitHubJsonResponse = { data: T; link: string | null; etag: string | null; lastModified: string | null }; +type GitHubJsonNotModifiedResponse = { notModified: true; link: string | null; etag: string | null; lastModified: string | null }; +type GitHubJsonConditionalResponse = GitHubJsonResponse | GitHubJsonNotModifiedResponse; +type GitHubSegmentConditionalRequest = { previous: RepoSyncSegmentRecord; validators: GitHubConditionalValidators }; export type BackfillRegisteredReposResult = { ok: true; @@ -479,7 +484,7 @@ export async function backfillRepositorySegment( { delaySeconds }, ); } - if (options.segment === "open_pull_requests" && result.status === "complete") { + if (options.segment === "open_pull_requests" && (result.status === "complete" || result.status === "not_modified")) { await env.JOBS.send({ type: "backfill-pr-details", requestedBy: "api", repoFullName: repo.fullName, mode: "resume", cursor: 0 }, { delaySeconds: 10 }); } await refreshRepoSyncStateFromSegments(env, repo, sourceKind); @@ -1243,6 +1248,25 @@ async function backfillRecentMergedSegment( ); } +function isNotModifiedResponse(result: GitHubJsonConditionalResponse): result is GitHubJsonNotModifiedResponse { + return "notModified" in result && result.notModified; +} + +function conditionalRequestForSegment( + previous: RepoSyncSegmentRecord | null, + expectedCount: number | undefined, + options: { allowEtag: boolean }, +): GitHubSegmentConditionalRequest | undefined { + if (!previous) return undefined; + if (!isFreshSegmentStatus(previous.status)) return undefined; + if (previous.lastCursor !== "1") return undefined; + if (previous.nextCursor) return undefined; + if (expectedCount !== undefined && previous.expectedCount !== expectedCount) return undefined; + const etag = options.allowEtag && previous.etag !== CURRENT_OPEN_SCAN_MARKER ? previous.etag : undefined; + const lastModified = previous.lastModified; + return etag || lastModified ? { previous, validators: { etag, lastModified } } : undefined; +} + async function fetchPagedSegment( env: Env, repo: RepositoryRecord, @@ -1286,13 +1310,38 @@ async function fetchPagedSegment( let pageCount = 0; let hasMore = false; let rateLimitResetAt: string | undefined; + let etag: string | null | undefined; + let lastModified: string | null | undefined; const warnings: string[] = []; let status: RepoSyncSegmentRecord["status"] = "complete"; + const conditionalRequest = + startPage === 1 + ? conditionalRequestForSegment(previous, expectedCount, { allowEtag: !requiresCurrentOpenScan }) + : undefined; try { for (let page = startPage; page < startPage + SEGMENT_PAGE_BUDGET[mode]; page += 1) { const separator = path.includes("?") ? "&" : "?"; const pagePath = `${path}${separator}per_page=100&page=${page}`; - const result = await githubJsonWithHeaders(env, repo.fullName, pagePath, token); + let result: GitHubJsonResponse; + if (conditionalRequest && page === 1) { + const conditionalResult = await githubJsonWithHeaders(env, repo.fullName, pagePath, token, { validators: conditionalRequest.validators, allowNotModified: true }); + if (isNotModifiedResponse(conditionalResult)) { + const previousSegment = conditionalRequest.previous; + status = "not_modified"; + lastCursor = "1"; + pageCount = previousSegment.pageCount; + etag = requiresCurrentOpenScan ? CURRENT_OPEN_SCAN_MARKER : conditionalResult.etag ?? previousSegment.etag; + lastModified = conditionalResult.lastModified ?? previousSegment.lastModified; + hasMore = false; + nextCursor = undefined; + break; + } + result = conditionalResult; + } else { + result = await githubJsonWithHeaders(env, repo.fullName, pagePath, token); + } + etag = result.etag ?? etag; + lastModified = result.lastModified ?? lastModified; lastCursor = String(page); pageCount += 1; fetchedThisRun += await persistPage(result.data, startedAt); @@ -1350,7 +1399,8 @@ async function fetchPagedSegment( pageCount, lastCursor, nextCursor, - etag: requiresCurrentOpenScan ? CURRENT_OPEN_SCAN_MARKER : undefined, + etag: requiresCurrentOpenScan ? CURRENT_OPEN_SCAN_MARKER : etag, + lastModified, warnings, errorSummary: status === "error" || status === "waiting_rate_limit" || status === "partial" ? warnings.at(-1) : undefined, rateLimitResetAt, @@ -1508,6 +1558,10 @@ function isTerminalSegmentStatus(status: RepoSyncSegmentRecord["status"]): boole return status === "complete" || status === "not_modified" || status === "sampled"; } +function isFreshSegmentStatus(status: RepoSyncSegmentRecord["status"]): boolean { + return status === "complete" || status === "not_modified"; +} + async function refreshRepoSyncStateFromSegments(env: Env, repo: RepositoryRecord, sourceKind: RepoSyncSegmentRecord["sourceKind"]): Promise { const [previous, totalsSnapshot, metadata, labels, openIssues, openPullRequests, recentMerged, files, reviews, checks] = await Promise.all([ getRepoSyncState(env, repo.fullName), @@ -1543,10 +1597,10 @@ async function refreshRepoSyncStateFromSegments(env: Env, repo: RepositoryRecord openIssuesCount: openIssues?.fetchedCount ?? previous?.openIssuesCount ?? totals?.openIssuesTotal ?? 0, openPullRequestsCount: openPullRequests?.fetchedCount ?? previous?.openPullRequestsCount ?? totals?.openPullRequestsTotal ?? 0, recentMergedPullRequestsCount: recentMerged?.fetchedCount ?? previous?.recentMergedPullRequestsCount ?? 0, - labelsSyncedAt: labels?.status === "complete" ? labels.completedAt : previous?.labelsSyncedAt, - issuesSyncedAt: openIssues?.status === "complete" ? openIssues.completedAt : previous?.issuesSyncedAt, - pullRequestsSyncedAt: openPullRequests?.status === "complete" ? openPullRequests.completedAt : previous?.pullRequestsSyncedAt, - mergedPullRequestsSyncedAt: recentMerged?.status === "complete" || recentMerged?.status === "sampled" ? recentMerged.completedAt : previous?.mergedPullRequestsSyncedAt, + labelsSyncedAt: labels && isFreshSegmentStatus(labels.status) ? labels.completedAt : previous?.labelsSyncedAt, + issuesSyncedAt: openIssues && isFreshSegmentStatus(openIssues.status) ? openIssues.completedAt : previous?.issuesSyncedAt, + pullRequestsSyncedAt: openPullRequests && isFreshSegmentStatus(openPullRequests.status) ? openPullRequests.completedAt : previous?.pullRequestsSyncedAt, + mergedPullRequestsSyncedAt: recentMerged && (isFreshSegmentStatus(recentMerged.status) || recentMerged.status === "sampled") ? recentMerged.completedAt : previous?.mergedPullRequestsSyncedAt, lastStartedAt: previous?.lastStartedAt, lastCompletedAt: completedAt, errorSummary: warnings.at(-1), @@ -2737,19 +2791,35 @@ async function githubJsonWithHeaders( repoFullName: string, path: string, token?: string, -): Promise<{ data: T; link: string | null; etag: string | null; lastModified: string | null }> { +): Promise>; +async function githubJsonWithHeaders( + env: Env, + repoFullName: string, + path: string, + token: string | undefined, + options: { validators?: GitHubConditionalValidators; allowNotModified: true }, +): Promise>; +async function githubJsonWithHeaders( + env: Env, + repoFullName: string, + path: string, + token?: string, + options?: { validators?: GitHubConditionalValidators; allowNotModified?: boolean }, +): Promise> { const { owner, name } = repoParts(repoFullName); const url = `https://api.github.com/repos/${encodeURIComponent(owner)}/${encodeURIComponent(name)}${path}`; - let response = await timeoutFetch(url, { headers: githubRestHeaders(token) }); + let response = await timeoutFetch(url, { headers: githubRestHeaders(token, options?.validators) }); if (!isGitHubResponseCacheReplay(response)) { await recordGitHubResponse(env, repoFullName, path, response, "rest"); } + if (response.status === 304 && options?.allowNotModified) return notModifiedResponse(response); if (response.status === 404 && token && token === env.GITHUB_PUBLIC_TOKEN) { - response = await timeoutFetch(url, { headers: githubRestHeaders() }); + response = await timeoutFetch(url, { headers: githubRestHeaders(undefined, options?.validators) }); // Do not persist unauthenticated fallback rate-limit headers into the shared REST backoff state. // GitHub's unauthenticated REST bucket is capped below LOW_REST_RATE_LIMIT_REMAINING, so recording // successful fallback responses can incorrectly stall later token-backed segment jobs. } + if (response.status === 304 && options?.allowNotModified) return notModifiedResponse(response); if (!response.ok) { const body = await response.text(); throw new GitHubApiError( @@ -2769,11 +2839,22 @@ async function githubJsonWithHeaders( }; } -function githubRestHeaders(token?: string): HeadersInit { +function notModifiedResponse(response: Response): GitHubJsonNotModifiedResponse { + return { + notModified: true, + link: response.headers.get("link"), + etag: response.headers.get("etag"), + lastModified: response.headers.get("last-modified"), + }; +} + +function githubRestHeaders(token?: string, validators?: GitHubConditionalValidators): HeadersInit { return { accept: "application/vnd.github+json", "user-agent": "gittensory/0.1", "x-github-api-version": "2022-11-28", + ...(validators?.etag ? { "if-none-match": validators.etag } : {}), + ...(validators?.lastModified ? { "if-modified-since": validators.lastModified } : {}), ...(token ? { authorization: `Bearer ${token}` } : {}), }; } diff --git a/test/unit/backfill.test.ts b/test/unit/backfill.test.ts index 8d10471c80..ff16a44d09 100644 --- a/test/unit/backfill.test.ts +++ b/test/unit/backfill.test.ts @@ -22,6 +22,7 @@ import { upsertPullRequestFile, upsertPullRequestFromGitHub, upsertIssueFromGitHub, + upsertRepoLabel, upsertRepositoryFromGitHub, upsertRepositorySettings, } from "../../src/db/repositories"; @@ -1224,6 +1225,263 @@ describe("GitHub backfill", () => { expect(await listRepoLabels(env, "JSONbored/gittensory")).toEqual(expect.arrayContaining([expect.objectContaining({ name: "bug" })])); }); + it("validates unchanged single-page label segments with conditional REST requests", async () => { + const env = createTestEnv({ GITHUB_PUBLIC_TOKEN: "public-token" }); + await seedRegisteredRepo(env); + const labelHeaders: Array<{ ifNoneMatch: string | null; ifModifiedSince: string | null }> = []; + let labelFetches = 0; + vi.stubGlobal("fetch", async (input: RequestInfo | URL, init?: RequestInit) => { + const url = input.toString(); + if (url === "https://api.github.com/graphql") return githubTotalsResponse({ openIssues: 0, openPullRequests: 0, mergedPullRequests: 0, closedPullRequests: 0, labels: 1 }); + if (url.includes("/labels?")) { + const headers = new Headers(init?.headers); + labelHeaders.push({ ifNoneMatch: headers.get("if-none-match"), ifModifiedSince: headers.get("if-modified-since") }); + labelFetches += 1; + if (labelFetches === 1) { + return Response.json([{ name: "bug", color: "cc0000", description: "Bug" }], { + headers: { etag: '"labels-v1"', "last-modified": "Tue, 26 May 2026 00:00:00 GMT" }, + }); + } + if (labelFetches === 2) { + return new Response(null, { status: 304, headers: { etag: '"labels-v1"', "last-modified": "Tue, 26 May 2026 00:00:00 GMT" } }); + } + return Response.json([{ name: "bug", color: "00cc00", description: "Bug" }], { + headers: { etag: '"labels-v2"', "last-modified": "Tue, 26 May 2026 00:05:00 GMT" }, + }); + } + return new Response("not found", { status: 404 }); + }); + + const first = await backfillRepositorySegment(env, { repoFullName: "JSONbored/gittensory", segment: "labels", mode: "resume", force: true }); + const second = await backfillRepositorySegment(env, { repoFullName: "JSONbored/gittensory", segment: "labels", mode: "resume", force: true }); + const third = await backfillRepositorySegment(env, { repoFullName: "JSONbored/gittensory", segment: "labels", mode: "resume", force: true }); + + expect(first).toMatchObject({ status: "complete", fetchedCount: 1, expectedCount: 1 }); + expect(second).toMatchObject({ status: "not_modified", fetchedCount: 1, expectedCount: 1 }); + expect(third).toMatchObject({ status: "complete", fetchedCount: 1, expectedCount: 1 }); + expect(labelHeaders).toEqual([ + { ifNoneMatch: null, ifModifiedSince: null }, + { ifNoneMatch: '"labels-v1"', ifModifiedSince: "Tue, 26 May 2026 00:00:00 GMT" }, + { ifNoneMatch: '"labels-v1"', ifModifiedSince: "Tue, 26 May 2026 00:00:00 GMT" }, + ]); + expect(await listRepoLabels(env, "JSONbored/gittensory")).toEqual(expect.arrayContaining([expect.objectContaining({ name: "bug", color: "00cc00" })])); + expect(await listRepoSyncSegments(env, "JSONbored/gittensory")).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + segment: "labels", + status: "complete", + fetchedCount: 1, + pageCount: 1, + lastCursor: "1", + etag: '"labels-v2"', + lastModified: "Tue, 26 May 2026 00:05:00 GMT", + }), + ]), + ); + expect(await listRepoSyncStates(env)).toEqual( + expect.arrayContaining([expect.objectContaining({ repoFullName: "JSONbored/gittensory", status: "success", labelsSyncedAt: expect.any(String) })]), + ); + }); + + it("preserves stored validators when an unauthenticated fallback returns not modified without validators", async () => { + const env = createTestEnv({ GITHUB_PUBLIC_TOKEN: "public-token" }); + await seedRegisteredRepo(env); + await upsertRepoLabel(env, { + repoFullName: "JSONbored/gittensory", + name: "bug", + color: "cc0000", + description: "Bug", + isConfigured: true, + observedCount: 0, + payload: {}, + lastSeenAt: "2026-05-26T00:00:00.000Z", + }); + await upsertRepoSyncSegment(env, { + repoFullName: "JSONbored/gittensory", + segment: "labels", + status: "complete", + sourceKind: "github", + mode: "resume", + fetchedCount: 1, + expectedCount: 1, + pageCount: 1, + lastCursor: "1", + etag: '"labels-v1"', + lastModified: "Tue, 26 May 2026 00:00:00 GMT", + warnings: [], + }); + const labelRequests: Array<{ auth: string | null; ifNoneMatch: string | null; ifModifiedSince: string | null }> = []; + vi.stubGlobal("fetch", async (input: RequestInfo | URL, init?: RequestInit) => { + const url = input.toString(); + const headers = new Headers(init?.headers); + if (url === "https://api.github.com/graphql") return githubTotalsResponse({ openIssues: 0, openPullRequests: 0, mergedPullRequests: 0, closedPullRequests: 0, labels: 1 }); + if (url.includes("/labels?")) { + labelRequests.push({ + auth: headers.get("authorization"), + ifNoneMatch: headers.get("if-none-match"), + ifModifiedSince: headers.get("if-modified-since"), + }); + if (headers.get("authorization") === "Bearer public-token") return new Response("", { status: 404 }); + return new Response(null, { status: 304 }); + } + return new Response("not found", { status: 404 }); + }); + + const result = await backfillRepositorySegment(env, { repoFullName: "JSONbored/gittensory", segment: "labels", mode: "resume", force: true }); + + expect(result).toMatchObject({ status: "not_modified", fetchedCount: 1, expectedCount: 1 }); + expect(labelRequests).toEqual([ + { auth: "Bearer public-token", ifNoneMatch: '"labels-v1"', ifModifiedSince: "Tue, 26 May 2026 00:00:00 GMT" }, + { auth: null, ifNoneMatch: '"labels-v1"', ifModifiedSince: "Tue, 26 May 2026 00:00:00 GMT" }, + ]); + expect(await listRepoSyncSegments(env, "JSONbored/gittensory")).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + segment: "labels", + status: "not_modified", + etag: '"labels-v1"', + lastModified: "Tue, 26 May 2026 00:00:00 GMT", + }), + ]), + ); + }); + + it("uses last-modified validators for unchanged current open PR scans while preserving the resume marker", async () => { + const sent: Array<{ message: import("../../src/types").JobMessage; options?: QueueSendOptions }> = []; + const env = createTestEnv({ + GITHUB_PUBLIC_TOKEN: "public-token", + JOBS: { + async send(message: import("../../src/types").JobMessage, options?: QueueSendOptions) { + sent.push(options ? { message, options } : { message }); + }, + } as unknown as Queue, + }); + await seedRegisteredRepo(env); + const prHeaders: Array<{ ifNoneMatch: string | null; ifModifiedSince: string | null }> = []; + let prFetches = 0; + vi.stubGlobal("fetch", async (input: RequestInfo | URL, init?: RequestInit) => { + const url = input.toString(); + if (url === "https://api.github.com/graphql") return githubTotalsResponse({ openIssues: 0, openPullRequests: 1, mergedPullRequests: 0, closedPullRequests: 0, labels: 0 }); + if (url.includes("/pulls?state=open")) { + const headers = new Headers(init?.headers); + prHeaders.push({ ifNoneMatch: headers.get("if-none-match"), ifModifiedSince: headers.get("if-modified-since") }); + prFetches += 1; + if (prFetches === 1) { + return Response.json( + [{ number: 7, title: "Current PR", state: "open", user: { login: "oktofeesh1" }, head: { sha: "sha7" }, labels: [], body: "" }], + { headers: { etag: '"open-prs-v1"', "last-modified": "Tue, 26 May 2026 01:00:00 GMT" } }, + ); + } + return new Response(null, { status: 304, headers: { "last-modified": "Tue, 26 May 2026 01:00:00 GMT" } }); + } + return Response.json([]); + }); + + const first = await backfillRepositorySegment(env, { repoFullName: "JSONbored/gittensory", segment: "open_pull_requests", mode: "resume", force: true }); + const second = await backfillRepositorySegment(env, { repoFullName: "JSONbored/gittensory", segment: "open_pull_requests", mode: "resume", force: true }); + + expect(first).toMatchObject({ status: "complete", fetchedCount: 1, expectedCount: 1 }); + expect(second).toMatchObject({ status: "not_modified", fetchedCount: 1, expectedCount: 1 }); + expect(prHeaders).toEqual([ + { ifNoneMatch: null, ifModifiedSince: null }, + { ifNoneMatch: null, ifModifiedSince: "Tue, 26 May 2026 01:00:00 GMT" }, + ]); + expect(await listRepoSyncSegments(env, "JSONbored/gittensory")).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + segment: "open_pull_requests", + status: "not_modified", + etag: "gittensory-current-open-scan-v1", + lastModified: "Tue, 26 May 2026 01:00:00 GMT", + }), + ]), + ); + expect(sent.filter((item) => item.message.type === "backfill-pr-details")).toHaveLength(2); + }); + + it("bypasses conditional validators when segment totals change", async () => { + const env = createTestEnv({ GITHUB_PUBLIC_TOKEN: "public-token" }); + await seedRegisteredRepo(env); + await upsertRepoSyncSegment(env, { + repoFullName: "JSONbored/gittensory", + segment: "labels", + status: "complete", + sourceKind: "github", + mode: "resume", + fetchedCount: 1, + expectedCount: 1, + pageCount: 1, + lastCursor: "1", + etag: '"labels-v1"', + lastModified: "Tue, 26 May 2026 00:00:00 GMT", + warnings: [], + }); + const labelHeaders: Array<{ ifNoneMatch: string | null; ifModifiedSince: string | null }> = []; + vi.stubGlobal("fetch", async (input: RequestInfo | URL, init?: RequestInit) => { + const url = input.toString(); + if (url === "https://api.github.com/graphql") return githubTotalsResponse({ openIssues: 0, openPullRequests: 0, mergedPullRequests: 0, closedPullRequests: 0, labels: 2 }); + if (url.includes("/labels?")) { + const headers = new Headers(init?.headers); + labelHeaders.push({ ifNoneMatch: headers.get("if-none-match"), ifModifiedSince: headers.get("if-modified-since") }); + return Response.json([ + { name: "bug", color: "cc0000", description: "Bug" }, + { name: "feature", color: "00cc00", description: "Feature" }, + ]); + } + return new Response("not found", { status: 404 }); + }); + + const result = await backfillRepositorySegment(env, { repoFullName: "JSONbored/gittensory", segment: "labels", mode: "resume", force: true }); + + expect(result).toMatchObject({ status: "complete", fetchedCount: 2, expectedCount: 2 }); + expect(labelHeaders).toEqual([{ ifNoneMatch: null, ifModifiedSince: null }]); + }); + + it("bypasses conditional validators for prior segment rows that are not a complete single page", async () => { + for (const previous of [ + { lastCursor: "2", nextCursor: undefined }, + { lastCursor: "1", nextCursor: "2" }, + ]) { + const env = createTestEnv({ GITHUB_PUBLIC_TOKEN: "public-token" }); + await seedRegisteredRepo(env); + await upsertRepoSyncSegment(env, { + repoFullName: "JSONbored/gittensory", + segment: "labels", + status: "complete", + sourceKind: "github", + mode: "resume", + fetchedCount: 2, + expectedCount: 2, + pageCount: 2, + lastCursor: previous.lastCursor, + nextCursor: previous.nextCursor, + etag: '"labels-v1"', + lastModified: "Tue, 26 May 2026 00:00:00 GMT", + warnings: [], + }); + const labelHeaders: Array<{ ifNoneMatch: string | null; ifModifiedSince: string | null }> = []; + vi.stubGlobal("fetch", async (input: RequestInfo | URL, init?: RequestInit) => { + const url = input.toString(); + if (url === "https://api.github.com/graphql") return githubTotalsResponse({ openIssues: 0, openPullRequests: 0, mergedPullRequests: 0, closedPullRequests: 0, labels: 2 }); + if (url.includes("/labels?")) { + const headers = new Headers(init?.headers); + labelHeaders.push({ ifNoneMatch: headers.get("if-none-match"), ifModifiedSince: headers.get("if-modified-since") }); + return Response.json([ + { name: "bug", color: "cc0000", description: "Bug" }, + { name: "feature", color: "00cc00", description: "Feature" }, + ]); + } + return new Response("not found", { status: 404 }); + }); + + const result = await backfillRepositorySegment(env, { repoFullName: "JSONbored/gittensory", segment: "labels", mode: "resume", force: true }); + + expect(result).toMatchObject({ status: "complete", fetchedCount: 2, expectedCount: 2 }); + expect(labelHeaders).toEqual([{ ifNoneMatch: null, ifModifiedSince: null }]); + vi.unstubAllGlobals(); + } + }); + it("resumes paginated segments from stored cursors instead of restarting from page one", async () => { const env = createTestEnv({ GITHUB_PUBLIC_TOKEN: "public-token" }); await seedRegisteredRepo(env);