feat: nightly mirror sweep of github-nango repos into shadow sync units

services/apps/connectors_worker/package.json

@@ -12,14 +12,16 @@
     "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:*",
     "@crowd/archetype-worker": "workspace:*",
     "@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": {

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,13 +43,16 @@ export {
   listShadowDiffChannels,
   listSwitchSweepRepos,
   logDispatchSummary,
+  logMirrorSweepSummary,
+  mirrorRepos,
   pauseNangoRepoSync,
   quiesceUnit,
   readNangoHandoverPoint,
   resolveSwitchTarget,
   resumeNangoRepoSync,
   runOctolensDiscovery,
   runShadowDiffForChannel,
+  selectMirrorBatch,
   setUnitEmitEnabled,
   startRun,
   takeOverUnit,

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<typeof import('@crowd/nango')>()),
+  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',
+    )
+  })
+})

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<number> {
+  return countShadowReposInDb(dbStoreQx(svc.postgres.writer))
+}
+
+export async function selectMirrorBatch(limit: number): Promise<IMirrorBatch> {
+  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<IMirrorReposResult> {
+  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<void> {
+  svc.log.info({ event: 'mirror_sweep_summary', ...summary }, 'mirror sweep summary')
+}