diff --git a/.env.example b/.env.example index f4540ab9..14077f34 100644 --- a/.env.example +++ b/.env.example @@ -39,6 +39,13 @@ ASSETS_DOMAIN="https://your-assets-domain.com" # Create token at: https://github.com/settings/tokens # Required scopes: repo, user GITHUB_TOKEN="github_pat_..." +# Extra commit-search scopes (optional, comma-separated). +# Orgs scope `author:@me` to private orgs whose repos appear in search but can't +# be read via REST — their commits are recorded from search metadata alone. +GITHUB_COMMIT_SEARCH_ORGS="acme-corp,acme-labs" +# Emails match commits authored under identities not linked to your account +# (e.g. a former work email). +GITHUB_COMMIT_AUTHOR_EMAILS="you@former-job.com" # Airtable Integration # IMPORTANT: This integration is configured specifically for the diff --git a/INTEGRATIONS.md b/INTEGRATIONS.md index d3b02398..715c5b26 100644 --- a/INTEGRATIONS.md +++ b/INTEGRATIONS.md @@ -55,12 +55,27 @@ Syncs your GitHub repositories, stars, and commits. - Recent commits, including AI-generated summaries and metadata for each commit - User profiles +Commits are matched against your authenticated account (`author:@me`) plus any +extra identities configured via `GITHUB_COMMIT_SEARCH_ORGS` and +`GITHUB_COMMIT_AUTHOR_EMAILS`. Contributions to repositories you don't own — and +the repositories themselves — are marked private by default. For a private org +repo that appears in commit search but can't be read via the REST API, the +commit is recorded from its search metadata (without per-file changes). + ### Sync Command ```bash rcr sync github ``` +To pull deep history beyond the live sync's forward cutoff and GitHub's +1000-result search cap, run the windowed backfill: + +```bash +bun scripts/backfill-github-commits.ts --dry-run # preview +bun scripts/backfill-github-commits.ts # writes to the configured database +``` + ## Airtable Integration Syncs records from Airtable bases with a specific structure. diff --git a/scripts/backfill-github-commits.ts b/scripts/backfill-github-commits.ts new file mode 100644 index 00000000..26575987 --- /dev/null +++ b/scripts/backfill-github-commits.ts @@ -0,0 +1,254 @@ +/** + * TEMPORARY backfill script — delete once the GitHub commit history is backfilled. + * + * The live sync (`rcr sync github`) only walks forward from the most recent + * commit and GitHub caps any search at 1000 results, so deep history is + * unreachable from a single query. This script windows each configured + * commit-search query by date — keeping every window under the cap — to pull the + * full history, then rebuilds records and marks non-owned repos private. + * + * It reuses the live sync's per-commit upsert, so it is idempotent (existing + * SHAs are skipped) and resumable: if it dies, rerun it. + * + * Usage: + * bun scripts/backfill-github-commits.ts [options] + * NODE_ENV=development bun scripts/backfill-github-commits.ts # target dev DB + * + * Options: + * --from=YYYY-MM-DD Start date (default: authenticated account creation) + * --to=YYYY-MM-DD End date (default: today) — bound a range to run in batches + * --window-days=N Search window size in days (default: 30) + * --query="..." Run a single search query instead of all configured ones + * --summaries Generate AI summaries for new commits afterward + * --dry-run Fetch and count only; no database writes + */ + +import { integrationRuns, IntegrationStatusSchema, records, RunTypeSchema } from '@hozo'; +import { Octokit } from '@octokit/rest'; +import { eq, inArray } from 'drizzle-orm'; +import { db } from '@/server/db/connections/postgres'; +import { + createRecordsFromGithubRepositories, + createRecordsFromGithubUsers, + getOwnerUserId, +} from '@/server/integrations/github/map'; +import { syncCommitSummaries } from '@/server/integrations/github/summarize-commits'; +import { + getCommitSearchQueries, + upsertCommitFromSearchItem, +} from '@/server/integrations/github/sync-commits'; + +const DAY_MS = 24 * 60 * 60 * 1000; +const PER_PAGE = 100; +const SEARCH_RESULT_CAP = 1000; +const REQUEST_DELAY_MS = 1000; +const SEARCH_DELAY_MS = 2000; + +function parseArgs() { + const args = process.argv.slice(2); + const value = (name: string): string | undefined => { + const match = args.find((arg) => arg.startsWith(`--${name}=`)); + return match ? match.slice(name.length + 3) : undefined; + }; + const flag = (name: string): boolean => args.includes(`--${name}`); + + const fromStr = value('from'); + const toStr = value('to'); + const windowDaysStr = value('window-days'); + return { + from: fromStr ? new Date(fromStr) : undefined, + to: toStr ? new Date(toStr) : undefined, + windowDays: windowDaysStr ? Number(windowDaysStr) : 30, + query: value('query'), + summaries: flag('summaries'), + dryRun: flag('dry-run'), + }; +} + +function toDateStr(date: Date): string { + return date.toISOString().split('T')[0] ?? ''; +} + +function* dateWindows( + from: Date, + to: Date, + windowDays: number +): Generator<{ start: Date; end: Date }> { + let start = new Date(from); + while (start.getTime() <= to.getTime()) { + const windowEnd = new Date(start.getTime() + (windowDays - 1) * DAY_MS); + const end = windowEnd.getTime() > to.getTime() ? new Date(to) : windowEnd; + yield { start, end }; + start = new Date(end.getTime() + DAY_MS); + } +} + +async function backfillQuery( + octokit: Octokit, + query: string, + from: Date, + to: Date, + windowDays: number, + integrationRunId: number, + dryRun: boolean +): Promise { + let inserted = 0; + + for (const { start, end } of dateWindows(from, to, windowDays)) { + const startStr = toDateStr(start); + const endStr = toDateStr(end); + let page = 1; + + while (true) { + const response = await octokit.rest.search.commits({ + q: `${query} committer-date:${startStr}..${endStr}`, + sort: 'committer-date', + order: 'asc', + per_page: PER_PAGE, + page, + }); + + if (page === 1) { + const total = response.data.total_count; + if (total === 0) break; + const remaining = response.headers['x-ratelimit-remaining']; + console.log( + `[${query}] ${startStr}..${endStr}: ${total} results (search rate left: ${remaining})` + ); + if (total > SEARCH_RESULT_CAP) { + console.warn( + ` ⚠ window exceeds the ${SEARCH_RESULT_CAP}-result cap; rerun with a smaller --window-days for this range` + ); + } + } + + const items = response.data.items; + if (items.length === 0) break; + + for (const item of items) { + if (!item) continue; + try { + const result = await upsertCommitFromSearchItem(octokit, item, integrationRunId, dryRun); + if (result === 'inserted') inserted++; + } catch (error) { + console.error(` error on ${item.sha} (${item.repository?.full_name})`, error); + } + await Bun.sleep(REQUEST_DELAY_MS); + } + + if (items.length < PER_PAGE) break; + page++; + await Bun.sleep(SEARCH_DELAY_MS); + } + + await Bun.sleep(SEARCH_DELAY_MS); + } + + return inserted; +} + +/** + * Marks records for repos the user doesn't own as private. The live mapping does + * this for newly created records; this fixes records created before the rule. + */ +async function markNonOwnedReposPrivate(ownerUserId: number): Promise { + const nonOwned = await db.query.githubRepositories.findMany({ + columns: { recordId: true }, + where: { + ownerId: { ne: ownerUserId }, + recordId: { isNotNull: true }, + }, + }); + + const recordIds = nonOwned.map((repo) => repo.recordId).filter((id): id is number => id !== null); + + if (recordIds.length > 0) { + await db + .update(records) + .set({ isPrivate: true, recordUpdatedAt: new Date() }) + .where(inArray(records.id, recordIds)); + } + + return recordIds.length; +} + +async function main(): Promise { + const options = parseArgs(); + const octokit = new Octokit({ auth: process.env.GITHUB_TOKEN }); + + const { data: me } = await octokit.rest.users.getAuthenticated(); + const from = options.from ?? new Date(me.created_at); + const to = options.to ?? new Date(); + const queries = options.query ? [options.query] : getCommitSearchQueries(); + + console.log( + `Backfilling ${queries.length} ${queries.length === 1 ? 'query' : 'queries'} from ${toDateStr(from)} to ${toDateStr(to)} in ${options.windowDays}-day windows${options.dryRun ? ' [DRY RUN]' : ''}:` + ); + for (const query of queries) console.log(` - ${query}`); + + if (options.dryRun) { + let total = 0; + for (const query of queries) { + total += await backfillQuery(octokit, query, from, to, options.windowDays, -1, true); + } + console.log(`[DRY RUN] would process ${total} commits (existing SHAs not deducted)`); + return; + } + + const [run] = await db + .insert(integrationRuns) + .values({ + integrationType: 'github', + runType: RunTypeSchema.enum.sync, + runStartTime: new Date(), + }) + .returning(); + if (!run) throw new Error('Failed to create integration run record'); + + let total = 0; + try { + for (const query of queries) { + total += await backfillQuery(octokit, query, from, to, options.windowDays, run.id, false); + } + await db + .update(integrationRuns) + .set({ + status: IntegrationStatusSchema.enum.success, + runEndTime: new Date(), + entriesCreated: total, + }) + .where(eq(integrationRuns.id, run.id)); + } catch (error) { + await db + .update(integrationRuns) + .set({ + status: IntegrationStatusSchema.enum.fail, + runEndTime: new Date(), + message: error instanceof Error ? error.message : String(error), + }) + .where(eq(integrationRuns.id, run.id)); + throw error; + } + + console.log(`Inserted ${total} new commits. Building records...`); + await createRecordsFromGithubUsers(); + await createRecordsFromGithubRepositories(); + + const ownerUserId = await getOwnerUserId(); + const privateCount = await markNonOwnedReposPrivate(ownerUserId); + console.log(`Marked ${privateCount} non-owned repo records private.`); + + if (options.summaries) { + console.log('Generating commit summaries...'); + await syncCommitSummaries(); + } + + console.log('Backfill complete.'); +} + +main() + .then(() => process.exit(0)) + .catch((error) => { + console.error(error); + process.exit(1); + }); diff --git a/src/server/integrations/github/map.ts b/src/server/integrations/github/map.ts index baa025f1..4a143998 100644 --- a/src/server/integrations/github/map.ts +++ b/src/server/integrations/github/map.ts @@ -6,6 +6,7 @@ import { type GithubUserSelect, type RecordInsert, } from '@hozo'; +import { Octokit } from '@octokit/rest'; import { eq } from 'drizzle-orm'; import { db } from '@/server/db/connections/postgres'; import { mapUrl } from '@/server/lib/url-utils'; @@ -14,6 +15,21 @@ import { createIntegrationLogger } from '../common/logging'; const logger = createIntegrationLogger('github', 'map'); +let cachedOwnerUserId: number | null = null; + +/** + * The GitHub user id of the authenticated account — the knowledge base owner. + * Repositories owned by anyone else are treated as private contributions. + */ +export async function getOwnerUserId(): Promise { + if (cachedOwnerUserId === null) { + const octokit = new Octokit({ auth: process.env.GITHUB_TOKEN }); + const { data } = await octokit.rest.users.getAuthenticated(); + cachedOwnerUserId = data.id; + } + return cachedOwnerUserId; +} + /** * Maps a GitHub user to a record * @@ -110,7 +126,10 @@ export async function createRecordsFromGithubUsers() { * @param repository - The GitHub repository to map * @returns A record insert object */ -const mapGithubRepositoryToRecord = (repository: GithubRepositorySelect): RecordInsert => { +const mapGithubRepositoryToRecord = ( + repository: GithubRepositorySelect, + ownerUserId: number +): RecordInsert => { return { id: repository.recordId ?? undefined, type: 'artifact', @@ -118,7 +137,8 @@ const mapGithubRepositoryToRecord = (repository: GithubRepositorySelect): Record summary: repository.description, url: repository.htmlUrl, isCurated: false, - isPrivate: repository.private, + // Repos the user doesn't own are private contributions by default. + isPrivate: repository.private || repository.ownerId !== ownerUserId, sources: ['github'], recordCreatedAt: repository.recordCreatedAt, recordUpdatedAt: repository.recordUpdatedAt, @@ -159,8 +179,10 @@ export async function createRecordsFromGithubRepositories() { logger.info(`Found ${unmappedRepositories.length} unmapped GitHub repositories`); + const ownerUserId = await getOwnerUserId(); + for (const repository of unmappedRepositories) { - const newRecordDefaults = mapGithubRepositoryToRecord(repository); + const newRecordDefaults = mapGithubRepositoryToRecord(repository, ownerUserId); const [newRecord] = await db .insert(records) diff --git a/src/server/integrations/github/sync-commits.ts b/src/server/integrations/github/sync-commits.ts index 0022232b..583b8204 100644 --- a/src/server/integrations/github/sync-commits.ts +++ b/src/server/integrations/github/sync-commits.ts @@ -21,6 +21,8 @@ const logger = createIntegrationLogger('github', 'sync-commits'); * Type definitions */ type GithubRepository = Endpoints['GET /repos/{owner}/{repo}']['response']['data']; +type GithubCommitSearchItem = Endpoints['GET /search/commits']['response']['data']['items'][number]; +type GithubCommitSearchRepository = GithubCommitSearchItem['repository']; /** * Configuration constants @@ -29,6 +31,43 @@ const MAX_PATCH_LENGTH = 2048; const PER_PAGE = 100; const REQUEST_DELAY_MS = 1000; +function splitEnvList(value: string | undefined): string[] { + return (value ?? '') + .split(',') + .map((entry) => entry.trim()) + .filter(Boolean); +} + +/** + * Builds the GitHub commit-search queries that define which contributions to sync. + * + * `author:@me` matches commits linked to the authenticated account, but only in + * public repos and the account's own private repos. Two env-driven extensions + * widen this: + * - GITHUB_COMMIT_SEARCH_ORGS scopes `author:@me` to a private org, surfacing + * contributions to repos the token can find via search but not read via REST. + * - GITHUB_COMMIT_AUTHOR_EMAILS matches commits authored under an email that + * isn't linked to the authenticated account (e.g. a former work identity). + */ +export function getCommitSearchQueries(): string[] { + const orgs = splitEnvList(process.env.GITHUB_COMMIT_SEARCH_ORGS); + const emails = splitEnvList(process.env.GITHUB_COMMIT_AUTHOR_EMAILS); + return [ + 'author:@me', + ...orgs.map((org) => `author:@me org:${org}`), + ...emails.map((email) => `author-email:${email}`), + ]; +} + +/** + * A 403/404 from a repo or commit endpoint means the token can't read the repo + * (typically a private org repo found via search but not granted via REST). + * These are recoverable: the commit is still recorded from its search metadata. + */ +function isRepoAccessError(error: unknown): boolean { + return error instanceof RequestError && (error.status === 403 || error.status === 404); +} + /** * Retrieves the most recent commit date from the database * @@ -65,24 +104,41 @@ async function getMostRecentCommitDate(): Promise { } /** - * Ensures a GitHub repository exists in the database - * - * This function creates or updates a repository record and ensures - * the repository owner exists in the database. + * Upserts a repository row, preserving an existing starredAt timestamp. + */ +async function upsertRepository(newRepo: GithubRepositoryInsert): Promise { + const existingRepo = await db.query.githubRepositories.findFirst({ + columns: { starredAt: true }, + where: { id: newRepo.id }, + }); + + await db + .insert(githubRepositories) + .values(newRepo) + .onConflictDoUpdate({ + target: githubRepositories.id, + set: { + ...newRepo, + recordUpdatedAt: new Date(), + // Only include starredAt if it exists in the database + ...(existingRepo?.starredAt && { starredAt: existingRepo.starredAt }), + }, + }); +} + +/** + * Ensures a GitHub repository exists in the database from full REST repo data, + * along with its owner. * - * @param repoData - Repository data from GitHub API - * @param integrationRunId - The ID of the current integration run * @returns The ID of the repository */ async function ensureRepositoryExists( repoData: GithubRepository, integrationRunId: number ): Promise { - // First ensure the owner exists await ensureGithubUserExists(repoData.owner, integrationRunId); - // Prepare repository data for insertion - const newRepo: GithubRepositoryInsert = { + await upsertRepository({ id: repoData.id, nodeId: repoData.node_id, name: repoData.name, @@ -98,86 +154,181 @@ async function ensureRepositoryExists( integrationRunId, contentCreatedAt: new Date(repoData.created_at), contentUpdatedAt: new Date(repoData.updated_at), - }; - - // First try to get existing repository to preserve starredAt - const existingRepo = await db.query.githubRepositories.findFirst({ - columns: { - starredAt: true, - }, - where: { - id: repoData.id, - }, }); - // Insert or update the repository - await db - .insert(githubRepositories) - .values(newRepo) - .onConflictDoUpdate({ - target: githubRepositories.id, - set: { - ...newRepo, - recordUpdatedAt: new Date(), - // Only include starredAt if it exists in the database - ...(existingRepo?.starredAt && { starredAt: existingRepo.starredAt }), - }, - }); - return repoData.id; } /** - * Synchronizes GitHub commits with the database - * - * This function: - * 1. Fetches commits from the GitHub API - * 2. Determines which commits are new since the last sync - * 3. Stores commit information and file changes in the database - * 4. Triggers commit summary and embedding generation + * Ensures a repository exists from the partial metadata carried by a commit + * search result, for repos the token can find via search but not read via REST. + * Fields only available from the full REST response (timestamps, license, + * homepage) are left null. * - * @param integrationRunId - The ID of the current integration run - * @param collectDebugData - Optional array to collect raw API data for debugging - * @param skipPersist - If true, only fetches data without writing to database (debug mode) - * @returns The number of new commits processed - * @throws Error if the GitHub API request fails + * @returns The ID of the repository */ -async function syncGitHubCommits( - integrationRunId: number, - collectDebugData?: unknown[], - skipPersist = false +async function ensureRepositoryFromSearch( + repo: GithubCommitSearchRepository, + integrationRunId: number ): Promise { - const octokit = new Octokit({ - auth: process.env.GITHUB_TOKEN, + await ensureGithubUserExists(repo.owner, integrationRunId); + + await upsertRepository({ + id: repo.id, + nodeId: repo.node_id, + name: repo.name, + fullName: repo.full_name, + ownerId: repo.owner.id, + private: repo.private, + htmlUrl: repo.html_url, + description: repo.description, + language: repo.language ?? null, + topics: repo.topics && repo.topics.length > 0 ? repo.topics : null, + integrationRunId, }); - logger.start('Fetching GitHub commits'); + return repo.id; +} - // Get the most recent commit date to use as a cutoff (skip DB query in debug mode) - const mostRecentCommitDate = skipPersist ? null : await getMostRecentCommitDate(); - if (mostRecentCommitDate) { +/** + * Fetches full detail for a searched commit and upserts it with its file changes. + * + * When the repository can't be read via REST (a private org repo surfaced by + * search but not granted to the token), the commit is recorded from its search + * metadata alone — without per-file changes or line stats. + * + * @returns 'inserted' when a new commit is written (or would be, in debug mode), + * 'skipped' when it already exists or predates a fork's creation + */ +export async function upsertCommitFromSearchItem( + octokit: Octokit, + item: GithubCommitSearchItem, + integrationRunId: number, + skipPersist: boolean +): Promise<'inserted' | 'skipped'> { + if (!skipPersist) { + const existingCommit = await db.query.githubCommits.findFirst({ + columns: { id: true }, + where: { sha: item.sha }, + }); + if (existingCommit) { + logger.info(`Skipping existing commit ${item.sha}`); + return 'skipped'; + } + } + + let repoData: GithubRepository | null = null; + let detail: Endpoints['GET /repos/{owner}/{repo}/commits/{ref}']['response']['data'] | null = + null; + + try { + const repoResponse = await octokit.rest.repos.get({ + owner: item.repository.owner.login, + repo: item.repository.name, + }); + logRateLimitInfo(repoResponse, logger); + repoData = repoResponse.data; + + // Skip commits that predate a fork's creation — they belong to the upstream repo. + if (repoData.fork) { + const commitDate = new Date(item.commit.author.date); + const forkDate = new Date(repoData.created_at); + if (commitDate < forkDate) { + logger.info(`Skipping commit ${item.sha} as it predates fork creation`); + return 'skipped'; + } + } + + const detailResponse = await octokit.rest.repos.getCommit({ + owner: item.repository.owner.login, + repo: item.repository.name, + ref: item.sha, + }); + logRateLimitInfo(detailResponse, logger); + detail = detailResponse.data; + } catch (error) { + if (!isRepoAccessError(error)) { + throw error; + } logger.info( - `Most recent commit activity in database: ${mostRecentCommitDate.toLocaleString()}` + `No REST access to ${item.repository.full_name}; recording ${item.sha} from search metadata` ); - } else if (!skipPersist) { - logger.info('No existing commits in database'); } + if (skipPersist) { + logger.info(`Would insert commit ${item.sha} for ${item.repository.full_name}`); + return 'inserted'; + } + + if (repoData) { + await ensureRepositoryExists(repoData, integrationRunId); + } else { + await ensureRepositoryFromSearch(item.repository, integrationRunId); + } + + const newCommit: GithubCommitInsert = { + id: item.node_id, + sha: item.sha, + message: item.commit.message, + htmlUrl: item.html_url, + repositoryId: item.repository.id, + committedAt: item.commit.committer?.date ? new Date(item.commit.committer.date) : null, + contentCreatedAt: new Date(item.commit.author.date), + integrationRunId, + changes: detail?.stats?.total ?? null, + additions: detail?.stats?.additions ?? null, + deletions: detail?.stats?.deletions ?? null, + }; + await db.insert(githubCommits).values(newCommit); + + for (const file of detail?.files ?? []) { + const newChange: GithubCommitChangeInsert = { + filename: file.filename, + status: file.status, + patch: file.patch ? file.patch.slice(0, MAX_PATCH_LENGTH) : '', + commitId: item.node_id, + changes: file.changes, + additions: file.additions, + deletions: file.deletions, + }; + await db.insert(githubCommitChanges).values(newChange); + } + + logger.info(`Inserted commit ${item.sha} for ${item.repository.full_name}`); + return 'inserted'; +} + +/** + * Paginates one commit-search query, upserting each new commit. + * + * Stops when results are exhausted or a page yields no new commits (the search + * is sorted by committer date ascending, so an all-existing page means we've + * caught up to previously synced history). + * + * @returns The number of new commits inserted for this query + */ +async function syncCommitsForQuery( + octokit: Octokit, + searchQuery: string, + mostRecentCommitDate: Date | null, + integrationRunId: number, + collectDebugData: unknown[] | undefined, + skipPersist: boolean +): Promise { let page = 1; - let hasMore = true; - let totalCommits = 0; + let insertedCount = 0; - while (hasMore) { - try { - logger.info(`Fetching page ${page}...`); + while (true) { + logger.info(`[${searchQuery}] Fetching page ${page}...`); - // Use the committer-date qualifier so that any push/merge update (reflected by committer date) - // after our mostRecentCommitDate will be included - const queryStr = mostRecentCommitDate - ? `author:@me committer-date:>=${mostRecentCommitDate.toISOString().split('T')[0]}` - : 'author:@me'; + // The committer-date qualifier limits to push/merge activity newer than the + // most recent commit already synced. + const queryStr = mostRecentCommitDate + ? `${searchQuery} committer-date:>=${mostRecentCommitDate.toISOString().split('T')[0]}` + : searchQuery; - // Search for commits authored by the authenticated user + let items: GithubCommitSearchItem[]; + try { const response = await octokit.rest.search.commits({ q: queryStr, sort: 'committer-date', @@ -185,159 +336,113 @@ async function syncGitHubCommits( per_page: PER_PAGE, page, }); - - // Log rate limit information for monitoring logRateLimitInfo(response, logger); - - // Collect debug data if requested - if (collectDebugData) { - collectDebugData.push(...response.data.items); - } - - // Check if we've reached the end of the results - if (response.data.items.length === 0) { - hasMore = false; - break; - } - - // Process commits sequentially with rate limiting - let newCommitsOnPage = 0; - let processedAnyNewCommits = false; - - for (let i = 0; i < response.data.items.length; i++) { - const item = response.data.items[i]; - if (!item) { - logger.warn(`Missing commit item at index ${i} on page ${page}`); - continue; - } - try { - // In debug mode, skip DB existence check - if (!skipPersist) { - const existingCommit = await db.query.githubCommits.findFirst({ - columns: { id: true, sha: true }, - where: { sha: item.sha }, - }); - - if (existingCommit) { - logger.info(`Skipping existing commit ${item.sha}`); - await Bun.sleep(REQUEST_DELAY_MS); - continue; - } - } - - // Get the full repository data - const repoResponse = await octokit.rest.repos.get({ - owner: item.repository.owner.login, - repo: item.repository.name, - }); - logRateLimitInfo(repoResponse, logger); - - // Skip if this is a fork and the commit is older than the fork date - if (repoResponse.data.fork) { - const commitDate = new Date(item.commit.author.date); - const forkDate = new Date(repoResponse.data.created_at); - - if (commitDate < forkDate) { - logger.info(`Skipping commit ${item.sha} as it predates fork creation`); - await Bun.sleep(REQUEST_DELAY_MS); - continue; - } - } - - // Get detailed commit info including file changes - const detailedCommit = await octokit.rest.repos.getCommit({ - owner: item.repository.owner.login, - repo: item.repository.name, - ref: item.sha, - }); - logRateLimitInfo(detailedCommit, logger); - - processedAnyNewCommits = true; - - // In debug mode, just count the commits without persisting - if (skipPersist) { - logger.info(`Would insert commit ${item.sha} for ${item.repository.full_name}`); - newCommitsOnPage++; - await Bun.sleep(REQUEST_DELAY_MS); - continue; - } - - // Ensure repository exists in database - await ensureRepositoryExists(repoResponse.data, integrationRunId); - - // Insert new commit - const newCommit: GithubCommitInsert = { - id: item.node_id, - sha: item.sha, - message: item.commit.message, - htmlUrl: item.html_url, - repositoryId: item.repository.id, - committedAt: item.commit.committer?.date ? new Date(item.commit.committer.date) : null, - contentCreatedAt: new Date(item.commit.author.date), - integrationRunId, - changes: detailedCommit.data.stats?.total ?? null, - additions: detailedCommit.data.stats?.additions ?? null, - deletions: detailedCommit.data.stats?.deletions ?? null, - }; - - await db.insert(githubCommits).values(newCommit); - - // Insert commit changes - for (const file of detailedCommit.data.files || []) { - const newChange: GithubCommitChangeInsert = { - filename: file.filename, - status: file.status, - patch: file.patch ? file.patch.slice(0, MAX_PATCH_LENGTH) : '', - commitId: item.node_id, - changes: file.changes, - additions: file.additions, - deletions: file.deletions, - }; - await db.insert(githubCommitChanges).values(newChange); - } - - logger.info(`Inserted commit ${item.sha} for ${item.repository.full_name}`); - newCommitsOnPage++; - await Bun.sleep(REQUEST_DELAY_MS); - } catch (error) { - logger.error(`Error processing commit ${item.sha}`, { - error: error instanceof Error ? error.message : String(error), - repository: item.repository?.full_name, - }); - await Bun.sleep(REQUEST_DELAY_MS); - } - - logger.info(`Processing commits: ${i + 1}/${response.data.items.length}`); - } - - totalCommits += newCommitsOnPage; - - // If we didn't process any new commits on this page, we can stop - if (!processedAnyNewCommits) { - logger.info('No new commits found on this page, stopping pagination'); - hasMore = false; - break; - } - - logger.info(`Processed new commits from page ${page}`); - page++; + items = response.data.items; } catch (error) { if (error instanceof RequestError) { - logger.error('GitHub API Error', { - status: error.status, - message: error.message, - headers: error.response?.headers, - }); + logger.error('GitHub search error', { status: error.status, message: error.message }); if (error.response) { logRateLimitInfo(error.response, logger); } - // If we hit rate limits, throw to stop the process if (error.status === 403 || error.status === 429) { throw new Error(`GitHub API rate limit exceeded: ${error.message}`); } } throw error; } + + if (collectDebugData) { + collectDebugData.push(...items); + } + if (items.length === 0) { + break; + } + + let processedAnyNew = false; + for (let i = 0; i < items.length; i++) { + const item = items[i]; + if (!item) { + logger.warn(`Missing commit item at index ${i} on page ${page}`); + continue; + } + try { + const result = await upsertCommitFromSearchItem( + octokit, + item, + integrationRunId, + skipPersist + ); + if (result === 'inserted') { + insertedCount++; + processedAnyNew = true; + } + } catch (error) { + logger.error(`Error processing commit ${item.sha}`, { + error: error instanceof Error ? error.message : String(error), + repository: item.repository?.full_name, + }); + } + await Bun.sleep(REQUEST_DELAY_MS); + logger.info(`[${searchQuery}] Processed ${i + 1}/${items.length}`); + } + + if (!processedAnyNew) { + logger.info(`[${searchQuery}] No new commits on page ${page}, stopping pagination`); + break; + } + page++; + } + + return insertedCount; +} + +/** + * Synchronizes GitHub commits with the database + * + * Runs each configured commit-search query (the authenticated user plus any + * extra org scopes and author emails), records new commits and their file + * changes, and triggers summary/embedding generation. + * + * @param integrationRunId - The ID of the current integration run + * @param collectDebugData - Optional array to collect raw API data for debugging + * @param skipPersist - If true, only fetches data without writing to database (debug mode) + * @returns The number of new commits processed + * @throws Error if the GitHub API request fails + */ +async function syncGitHubCommits( + integrationRunId: number, + collectDebugData?: unknown[], + skipPersist = false +): Promise { + const octokit = new Octokit({ + auth: process.env.GITHUB_TOKEN, + }); + + logger.start('Fetching GitHub commits'); + + // Get the most recent commit date to use as a cutoff (skip DB query in debug mode) + const mostRecentCommitDate = skipPersist ? null : await getMostRecentCommitDate(); + if (mostRecentCommitDate) { + logger.info( + `Most recent commit activity in database: ${mostRecentCommitDate.toLocaleString()}` + ); + } else if (!skipPersist) { + logger.info('No existing commits in database'); + } + + const searchQueries = getCommitSearchQueries(); + logger.info(`Searching commits across ${searchQueries.length} queries`); + + let totalCommits = 0; + for (const searchQuery of searchQueries) { + totalCommits += await syncCommitsForQuery( + octokit, + searchQuery, + mostRecentCommitDate, + integrationRunId, + collectDebugData, + skipPersist + ); } logger.complete('Synced commits', totalCommits);