feat: run octolens discovery on a schedule with an integration per subproject

services/apps/connectors_worker/src/activities.ts

@@ -6,6 +6,7 @@ import {
   startRun,
   touchHeartbeat,
 } from './activities/dispatcherActivities'
+import { runOctolensDiscovery } from './activities/octolensDiscoveryActivities'
 import { listShadowDiffChannels, runShadowDiffForChannel } from './activities/shadowDiffActivities'
 import { executeSync } from './activities/syncRunActivities'
 
@@ -16,6 +17,7 @@ export {
   executeSync,
   listShadowDiffChannels,
   logDispatchSummary,
+  runOctolensDiscovery,
   runShadowDiffForChannel,
   startRun,
   touchHeartbeat,

services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts

@@ -0,0 +1,151 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+import { runOctolensDiscovery } from './octolensDiscoveryActivities'
+
+const calls: string[] = []
+const record =
+  <T>(name: string, value: T) =>
+  async (...args: unknown[]) => {
+    calls.push(`${name}${args.length > 1 ? ` ${JSON.stringify(args.slice(1))}` : ''}`)
+    return value
+  }
+
+const mocks = vi.hoisted(() => ({
+  findOrCreateOctolensAnchorIntegration: vi.fn(),
+  reconcileOctolensKeywordMappings: vi.fn(),
+  upsertSyncUnits: vi.fn(),
+  reviveDeadLetteredSyncUnits: vi.fn(),
+  getCredential: vi.fn(),
+  discover: vi.fn(),
+  runMappingDiscovery: vi.fn(),
+  isOctolensCredentialConfigured: vi.fn(),
+}))
+
+vi.mock('../main', () => ({
+  svc: {
+    postgres: { writer: {} },
+    log: { info: vi.fn(), warn: vi.fn(), error: vi.fn() },
+  },
+}))
+
+vi.mock('@crowd/data-access-layer/src/queryExecutor', () => ({
+  dbStoreQx: vi.fn(() => ({})),
+}))
+
+vi.mock('@crowd/data-access-layer/src/octolens', () => ({
+  findOrCreateOctolensAnchorIntegration: mocks.findOrCreateOctolensAnchorIntegration,
+  reconcileOctolensKeywordMappings: mocks.reconcileOctolensKeywordMappings,
+}))
+
+vi.mock('@crowd/data-access-layer/src/connectors', () => ({
+  upsertSyncUnits: mocks.upsertSyncUnits,
+  reviveDeadLetteredSyncUnits: mocks.reviveDeadLetteredSyncUnits,
+}))
+
+vi.mock('@crowd/connectors', async (importOriginal) => ({
+  ...(await importOriginal<typeof import('@crowd/connectors')>()),
+  getCredential: mocks.getCredential,
+  getManifest: () => ({ discover: mocks.discover }),
+  isOctolensCredentialConfigured: mocks.isOctolensCredentialConfigured,
+}))
+
+vi.mock('@crowd/connectors/src/connectors/octolens/deps', () => ({
+  resolveOctolensProjectSegmentId: vi.fn(),
+}))
+
+vi.mock('@crowd/connectors/src/connectors/octolens/mappingDiscovery', () => ({
+  runMappingDiscovery: mocks.runMappingDiscovery,
+}))
+
+const MAPPINGS = [
+  { keywordId: 1, keyword: 'ebpf', segmentId: 'segment-ebpf' },
+  { keywordId: 3, keyword: 'valkey', segmentId: 'segment-valkey' },
+]
+
+beforeEach(() => {
+  vi.clearAllMocks()
+  calls.length = 0
+  mocks.isOctolensCredentialConfigured.mockReturnValue(true)
+  mocks.findOrCreateOctolensAnchorIntegration.mockImplementation(
+    record('anchor', { id: 'anchor', created: true, restored: false }),
+  )
+  mocks.getCredential.mockResolvedValue({
+    platform: 'octolens',
+    kind: 'octolens-token',
+    data: { token: 'token' },
+  })
+  mocks.discover.mockResolvedValue([
+    { channelId: 'octolens-global', channelName: 'octolens-global' },
+  ])
+  mocks.upsertSyncUnits.mockImplementation(record('upsertSyncUnits', 1))
+  mocks.reviveDeadLetteredSyncUnits.mockImplementation(record('revive', 0))
+  mocks.reconcileOctolensKeywordMappings.mockImplementation(
+    record('reconcile', { deactivated: 4, created: [], restored: [], softDeleted: [] }),
+  )
+  mocks.runMappingDiscovery.mockImplementation(
+    async ({
+      reconcileMappings,
+    }: {
+      reconcileMappings: (m: typeof MAPPINGS) => Promise<number>
+    }) => {
+      const deactivated = await reconcileMappings(MAPPINGS)
+      return { complete: true, mappings: MAPPINGS.length, deactivated }
+    },
+  )
+})
+
+describe('runOctolensDiscovery', () => {
+  it('ensures the anchor and its mention poll unit with emitEnabled, so mentions become activities', async () => {
+    await runOctolensDiscovery()
+
+    expect(mocks.getCredential).toHaveBeenCalledWith(expect.anything(), 'anchor')
+    expect(mocks.upsertSyncUnits).toHaveBeenCalledWith(expect.anything(), [
+      {
+        integrationId: 'anchor',
+        platform: 'octolens',
+        channelId: 'octolens-global',
+        channelName: 'octolens-global',
+        syncName: 'mentionPoll',
+        emitEnabled: true,
+      },
+    ])
+  })
+
+  it('reconciles the discovered mappings, then revives the mention poll', async () => {
+    const result = await runOctolensDiscovery()
+
+    expect(calls).toEqual([
+      'anchor',
+      'upsertSyncUnits [[{"integrationId":"anchor","platform":"octolens","channelId":"octolens-global","channelName":"octolens-global","syncName":"mentionPoll","emitEnabled":true}]]',
+      `reconcile [${JSON.stringify(MAPPINGS)}]`,
+      'revive ["anchor","mentionPoll"]',
+    ])
+    expect(result).toEqual({ complete: true, mappings: 2, deactivated: 4 })
+  })
+
+  it('does nothing at all when no Octolens token is configured', async () => {
+    mocks.isOctolensCredentialConfigured.mockReturnValue(false)
+
+    const result = await runOctolensDiscovery()
+
+    expect(result).toEqual({ complete: false, mappings: 0, deactivated: 0 })
+    expect(mocks.findOrCreateOctolensAnchorIntegration).not.toHaveBeenCalled()
+    expect(mocks.upsertSyncUnits).not.toHaveBeenCalled()
+    expect(mocks.runMappingDiscovery).not.toHaveBeenCalled()
+  })
+
+  it('does not revive the mention poll when Octolens rejects the token', async () => {
+    mocks.runMappingDiscovery.mockRejectedValue(new Error('provider authentication failed'))
+
+    await expect(runOctolensDiscovery()).rejects.toThrow('provider authentication failed')
+    expect(mocks.reviveDeadLetteredSyncUnits).not.toHaveBeenCalled()
+  })
+
+  it('makes no mapping writes when discovery does not reconcile (incomplete snapshot)', async () => {
+    mocks.runMappingDiscovery.mockResolvedValue({ complete: false, mappings: 0, deactivated: 0 })
+
+    await runOctolensDiscovery()
+
+    expect(mocks.reconcileOctolensKeywordMappings).not.toHaveBeenCalled()
+  })
+})

services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts

@@ -0,0 +1,99 @@
+import {
+  assertCredentialKind,
+  getCredential,
+  getManifest,
+  isOctolensCredentialConfigured,
+} from '@crowd/connectors'
+import { resolveOctolensProjectSegmentId } from '@crowd/connectors/src/connectors/octolens/deps'
+import { runMappingDiscovery } from '@crowd/connectors/src/connectors/octolens/mappingDiscovery'
+import type {
+  OctolensKeywordMappingInput,
+  OctolensMappingDiscoveryResult,
+} from '@crowd/connectors/src/connectors/octolens/mappingDiscovery'
+import {
+  reviveDeadLetteredSyncUnits,
+  upsertSyncUnits,
+} from '@crowd/data-access-layer/src/connectors'
+import {
+  findOrCreateOctolensAnchorIntegration,
+  reconcileOctolensKeywordMappings,
+} from '@crowd/data-access-layer/src/octolens'
+import type { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
+import { dbStoreQx } from '@crowd/data-access-layer/src/queryExecutor'
+import { PlatformType } from '@crowd/types'
+
+import { svc } from '../main'
+
+const MENTION_POLL_SYNC_NAME = 'mentionPoll'
+
+async function reconcileOctolensMappings(
+  qx: QueryExecutor,
+  mappings: OctolensKeywordMappingInput[],
+): Promise<number> {
+  const outcome = await reconcileOctolensKeywordMappings(qx, mappings)
+
+  if (outcome.created.length > 0 || outcome.restored.length > 0) {
+    svc.log.info(
+      { created: outcome.created, restored: outcome.restored },
+      'octolens discovery: subproject integrations created or restored',
+    )
+  }
+  if (outcome.softDeleted.length > 0) {
+    svc.log.info(
+      { integrationIds: outcome.softDeleted },
+      'octolens discovery: soft-deleted subproject integrations with no active keywords',
+    )
+  }
+
+  return outcome.deactivated
+}
+
+export async function runOctolensDiscovery(): Promise<OctolensMappingDiscoveryResult> {
+  if (!isOctolensCredentialConfigured()) {
+    svc.log.info('octolens discovery skipped: CROWD_OCTOLENS_API_TOKEN is not set')
+    return { complete: false, mappings: 0, deactivated: 0 }
+  }
+
+  const qx = dbStoreQx(svc.postgres.writer)
+
+  // The anchor owns the single mention poll: Octolens's 500 req/hr budget is shared org-wide.
+  const anchor = await findOrCreateOctolensAnchorIntegration(qx)
+  const credential = await getCredential(qx, anchor.id)
+  assertCredentialKind(credential, 'octolens-token')
+
+  const channels = await getManifest(PlatformType.OCTOLENS).discover(credential)
+  // emitEnabled is insert-only and must be true: false is shadow mode, which writes no activities.
+  const syncUnitsUpserted = await upsertSyncUnits(
+    qx,
+    channels.map((channel) => ({
+      integrationId: anchor.id,
+      platform: PlatformType.OCTOLENS,
+      channelId: channel.channelId,
+      channelName: channel.channelName,
+      syncName: MENTION_POLL_SYNC_NAME,
+      emitEnabled: true,
+    })),
+  )
+  svc.log.info(
+    { anchorIntegrationId: anchor.id, anchorCreated: anchor.created, syncUnitsUpserted },
+    'octolens discovery: ensured the mention poll sync unit (emitEnabled on insert)',
+  )
+
+  const result = await runMappingDiscovery({
+    token: credential.data.token,
+    log: svc.log,
+    resolveProjectSegmentId: (slug) => resolveOctolensProjectSegmentId(qx, slug),
+    reconcileMappings: (mappings) => reconcileOctolensMappings(qx, mappings),
+  })
+
+  // Discovery's own Octolens calls just succeeded with this token, so an auth dead-letter is stale.
+  const revived = await reviveDeadLetteredSyncUnits(qx, anchor.id, MENTION_POLL_SYNC_NAME)
+  if (revived > 0) {
+    svc.log.info(
+      { anchorIntegrationId: anchor.id },
+      'octolens discovery: revived the dead-lettered mention poll sync unit',
+    )
+  }
+
+  return result
+}

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 { scheduleOctolensDiscovery } from './schedules/octolensDiscovery'
 import { scheduleShadowDiff } from './schedules/shadowDiff'
 
 const config: Config = {
@@ -46,6 +47,7 @@ setImmediate(async () => {
 
   await scheduleDispatcher()
   await scheduleShadowDiff()
+  await scheduleOctolensDiscovery()
 
   await svc.start()
 })

services/apps/connectors_worker/src/schedules/octolensDiscovery.ts

@@ -0,0 +1,33 @@
+import { ScheduleAlreadyRunning, ScheduleOverlapPolicy } from '@temporalio/client'
+
+import { svc } from '../main'
+import { octolensDiscovery } from '../workflows/octolensDiscovery'
+
+// SKIP overlap is required: concurrent runs could soft-delete each other's new integrations.
+export async function scheduleOctolensDiscovery(): Promise<void> {
+  try {
+    await svc.temporal.schedule.create({
+      scheduleId: 'connectors-octolens-discovery',
+      spec: {
+        intervals: [{ every: '1 hour' }],
+      },
+      policies: {
+        overlap: ScheduleOverlapPolicy.SKIP,
+        catchupWindow: '1 hour',
+      },
+      action: {
+        type: 'startWorkflow',
+        workflowType: octolensDiscovery,
+        taskQueue: 'connectors',
+        args: [],
+        workflowExecutionTimeout: '30 minutes',
+      },
+    })
+  } catch (err) {
+    if (err instanceof ScheduleAlreadyRunning) {
+      svc.log.info('Octolens discovery schedule already registered in Temporal.')
+    } else {
+      throw new Error(err)
+    }
+  }
+}

services/apps/connectors_worker/src/workflows.ts

@@ -1,5 +1,6 @@
 import { dispatcher } from './workflows/dispatcher'
+import { octolensDiscovery } from './workflows/octolensDiscovery'
 import { shadowDiff } from './workflows/shadowDiff'
 import { syncRun } from './workflows/syncRun'
 
-export { dispatcher, shadowDiff, syncRun }
+export { dispatcher, octolensDiscovery, shadowDiff, syncRun }

services/apps/connectors_worker/src/workflows/octolensDiscovery.ts

@@ -0,0 +1,19 @@
+import { log, proxyActivities } from '@temporalio/workflow'
+
+import type * as activities from '../activities/octolensDiscoveryActivities'
+
+// One retry only: a failed run (Octolens down, rate limited) is simply retried by the next hourly
+// run, and every request here counts against the shared 500/hour Octolens budget.
+const activity = proxyActivities<typeof activities>({
+  startToCloseTimeout: '10 minutes',
+  retry: {
+    maximumAttempts: 2,
+    backoffCoefficient: 2,
+    nonRetryableErrorTypes: ['ProviderAuthError'],
+  },
+})
+
+export async function octolensDiscovery(): Promise<void> {
+  const result = await activity.runOctolensDiscovery()
+  log.info('octolens discovery run complete', { ...result })
+}