diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 3d9441fd53..52ea0c0abc 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -842,6 +842,9 @@ importers: '@crowd/data-access-layer': specifier: workspace:* version: link:../../libs/data-access-layer + '@crowd/database': + specifier: workspace:* + version: link:../../libs/database '@crowd/logging': specifier: workspace:* version: link:../../libs/logging @@ -869,6 +872,9 @@ importers: '@temporalio/workflow': specifier: ~1.17.2 version: 1.17.2 + axios: + specifier: 'catalog:' + version: 1.20.0(debug@4.4.3(supports-color@8.1.1))(supports-color@8.1.1) tsx: specifier: ^4.7.1 version: 4.23.12 @@ -1158,7 +1164,7 @@ importers: version: link:../../libs/types '@sendgrid/mail': specifier: ~8.1.0 - version: 8.1.3(debug@4.4.3(supports-color@8.1.1))(supports-color@8.1.1) + version: 8.1.3 '@temporalio/workflow': specifier: 'catalog:' version: 1.17.2 @@ -7348,9 +7354,6 @@ packages: axios@1.12.0: resolution: {integrity: sha512-oXTDccv8PcfjZmPGlWsPSwtOJCZ/b6W5jAMCNcfwJbCzDckwG0jrYJFaWH1yvivfCXjVzV/SPDEhMB3Q+DSurg==} - axios@1.19.0: - resolution: {integrity: sha512-ht/iuYZXEjFxLH/Hkezgd7m6JKlHHXEUSneaDz8uZe1Gj5QZtCnpyDsckvAiEnT89OEbCLmnte4R4sn7P0EKFw==} - axios@1.20.0: resolution: {integrity: sha512-r8aOh8j9cGKpgQAqpzrUHnSIc6a59Y3Xf/cv8sy1DrHCkZHzQGEuoq1tARk6qSyDdtQGSDgpb9kFlruzPvrgwg==} @@ -16694,10 +16697,10 @@ snapshots: domhandler: 4.3.1 selderee: 0.6.0 - '@sendgrid/client@8.1.3(debug@4.4.3(supports-color@8.1.1))(supports-color@8.1.1)': + '@sendgrid/client@8.1.3': dependencies: '@sendgrid/helpers': 8.0.0 - axios: 1.19.0(debug@4.4.3(supports-color@8.1.1))(supports-color@8.1.1) + axios: 1.20.0 transitivePeerDependencies: - debug - supports-color @@ -16706,9 +16709,9 @@ snapshots: dependencies: deepmerge: 4.3.1 - '@sendgrid/mail@8.1.3(debug@4.4.3(supports-color@8.1.1))(supports-color@8.1.1)': + '@sendgrid/mail@8.1.3': dependencies: - '@sendgrid/client': 8.1.3(debug@4.4.3(supports-color@8.1.1))(supports-color@8.1.1) + '@sendgrid/client': 8.1.3 '@sendgrid/helpers': 8.0.0 transitivePeerDependencies: - debug @@ -18580,6 +18583,12 @@ snapshots: turndown-plugin-gfm: 1.0.2 yaml: 2.7.0 + agent-base@6.0.2: + dependencies: + debug: 4.4.3(supports-color@8.1.1) + transitivePeerDependencies: + - supports-color + agent-base@6.0.2(supports-color@5.5.0): dependencies: debug: 4.4.3(supports-color@5.5.0) @@ -18830,11 +18839,11 @@ snapshots: transitivePeerDependencies: - debug - axios@1.19.0(debug@4.4.3(supports-color@8.1.1))(supports-color@8.1.1): + axios@1.20.0: dependencies: follow-redirects: 1.16.0(debug@4.4.3(supports-color@8.1.1)) form-data: 4.0.6 - https-proxy-agent: 5.0.1(supports-color@8.1.1) + https-proxy-agent: 5.0.1 proxy-from-env: 2.1.0 transitivePeerDependencies: - debug @@ -21387,6 +21396,13 @@ snapshots: http2-client@1.3.5: {} + https-proxy-agent@5.0.1: + dependencies: + agent-base: 6.0.2 + debug: 4.4.3(supports-color@8.1.1) + transitivePeerDependencies: + - supports-color + https-proxy-agent@5.0.1(supports-color@5.5.0): dependencies: agent-base: 6.0.2(supports-color@5.5.0) diff --git a/services/apps/connectors_worker/package.json b/services/apps/connectors_worker/package.json index 5523b35e41..34f1f0d031 100644 --- a/services/apps/connectors_worker/package.json +++ b/services/apps/connectors_worker/package.json @@ -12,7 +12,8 @@ "script:diff-nango-shadow": "tsx src/bin/diff-nango-shadow.ts", "script:diff-shadow-unit-detail": "tsx src/bin/diff-shadow-unit-detail.ts", "script:backfill-octolens-mentions": "tsx src/bin/backfill-octolens-mentions.ts", - "script:switch-sync-pipeline": "tsx src/bin/switch-sync-pipeline.ts" + "script:switch-sync-pipeline": "tsx src/bin/switch-sync-pipeline.ts", + "script:mirror-nango-repos": "tsx src/bin/mirror-nango-repos-to-shadow-sync-units.ts" }, "dependencies": { "@crowd/archetype-standard": "workspace:*", @@ -20,6 +21,7 @@ "@crowd/common": "workspace:*", "@crowd/connectors": "workspace:*", "@crowd/data-access-layer": "workspace:*", + "@crowd/database": "workspace:*", "@crowd/logging": "workspace:*", "@crowd/nango": "workspace:*", "@crowd/redis": "workspace:*", @@ -29,6 +31,7 @@ "@temporalio/activity": "~1.17.2", "@temporalio/client": "~1.17.2", "@temporalio/workflow": "~1.17.2", + "axios": "catalog:", "tsx": "^4.7.1" }, "devDependencies": { diff --git a/services/apps/connectors_worker/src/activities.ts b/services/apps/connectors_worker/src/activities.ts index fad4b0612b..a62d13a5b7 100644 --- a/services/apps/connectors_worker/src/activities.ts +++ b/services/apps/connectors_worker/src/activities.ts @@ -6,6 +6,12 @@ import { startRun, touchHeartbeat, } from './activities/dispatcherActivities' +import { + countShadowRepos, + logMirrorSweepSummary, + mirrorRepos, + selectMirrorBatch, +} from './activities/mirrorSweepActivities' import { runOctolensDiscovery } from './activities/octolensDiscoveryActivities' import { listShadowDiffChannels, runShadowDiffForChannel } from './activities/shadowDiffActivities' import { @@ -28,6 +34,7 @@ export { activateUnit, admitByBudget, claimDue, + countShadowRepos, deferUnit, evaluateBackGate, evaluateForwardGate, @@ -36,6 +43,8 @@ export { listShadowDiffChannels, listSwitchSweepRepos, logDispatchSummary, + logMirrorSweepSummary, + mirrorRepos, pauseNangoRepoSync, quiesceUnit, readNangoHandoverPoint, @@ -43,6 +52,7 @@ export { resumeNangoRepoSync, runOctolensDiscovery, runShadowDiffForChannel, + selectMirrorBatch, setUnitEmitEnabled, startRun, takeOverUnit, diff --git a/services/apps/connectors_worker/src/activities/mirrorSweepActivities.test.ts b/services/apps/connectors_worker/src/activities/mirrorSweepActivities.test.ts new file mode 100644 index 0000000000..cbbf11f065 --- /dev/null +++ b/services/apps/connectors_worker/src/activities/mirrorSweepActivities.test.ts @@ -0,0 +1,123 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +import type { IMirrorCandidate } from '../mirrorSweep' +import { mirrorRepos, selectMirrorBatch } from './mirrorSweepActivities' + +const mocks = vi.hoisted(() => ({ + listUnmirroredNangoRepos: vi.fn(), + countShadowRepos: vi.fn(), + executeSql: vi.fn(), + mirrorRepo: vi.fn(), + initNangoCloudClient: vi.fn(), + log: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, +})) + +vi.mock('../main', () => ({ + svc: { + postgres: { writer: {} }, + log: mocks.log, + }, +})) +vi.mock('@crowd/data-access-layer/src/queryExecutor', () => ({ dbStoreQx: vi.fn(() => 'qx') })) +vi.mock('@crowd/data-access-layer/src/connectors', () => ({ + countShadowRepos: mocks.countShadowRepos, +})) +vi.mock('@crowd/data-access-layer/src/integrations', () => ({ + listUnmirroredNangoRepos: mocks.listUnmirroredNangoRepos, +})) +vi.mock('@crowd/database', () => ({ + TinybirdClient: class { + executeSql = mocks.executeSql + }, +})) +vi.mock('../mirrorRepo', () => ({ mirrorRepo: mocks.mirrorRepo })) +vi.mock('@crowd/nango', async (importOriginal) => ({ + ...(await importOriginal()), + initNangoCloudClient: mocks.initNangoCloudClient, +})) + +const repo = (repoName: string) => ({ + integrationId: 'int-1', + connectionId: 'conn-1', + owner: 'acme', + repoName, + enabled: true, + archived: false, + excluded: false, +}) + +const candidate = (repoName: string): IMirrorCandidate => ({ + integrationId: 'int-1', + connectionId: 'conn-1', + owner: 'acme', + repoName, +}) + +describe('mirror sweep activities', () => { + beforeEach(() => { + vi.clearAllMocks() + vi.useFakeTimers({ now: new Date('2026-10-20T10:00:00Z'), toFake: ['Date'] }) + }) + afterEach(() => vi.useRealTimers()) + + it('selects the most active candidates up to the limit and lets tinybird errors fail the batch', async () => { + mocks.listUnmirroredNangoRepos.mockResolvedValue([repo('c'), repo('a'), repo('b')]) + mocks.executeSql.mockResolvedValue({ + data: [ + { channelKey: 'https://github.com/acme/b', activities: 50 }, + { channelKey: 'https://github.com/acme/a', activities: 10 }, + ], + }) + + await expect(selectMirrorBatch(2)).resolves.toEqual({ + since: '2026-10-18T00:00:00.000Z', + candidatesTotal: 3, + batch: [candidate('b'), candidate('a')], + }) + + vi.clearAllMocks() + mocks.listUnmirroredNangoRepos.mockResolvedValue([repo('a')]) + mocks.executeSql.mockRejectedValue(new Error('tinybird down')) + + await expect(selectMirrorBatch(2)).rejects.toThrow('tinybird down') + expect(mocks.listUnmirroredNangoRepos).toHaveBeenCalledTimes(1) + expect(mocks.countShadowRepos).not.toHaveBeenCalled() + }) + + it('mirrors the batch at most five repos at a time and logs every skip', async () => { + let inFlight = 0 + let peak = 0 + mocks.mirrorRepo.mockImplementation(async (_qx: unknown, c: IMirrorCandidate) => { + inFlight++ + peak = Math.max(peak, inFlight) + await Promise.resolve() + inFlight-- + return ['r2', 'r5'].includes(c.repoName) + ? { outcome: 'skipped', reason: 'repo_not_found', message: '404' } + : { outcome: 'mirrored', unitsInserted: 8 } + }) + const batch = ['r1', 'r2', 'r3', 'r4', 'r5', 'r6', 'r7'].map(candidate) + + await expect(mirrorRepos(batch, '2026-10-18T00:00:00.000Z')).resolves.toEqual({ + mirrored: 5, + skipped: 2, + unitsInserted: 40, + }) + + expect(peak).toBe(5) + expect(mocks.initNangoCloudClient).toHaveBeenCalledTimes(1) + expect(mocks.mirrorRepo).toHaveBeenCalledWith('qx', batch[0], '2026-10-18T00:00:00.000Z') + expect(mocks.log.warn).toHaveBeenCalledTimes(2) + expect(mocks.log.warn).toHaveBeenCalledWith( + { + event: 'mirror_repo_skipped', + integrationId: 'int-1', + owner: 'acme', + repoName: 'r2', + reason: 'repo_not_found', + message: '404', + }, + 'repo skipped by mirror sweep', + ) + }) +}) diff --git a/services/apps/connectors_worker/src/activities/mirrorSweepActivities.ts b/services/apps/connectors_worker/src/activities/mirrorSweepActivities.ts new file mode 100644 index 0000000000..f402d2b0e3 --- /dev/null +++ b/services/apps/connectors_worker/src/activities/mirrorSweepActivities.ts @@ -0,0 +1,84 @@ +import { countShadowRepos as countShadowReposInDb } from '@crowd/data-access-layer/src/connectors' +import { listUnmirroredNangoRepos } from '@crowd/data-access-layer/src/integrations' +import { dbStoreQx } from '@crowd/data-access-layer/src/queryExecutor' +import { TinybirdClient } from '@crowd/database' +import { initNangoCloudClient } from '@crowd/nango' + +import { readGithubChannelActivity } from '../githubChannelActivity' +import { svc } from '../main' +import { mirrorRepo } from '../mirrorRepo' +import { + type IMirrorCandidate, + type IMirrorSweepSummary, + MIRROR_RESOLVE_CONCURRENCY, + chunk, + mirrorSince, + orderMirrorCandidates, +} from '../mirrorSweep' + +export interface IMirrorBatch { + since: string + candidatesTotal: number + batch: IMirrorCandidate[] +} + +export interface IMirrorReposResult { + mirrored: number + skipped: number + unitsInserted: number +} + +export async function countShadowRepos(): Promise { + return countShadowReposInDb(dbStoreQx(svc.postgres.writer)) +} + +export async function selectMirrorBatch(limit: number): Promise { + const now = new Date() + const repos = await listUnmirroredNangoRepos(dbStoreQx(svc.postgres.writer)) + const activityByChannel = await readGithubChannelActivity(new TinybirdClient(), now) + const ordered = orderMirrorCandidates(repos, activityByChannel) + return { + since: mirrorSince(now), + candidatesTotal: ordered.length, + batch: ordered.slice(0, limit), + } +} + +export async function mirrorRepos( + batch: IMirrorCandidate[], + since: string, +): Promise { + await initNangoCloudClient() + const qx = dbStoreQx(svc.postgres.writer) + const result: IMirrorReposResult = { mirrored: 0, skipped: 0, unitsInserted: 0 } + + for (const group of chunk(batch, MIRROR_RESOLVE_CONCURRENCY)) { + const outcomes = await Promise.all(group.map((candidate) => mirrorRepo(qx, candidate, since))) + outcomes.forEach((outcome, i) => { + if (outcome.outcome === 'mirrored') { + result.mirrored++ + result.unitsInserted += outcome.unitsInserted + return + } + const { integrationId, owner, repoName } = group[i] + result.skipped++ + svc.log.warn( + { + event: 'mirror_repo_skipped', + integrationId, + owner, + repoName, + reason: outcome.reason, + message: outcome.message, + }, + 'repo skipped by mirror sweep', + ) + }) + } + + return result +} + +export async function logMirrorSweepSummary(summary: IMirrorSweepSummary): Promise { + svc.log.info({ event: 'mirror_sweep_summary', ...summary }, 'mirror sweep summary') +} diff --git a/services/apps/connectors_worker/src/bin/mirror-nango-repos-to-shadow-sync-units.ts b/services/apps/connectors_worker/src/bin/mirror-nango-repos-to-shadow-sync-units.ts index 717b8b206b..ff776569db 100644 --- a/services/apps/connectors_worker/src/bin/mirror-nango-repos-to-shadow-sync-units.ts +++ b/services/apps/connectors_worker/src/bin/mirror-nango-repos-to-shadow-sync-units.ts @@ -1,24 +1,22 @@ -import { getCredential } from '@crowd/connectors' -import { githubConnector } from '@crowd/connectors/src/connectors/github' -import { mintInstallationToken } from '@crowd/connectors/src/connectors/github/appToken' -import { resolveRepoChannel } from '@crowd/connectors/src/connectors/github/discover' -import type { SyncUnitUpsert } from '@crowd/data-access-layer/src/connectors' -import { upsertSyncUnits } from '@crowd/data-access-layer/src/connectors' import { WRITE_DB_CONFIG, getDbConnection } from '@crowd/data-access-layer/src/database' +import { findActiveNangoMappingForRepo } from '@crowd/data-access-layer/src/integrations' import type { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor' import { getServiceLogger } from '@crowd/logging' -import { NangoIntegration, getNangoConnectionData, initNangoCloudClient } from '@crowd/nango' +import { initNangoCloudClient } from '@crowd/nango' +import { mirrorRepo } from '../mirrorRepo' +import { mirrorSince } from '../mirrorSweep' import { type RepoRef, parseRepoArg } from './repoArg' const log = getServiceLogger() -const MIRROR_INITIAL_LOOKBACK_MS = 48 * 60 * 60 * 1000 - -interface NangoMappingRow { - integrationId: string - connectionId: string +interface MirrorReportRow { + owner: string + repoName: string + outcome: 'mirrored' | 'skipped' + reason?: string + unitsInserted?: number } function usage(): never { @@ -36,27 +34,39 @@ function parseArgs(rawArgv: string[]): RepoRef[] { return argv.map(parseRepoArg) } -async function getNangoMappingForOwnerRepo( +async function mirrorOne( qx: QueryExecutor, - owner: string, - repoName: string, -): Promise { - const row: NangoMappingRow | null = await qx.selectOneOrNone( - `SELECT nm."integrationId", nm."connectionId" - FROM integration.nango_mapping nm - JOIN integrations i ON i.id = nm."integrationId" - WHERE lower(nm.owner) = lower($(owner)) - AND lower(nm."repoName") = lower($(repoName)) - AND i.platform = 'github-nango' - AND i."deletedAt" IS NULL - ORDER BY nm."updatedAt" DESC - LIMIT 1`, - { owner, repoName }, - ) - if (!row) { - throw new Error(`no active github-nango integration found for ${owner}/${repoName}`) + repo: RepoRef, + since: string, +): Promise { + const mapping = await findActiveNangoMappingForRepo(qx, repo.owner, repo.name) + if (!mapping) { + return { owner: repo.owner, repoName: repo.name, outcome: 'skipped', reason: 'no mapping' } } - return row + + const result = await mirrorRepo( + qx, + { + integrationId: mapping.integrationId, + connectionId: mapping.connectionId, + owner: mapping.owner, + repoName: mapping.repoName, + }, + since, + ) + return result.outcome === 'mirrored' + ? { + owner: repo.owner, + repoName: repo.name, + outcome: 'mirrored', + unitsInserted: result.unitsInserted, + } + : { + owner: repo.owner, + repoName: repo.name, + outcome: 'skipped', + reason: result.message ? `${result.reason}: ${result.message}` : result.reason, + } } setImmediate(async () => { @@ -68,51 +78,14 @@ setImmediate(async () => { await initNangoCloudClient() - const syncNames = githubConnector.syncs.map((s) => s.name) - if (syncNames.length === 0) { - log.warn('github manifest has no syncs registered yet - mirroring channels only') - } - - const cutoff = new Date(Date.now() - MIRROR_INITIAL_LOOKBACK_MS) - const since = new Date( - Date.UTC(cutoff.getUTCFullYear(), cutoff.getUTCMonth(), cutoff.getUTCDate()), - ).toISOString() - const watermark = { phase: 'incremental', since, cursor: null } - - const units: SyncUnitUpsert[] = [] + const since = mirrorSince(new Date()) + const rows: MirrorReportRow[] = [] for (const repo of repos) { - const mapping = await getNangoMappingForOwnerRepo(qx, repo.owner, repo.name) - const credential = await getCredential(qx, mapping.integrationId) - - const connectionData = await getNangoConnectionData( - NangoIntegration.GITHUB, - mapping.connectionId, - ) - const installationId: string = connectionData.connection_config.installation_id - const { token } = await mintInstallationToken(credential, installationId) - - const channel = await resolveRepoChannel(token, repo.owner, repo.name) - log.info( - { owner: repo.owner, name: repo.name, integrationId: mapping.integrationId, ...channel }, - 'repo channel resolved', - ) - - for (const syncName of syncNames) { - units.push({ - integrationId: mapping.integrationId, - platform: githubConnector.platform, - channelId: channel.channelId, - channelName: channel.channelName, - syncName, - watermark, - }) - } + rows.push(await mirrorOne(qx, repo, since)) } - const unitsUpserted = await upsertSyncUnits(qx, units) - - log.info({ repos: repos.length, syncNames, unitsUpserted }, 'mirroring complete') - process.exit(0) + console.table(rows) + process.exit(rows.some((row) => row.outcome === 'skipped') ? 1 : 0) } catch (err) { log.error(err, 'mirroring failed') process.exit(1) diff --git a/services/apps/connectors_worker/src/githubChannelActivity.test.ts b/services/apps/connectors_worker/src/githubChannelActivity.test.ts new file mode 100644 index 0000000000..636ad7c584 --- /dev/null +++ b/services/apps/connectors_worker/src/githubChannelActivity.test.ts @@ -0,0 +1,24 @@ +import { describe, expect, it, vi } from 'vitest' + +import type { TinybirdClient } from '@crowd/database' + +import { readGithubChannelActivity } from './githubChannelActivity' + +describe('readGithubChannelActivity', () => { + it('keys 30-day activity by lower-cased channel', async () => { + const executeSql = vi.fn().mockResolvedValue({ + data: [{ channelKey: 'https://github.com/a/b', activities: 12 }], + }) + + const result = await readGithubChannelActivity( + { executeSql } as unknown as TinybirdClient, + new Date('2026-10-22T03:00:00Z'), + ) + + expect(result.get('https://github.com/a/b')).toBe(12) + const [query, params] = executeSql.mock.calls[0] + expect(query).toContain('uniqExact(activityId)') + expect(query).toContain("platform = 'github'") + expect(params.since).toBe('2026-09-22 03:00:00.000') + }) +}) diff --git a/services/apps/connectors_worker/src/githubChannelActivity.ts b/services/apps/connectors_worker/src/githubChannelActivity.ts new file mode 100644 index 0000000000..080e7515c8 --- /dev/null +++ b/services/apps/connectors_worker/src/githubChannelActivity.ts @@ -0,0 +1,34 @@ +import type { TinybirdClient } from '@crowd/database' + +const ACTIVITY_WINDOW_DAYS = 30 +const CREATED_AT_BOUND_DAYS = 400 +const DAY_MS = 24 * 60 * 60 * 1000 + +const QUERY = `% +SELECT lower(channel) AS channelKey, uniqExact(activityId) AS activities +FROM activityRelations +WHERE platform = 'github' + AND channel != '' + AND timestamp >= parseDateTime64BestEffort({{String(since)}}) + AND createdAt >= parseDateTime64BestEffort({{String(createdSince)}}) +GROUP BY channelKey +FORMAT JSON +` + +type ChannelActivityRow = { channelKey: string; activities: number } + +function formatTinybirdDate(date: Date): string { + return date.toISOString().replace('T', ' ').replace('Z', '') +} + +export async function readGithubChannelActivity( + tb: TinybirdClient, + now: Date, +): Promise> { + const result = await tb.executeSql<{ data: ChannelActivityRow[] }>(QUERY, { + since: formatTinybirdDate(new Date(now.getTime() - ACTIVITY_WINDOW_DAYS * DAY_MS)), + createdSince: formatTinybirdDate(new Date(now.getTime() - CREATED_AT_BOUND_DAYS * DAY_MS)), + }) + + return new Map(result.data.map((row) => [row.channelKey, Number(row.activities)])) +} diff --git a/services/apps/connectors_worker/src/main.ts b/services/apps/connectors_worker/src/main.ts index 2cbfae1ff7..4833d4f6db 100644 --- a/services/apps/connectors_worker/src/main.ts +++ b/services/apps/connectors_worker/src/main.ts @@ -7,6 +7,7 @@ import { octolensConnector } from '@crowd/connectors/src/connectors/octolens' import { dummyConnector } from '@crowd/connectors/src/testing/dummyConnector' import { scheduleDispatcher } from './schedules/dispatcher' +import { scheduleMirrorSweep } from './schedules/mirrorSweep' import { scheduleOctolensDiscovery } from './schedules/octolensDiscovery' import { scheduleShadowDiff } from './schedules/shadowDiff' import { scheduleSwitchSyncPipelineSweep } from './schedules/switchSyncPipelineSweep' @@ -50,6 +51,7 @@ setImmediate(async () => { await scheduleShadowDiff() await scheduleOctolensDiscovery() await scheduleSwitchSyncPipelineSweep() + await scheduleMirrorSweep() await svc.start() }) diff --git a/services/apps/connectors_worker/src/mirrorRepo.test.ts b/services/apps/connectors_worker/src/mirrorRepo.test.ts new file mode 100644 index 0000000000..eec14b5333 --- /dev/null +++ b/services/apps/connectors_worker/src/mirrorRepo.test.ts @@ -0,0 +1,125 @@ +import axios from 'axios' +import { beforeEach, describe, expect, it, vi } from 'vitest' + +import type { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' + +import { mirrorRepo } from './mirrorRepo' + +const mocks = vi.hoisted(() => ({ + getCredential: vi.fn(), + mintInstallationToken: vi.fn(), + resolveRepoChannel: vi.fn(), + getNangoConnectionData: vi.fn(), + hasSyncUnitsForChannel: vi.fn(), + upsertSyncUnits: vi.fn(), +})) + +vi.mock('@crowd/connectors', () => ({ getCredential: mocks.getCredential })) +vi.mock('@crowd/connectors/src/connectors/github', () => ({ + githubConnector: { + platform: 'github', + syncs: [ + 'discussions', + 'forks', + 'issues', + 'issue-comments', + 'pull-requests', + 'pull-request-comments', + 'pull-request-review-comments', + 'pull-request-commits', + ].map((name) => ({ name })), + }, +})) +vi.mock('@crowd/connectors/src/connectors/github/appToken', () => ({ + mintInstallationToken: mocks.mintInstallationToken, +})) +vi.mock('@crowd/connectors/src/connectors/github/discover', () => ({ + resolveRepoChannel: mocks.resolveRepoChannel, +})) +vi.mock('@crowd/nango', async (importOriginal) => ({ + ...(await importOriginal()), + getNangoConnectionData: mocks.getNangoConnectionData, +})) +vi.mock('@crowd/data-access-layer/src/connectors', () => ({ + hasSyncUnitsForChannel: mocks.hasSyncUnitsForChannel, + upsertSyncUnits: mocks.upsertSyncUnits, +})) + +const qx = {} as QueryExecutor +const candidate = { integrationId: 'i-1', connectionId: 'c-1', owner: 'acme', repoName: 'widget' } +const since = '2026-10-18T00:00:00.000Z' + +describe('mirrorRepo', () => { + beforeEach(() => { + vi.clearAllMocks() + mocks.getCredential.mockResolvedValue({ appId: 'app' }) + mocks.getNangoConnectionData.mockResolvedValue({ + connection_config: { installation_id: 'inst-1' }, + }) + mocks.mintInstallationToken.mockResolvedValue({ token: 'tok' }) + mocks.resolveRepoChannel.mockResolvedValue({ + channelId: 'R_1', + channelName: 'https://github.com/Acme/Widget', + }) + mocks.hasSyncUnitsForChannel.mockResolvedValue(false) + mocks.upsertSyncUnits.mockImplementation(async (_qx, units: unknown[]) => units.length) + }) + + it('mirrors one incremental unit per github sync', async () => { + await expect(mirrorRepo(qx, candidate, since)).resolves.toEqual({ + outcome: 'mirrored', + unitsInserted: 8, + }) + + expect(mocks.mintInstallationToken).toHaveBeenCalledWith({ appId: 'app' }, 'inst-1') + expect(mocks.resolveRepoChannel).toHaveBeenCalledWith('tok', 'acme', 'widget') + expect(mocks.hasSyncUnitsForChannel).toHaveBeenCalledWith(qx, 'i-1', 'R_1') + const units = mocks.upsertSyncUnits.mock.calls[0][1] + expect(units).toHaveLength(8) + expect(new Set(units.map((u: { syncName: string }) => u.syncName)).size).toBe(8) + expect(units[0]).toEqual({ + integrationId: 'i-1', + platform: 'github', + channelId: 'R_1', + channelName: 'https://github.com/Acme/Widget', + syncName: 'discussions', + watermark: { phase: 'incremental', since, cursor: null }, + }) + }) + + it('skips a repo whose channel resolves to another name', async () => { + mocks.resolveRepoChannel.mockResolvedValue({ + channelId: 'R_1', + channelName: 'https://github.com/acme/widget-next', + }) + + await expect(mirrorRepo(qx, candidate, since)).resolves.toEqual({ + outcome: 'skipped', + reason: 'repo_renamed', + }) + + mocks.getCredential.mockRejectedValue(new Error('connection terminated')) + await expect(mirrorRepo(qx, candidate, since)).rejects.toThrow('connection terminated') + expect(mocks.hasSyncUnitsForChannel).not.toHaveBeenCalled() + expect(mocks.upsertSyncUnits).not.toHaveBeenCalled() + }) + + it('skips a channel already mirrored and a repo github no longer finds', async () => { + mocks.hasSyncUnitsForChannel.mockResolvedValue(true) + await expect(mirrorRepo(qx, candidate, since)).resolves.toEqual({ + outcome: 'skipped', + reason: 'already_mirrored', + message: 'units exist for https://github.com/Acme/Widget', + }) + + mocks.resolveRepoChannel.mockRejectedValue( + new axios.AxiosError('Not Found', '404', undefined, undefined, { status: 404 } as never), + ) + await expect(mirrorRepo(qx, candidate, since)).resolves.toEqual({ + outcome: 'skipped', + reason: 'repo_not_found', + message: 'Not Found', + }) + expect(mocks.upsertSyncUnits).not.toHaveBeenCalled() + }) +}) diff --git a/services/apps/connectors_worker/src/mirrorRepo.ts b/services/apps/connectors_worker/src/mirrorRepo.ts new file mode 100644 index 0000000000..7e4f99c744 --- /dev/null +++ b/services/apps/connectors_worker/src/mirrorRepo.ts @@ -0,0 +1,100 @@ +import { getCredential } from '@crowd/connectors' +import { githubConnector } from '@crowd/connectors/src/connectors/github' +import { mintInstallationToken } from '@crowd/connectors/src/connectors/github/appToken' +import { resolveRepoChannel } from '@crowd/connectors/src/connectors/github/discover' +import { hasSyncUnitsForChannel, upsertSyncUnits } from '@crowd/data-access-layer/src/connectors' +import type { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' +import { NangoIntegration, getNangoConnectionData } from '@crowd/nango' + +import { + type IMirrorCandidate, + type MirrorRepoResult, + type MirrorSkipReason, + type MirrorStage, + isRenamedChannel, + mirrorSkipReasonFor, +} from './mirrorSweep' + +interface IRepoChannel { + channelId: string + channelName: string +} + +class MirrorSkip extends Error { + constructor( + readonly reason: MirrorSkipReason, + readonly detail?: string, + ) { + super(reason) + } +} + +async function atStage(stage: MirrorStage, run: () => Promise): Promise { + try { + return await run() + } catch (err) { + throw new MirrorSkip( + mirrorSkipReasonFor(stage, err), + err instanceof Error ? err.message : undefined, + ) + } +} + +async function resolveCandidateChannel( + qx: QueryExecutor, + candidate: IMirrorCandidate, +): Promise { + const credential = await getCredential(qx, candidate.integrationId) + const connection = await atStage('nango', () => + getNangoConnectionData(NangoIntegration.GITHUB, candidate.connectionId), + ) + const installationId: string | undefined = connection.connection_config?.installation_id + if (!installationId) { + throw new MirrorSkip('no_installation_id') + } + const { token } = await atStage('mint', () => mintInstallationToken(credential, installationId)) + return atStage('resolve', () => resolveRepoChannel(token, candidate.owner, candidate.repoName)) +} + +export async function mirrorRepo( + qx: QueryExecutor, + candidate: IMirrorCandidate, + since: string, +): Promise { + let channel: IRepoChannel + try { + channel = await resolveCandidateChannel(qx, candidate) + } catch (err) { + if (err instanceof MirrorSkip) { + return err.detail === undefined + ? { outcome: 'skipped', reason: err.reason } + : { outcome: 'skipped', reason: err.reason, message: err.detail } + } + throw err + } + + if (isRenamedChannel(candidate, channel.channelName)) { + return { outcome: 'skipped', reason: 'repo_renamed' } + } + if (await hasSyncUnitsForChannel(qx, candidate.integrationId, channel.channelId)) { + return { + outcome: 'skipped', + reason: 'already_mirrored', + message: `units exist for ${channel.channelName}`, + } + } + + const watermark = { phase: 'incremental', since, cursor: null } + const unitsInserted = await upsertSyncUnits( + qx, + githubConnector.syncs.map(({ name }) => ({ + integrationId: candidate.integrationId, + platform: githubConnector.platform, + channelId: channel.channelId, + channelName: channel.channelName, + syncName: name, + watermark, + })), + ) + return { outcome: 'mirrored', unitsInserted } +} diff --git a/services/apps/connectors_worker/src/mirrorSweep.test.ts b/services/apps/connectors_worker/src/mirrorSweep.test.ts new file mode 100644 index 0000000000..4b1993ed13 --- /dev/null +++ b/services/apps/connectors_worker/src/mirrorSweep.test.ts @@ -0,0 +1,122 @@ +import axios from 'axios' +import { describe, expect, it } from 'vitest' + +import type { IUnmirroredNangoRepo } from '@crowd/data-access-layer/src/integrations' + +import { + type IMirrorCandidate, + chunk, + isRenamedChannel, + mirrorRoom, + mirrorSince, + mirrorSkipReasonFor, + orderMirrorCandidates, +} from './mirrorSweep' + +const repo = (overrides: Partial = {}): IUnmirroredNangoRepo => ({ + integrationId: 'i-1', + connectionId: 'c-1', + owner: 'owner', + repoName: 'repo', + enabled: true, + archived: false, + excluded: false, + ...overrides, +}) + +const axiosError = (status: number) => + new axios.AxiosError('failed', String(status), undefined, undefined, { + status, + } as never) + +describe('mirrorRoom', () => { + it('caps the batch by the remaining shadow room and never goes negative', () => { + expect(mirrorRoom({ batchSize: 500, shadowCap: 5000 }, 110)).toBe(500) + expect(mirrorRoom({ batchSize: 2000, shadowCap: 5000 }, 4600)).toBe(400) + expect(mirrorRoom({ batchSize: 500, shadowCap: 5000 }, 5200)).toBe(0) + }) +}) + +describe('mirrorSince', () => { + it('returns midnight UTC of the day 48 hours before now', () => { + expect(mirrorSince(new Date('2026-10-22T03:00:00Z'))).toBe('2026-10-20T00:00:00.000Z') + }) +}) + +describe('orderMirrorCandidates', () => { + it('puts active repos first, then activity desc, then owner and repo name', () => { + const ordered = orderMirrorCandidates( + [ + repo({ owner: 'a', repoName: 'archived', archived: true }), + repo({ owner: 'b', repoName: 'quiet' }), + repo({ owner: 'c', repoName: 'busy' }), + repo({ owner: 'z', repoName: 'tie' }), + repo({ owner: 'y', repoName: 'tie' }), + repo({ owner: 'y', repoName: 'alpha' }), + ], + new Map([ + ['https://github.com/a/archived', 1000], + ['https://github.com/b/quiet', 10], + ['https://github.com/c/busy', 50], + ['https://github.com/z/tie', 5], + ['https://github.com/y/tie', 5], + ['https://github.com/y/alpha', 5], + ]), + ) + expect(ordered.map((c) => `${c.owner}/${c.repoName}`)).toEqual([ + 'c/busy', + 'b/quiet', + 'y/alpha', + 'y/tie', + 'z/tie', + 'a/archived', + ]) + expect(ordered[0]).toEqual({ + integrationId: 'i-1', + connectionId: 'c-1', + owner: 'c', + repoName: 'busy', + }) + }) + + it('matches activity case-insensitively', () => { + const ordered = orderMirrorCandidates( + [repo({ owner: 'A', repoName: 'Low' }), repo({ owner: 'B', repoName: 'High' })], + new Map([['https://github.com/b/high', 9]]), + ) + expect(ordered.map((c) => c.repoName)).toEqual(['High', 'Low']) + }) +}) + +describe('isRenamedChannel', () => { + it('compares the channel to the mapped owner and repo ignoring case', () => { + const candidate: IMirrorCandidate = { + integrationId: 'i-1', + connectionId: 'c-1', + owner: 'owner', + repoName: 'repo', + } + expect(isRenamedChannel(candidate, 'https://github.com/Owner/Repo')).toBe(false) + expect(isRenamedChannel(candidate, 'https://github.com/other/repo')).toBe(true) + }) +}) + +describe('mirrorSkipReasonFor', () => { + it('classifies each stage and tells a 404 from other resolve errors', () => { + expect(mirrorSkipReasonFor('resolve', axiosError(404))).toBe('repo_not_found') + expect(mirrorSkipReasonFor('resolve', axiosError(500))).toBe('github_error') + expect(mirrorSkipReasonFor('mint', new Error('x'))).toBe('token_mint_failed') + expect(mirrorSkipReasonFor('nango', new Error('x'))).toBe('nango_error') + }) +}) + +describe('chunk', () => { + it('splits into fixed-size groups and returns nothing for empty input', () => { + const sizes = chunk( + Array.from({ length: 250 }, (_, i) => i), + 100, + ).map((c) => c.length) + expect(sizes).toEqual([100, 100, 50]) + expect(chunk([], 100)).toEqual([]) + }) +}) diff --git a/services/apps/connectors_worker/src/mirrorSweep.ts b/services/apps/connectors_worker/src/mirrorSweep.ts new file mode 100644 index 0000000000..14438c472b --- /dev/null +++ b/services/apps/connectors_worker/src/mirrorSweep.ts @@ -0,0 +1,106 @@ +import axios from 'axios' + +import type { IUnmirroredNangoRepo } from '@crowd/data-access-layer/src/integrations' +import { githubChannelName } from '@crowd/nango' + +export { + MIRROR_CHUNK_SIZE, + MIRROR_RESOLVE_CONCURRENCY, + chunk, + mirrorRoom, +} from './mirrorSweepLimits' + +export type MirrorSkipReason = + | 'no_installation_id' + | 'token_mint_failed' + | 'repo_not_found' + | 'repo_renamed' + | 'github_error' + | 'nango_error' + | 'already_mirrored' + +export interface IMirrorCandidate { + integrationId: string + connectionId: string + owner: string + repoName: string +} + +export interface IMirrorSweepInput { + batchSize: number + shadowCap: number +} + +export interface IMirrorSweepSummary { + batchSize: number + shadowCap: number + inShadow: number + candidates: number + attempted: number + mirrored: number + skipped: number + unitsInserted: number + chunksFailed: number + durationMs: number +} + +export type MirrorRepoResult = + | { outcome: 'mirrored'; unitsInserted: number } + | { outcome: 'skipped'; reason: MirrorSkipReason; message?: string } + +export type MirrorStage = 'nango' | 'mint' | 'resolve' + +const LOOKBACK_MS = 48 * 60 * 60 * 1000 + +export function mirrorSince(now: Date): string { + const lookback = new Date(now.getTime() - LOOKBACK_MS) + return new Date( + Date.UTC(lookback.getUTCFullYear(), lookback.getUTCMonth(), lookback.getUTCDate()), + ).toISOString() +} + +const channelKey = (owner: string, repoName: string): string => + githubChannelName(owner, repoName).toLowerCase() + +const isActive = (repo: IUnmirroredNangoRepo): boolean => + repo.enabled && !repo.archived && !repo.excluded + +export function orderMirrorCandidates( + repos: IUnmirroredNangoRepo[], + activityByChannel: Map, +): IMirrorCandidate[] { + const activityOf = (repo: IUnmirroredNangoRepo): number => + activityByChannel.get(channelKey(repo.owner, repo.repoName)) ?? 0 + + return [...repos] + .sort( + (a, b) => + Number(isActive(b)) - Number(isActive(a)) || + activityOf(b) - activityOf(a) || + a.owner.localeCompare(b.owner) || + a.repoName.localeCompare(b.repoName), + ) + .map(({ integrationId, connectionId, owner, repoName }) => ({ + integrationId, + connectionId, + owner, + repoName, + })) +} + +export function isRenamedChannel(candidate: IMirrorCandidate, channelName: string): boolean { + return channelName.toLowerCase() !== channelKey(candidate.owner, candidate.repoName) +} + +export function mirrorSkipReasonFor(stage: MirrorStage, err: unknown): MirrorSkipReason { + switch (stage) { + case 'nango': + return 'nango_error' + case 'mint': + return 'token_mint_failed' + case 'resolve': + return axios.isAxiosError(err) && err.response?.status === 404 + ? 'repo_not_found' + : 'github_error' + } +} diff --git a/services/apps/connectors_worker/src/mirrorSweepLimits.ts b/services/apps/connectors_worker/src/mirrorSweepLimits.ts new file mode 100644 index 0000000000..709fea33be --- /dev/null +++ b/services/apps/connectors_worker/src/mirrorSweepLimits.ts @@ -0,0 +1,16 @@ +import type { IMirrorSweepInput } from './mirrorSweep' + +export const MIRROR_CHUNK_SIZE = 100 +export const MIRROR_RESOLVE_CONCURRENCY = 5 + +export function mirrorRoom(input: IMirrorSweepInput, inShadow: number): number { + return Math.max(0, Math.min(input.batchSize, input.shadowCap - inShadow)) +} + +export function chunk(items: T[], size: number): T[][] { + const chunks: T[][] = [] + for (let i = 0; i < items.length; i += size) { + chunks.push(items.slice(i, i + size)) + } + return chunks +} diff --git a/services/apps/connectors_worker/src/schedules/mirrorSweep.ts b/services/apps/connectors_worker/src/schedules/mirrorSweep.ts new file mode 100644 index 0000000000..6718c4eb98 --- /dev/null +++ b/services/apps/connectors_worker/src/schedules/mirrorSweep.ts @@ -0,0 +1,32 @@ +import { ScheduleAlreadyRunning, ScheduleOverlapPolicy } from '@temporalio/client' + +import { svc } from '../main' +import { mirrorSweep } from '../workflows/mirrorSweep' + +export async function scheduleMirrorSweep(): Promise { + try { + await svc.temporal.schedule.create({ + scheduleId: 'connectors-mirror-sweep', + spec: { + calendars: [{ hour: 3, minute: 0 }], + }, + policies: { + overlap: ScheduleOverlapPolicy.SKIP, + catchupWindow: '1 hour', + }, + action: { + type: 'startWorkflow', + workflowType: mirrorSweep, + taskQueue: 'connectors', + args: [{ batchSize: 500, shadowCap: 5000 }], + workflowExecutionTimeout: '2 hours', + }, + }) + } catch (err) { + if (err instanceof ScheduleAlreadyRunning) { + svc.log.info('Mirror sweep schedule already registered in Temporal.') + } else { + throw new Error(err) + } + } +} diff --git a/services/apps/connectors_worker/src/workflows.ts b/services/apps/connectors_worker/src/workflows.ts index 98ccc54b03..2454554603 100644 --- a/services/apps/connectors_worker/src/workflows.ts +++ b/services/apps/connectors_worker/src/workflows.ts @@ -1,4 +1,5 @@ import { dispatcher } from './workflows/dispatcher' +import { mirrorSweep } from './workflows/mirrorSweep' import { octolensDiscovery } from './workflows/octolensDiscovery' import { shadowDiff } from './workflows/shadowDiff' import { switchSyncPipeline } from './workflows/switchSyncPipeline' @@ -7,6 +8,7 @@ import { syncRun } from './workflows/syncRun' export { dispatcher, + mirrorSweep, octolensDiscovery, shadowDiff, switchSyncPipeline, diff --git a/services/apps/connectors_worker/src/workflows/mirrorSweep.test.ts b/services/apps/connectors_worker/src/workflows/mirrorSweep.test.ts new file mode 100644 index 0000000000..55bb74d81e --- /dev/null +++ b/services/apps/connectors_worker/src/workflows/mirrorSweep.test.ts @@ -0,0 +1,88 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +import { mirrorSweep } from './mirrorSweep' + +const mocks = vi.hoisted(() => ({ + countShadowRepos: vi.fn(), + selectMirrorBatch: vi.fn(), + mirrorRepos: vi.fn(), + logMirrorSweepSummary: vi.fn(), + log: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, +})) + +vi.mock('@temporalio/workflow', () => ({ + proxyActivities: () => ({ + countShadowRepos: mocks.countShadowRepos, + selectMirrorBatch: mocks.selectMirrorBatch, + mirrorRepos: mocks.mirrorRepos, + logMirrorSweepSummary: mocks.logMirrorSweepSummary, + }), + log: mocks.log, + ApplicationFailure: { nonRetryable: (message: string) => new Error(message) }, +})) + +const candidates = (count: number) => + Array.from({ length: count }, (_, i) => ({ + integrationId: 'i', + connectionId: `c-${i}`, + owner: 'o', + repoName: `r-${i}`, + })) + +describe('mirrorSweep', () => { + beforeEach(() => { + vi.clearAllMocks() + }) + + it('mirrors in sequential chunks of 100, stops at the first failed chunk and still logs the summary', async () => { + const batch = candidates(250) + mocks.countShadowRepos.mockResolvedValue(110) + mocks.selectMirrorBatch.mockResolvedValue({ + since: '2026-10-18T00:00:00.000Z', + candidatesTotal: 900, + batch, + }) + mocks.mirrorRepos + .mockResolvedValueOnce({ mirrored: 90, skipped: 10, unitsInserted: 800 }) + .mockRejectedValueOnce(new Error('chunk timed out')) + + const summary = await mirrorSweep({ batchSize: 500, shadowCap: 5000 }) + + expect(mocks.selectMirrorBatch).toHaveBeenCalledWith(500) + expect(mocks.mirrorRepos.mock.calls.map(([repos, since]) => [repos.length, since])).toEqual([ + [100, '2026-10-18T00:00:00.000Z'], + [100, '2026-10-18T00:00:00.000Z'], + ]) + expect(summary).toMatchObject({ + batchSize: 500, + shadowCap: 5000, + inShadow: 110, + candidates: 900, + attempted: 100, + mirrored: 90, + skipped: 10, + unitsInserted: 800, + chunksFailed: 1, + }) + expect(mocks.logMirrorSweepSummary).toHaveBeenCalledTimes(1) + expect(mocks.logMirrorSweepSummary).toHaveBeenCalledWith(summary) + }) + + it('does not select or mirror anything when the shadow pool is full, but still logs the summary', async () => { + mocks.countShadowRepos.mockResolvedValue(5000) + + const summary = await mirrorSweep({ batchSize: 500, shadowCap: 5000 }) + + expect(mocks.selectMirrorBatch).not.toHaveBeenCalled() + expect(mocks.mirrorRepos).not.toHaveBeenCalled() + expect(summary).toMatchObject({ inShadow: 5000, candidates: 0, attempted: 0, chunksFailed: 0 }) + expect(mocks.logMirrorSweepSummary).toHaveBeenCalledWith(summary) + + vi.clearAllMocks() + await expect(mirrorSweep({ batchSize: 0, shadowCap: 5000 })).rejects.toThrow('batchSize') + await expect(mirrorSweep({ shadowCap: 5000 } as never)).rejects.toThrow('batchSize') + await expect(mirrorSweep({ batchSize: 600, shadowCap: 500 })).rejects.toThrow('shadowCap') + expect(mocks.countShadowRepos).not.toHaveBeenCalled() + expect(mocks.logMirrorSweepSummary).not.toHaveBeenCalled() + }) +}) diff --git a/services/apps/connectors_worker/src/workflows/mirrorSweep.ts b/services/apps/connectors_worker/src/workflows/mirrorSweep.ts new file mode 100644 index 0000000000..599267bcf2 --- /dev/null +++ b/services/apps/connectors_worker/src/workflows/mirrorSweep.ts @@ -0,0 +1,75 @@ +import { ApplicationFailure, log, proxyActivities } from '@temporalio/workflow' + +import type * as activities from '../activities/mirrorSweepActivities' +import type { IMirrorSweepInput, IMirrorSweepSummary } from '../mirrorSweep' +import { MIRROR_CHUNK_SIZE, chunk, mirrorRoom } from '../mirrorSweepLimits' + +const activity = proxyActivities({ + startToCloseTimeout: '10 minutes', + retry: { + maximumAttempts: 3, + backoffCoefficient: 2, + }, +}) + +const isPositiveInteger = (value: unknown): boolean => Number.isInteger(value) && Number(value) > 0 + +function assertValidInput(input: IMirrorSweepInput): void { + if (!isPositiveInteger(input?.batchSize)) { + throw ApplicationFailure.nonRetryable( + `mirrorSweep input batchSize must be a positive integer, got ${JSON.stringify(input?.batchSize)}`, + ) + } + if (!isPositiveInteger(input.shadowCap)) { + throw ApplicationFailure.nonRetryable( + `mirrorSweep input shadowCap must be a positive integer, got ${JSON.stringify(input.shadowCap)}`, + ) + } + if (input.batchSize > input.shadowCap) { + throw ApplicationFailure.nonRetryable( + `mirrorSweep input batchSize ${input.batchSize} must not exceed shadowCap ${input.shadowCap}`, + ) + } +} + +export async function mirrorSweep(input: IMirrorSweepInput): Promise { + assertValidInput(input) + const startedAt = Date.now() + const inShadow = await activity.countShadowRepos() + const summary: IMirrorSweepSummary = { + batchSize: input.batchSize, + shadowCap: input.shadowCap, + inShadow, + candidates: 0, + attempted: 0, + mirrored: 0, + skipped: 0, + unitsInserted: 0, + chunksFailed: 0, + durationMs: 0, + } + + const room = mirrorRoom(input, inShadow) + if (room > 0) { + const { since, candidatesTotal, batch } = await activity.selectMirrorBatch(room) + summary.candidates = candidatesTotal + + for (const repos of chunk(batch, MIRROR_CHUNK_SIZE)) { + try { + const result = await activity.mirrorRepos(repos, since) + summary.attempted += repos.length + summary.mirrored += result.mirrored + summary.skipped += result.skipped + summary.unitsInserted += result.unitsInserted + } catch (err) { + log.error('mirror sweep chunk failed', { err }) + summary.chunksFailed += 1 + break + } + } + } + + summary.durationMs = Date.now() - startedAt + await activity.logMirrorSweepSummary(summary) + return summary +} diff --git a/services/apps/connectors_worker/tsconfig.json b/services/apps/connectors_worker/tsconfig.json index 52b3ed9336..78ff18dd1d 100644 --- a/services/apps/connectors_worker/tsconfig.json +++ b/services/apps/connectors_worker/tsconfig.json @@ -17,6 +17,9 @@ { "path": "../../libs/data-access-layer" }, + { + "path": "../../libs/database" + }, { "path": "../../libs/logging" }, diff --git a/services/libs/data-access-layer/src/connectors/syncUnits.test.ts b/services/libs/data-access-layer/src/connectors/syncUnits.test.ts index 79912c4a1f..c8fa9fa9b5 100644 --- a/services/libs/data-access-layer/src/connectors/syncUnits.test.ts +++ b/services/libs/data-access-layer/src/connectors/syncUnits.test.ts @@ -7,6 +7,8 @@ import { withQx } from '@crowd/test-kit/db' import { recordShadowRecords } from './shadowRecords' import { claimDueUnits, + countShadowRepos, + hasSyncUnitsForChannel, listShadowDiffUnits, reassertUnitClaim, renewRunningUnitLease, @@ -244,6 +246,47 @@ describe('upsertSyncUnits', () => { }) }) +describe('countShadowRepos and hasSyncUnitsForChannel', () => { + test('counts repos with a live shadow unit, and finds units by channel id', async ({ qx }) => { + const integrationId = await createIntegration(qx) + + await createSyncUnit(qx, integrationId, { + channelName: 'org/emitting', + syncName: 'issues', + emitEnabled: true, + }) + await createSyncUnit(qx, integrationId, { + channelName: 'org/emitting', + syncName: 'forks', + emitEnabled: true, + }) + await createSyncUnit(qx, integrationId, { + channelName: 'org/mixed', + syncName: 'issues', + emitEnabled: true, + }) + await createSyncUnit(qx, integrationId, { + channelName: 'org/retired', + syncName: 'issues', + status: 'decommissioned', + }) + await createSyncUnit(qx, integrationId, { channelName: 'org/mixed', syncName: 'forks' }) + await createSyncUnit(qx, integrationId, { channelName: 'org/shadow', syncName: 'issues' }) + await createSyncUnit(qx, integrationId, { channelName: 'org/shadow', syncName: 'forks' }) + const deletedIntegrationId = await createIntegration(qx, { + deletedAt: '2026-09-01T00:00:00.000Z', + }) + await createSyncUnit(qx, deletedIntegrationId, { + channelName: 'org/orphan', + syncName: 'issues', + }) + + expect(await countShadowRepos(qx)).toBe(2) + expect(await hasSyncUnitsForChannel(qx, integrationId, 'org/shadow')).toBe(true) + expect(await hasSyncUnitsForChannel(qx, integrationId, 'org/unknown')).toBe(false) + }) +}) + describe('reviveDeadLetteredSyncUnits', () => { async function readUnit(qx: QueryExecutor, id: string) { return qx.selectOne( diff --git a/services/libs/data-access-layer/src/connectors/syncUnits.ts b/services/libs/data-access-layer/src/connectors/syncUnits.ts index cfb393251c..ec48059ae8 100644 --- a/services/libs/data-access-layer/src/connectors/syncUnits.ts +++ b/services/libs/data-access-layer/src/connectors/syncUnits.ts @@ -391,3 +391,36 @@ export async function listShadowDiffUnits(qx: QueryExecutor): Promise { + const row = await qx.selectOne( + `SELECT COUNT(DISTINCT (su."integrationId", su."channelId"))::int AS count + FROM integration.sync_units su + WHERE platform = 'github' + AND "emitEnabled" IS NOT TRUE + AND status <> 'decommissioned' + AND EXISTS ( + SELECT 1 + FROM public.integrations i + WHERE i.id = su."integrationId" AND i."deletedAt" IS NULL + )`, + ) + return row.count +} + +export async function hasSyncUnitsForChannel( + qx: QueryExecutor, + integrationId: string, + channelId: string, +): Promise { + const row = await qx.selectOne( + `SELECT EXISTS ( + SELECT 1 + FROM integration.sync_units + WHERE "integrationId" = $(integrationId) + AND "channelId" = $(channelId) + ) AS exists`, + { integrationId, channelId }, + ) + return row.exists === true +} diff --git a/services/libs/data-access-layer/src/integrations/index.ts b/services/libs/data-access-layer/src/integrations/index.ts index 32e037ef58..d513382720 100644 --- a/services/libs/data-access-layer/src/integrations/index.ts +++ b/services/libs/data-access-layer/src/integrations/index.ts @@ -620,6 +620,45 @@ export async function findActiveNangoMappingForRepo( ) } +export interface IUnmirroredNangoRepo { + integrationId: string + connectionId: string + owner: string + repoName: string + enabled: boolean + archived: boolean + excluded: boolean +} + +export async function listUnmirroredNangoRepos(qx: QueryExecutor): Promise { + return qx.select( + ` + SELECT + nm."integrationId", + nm."connectionId", + nm.owner, + nm."repoName", + COALESCE(r.enabled, true) AS enabled, + COALESCE(r.archived, false) AS archived, + COALESCE(r.excluded, false) AS excluded + FROM integration.nango_mapping nm + JOIN integrations i ON i.id = nm."integrationId" + LEFT JOIN repositories r ON r.id = nm."repositoryId" + WHERE i.platform = $(nangoPlatform) + AND i."deletedAt" IS NULL + AND i.status = 'done' + AND NOT EXISTS ( + SELECT 1 + FROM integration.sync_units su + WHERE su."integrationId" = nm."integrationId" + AND lower(su."channelName") = lower('https://github.com/' || nm.owner || '/' || nm."repoName") + ) + ORDER BY nm."createdAt", nm.owner, nm."repoName" + `, + { nangoPlatform: PlatformType.GITHUB_NANGO }, + ) +} + export async function getNangoMappingForRepo( qx: QueryExecutor, integrationId: string, diff --git a/services/libs/data-access-layer/src/integrations/nangoMapping.test.ts b/services/libs/data-access-layer/src/integrations/nangoMapping.test.ts index a3e8ced2f2..b16162ce65 100644 --- a/services/libs/data-access-layer/src/integrations/nangoMapping.test.ts +++ b/services/libs/data-access-layer/src/integrations/nangoMapping.test.ts @@ -5,7 +5,7 @@ import type { QueryExecutor } from '@crowd/database' import { withQx } from '@crowd/test-kit/db' import { createSegments } from '@crowd/test-kit/factories' -import { addGithubNangoConnection, getNangoMappingForRepo } from './index' +import { addGithubNangoConnection, getNangoMappingForRepo, listUnmirroredNangoRepos } from './index' const test = withQx(base) @@ -89,3 +89,77 @@ describe('getNangoMappingForRepo', () => { expect(result).toBeNull() }) }) + +describe('listUnmirroredNangoRepos', () => { + test('returns active mappings without a unit, flags archived repos, and orders by creation', async ({ + qx, + }) => { + const [segmentGroup] = ( + await createSegments(qx, [{ name: generateUUIDv1(), slug: generateUUIDv1() }]) + ).projectGroups + const segmentId = segmentGroup.row.id + + const gitIntegrationId = await createIntegration(qx, 'git', segmentId) + const nangoIntegrationId = await createIntegration(qx, 'github-nango', segmentId) + const deletedIntegrationId = await createIntegration(qx, 'github-nango', segmentId) + const pendingIntegrationId = await createIntegration(qx, 'github-nango', segmentId) + await qx.result(`UPDATE public.integrations SET status = 'in-progress' WHERE id = $(id)`, { + id: pendingIntegrationId, + }) + await qx.result(`UPDATE public.integrations SET "deletedAt" = NOW() WHERE id = $(id)`, { + id: deletedIntegrationId, + }) + + const insightsProjectId = generateUUIDv1() + await qx.result( + `INSERT INTO "insightsProjects" (id, name, slug, "segmentId", "isLF") VALUES ($(id), $(id), $(id), $(segmentId), false)`, + { id: insightsProjectId, segmentId }, + ) + const archivedRepositoryId = generateUUIDv1() + await qx.result( + ` + INSERT INTO public.repositories (id, url, "segmentId", "gitIntegrationId", "sourceIntegrationId", "insightsProjectId", archived) + VALUES ($(id), $(url), $(segmentId), $(gitIntegrationId), $(nangoIntegrationId), $(insightsProjectId), true) + `, + { + id: archivedRepositoryId, + url: `https://github.com/${archivedRepositoryId}`, + segmentId, + gitIntegrationId, + nangoIntegrationId, + insightsProjectId, + }, + ) + + await addGithubNangoConnection(qx, nangoIntegrationId, 'conn-mirrored', 'owner', 'repo') + await addGithubNangoConnection(qx, nangoIntegrationId, 'conn-plain', 'plain', 'repo') + await addGithubNangoConnection(qx, nangoIntegrationId, 'conn-archived', 'old', 'repo') + await addGithubNangoConnection(qx, deletedIntegrationId, 'conn-deleted', 'gone', 'repo') + await addGithubNangoConnection(qx, pendingIntegrationId, 'conn-pending', 'wait', 'repo') + await qx.result( + `UPDATE integration.nango_mapping SET "repositoryId" = $(archivedRepositoryId) WHERE "connectionId" = 'conn-archived'`, + { archivedRepositoryId }, + ) + await qx.result( + `UPDATE integration.nango_mapping SET "createdAt" = NOW() - interval '1 day' WHERE "connectionId" = 'conn-archived'`, + ) + await qx.result( + ` + INSERT INTO integration.sync_units (id, "integrationId", platform, "channelId", "channelName", "syncName") + VALUES ($(id), $(nangoIntegrationId), 'github', 'owner/repo', 'https://github.com/Owner/Repo', 'issues') + `, + { id: generateUUIDv1(), nangoIntegrationId }, + ) + + const result = await listUnmirroredNangoRepos(qx) + const ours = result.filter((r) => r.integrationId === nangoIntegrationId) + + expect( + result.some((r) => [deletedIntegrationId, pendingIntegrationId].includes(r.integrationId)), + ).toBe(false) + expect(ours.map((r) => [r.connectionId, r.archived, r.enabled, r.excluded])).toEqual([ + ['conn-archived', true, true, false], + ['conn-plain', false, true, false], + ]) + }) +})