diff --git a/src/inngest/functions/issues-sweep.test.ts b/src/inngest/functions/issues-sweep.test.ts index 8fd49cd6..ea295e17 100644 --- a/src/inngest/functions/issues-sweep.test.ts +++ b/src/inngest/functions/issues-sweep.test.ts @@ -24,6 +24,37 @@ vi.mock('../client', () => ({ const run = issuesSweep as unknown as (ctx: { step: typeof step }) => Promise; +const installsMock = () => + sb({ limit: vi.fn().mockResolvedValue({ data: [{ id: 1, account_login: 'test-org' }] }) }); + +const reposMock = (rows: Array<{ repo_full_name: string }>) => + sb({ limit: vi.fn().mockResolvedValue({ data: rows }) }); + +const baseOctokit = () => ({ + repos: { + get: vi.fn().mockResolvedValue({ data: { fork: false, parent: null } }), + }, + issues: { + listForRepo: vi.fn().mockResolvedValue({ + data: [ + { + number: 101, + title: 'Fix bug', + body: 'Bug description', + html_url: 'https://github.com/test-org/repo-1/issues/101', + comments: 2, + labels: ['bug'], + }, + { + number: 102, + title: 'Is a PR', + pull_request: {}, // Should be skipped + }, + ], + }), + }, +}); + describe('issuesSweep', () => { beforeEach(() => { vi.clearAllMocks(); @@ -32,48 +63,12 @@ describe('issuesSweep', () => { it('sweeps issues and triggers recommendations build', async () => { const issues = sb({ upsert: vi.fn().mockResolvedValue({}) }); wire({ - github_installations: sb({ - select: vi.fn().mockReturnThis(), - is: vi.fn().mockReturnValue({ - is: vi.fn().mockResolvedValue({ data: [{ id: 1, account_login: 'test-org' }] }), - }), - }), - installation_repositories: sb({ - select: vi.fn().mockReturnThis(), - eq: vi.fn().mockResolvedValue({ - data: [{ repo_full_name: 'test-org/repo-1' }], - }), - }), + github_installations: installsMock(), + installation_repositories: reposMock([{ repo_full_name: 'test-org/repo-1' }]), issues, }); - const octokit = { - repos: { - get: vi.fn().mockResolvedValue({ - data: { fork: false, parent: null }, - }), - }, - issues: { - listForRepo: vi.fn().mockResolvedValue({ - data: [ - { - number: 101, - title: 'Fix bug', - body: 'Bug description', - html_url: 'https://github.com/test-org/repo-1/issues/101', - comments: 2, - labels: ['bug'], - }, - { - number: 102, - title: 'Is a PR', - pull_request: {}, // Should be skipped - }, - ], - }), - }, - }; - vi.mocked(getInstallOctokit).mockResolvedValue(octokit as never); + vi.mocked(getInstallOctokit).mockResolvedValue(baseOctokit() as never); vi.mocked(fetchRepoMetrics).mockResolvedValue({ language: 'TypeScript' } as never); vi.mocked(repoHealth).mockReturnValue(85); vi.mocked(scoreDifficulty).mockResolvedValue({ @@ -108,18 +103,8 @@ describe('issuesSweep', () => { it('handles github api errors gracefully', async () => { wire({ - github_installations: sb({ - select: vi.fn().mockReturnThis(), - is: vi.fn().mockReturnValue({ - is: vi.fn().mockResolvedValue({ data: [{ id: 1, account_login: 'test-org' }] }), - }), - }), - installation_repositories: sb({ - select: vi.fn().mockReturnThis(), - eq: vi.fn().mockResolvedValue({ - data: [{ repo_full_name: 'test-org/repo-1' }], - }), - }), + github_installations: installsMock(), + installation_repositories: reposMock([{ repo_full_name: 'test-org/repo-1' }]), }); vi.mocked(getInstallOctokit).mockRejectedValue(new Error('Bad credentials')); @@ -142,15 +127,9 @@ describe('issuesSweep', () => { it('self-heals empty installation_repositories by discovering repos via GitHub API and triggering pr-backfill', async () => { const reposUpsert = vi.fn().mockResolvedValue({}); wire({ - github_installations: sb({ - select: vi.fn().mockReturnThis(), - is: vi.fn().mockReturnValue({ - is: vi.fn().mockResolvedValue({ data: [{ id: 1, account_login: 'test-org' }] }), - }), - }), + github_installations: installsMock(), installation_repositories: sb({ - select: vi.fn().mockReturnThis(), - eq: vi.fn().mockResolvedValue({ data: [] }), // Initially empty! + limit: vi.fn().mockResolvedValue({ data: [] }), upsert: reposUpsert, }), }); @@ -168,6 +147,8 @@ describe('issuesSweep', () => { }, }; vi.mocked(getInstallOctokit).mockResolvedValue(octokit as never); + vi.mocked(fetchRepoMetrics).mockResolvedValue({ language: 'TypeScript' } as never); + vi.mocked(repoHealth).mockReturnValue(85); const result = await run({ step }); @@ -191,4 +172,125 @@ describe('issuesSweep', () => { }), ); }); + + it('reuses cached difficulty without invoking the LLM', async () => { + const issuesUpsert = vi.fn().mockResolvedValue({}); + wire({ + github_installations: installsMock(), + installation_repositories: reposMock([{ repo_full_name: 'test-org/repo-1' }]), + issues: sb({ + select: vi.fn().mockReturnThis(), + eq: vi.fn().mockReturnThis(), + in: vi.fn().mockResolvedValue({ + data: [ + { + github_issue_number: 101, + difficulty: 'H', + difficulty_source: 'label', + xp_reward: 250, + scored_at: new Date(Date.now() - 48 * 60 * 60 * 1000).toISOString(), + }, + ], + }), + upsert: issuesUpsert, + }), + }); + + vi.mocked(getInstallOctokit).mockResolvedValue(baseOctokit() as never); + vi.mocked(fetchRepoMetrics).mockResolvedValue({ language: 'TypeScript' } as never); + vi.mocked(repoHealth).mockReturnValue(85); + + await run({ step }); + + expect(scoreDifficulty).not.toHaveBeenCalled(); + expect(issuesUpsert).toHaveBeenCalledWith( + expect.objectContaining({ + github_issue_number: 101, + difficulty: 'H', + difficulty_source: 'label', + xp_reward: 250, + }), + { onConflict: 'repo_full_name,github_issue_number' }, + ); + }); + + it('skips re-scoring issues attempted within the last 24h', async () => { + const issuesUpsert = vi.fn().mockResolvedValue({}); + wire({ + github_installations: installsMock(), + installation_repositories: reposMock([{ repo_full_name: 'test-org/repo-1' }]), + issues: sb({ + select: vi.fn().mockReturnThis(), + eq: vi.fn().mockReturnThis(), + in: vi.fn().mockResolvedValue({ + data: [ + { + github_issue_number: 101, + difficulty: null, + difficulty_source: null, + xp_reward: 0, + scored_at: new Date().toISOString(), + }, + ], + }), + upsert: issuesUpsert, + }), + }); + + vi.mocked(getInstallOctokit).mockResolvedValue(baseOctokit() as never); + vi.mocked(fetchRepoMetrics).mockResolvedValue({ language: 'TypeScript' } as never); + vi.mocked(repoHealth).mockReturnValue(85); + + await run({ step }); + + expect(scoreDifficulty).not.toHaveBeenCalled(); + expect(issuesUpsert).not.toHaveBeenCalled(); + }); + + it('does not advance the per-repo cursor when the issue budget truncates a full page', async () => { + const cursorUpsert = vi.fn().mockResolvedValue({}); + const cursorDelete = vi.fn().mockResolvedValue({}); + wire({ + github_installations: installsMock(), + installation_repositories: reposMock([ + { repo_full_name: 'test-org/repo-1' }, + { repo_full_name: 'test-org/repo-2' }, + { repo_full_name: 'test-org/repo-3' }, + { repo_full_name: 'test-org/repo-4' }, + ]), + issues: sb({ upsert: vi.fn().mockResolvedValue({}) }), + repo_sync_cursors: sb({ upsert: cursorUpsert, delete: cursorDelete }), + }); + + const fullPage = Array.from({ length: 30 }, (_, i) => ({ + number: 100 + i, + title: `issue ${i}`, + body: 'body', + html_url: 'https://github.com/test-org/repo-1/issues/100', + comments: 0, + labels: [], + })); + + const octokit = { + repos: { get: vi.fn().mockResolvedValue({ data: { fork: false, parent: null } }) }, + issues: { listForRepo: vi.fn().mockResolvedValue({ data: fullPage }) }, + }; + vi.mocked(getInstallOctokit).mockResolvedValue(octokit as never); + vi.mocked(fetchRepoMetrics).mockResolvedValue({ language: 'TypeScript' } as never); + vi.mocked(repoHealth).mockReturnValue(85); + vi.mocked(scoreDifficulty).mockResolvedValue({ + difficulty: 'M', + source: 'llm', + confidence: 1, + xpReward: 100, + }); + + await run({ step }); + + // Repos 1-3 drain full pages (30/30) and advance their cursors; repo-4 + // hits the install-wide budget mid-page (10/30), so its cursor must be + // left in place so the unprocessed issues are retried next sweep. + expect(cursorUpsert).toHaveBeenCalledTimes(3); + expect(cursorDelete).not.toHaveBeenCalled(); + }); }); diff --git a/src/inngest/functions/issues-sweep.ts b/src/inngest/functions/issues-sweep.ts index 0072d7cd..b3af550a 100644 --- a/src/inngest/functions/issues-sweep.ts +++ b/src/inngest/functions/issues-sweep.ts @@ -6,21 +6,52 @@ import { scoreDifficulty, repoHealth } from '@/lib/pipeline/score'; import { fetchRepoMetrics } from '@/lib/github/repo-meta'; import { llmCall } from '@/lib/llm/router'; import { DifficultySchema } from '@/lib/llm/schemas'; +import { getSyncCursor, setSyncCursor, clearSyncCursor } from '@/lib/maintainer/sync-cursor'; /** * Pulls open issues from every active GitHub App install, scores difficulty, * upserts into the issues table. * - * Cron: every 30 min. The function is split into named steps so the run + * Cron: every 12 hours. The function is split into named steps so the run * trace shows where rows drop. Each step returns counts + a sample so a * single Inngest run trace tells us exactly what's happening. + * + * Cost bounds (per run): + * - installs capped (rotated round-robin by least-recently-swept) + * - repos per install capped (rotated round-robin by least-recently-swept) + * - issues per install and per sweep capped + * - issues scored within the last 24h are skipped (scored_at cooldown) + * - each repo is its own step.run so a failure only retries that repo */ +const MAX_INSTALLS_PER_SWEEP = 50; +const MAX_REPOS_PER_INSTALL = 20; +const MAX_ISSUES_PER_INSTALL = 100; +const MAX_ISSUES_PER_SWEEP = 1000; +const ISSUES_PER_REPO_PAGE = 30; +const SCORE_COOLDOWN_MS = 24 * 60 * 60 * 1000; +const SYNC_TYPE = 'issues_sweep'; + type RepoRow = { repo_full_name: string }; -type ResolvedTarget = { target: string; via: string; isFork: boolean }; + +type RepoReport = { + repo: string; + target: string; + skipped: boolean; + targets: number; + sampleTargets: string[]; + issues: number; + upserts: number; + errors: string[]; +}; export const issuesSweep = inngest.createFunction( - { id: 'issues-sweep' }, + { + id: 'issues-sweep', + // Overlapping sweeps would double GitHub + LLM spend, so only one run at + // a time (cron is every 12h and each run is capped, so it always drains). + concurrency: { key: 'issues-sweep', limit: 1 }, + }, { cron: '0 */12 * * *' }, async ({ step }) => { const installs = await step.run('list-installs', async () => { @@ -30,11 +61,17 @@ export const issuesSweep = inngest.createFunction( .from('github_installations') .select('id, account_login') .is('uninstalled_at', null) - .is('suspended_at', null); + .is('suspended_at', null) + // Least-recently-swept first so every install rotates through the cap + // instead of the same top-50 by id being swept every run. + .order('last_swept_at', { ascending: true, nullsFirst: true }) + .order('id', { ascending: true }) + .limit(MAX_INSTALLS_PER_SWEEP); return data ?? []; }); let totalUpserts = 0; + let totalIssuesSeen = 0; const perInstallReport: Array<{ install: number; account: string; @@ -47,66 +84,44 @@ export const issuesSweep = inngest.createFunction( }> = []; for (const install of installs) { - const budget = await step.run(`check-budget-install-${install.id}`, () => - checkRateBudget(install.id), - ); - if (!budget.ok) { - await step.sleepUntil( - `sleep-budget-install-${install.id}`, - new Date(budget.resetAt * 1000 + 5000), - ); - } - - // Each install is its own checkpoint so we can see the boundary in - // the trace if one install blows up. - const report = await step.run(`process-install-${install.id}`, async () => { + const setup = await step.run(`setup-install-${install.id}`, async () => { const sb = getServiceSupabase(); if (!sb) throw new Error('service role missing'); - const errors: string[] = []; - const { data: repoRows } = await sb .from('installation_repositories') .select('repo_full_name') - .eq('installation_id', install.id); + .eq('installation_id', install.id) + // Least-recently-swept first so repos past the per-install cap still + // get swept on later runs instead of an alphabetical top-20 forever. + .order('last_swept_at', { ascending: true, nullsFirst: true }) + .order('repo_full_name', { ascending: true }) + .limit(MAX_REPOS_PER_INSTALL); let repos = (repoRows ?? []) as RepoRow[]; - let octokit; - try { - octokit = await getInstallOctokit(install.id); - } catch (e) { - return { - install: install.id, - account: install.account_login, - repos: repos.length, - targets: 0, - sampleTargets: [], - issues: 0, - upserts: 0, - errors: [`install-token: ${(e as Error).message}`], - }; - } - // Self-healing: if installation_repositories is empty for an active install, // re-discover accessible repos via GitHub API and populate table. if (repos.length === 0) { try { + const octokit = await getInstallOctokit(install.id); const res = await octokit.paginate(octokit.apps.listReposAccessibleToInstallation, { per_page: 100, }); - const discovered = (res as unknown as Array<{ full_name: string }>).map((r) => ({ - full_name: r.full_name, - })); + const discovered = (res as unknown as Array<{ full_name: string }>).map( + (r) => r.full_name, + ); if (discovered.length > 0) { await sb.from('installation_repositories').upsert( - discovered.map((r) => ({ + discovered.map((fullName) => ({ installation_id: install.id, - repo_full_name: r.full_name, + repo_full_name: fullName, })), { onConflict: 'installation_id,repo_full_name' }, ); - repos = discovered.map((r) => ({ repo_full_name: r.full_name })); + repos = discovered + .map((fullName) => ({ repo_full_name: fullName })) + .slice(0, MAX_REPOS_PER_INSTALL); await inngest.send({ name: 'pr-backfill/installation', @@ -114,165 +129,102 @@ export const issuesSweep = inngest.createFunction( }); } } catch (e) { - errors.push(`repo-discovery: ${(e as Error).message}`); + return { repos: [], error: `repo-discovery: ${(e as Error).message}` }; } } - // Resolve fork → upstream. The interesting issues live on the - // upstream a user forked from, not on the fork itself. Dedup - // so two users forking the same project don't sweep it twice. - const resolved: ResolvedTarget[] = []; - const targetSet = new Set(); - for (const repo of repos) { - const [owner, name] = repo.repo_full_name.split('/'); - if (!owner || !name) continue; - try { - const meta = await octokit.repos.get({ owner, repo: name }); - const isFork = Boolean(meta.data.fork); - const upstream = isFork ? (meta.data.parent?.full_name ?? null) : repo.repo_full_name; - if (upstream && !targetSet.has(upstream)) { - targetSet.add(upstream); - resolved.push({ target: upstream, via: repo.repo_full_name, isFork }); - } - } catch (e) { - errors.push(`repos.get ${repo.repo_full_name}: ${(e as Error).message}`); - } - } + return { repos, error: null as string | null }; + }); - let issuesSeen = 0; - let upserts = 0; - - for (const t of resolved) { - const [owner, name] = t.target.split('/'); - if (!owner || !name) continue; - - // Real repo health signals + primary language (cached 24h) instead - // of the prior hardcoded constants. - const metrics = await fetchRepoMetrics(octokit, owner, name); - const healthScore = repoHealth(metrics); - const repoLanguage = metrics.language; - - let issues: Array<{ - number: number; - title: string; - body: string | null; - html_url: string; - comments: number; - labels: Array; - pull_request?: unknown; - }> = []; - try { - const res = await octokit.issues.listForRepo({ - owner, - repo: name, - state: 'open', - per_page: 30, - sort: 'updated', - }); - issues = res.data as typeof issues; - } catch (e) { - errors.push(`issues.list ${t.target}: ${(e as Error).message}`); - continue; - } + if (setup.error) { + perInstallReport.push({ + install: install.id, + account: install.account_login, + repos: 0, + targets: 0, + sampleTargets: [], + issues: 0, + upserts: 0, + errors: [setup.error], + }); + continue; + } - // Pre-fetch existing issues for this repository to avoid redundant LLM scoring - const issueNumbers = issues.filter((i) => !i.pull_request).map((i) => i.number); - const existingIssuesMap = new Map< - number, - { difficulty: string; difficulty_source: string; xp_reward: number } - >(); - if (issueNumbers.length > 0) { - const { data: existingIssues } = await sb - .from('issues') - .select('github_issue_number, difficulty, difficulty_source, xp_reward') - .eq('repo_full_name', t.target) - .in('github_issue_number', issueNumbers); - - if (existingIssues) { - for (const ex of existingIssues) { - existingIssuesMap.set(ex.github_issue_number, ex); - } - } - } + // Best-effort dedup of resolved upstream targets within this install. + // Cheap: repos.get is still called per repo, but the LLM + issues work + // is skipped for a fork whose upstream was already swept this run. + const seenTargets = new Set(); + const reports: RepoReport[] = []; + let issuesThisInstall = 0; + let reposProcessed = 0; - for (const issue of issues) { - if (issue.pull_request) continue; - issuesSeen += 1; + for (const repo of setup.repos) { + if ( + totalIssuesSeen >= MAX_ISSUES_PER_SWEEP || + issuesThisInstall >= MAX_ISSUES_PER_INSTALL + ) { + break; + } - const labels = (issue.labels ?? []).map((l) => - typeof l === 'string' ? l : (l.name ?? ''), - ); + // Budget re-checked before every repo so a rate-limited install stops + // making GitHub calls mid-loop instead of hammering the API. + const budget = await step.run( + `check-budget-${install.id}-${repo.repo_full_name.replace('/', '-')}`, + () => checkRateBudget(install.id), + ); + if (!budget.ok) { + await step.sleepUntil( + `sleep-budget-${install.id}-${repo.repo_full_name.replace('/', '-')}`, + new Date(budget.resetAt * 1000 + 5000), + ); + } - let scored; - const existing = existingIssuesMap.get(issue.number); - if (existing?.difficulty && existing?.difficulty_source) { - scored = { - difficulty: existing.difficulty as 'E' | 'M' | 'H', - source: existing.difficulty_source as 'label' | 'heuristic' | 'llm' | 'maintainer', - xpReward: existing.xp_reward, - }; - } else { - scored = await scoreDifficulty( - { - title: issue.title, - body: issue.body ?? undefined, - labels, - commentCount: issue.comments, - }, - { - llmFallback: async (i) => - llmCall({ - prompt: `Rate this OSS issue's difficulty as E/M/H.\nTitle: ${i.title}\nLabels: ${i.labels.join(', ')}\nBody: ${(i.body ?? '').slice(0, 800)}\n\nReturn JSON: {"difficulty":"E"|"M"|"H","confidence":0..1,"reason":"..."}`, - schema: DifficultySchema, - }), - }, - ); - } + const report = await step.run( + `sweep-${install.id}-${repo.repo_full_name.replace('/', '-')}`, + async () => + sweepRepo( + install.id, + repo.repo_full_name, + Math.min( + MAX_ISSUES_PER_SWEEP - totalIssuesSeen, + MAX_ISSUES_PER_INSTALL - issuesThisInstall, + ), + seenTargets, + ), + ); + reports.push(report); + reposProcessed += 1; + issuesThisInstall += report.issues; + totalIssuesSeen += report.issues; + } - const { error } = await sb.from('issues').upsert( - { - repo_full_name: t.target, - github_issue_number: issue.number, - title: issue.title, - body_excerpt: (issue.body ?? '').slice(0, 500), - difficulty: scored.difficulty, - difficulty_source: scored.source, - xp_reward: scored.xpReward, - labels: labels.filter((l): l is string => Boolean(l)), - state: 'open', - url: issue.html_url, - repo_health_score: healthScore, - repo_language: repoLanguage, - scored_at: new Date().toISOString(), - }, - { onConflict: 'repo_full_name,github_issue_number' }, - ); - if (error) { - errors.push( - `upsert ${t.target}#${issue.number}: ${error.code ?? ''} ${error.message}`, - ); - } else { - upserts += 1; - } - } - } + // Rotate the install to the back of the queue so it isn't re-picked until + // the other installs have had a turn. Only if at least one repo was + // actually processed — a fully budget-blocked install stays eligible so + // its repos get reached on the next sweep. + if (reposProcessed > 0) { + await step.run(`mark-install-swept-${install.id}`, async () => { + const sb = getServiceSupabase(); + if (!sb) return; + await sb + .from('github_installations') + .update({ last_swept_at: new Date().toISOString() }) + .eq('id', install.id); + }); + } - return { - install: install.id, - account: install.account_login, - repos: repos.length, - targets: resolved.length, - sampleTargets: resolved - .slice(0, 10) - .map((r) => `${r.target} (via ${r.via}${r.isFork ? ', fork' : ''})`), - issues: issuesSeen, - upserts, - errors: errors.slice(0, 10), - }; + perInstallReport.push({ + install: install.id, + account: install.account_login, + repos: setup.repos.length, + targets: reports.reduce((acc, r) => acc + r.targets, 0), + sampleTargets: reports.flatMap((r) => r.sampleTargets).slice(0, 10), + issues: issuesThisInstall, + upserts: reports.reduce((acc, r) => acc + r.upserts, 0), + errors: reports.flatMap((r) => r.errors).slice(0, 10), }); - perInstallReport.push(report); - totalUpserts += report.upserts; + totalUpserts += reports.reduce((acc, r) => acc + r.upserts, 0); } await step.run('build-recommendations', async () => { @@ -286,3 +238,293 @@ export const issuesSweep = inngest.createFunction( }; }, ); + +async function markRepoSwept(installationId: number, repoFullName: string): Promise { + const sb = getServiceSupabase(); + if (!sb) return; + await sb + .from('installation_repositories') + .update({ last_swept_at: new Date().toISOString() }) + .eq('installation_id', installationId) + .eq('repo_full_name', repoFullName); +} + +async function sweepRepo( + installationId: number, + repoFullName: string, + issueBudget: number, + seenTargets: Set, +): Promise { + const errors: string[] = []; + const sb = getServiceSupabase(); + if (!sb) { + return { + repo: repoFullName, + target: repoFullName, + skipped: false, + targets: 0, + sampleTargets: [], + issues: 0, + upserts: 0, + errors: ['service role missing'], + }; + } + + let octokit; + try { + octokit = await getInstallOctokit(installationId); + } catch (e) { + return { + repo: repoFullName, + target: repoFullName, + skipped: false, + targets: 0, + sampleTargets: [], + issues: 0, + upserts: 0, + errors: [`install-token: ${(e as Error).message}`], + }; + } + + const [owner, name] = repoFullName.split('/'); + if (!owner || !name) { + return { + repo: repoFullName, + target: repoFullName, + skipped: false, + targets: 0, + sampleTargets: [], + issues: 0, + upserts: 0, + errors: ['bad repo name'], + }; + } + + // Resolve fork → upstream. The interesting issues live on the upstream a + // user forked from, not on the fork itself. + let target = repoFullName; + const via = repoFullName; + let isFork = false; + try { + const meta = await octokit.repos.get({ owner, repo: name }); + isFork = Boolean(meta.data.fork); + target = isFork ? (meta.data.parent?.full_name ?? repoFullName) : repoFullName; + } catch (e) { + errors.push(`repos.get ${repoFullName}: ${(e as Error).message}`); + return { + repo: repoFullName, + target: repoFullName, + skipped: false, + targets: 0, + sampleTargets: [], + issues: 0, + upserts: 0, + errors, + }; + } + + if (seenTargets.has(target)) { + // Handled (cheap repos.get only) — rotate it so it isn't re-picked every + // run while the upstream is already being swept elsewhere in the install. + await markRepoSwept(installationId, repoFullName); + return { + repo: repoFullName, + target, + skipped: true, + targets: 0, + sampleTargets: [], + issues: 0, + upserts: 0, + errors, + }; + } + seenTargets.add(target); + + const [tOwner, tName] = target.split('/'); + if (!tOwner || !tName) { + return { + repo: repoFullName, + target, + skipped: false, + targets: 0, + sampleTargets: [], + issues: 0, + upserts: 0, + errors, + }; + } + + // Real repo health signals + primary language (cached 24h) instead of the + // prior hardcoded constants. + const metrics = await fetchRepoMetrics(octokit, tOwner, tName); + const healthScore = repoHealth(metrics); + const repoLanguage = metrics.language; + + // Per-repo cursor: resume from the last page we processed so a large repo + // is walked incrementally across sweeps instead of always re-sweeping page 1. + const lastPage = await getSyncCursor(installationId, target, SYNC_TYPE); + const startingPage = (lastPage ?? 0) + 1; + + let issues: Array<{ + number: number; + title: string; + body: string | null; + html_url: string; + comments: number; + labels: Array; + pull_request?: unknown; + }> = []; + try { + const res = await octokit.issues.listForRepo({ + owner: tOwner, + repo: tName, + state: 'open', + per_page: ISSUES_PER_REPO_PAGE, + page: startingPage, + sort: 'updated', + }); + issues = res.data as typeof issues; + } catch (e) { + errors.push(`issues.list ${target}: ${(e as Error).message}`); + return { + repo: repoFullName, + target, + skipped: false, + targets: 0, + sampleTargets: [], + issues: 0, + upserts: 0, + errors, + }; + } + + // Pre-fetch existing issues for this repository to reuse cached difficulty + // and respect the 24h re-score cooldown (keeps LLM spend bounded). + const issueNumbers = issues.filter((i) => !i.pull_request).map((i) => i.number); + const existingIssuesMap = new Map< + number, + { + difficulty: string | null; + difficulty_source: string | null; + xp_reward: number; + scored_at: string | null; + } + >(); + if (issueNumbers.length > 0) { + const { data: existingIssues } = await sb + .from('issues') + .select('github_issue_number, difficulty, difficulty_source, xp_reward, scored_at') + .eq('repo_full_name', target) + .in('github_issue_number', issueNumbers); + + if (existingIssues) { + for (const ex of existingIssues) { + existingIssuesMap.set(ex.github_issue_number, ex); + } + } + } + + const now = Date.now(); + let issuesSeen = 0; + let upserts = 0; + let budgetTruncated = false; + + for (const issue of issues) { + if (issue.pull_request) continue; + if (issuesSeen >= issueBudget) { + // Hit the per-repo budget mid-page. A full page can still hold + // unprocessed issues past the budget, so flag it and keep the cursor. + budgetTruncated = true; + break; + } + issuesSeen += 1; + + const labels = (issue.labels ?? []).map((l) => (typeof l === 'string' ? l : (l.name ?? ''))); + + let scored; + const existing = existingIssuesMap.get(issue.number); + if (existing?.difficulty && existing?.difficulty_source) { + // Cached forever — no LLM call. + scored = { + difficulty: existing.difficulty as 'E' | 'M' | 'H', + source: existing.difficulty_source as 'label' | 'heuristic' | 'llm' | 'maintainer', + xpReward: existing.xp_reward, + }; + } else if ( + existing?.scored_at && + now - new Date(existing.scored_at).getTime() < SCORE_COOLDOWN_MS + ) { + // Attempted within the last 24h — skip re-scoring so LLM spend stays + // bounded. Revisited on a later sweep if it still has no difficulty. + continue; + } else { + scored = await scoreDifficulty( + { + title: issue.title, + body: issue.body ?? undefined, + labels, + commentCount: issue.comments, + }, + { + llmFallback: async (i) => + llmCall({ + prompt: `Rate this OSS issue's difficulty as E/M/H.\nTitle: ${i.title}\nLabels: ${i.labels.join(', ')}\nBody: ${(i.body ?? '').slice(0, 800)}\n\nReturn JSON: {"difficulty":"E"|"M"|"H","confidence":0..1,"reason":"..."}`, + schema: DifficultySchema, + }), + }, + ); + } + + const { error } = await sb.from('issues').upsert( + { + repo_full_name: target, + github_issue_number: issue.number, + title: issue.title, + body_excerpt: (issue.body ?? '').slice(0, 500), + difficulty: scored.difficulty, + difficulty_source: scored.source, + xp_reward: scored.xpReward, + labels: labels.filter((l): l is string => Boolean(l)), + state: 'open', + url: issue.html_url, + repo_health_score: healthScore, + repo_language: repoLanguage, + scored_at: new Date().toISOString(), + }, + { onConflict: 'repo_full_name,github_issue_number' }, + ); + if (error) { + errors.push(`upsert ${target}#${issue.number}: ${error.code ?? ''} ${error.message}`); + } else { + upserts += 1; + } + } + + // Advance the cursor; clear it once we've caught up so the next sweep + // starts from page 1 again. Only when the page was fully drained: if the + // loop was budget-truncated, a full page still has unprocessed issues on + // it, so leave the cursor in place and retry the same page next sweep + // instead of skipping whatever was left over. + if (!budgetTruncated) { + if (issues.length < ISSUES_PER_REPO_PAGE) { + await clearSyncCursor(installationId, target, SYNC_TYPE); + } else { + await setSyncCursor(installationId, target, SYNC_TYPE, startingPage); + } + } + + // Rotate the repo to the back of the per-install queue so repos past the + // cap still get swept on later runs. + await markRepoSwept(installationId, repoFullName); + + return { + repo: repoFullName, + target, + skipped: false, + targets: 1, + sampleTargets: [`${target} (via ${via}${isFork ? ', fork' : ''})`], + issues: issuesSeen, + upserts, + errors: errors.slice(0, 10), + }; +} diff --git a/src/lib/db/schema.ts b/src/lib/db/schema.ts index eb580be3..1cbdc7f3 100644 --- a/src/lib/db/schema.ts +++ b/src/lib/db/schema.ts @@ -100,10 +100,12 @@ export const githubInstallations = pgTable( installedAt: timestamp('installed_at', { withTimezone: true }).notNull().defaultNow(), suspendedAt: timestamp('suspended_at', { withTimezone: true }), uninstalledAt: timestamp('uninstalled_at', { withTimezone: true }), + lastSweptAt: timestamp('last_swept_at', { withTimezone: true }), }, (t) => ({ userIdx: index('github_installations_user_idx').on(t.userId), accountIdx: index('github_installations_account_idx').on(t.accountLogin), + lastSweptIdx: index('github_installations_last_swept_idx').on(t.lastSweptAt), }), ); @@ -119,9 +121,14 @@ export const installationRepositories = pgTable( // onboarding repo picker). Distinct from "installed" — GitHub tells us what's // installed; this is the opt-in. Defaults true so existing installs are unaffected. managed: boolean('managed').notNull().default(true), + lastSweptAt: timestamp('last_swept_at', { withTimezone: true }), }, (t) => ({ pk: primaryKey({ columns: [t.installationId, t.repoFullName] }), + lastSweptIdx: index('installation_repositories_last_swept_idx').on( + t.installationId, + t.lastSweptAt, + ), }), ); diff --git a/supabase/migrations/0046_issues_sweep_rotation.sql b/supabase/migrations/0046_issues_sweep_rotation.sql new file mode 100644 index 00000000..3e9bc9ea --- /dev/null +++ b/supabase/migrations/0046_issues_sweep_rotation.sql @@ -0,0 +1,15 @@ +-- Round-robin rotation for the bounded issues-sweep (#860 / #865). +-- The sweep orders installs and repos by least-recently-swept and updates +-- this column after processing, so every install/repo rotates through the +-- per-run caps instead of the same top-N being swept every run. +alter table github_installations + add column if not exists last_swept_at timestamptz; + +alter table installation_repositories + add column if not exists last_swept_at timestamptz; + +create index if not exists github_installations_last_swept_idx + on github_installations (last_swept_at); + +create index if not exists installation_repositories_last_swept_idx + on installation_repositories (installation_id, last_swept_at);