diff --git a/services/apps/connectors_worker/src/activities.ts b/services/apps/connectors_worker/src/activities.ts index b0864f3bb8..b44c0274ce 100644 --- a/services/apps/connectors_worker/src/activities.ts +++ b/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, diff --git a/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts new file mode 100644 index 0000000000..3f596c4623 --- /dev/null +++ b/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 = + (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()), + 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 + }) => { + 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() + }) +}) diff --git a/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts new file mode 100644 index 0000000000..eb438d58b5 --- /dev/null +++ b/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 { + 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 { + 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 +} diff --git a/services/apps/connectors_worker/src/main.ts b/services/apps/connectors_worker/src/main.ts index e4ae4f4270..4bb3949a3c 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 { 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() }) diff --git a/services/apps/connectors_worker/src/schedules/octolensDiscovery.ts b/services/apps/connectors_worker/src/schedules/octolensDiscovery.ts new file mode 100644 index 0000000000..ec029f7d81 --- /dev/null +++ b/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 { + 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) + } + } +} diff --git a/services/apps/connectors_worker/src/workflows.ts b/services/apps/connectors_worker/src/workflows.ts index dca7688938..4abfbc1318 100644 --- a/services/apps/connectors_worker/src/workflows.ts +++ b/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 } diff --git a/services/apps/connectors_worker/src/workflows/octolensDiscovery.ts b/services/apps/connectors_worker/src/workflows/octolensDiscovery.ts new file mode 100644 index 0000000000..c88181b78c --- /dev/null +++ b/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({ + startToCloseTimeout: '10 minutes', + retry: { + maximumAttempts: 2, + backoffCoefficient: 2, + nonRetryableErrorTypes: ['ProviderAuthError'], + }, +}) + +export async function octolensDiscovery(): Promise { + const result = await activity.runOctolensDiscovery() + log.info('octolens discovery run complete', { ...result }) +} diff --git a/services/libs/connectors/src/connectors/octolens/deps.test.ts b/services/libs/connectors/src/connectors/octolens/deps.test.ts index afaedc4618..dd902098ad 100644 --- a/services/libs/connectors/src/connectors/octolens/deps.test.ts +++ b/services/libs/connectors/src/connectors/octolens/deps.test.ts @@ -1,20 +1,14 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' import { queryInsightsProjects } from '@crowd/data-access-layer/src/collections' -import { - deactivateOctolensKeywordMappingsNotSeen, - listActiveOctolensKeywordMappings, - upsertOctolensKeywordMapping, -} from '@crowd/data-access-layer/src/octolens' +import { listActiveOctolensKeywordMappings } from '@crowd/data-access-layer/src/octolens' import type { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' import { findSegmentById } from '@crowd/data-access-layer/src/segments' -import { createOctolensDeps } from './deps' +import { createOctolensDeps, resolveOctolensProjectSegmentId } from './deps' vi.mock('@crowd/data-access-layer/src/octolens', () => ({ listActiveOctolensKeywordMappings: vi.fn(), - upsertOctolensKeywordMapping: vi.fn(), - deactivateOctolensKeywordMappingsNotSeen: vi.fn(), })) vi.mock('@crowd/data-access-layer/src/collections', () => ({ InsightsProjectField: { SEGMENT_ID: 'segmentId' }, @@ -88,60 +82,41 @@ describe('createOctolensDeps', () => { expect(mockedListMappings).toHaveBeenCalledTimes(2) }) +}) - describe('resolveProjectSegmentId', () => { - const deps = () => createOctolensDeps({ qx, integrationId: 'integration-1' }) - - it('returns the insights project segment when it is a subproject', async () => { - mockedQueryProjects.mockResolvedValue([{ segmentId: 'segment-sub' }] as never) - mockedFindSegment.mockResolvedValue({ - id: 'segment-sub', - slug: 'ebpf', - parentSlug: 'ebpf-parent', - grandparentSlug: 'lf', - } as Segment) - - expect(await deps().resolveProjectSegmentId('ebpf')).toBe('segment-sub') - expect(mockedQueryProjects).toHaveBeenCalledWith( - qx, - expect.objectContaining({ filter: { slug: { eq: 'ebpf' } } }), - ) - }) - - it('returns null when the segment is not a subproject', async () => { - mockedQueryProjects.mockResolvedValue([{ segmentId: 'segment-group' }] as never) - mockedFindSegment.mockResolvedValue({ - id: 'segment-group', - slug: 'lf', - parentSlug: null, - grandparentSlug: null, - } as Segment) - - expect(await deps().resolveProjectSegmentId('lf')).toBeNull() - }) +describe('resolveOctolensProjectSegmentId', () => { + it('returns the insights project segment when it is a subproject', async () => { + mockedQueryProjects.mockResolvedValue([{ segmentId: 'segment-sub' }] as never) + mockedFindSegment.mockResolvedValue({ + id: 'segment-sub', + slug: 'ebpf', + parentSlug: 'ebpf-parent', + grandparentSlug: 'lf', + } as Segment) + + expect(await resolveOctolensProjectSegmentId(qx, 'ebpf')).toBe('segment-sub') + expect(mockedQueryProjects).toHaveBeenCalledWith( + qx, + expect.objectContaining({ filter: { slug: { eq: 'ebpf' } } }), + ) + }) - it('returns null when no insights project has the slug', async () => { - mockedQueryProjects.mockResolvedValue([]) + it('returns null when the segment is not a subproject', async () => { + mockedQueryProjects.mockResolvedValue([{ segmentId: 'segment-group' }] as never) + mockedFindSegment.mockResolvedValue({ + id: 'segment-group', + slug: 'lf', + parentSlug: null, + grandparentSlug: null, + } as Segment) - expect(await deps().resolveProjectSegmentId('missing')).toBeNull() - expect(mockedFindSegment).not.toHaveBeenCalled() - }) + expect(await resolveOctolensProjectSegmentId(qx, 'lf')).toBeNull() }) - it('writes and deactivates mappings for its own integration', async () => { - vi.mocked(deactivateOctolensKeywordMappingsNotSeen).mockResolvedValue(2) - const deps = createOctolensDeps({ qx, integrationId: 'integration-1' }) - - await deps.upsertMapping({ keywordId: 1, keyword: 'ebpf', segmentId: 'segment-a' }) - const deactivated = await deps.deactivateMappingsNotSeen([1]) + it('returns null when no insights project has the slug', async () => { + mockedQueryProjects.mockResolvedValue([]) - expect(upsertOctolensKeywordMapping).toHaveBeenCalledWith(qx, { - integrationId: 'integration-1', - keywordId: 1, - keyword: 'ebpf', - segmentId: 'segment-a', - }) - expect(deactivateOctolensKeywordMappingsNotSeen).toHaveBeenCalledWith(qx, 'integration-1', [1]) - expect(deactivated).toBe(2) + expect(await resolveOctolensProjectSegmentId(qx, 'missing')).toBeNull() + expect(mockedFindSegment).not.toHaveBeenCalled() }) }) diff --git a/services/libs/connectors/src/connectors/octolens/deps.ts b/services/libs/connectors/src/connectors/octolens/deps.ts index 6c98a19d19..85021a44ee 100644 --- a/services/libs/connectors/src/connectors/octolens/deps.ts +++ b/services/libs/connectors/src/connectors/octolens/deps.ts @@ -2,21 +2,12 @@ import { InsightsProjectField, queryInsightsProjects, } from '@crowd/data-access-layer/src/collections' -import { - deactivateOctolensKeywordMappingsNotSeen, - listActiveOctolensKeywordMappings, - upsertOctolensKeywordMapping, -} from '@crowd/data-access-layer/src/octolens' +import { listActiveOctolensKeywordMappings } from '@crowd/data-access-layer/src/octolens' +import type { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' import { findSegmentById, isSegmentSubproject } from '@crowd/data-access-layer/src/segments' import type { DepsFactoryInput } from '../../types' -export interface OctolensKeywordMappingInput { - keywordId: number - keyword: string - segmentId: string -} - export interface OctolensKeywordTarget { segmentId: string integrationId: string @@ -25,13 +16,25 @@ export interface OctolensKeywordTarget { export interface OctolensDeps { // Where a keyword's mentions land: its subproject's Octolens integration, never the poll's own. resolveKeywordTarget(keywordId: number): Promise - // Activities only land on subproject segments, so a slug resolving to anything else is unusable. - resolveProjectSegmentId(slug: string): Promise - upsertMapping(mapping: OctolensKeywordMappingInput): Promise - deactivateMappingsNotSeen(seenKeywordIds: number[]): Promise } -export function createOctolensDeps({ qx, integrationId }: DepsFactoryInput): OctolensDeps { +// Activities only land on subproject segments, so a slug resolving to anything else is unusable. +export async function resolveOctolensProjectSegmentId( + qx: QueryExecutor, + slug: string, +): Promise { + const [project] = await queryInsightsProjects(qx, { + fields: [InsightsProjectField.SEGMENT_ID], + filter: { slug: { eq: slug } }, + limit: 1, + }) + if (!project?.segmentId) return null + + const segment = await findSegmentById(qx, project.segmentId) + return segment && isSegmentSubproject(segment) ? segment.id : null +} + +export function createOctolensDeps({ qx }: DepsFactoryInput): OctolensDeps { let keywordTargets: Map | null = null return { @@ -52,25 +55,5 @@ export function createOctolensDeps({ qx, integrationId }: DepsFactoryInput): Oct } return keywordTargets.get(keywordId) ?? null }, - - async resolveProjectSegmentId(slug) { - const [project] = await queryInsightsProjects(qx, { - fields: [InsightsProjectField.SEGMENT_ID], - filter: { slug: { eq: slug } }, - limit: 1, - }) - if (!project?.segmentId) return null - - const segment = await findSegmentById(qx, project.segmentId) - return segment && isSegmentSubproject(segment) ? segment.id : null - }, - - async upsertMapping(mapping) { - await upsertOctolensKeywordMapping(qx, { integrationId, ...mapping }) - }, - - async deactivateMappingsNotSeen(seenKeywordIds) { - return deactivateOctolensKeywordMappingsNotSeen(qx, integrationId, seenKeywordIds) - }, } } diff --git a/services/libs/connectors/src/connectors/octolens/index.ts b/services/libs/connectors/src/connectors/octolens/index.ts index 4fd87c3595..152c5cb850 100644 --- a/services/libs/connectors/src/connectors/octolens/index.ts +++ b/services/libs/connectors/src/connectors/octolens/index.ts @@ -3,7 +3,6 @@ import { createOctolensDeps } from './deps' import type { OctolensDeps } from './deps' import { discoverOctolens } from './discover' import { octolensActivitySchema } from './mapper' -import { runMappingDiscovery } from './syncs/mappingDiscovery' import { runMentionPoll } from './syncs/mentionPoll' export const octolensConnector: Manifest = { @@ -17,14 +16,6 @@ export const octolensConnector: Manifest = { schema: octolensActivitySchema, run: runMentionPoll, }, - { - // New projects and webhooks appear rarely; 3 requests an hour leaves the shared budget alone. - // It emits no activities, so the schema is only there to satisfy SyncDefinition. - name: 'mappingDiscovery', - cadenceMinutes: 60, - schema: octolensActivitySchema, - run: runMappingDiscovery, - }, ], discover: discoverOctolens, createDeps: createOctolensDeps, diff --git a/services/libs/connectors/src/connectors/octolens/syncs/mappingDiscovery.test.ts b/services/libs/connectors/src/connectors/octolens/mappingDiscovery.test.ts similarity index 90% rename from services/libs/connectors/src/connectors/octolens/syncs/mappingDiscovery.test.ts rename to services/libs/connectors/src/connectors/octolens/mappingDiscovery.test.ts index 3b40129731..c241ee10a8 100644 --- a/services/libs/connectors/src/connectors/octolens/syncs/mappingDiscovery.test.ts +++ b/services/libs/connectors/src/connectors/octolens/mappingDiscovery.test.ts @@ -2,14 +2,13 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' import type { Logger } from '@crowd/logging' -import { RateLimitError } from '../../../http/errors' -import type { Credential, SyncContext } from '../../../types' -import { fetchFeeds, fetchKeywords, fetchNotifications } from '../configClient' -import type { OctolensFeed, OctolensKeyword, OctolensNotification } from '../configClient' -import type { OctolensDeps, OctolensKeywordMappingInput } from '../deps' +import { RateLimitError } from '../../http/errors' +import { fetchFeeds, fetchKeywords, fetchNotifications } from './configClient' +import type { OctolensFeed, OctolensKeyword, OctolensNotification } from './configClient' import { runMappingDiscovery } from './mappingDiscovery' +import type { OctolensKeywordMappingInput, OctolensMappingDiscoveryInput } from './mappingDiscovery' -vi.mock('../configClient', () => ({ +vi.mock('./configClient', () => ({ fetchNotifications: vi.fn(), fetchFeeds: vi.fn(), fetchKeywords: vi.fn(), @@ -19,12 +18,6 @@ const mockedNotifications = vi.mocked(fetchNotifications) const mockedFeeds = vi.mocked(fetchFeeds) const mockedKeywords = vi.mocked(fetchKeywords) -const OCTOLENS_CREDENTIAL: Credential = { - platform: 'octolens', - kind: 'octolens-token', - data: { token: 'test-token' }, -} - const SEGMENTS: Record = { ebpf: 'segment-ebpf', valkey: 'segment-valkey' } function webhook(slug: string) { @@ -64,24 +57,17 @@ function makeCtx() { const upserts: OctolensKeywordMappingInput[] = [] const deactivateCalls: number[][] = [] const log = { info: vi.fn(), warn: vi.fn(), error: vi.fn() } - const deps: OctolensDeps = { - resolveKeywordTarget: async () => null, + const ctx: OctolensMappingDiscoveryInput = { + token: 'test-token', + log: log as unknown as Logger, resolveProjectSegmentId: vi.fn(async (slug: string) => SEGMENTS[slug] ?? null), - upsertMapping: async (mapping) => { - upserts.push(mapping) - }, - deactivateMappingsNotSeen: async (seen) => { - deactivateCalls.push(seen) + reconcileMappings: async (mappings) => { + upserts.push(...mappings) + deactivateCalls.push(mappings.map((m) => m.keywordId)) return 0 }, } - const ctx = { - integrationId: 'integration-1', - log: log as unknown as Logger, - getCredential: async () => OCTOLENS_CREDENTIAL, - deps, - } as unknown as SyncContext - return { ctx, deps, upserts, deactivateCalls, log } + return { ctx, deps: ctx, upserts, deactivateCalls, log } } function givenOctolens( @@ -111,7 +97,7 @@ describe('runMappingDiscovery', () => { const outcome = await runMappingDiscovery(ctx) - expect(outcome).toEqual({ complete: true }) + expect(outcome).toEqual({ complete: true, mappings: 3, deactivated: 0 }) expect(upserts).toEqual([ { keywordId: 1, keyword: 'keyword-1', segmentId: 'segment-ebpf' }, { keywordId: 2, keyword: 'keyword-2', segmentId: 'segment-ebpf' }, diff --git a/services/libs/connectors/src/connectors/octolens/syncs/mappingDiscovery.ts b/services/libs/connectors/src/connectors/octolens/mappingDiscovery.ts similarity index 81% rename from services/libs/connectors/src/connectors/octolens/syncs/mappingDiscovery.ts rename to services/libs/connectors/src/connectors/octolens/mappingDiscovery.ts index 3d9821c07c..be6c861756 100644 --- a/services/libs/connectors/src/connectors/octolens/syncs/mappingDiscovery.ts +++ b/services/libs/connectors/src/connectors/octolens/mappingDiscovery.ts @@ -1,13 +1,31 @@ -import { assertCredentialKind } from '../../../types' -import type { SyncContext, SyncOutcome } from '../../../types' -import { fetchFeeds, fetchKeywords, fetchNotifications } from '../configClient' -import type { OctolensDeps, OctolensKeywordMappingInput } from '../deps' -import { feedKeywordIds, insightsSlugFromWebhookUrl } from '../mappingRules' +import type { Logger } from '@crowd/logging' -export async function runMappingDiscovery(ctx: SyncContext): Promise { - const credential = await ctx.getCredential() - assertCredentialKind(credential, 'octolens-token') - const { token } = credential.data +import { fetchFeeds, fetchKeywords, fetchNotifications } from './configClient' +import { feedKeywordIds, insightsSlugFromWebhookUrl } from './mappingRules' + +export interface OctolensKeywordMappingInput { + keywordId: number + keyword: string + segmentId: string +} + +export interface OctolensMappingDiscoveryInput { + token: string + log: Logger + resolveProjectSegmentId(slug: string): Promise + reconcileMappings(mappings: OctolensKeywordMappingInput[]): Promise +} + +export interface OctolensMappingDiscoveryResult { + complete: boolean + mappings: number + deactivated: number +} + +export async function runMappingDiscovery( + ctx: OctolensMappingDiscoveryInput, +): Promise { + const { token } = ctx const notifications = await fetchNotifications(token) const feeds = new Map((await fetchFeeds(token)).map((feed) => [feed.id, feed])) @@ -76,7 +94,7 @@ export async function runMappingDiscovery(ctx: SyncContext): Promi const [slug] = slugs if (!segmentBySlug.has(slug)) { - segmentBySlug.set(slug, await ctx.deps.resolveProjectSegmentId(slug)) + segmentBySlug.set(slug, await ctx.resolveProjectSegmentId(slug)) } const segmentId = segmentBySlug.get(slug) if (!segmentId) { @@ -116,10 +134,7 @@ export async function runMappingDiscovery(ctx: SyncContext): Promi } else if (mappings.length === 0) { ctx.log.warn('octolens mapping discovery: no mappings found, skipping deactivation') } else { - for (const mapping of mappings) { - await ctx.deps.upsertMapping(mapping) - } - deactivated = await ctx.deps.deactivateMappingsNotSeen(mappings.map((m) => m.keywordId)) + deactivated = await ctx.reconcileMappings(mappings) } ctx.log.info( @@ -134,5 +149,5 @@ export async function runMappingDiscovery(ctx: SyncContext): Promi }, 'octolens mapping discovery complete', ) - return { complete: true } + return { complete, mappings: mappings.length, deactivated } } diff --git a/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts b/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts index 6ad6637cfe..f0e092a7d9 100644 --- a/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts +++ b/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts @@ -93,9 +93,6 @@ function makeCtx( deps: { resolveKeywordTarget: resolveKeywordTarget ?? (async (keywordId) => target(`segment-${keywordId}`)), - resolveProjectSegmentId: async () => null, - upsertMapping: async () => undefined, - deactivateMappingsNotSeen: async () => 0, }, ...ctxOverrides, } diff --git a/services/libs/connectors/src/credentials.ts b/services/libs/connectors/src/credentials.ts index a37f527304..88641c2888 100644 --- a/services/libs/connectors/src/credentials.ts +++ b/services/libs/connectors/src/credentials.ts @@ -46,6 +46,11 @@ function githubAppCredential(): Credential { } } +// Lets scheduled Octolens work skip environments that never got a token, instead of failing hourly. +export function isOctolensCredentialConfigured(): boolean { + return Boolean(process.env.CROWD_OCTOLENS_API_TOKEN) +} + // Octolens is one shared org-wide bearer token, not per-integration like GitHub's app // installations — every octolens integration resolves to the same env-provided token. function octolensTokenCredential(): Credential { 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 3b9b9acd9e..655797dca7 100644 --- a/services/libs/data-access-layer/src/connectors/syncUnits.test.ts +++ b/services/libs/data-access-layer/src/connectors/syncUnits.test.ts @@ -4,7 +4,7 @@ import { DEFAULT_TENANT_ID, generateUUIDv1 } from '@crowd/common' import type { QueryExecutor } from '@crowd/database' import { withQx } from '@crowd/test-kit/db' -import { listShadowDiffUnits, upsertSyncUnits } from './syncUnits' +import { listShadowDiffUnits, reviveDeadLetteredSyncUnits, upsertSyncUnits } from './syncUnits' const test = withQx(base) @@ -234,3 +234,60 @@ describe('upsertSyncUnits', () => { }) }) }) + +describe('reviveDeadLetteredSyncUnits', () => { + async function readUnit(qx: QueryExecutor, id: string) { + return qx.selectOne( + `SELECT status, "consecutiveFailures", "lastErrorClass", watermark FROM integration.sync_units WHERE id = $(id)`, + { id }, + ) + } + + test('reactivates only the dead-lettered unit of that integration and sync, keeping its watermark', async ({ + qx, + }) => { + const integrationId = await createIntegration(qx) + const otherIntegrationId = await createIntegration(qx) + const dead = await createSyncUnit(qx, integrationId, { + channelName: 'octolens-global', + syncName: 'mentionPoll', + status: 'dead_letter', + watermark: { startDate: '2026-10-01T00:00:00.000Z' }, + }) + const otherSync = await createSyncUnit(qx, integrationId, { + channelName: 'octolens-global', + syncName: 'otherSync', + status: 'dead_letter', + }) + const otherIntegration = await createSyncUnit(qx, otherIntegrationId, { + channelName: 'octolens-global', + syncName: 'mentionPoll', + status: 'dead_letter', + }) + await qx.result( + `UPDATE integration.sync_units SET "consecutiveFailures" = 5, "lastErrorClass" = 'provider.auth' WHERE id = $(dead)`, + { dead }, + ) + + expect(await reviveDeadLetteredSyncUnits(qx, integrationId, 'mentionPoll')).toBe(1) + + expect(await readUnit(qx, dead)).toEqual({ + status: 'active', + consecutiveFailures: 0, + lastErrorClass: null, + watermark: { startDate: '2026-10-01T00:00:00.000Z' }, + }) + expect((await readUnit(qx, otherSync)).status).toBe('dead_letter') + expect((await readUnit(qx, otherIntegration)).status).toBe('dead_letter') + }) + + test('leaves an active unit untouched', async ({ qx }) => { + const integrationId = await createIntegration(qx) + await createSyncUnit(qx, integrationId, { + channelName: 'octolens-global', + syncName: 'mentionPoll', + }) + + expect(await reviveDeadLetteredSyncUnits(qx, integrationId, 'mentionPoll')).toBe(0) + }) +}) diff --git a/services/libs/data-access-layer/src/connectors/syncUnits.ts b/services/libs/data-access-layer/src/connectors/syncUnits.ts index 1760d16548..3cf5485b6b 100644 --- a/services/libs/data-access-layer/src/connectors/syncUnits.ts +++ b/services/libs/data-access-layer/src/connectors/syncUnits.ts @@ -210,6 +210,29 @@ export async function recordRunFailure( ) } +// Only provider.auth failures dead-letter a unit, so callers revive it once that +// credential works again. +export async function reviveDeadLetteredSyncUnits( + qx: QueryExecutor, + integrationId: string, + syncName: string, +): Promise { + return qx.result( + `UPDATE integration.sync_units + SET status = 'active', + "consecutiveFailures" = 0, + "lastErrorClass" = NULL, + "lastErrorMessage" = NULL, + "nextRunAt" = now(), + "lockedAt" = NULL, + "updatedAt" = now() + WHERE "integrationId" = $(integrationId) + AND "syncName" = $(syncName) + AND status = 'dead_letter'`, + { integrationId, syncName }, + ) +} + export async function getUnitById(qx: QueryExecutor, id: string): Promise { return qx.selectOneOrNone( `SELECT * diff --git a/services/libs/data-access-layer/src/octolens/index.test.ts b/services/libs/data-access-layer/src/octolens/index.test.ts index c3eb95fb6c..390809e40c 100644 --- a/services/libs/data-access-layer/src/octolens/index.test.ts +++ b/services/libs/data-access-layer/src/octolens/index.test.ts @@ -5,13 +5,11 @@ import type { QueryExecutor } from '@crowd/database' import { withQx } from '@crowd/test-kit/db' import { - deactivateOctolensKeywordMappingsNotSeen, deactivateOctolensKeywordMappingsNotSeenAcross, - findOctolensKeywordMapping, findOrCreateOctolensAnchorIntegration, findOrCreateOctolensSubprojectIntegration, listActiveOctolensKeywordMappings, - listOctolensKeywordMappings, + reconcileOctolensKeywordMappings, softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings, upsertOctolensKeywordMapping, } from './index' @@ -64,6 +62,43 @@ async function readRowIncludingDeactivated( ) } +async function findActiveMapping( + qx: QueryExecutor, + integrationId: string, + keywordId: number, +): Promise<{ + id: string + integrationId: string + segmentId: string + keywordId: number + keyword: string +} | null> { + return qx.selectOneOrNone( + ` + SELECT id, "integrationId", "segmentId", "keywordId", keyword + FROM integration."octolensKeywordMappings" + WHERE "integrationId" = $(integrationId) AND "keywordId" = $(keywordId) + AND "deactivatedAt" IS NULL + `, + { integrationId, keywordId }, + ) +} + +async function deactivateMapping( + qx: QueryExecutor, + integrationId: string, + keywordId: number, +): Promise { + await qx.result( + ` + UPDATE integration."octolensKeywordMappings" + SET "deactivatedAt" = now() + WHERE "integrationId" = $(integrationId) AND "keywordId" = $(keywordId) + `, + { integrationId, keywordId }, + ) +} + describe('octolens keyword mappings', () => { test('upsertOctolensKeywordMapping inserts a new mapping', async ({ qx }) => { const integrationId = await createIntegration(qx) @@ -76,7 +111,7 @@ describe('octolens keyword mappings', () => { keyword: 'crowd.dev', }) - const mapping = await findOctolensKeywordMapping(qx, integrationId, 101) + const mapping = await findActiveMapping(qx, integrationId, 101) expect(mapping).toMatchObject({ integrationId, @@ -99,7 +134,7 @@ describe('octolens keyword mappings', () => { keywordId: 5, keyword: 'old', }) - const before = await findOctolensKeywordMapping(qx, integrationId, 5) + const before = await findActiveMapping(qx, integrationId, 5) await qx.result( ` UPDATE integration."octolensKeywordMappings" SET "updatedAt" = '2020-01-01T00:00:00Z' @@ -115,7 +150,7 @@ describe('octolens keyword mappings', () => { keyword: 'new', }) - const after = await findOctolensKeywordMapping(qx, integrationId, 5) + const after = await findActiveMapping(qx, integrationId, 5) expect(after).toMatchObject({ id: before?.id, segmentId: otherSegmentId, keyword: 'new' }) const row = await readRowIncludingDeactivated(qx, integrationId, 5) expect(new Date(row.updatedAt).getTime()).toBeGreaterThan(Date.parse('2020-01-02')) @@ -152,121 +187,16 @@ describe('octolens keyword mappings', () => { await upsertOctolensKeywordMapping(qx, data) await upsertOctolensKeywordMapping(qx, { ...data, keywordId: 8, keyword: 'other' }) - await deactivateOctolensKeywordMappingsNotSeen(qx, integrationId, [8]) - expect(await findOctolensKeywordMapping(qx, integrationId, 7)).toBeNull() + await deactivateMapping(qx, integrationId, 7) + expect(await findActiveMapping(qx, integrationId, 7)).toBeNull() await upsertOctolensKeywordMapping(qx, data) - expect(await findOctolensKeywordMapping(qx, integrationId, 7)).toMatchObject({ + expect(await findActiveMapping(qx, integrationId, 7)).toMatchObject({ keyword: 'back', }) expect((await readRowIncludingDeactivated(qx, integrationId, 7)).deactivatedAt).toBeNull() }) - - test('findOctolensKeywordMapping returns null when no mapping exists', async ({ qx }) => { - const integrationId = await createIntegration(qx) - - const mapping = await findOctolensKeywordMapping(qx, integrationId, 999) - - expect(mapping).toBeNull() - }) - - test('listOctolensKeywordMappings returns only active mappings of the given integration', async ({ - qx, - }) => { - const integrationId = await createIntegration(qx) - const otherIntegrationId = await createIntegration(qx) - const segmentId = await createSegment(qx) - - for (const [keywordId, keyword] of [ - [1, 'alpha'], - [2, 'beta'], - [4, 'gone'], - ] as const) { - await upsertOctolensKeywordMapping(qx, { integrationId, segmentId, keywordId, keyword }) - } - await upsertOctolensKeywordMapping(qx, { - integrationId: otherIntegrationId, - segmentId, - keywordId: 3, - keyword: 'gamma', - }) - await deactivateOctolensKeywordMappingsNotSeen(qx, integrationId, [1, 2]) - - const mappings = await listOctolensKeywordMappings(qx, integrationId) - - expect(mappings.map((m) => m.keyword)).toEqual(['alpha', 'beta']) - }) - - test('deactivateOctolensKeywordMappingsNotSeen only deactivates unseen mappings of the integration', async ({ - qx, - }) => { - const integrationId = await createIntegration(qx) - const otherIntegrationId = await createIntegration(qx) - const segmentId = await createSegment(qx) - - await upsertOctolensKeywordMapping(qx, { - integrationId, - segmentId, - keywordId: 1, - keyword: 'seen', - }) - await upsertOctolensKeywordMapping(qx, { - integrationId, - segmentId, - keywordId: 2, - keyword: 'unseen', - }) - await upsertOctolensKeywordMapping(qx, { - integrationId: otherIntegrationId, - segmentId, - keywordId: 2, - keyword: 'other integration', - }) - - const deactivated = await deactivateOctolensKeywordMappingsNotSeen(qx, integrationId, [1]) - - expect(deactivated).toBe(1) - expect(await findOctolensKeywordMapping(qx, integrationId, 1)).not.toBeNull() - expect(await findOctolensKeywordMapping(qx, integrationId, 2)).toBeNull() - expect(await findOctolensKeywordMapping(qx, otherIntegrationId, 2)).not.toBeNull() - }) - - test('deactivateOctolensKeywordMappingsNotSeen keeps the first deactivation timestamp', async ({ - qx, - }) => { - const integrationId = await createIntegration(qx) - const segmentId = await createSegment(qx) - - await upsertOctolensKeywordMapping(qx, { integrationId, segmentId, keywordId: 1, keyword: 'a' }) - await upsertOctolensKeywordMapping(qx, { integrationId, segmentId, keywordId: 2, keyword: 'b' }) - await deactivateOctolensKeywordMappingsNotSeen(qx, integrationId, [1]) - const first = (await readRowIncludingDeactivated(qx, integrationId, 2)).deactivatedAt - - const deactivated = await deactivateOctolensKeywordMappingsNotSeen(qx, integrationId, [1]) - - expect(deactivated).toBe(0) - expect((await readRowIncludingDeactivated(qx, integrationId, 2)).deactivatedAt).toEqual(first) - }) - - test('deactivateOctolensKeywordMappingsNotSeen with no seen keywords deactivates nothing', async ({ - qx, - }) => { - const integrationId = await createIntegration(qx) - const segmentId = await createSegment(qx) - - await upsertOctolensKeywordMapping(qx, { - integrationId, - segmentId, - keywordId: 1, - keyword: 'keep', - }) - - const deactivated = await deactivateOctolensKeywordMappingsNotSeen(qx, integrationId, []) - - expect(deactivated).toBe(0) - expect(await findOctolensKeywordMapping(qx, integrationId, 1)).not.toBeNull() - }) }) async function createPlatformIntegration( @@ -463,7 +393,7 @@ describe('octolens subproject integrations', () => { await upsert(deletedIntegration, segmentA, 4) await upsert(githubIntegration, segmentA, 5) await upsert(anchor, segmentA, 6) - await deactivateOctolensKeywordMappingsNotSeen(qx, integrationB, [1]) + await deactivateMapping(qx, integrationB, 3) const ours = new Set([ integrationA, @@ -520,11 +450,11 @@ describe('octolens subproject integrations', () => { ]) expect(deactivated).toBe(2) - expect(await findOctolensKeywordMapping(qx, integrationA, 1)).not.toBeNull() - expect(await findOctolensKeywordMapping(qx, integrationA, 2)).toBeNull() - expect(await findOctolensKeywordMapping(qx, integrationB, 1)).toBeNull() - expect(await findOctolensKeywordMapping(qx, githubIntegration, 9)).not.toBeNull() - expect(await findOctolensKeywordMapping(qx, anchor, 7)).not.toBeNull() + expect(await findActiveMapping(qx, integrationA, 1)).not.toBeNull() + expect(await findActiveMapping(qx, integrationA, 2)).toBeNull() + expect(await findActiveMapping(qx, integrationB, 1)).toBeNull() + expect(await findActiveMapping(qx, githubIntegration, 9)).not.toBeNull() + expect(await findActiveMapping(qx, anchor, 7)).not.toBeNull() }) test('deactivateOctolensKeywordMappingsNotSeenAcross with nothing seen deactivates nothing', async ({ @@ -535,7 +465,7 @@ describe('octolens subproject integrations', () => { await upsertOctolensKeywordMapping(qx, { integrationId, segmentId, keywordId: 1, keyword: 'k' }) expect(await deactivateOctolensKeywordMappingsNotSeenAcross(qx, [])).toBe(0) - expect(await findOctolensKeywordMapping(qx, integrationId, 1)).not.toBeNull() + expect(await findActiveMapping(qx, integrationId, 1)).not.toBeNull() }) test('softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings deletes only empty subproject integrations', async ({ @@ -626,7 +556,7 @@ describe('octolens subproject integrations', () => { keyword: `kw-${keywordId}`, }) } - await deactivateOctolensKeywordMappingsNotSeen(qx, owner, [2]) + await deactivateMapping(qx, owner, 1) const other = await createPlatformIntegration(qx, { segmentId: await createSegment(qx) }) const listed = await listActiveOctolensKeywordMappings(qx) @@ -635,12 +565,112 @@ describe('octolens subproject integrations', () => { ]) expect(listed.map((m) => m.integrationId)).not.toContain(owner) - expect(await findOctolensKeywordMapping(qx, owner, 2)).not.toBeNull() + expect(await findActiveMapping(qx, owner, 2)).not.toBeNull() - await deactivateOctolensKeywordMappingsNotSeen(qx, owner, [99]) + await deactivateMapping(qx, owner, 2) const deleted = await softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings(qx) expect(deleted).not.toContain(owner) expect((await readIntegration(qx, owner)).deletedAt).toBeNull() }) }) + +describe('reconcileOctolensKeywordMappings', () => { + async function liveIntegrationFor(qx: QueryExecutor, segmentId: string): Promise { + const row: { id: string } | null = await qx.selectOneOrNone( + `SELECT id FROM public.integrations + WHERE platform = 'octolens' AND "segmentId" = $(segmentId) AND "deletedAt" IS NULL`, + { segmentId }, + ) + return row?.id ?? null + } + + test('creates one integration per subproject and maps each keyword under it', async ({ qx }) => { + const ebpf = await createSegment(qx) + const valkey = await createSegment(qx) + + const outcome = await reconcileOctolensKeywordMappings(qx, [ + { keywordId: 1, keyword: 'ebpf', segmentId: ebpf }, + { keywordId: 2, keyword: 'cilium', segmentId: ebpf }, + { keywordId: 3, keyword: 'valkey', segmentId: valkey }, + ]) + + const ebpfIntegration = await liveIntegrationFor(qx, ebpf) + const valkeyIntegration = await liveIntegrationFor(qx, valkey) + expect(outcome.created.sort()).toEqual([ebpfIntegration, valkeyIntegration].sort()) + expect(outcome).toMatchObject({ deactivated: 0, restored: [], softDeleted: [] }) + expect(await findActiveMapping(qx, ebpfIntegration as string, 2)).toMatchObject({ + segmentId: ebpf, + keyword: 'cilium', + }) + expect(await findActiveMapping(qx, valkeyIntegration as string, 3)).not.toBeNull() + }) + + test('moves a keyword to its new subproject and soft-deletes the emptied integration', async ({ + qx, + }) => { + const from = await createSegment(qx) + const to = await createSegment(qx) + await reconcileOctolensKeywordMappings(qx, [{ keywordId: 1, keyword: 'ebpf', segmentId: from }]) + const fromIntegration = (await liveIntegrationFor(qx, from)) as string + + const outcome = await reconcileOctolensKeywordMappings(qx, [ + { keywordId: 1, keyword: 'ebpf', segmentId: to }, + ]) + + const toIntegration = (await liveIntegrationFor(qx, to)) as string + expect(outcome).toEqual({ + deactivated: 1, + created: [toIntegration], + restored: [], + softDeleted: [fromIntegration], + }) + expect(await findActiveMapping(qx, fromIntegration, 1)).toBeNull() + expect(await findActiveMapping(qx, toIntegration, 1)).not.toBeNull() + expect(await listActiveOctolensKeywordMappings(qx)).toContainEqual( + expect.objectContaining({ keywordId: 1, integrationId: toIntegration, segmentId: to }), + ) + }) + + test('restores the same integration when a soft-deleted subproject gets its keywords back', async ({ + qx, + }) => { + const segment = await createSegment(qx) + const other = await createSegment(qx) + await reconcileOctolensKeywordMappings(qx, [ + { keywordId: 1, keyword: 'ebpf', segmentId: segment }, + ]) + const original = (await liveIntegrationFor(qx, segment)) as string + await reconcileOctolensKeywordMappings(qx, [ + { keywordId: 2, keyword: 'other', segmentId: other }, + ]) + expect(await liveIntegrationFor(qx, segment)).toBeNull() + + const outcome = await reconcileOctolensKeywordMappings(qx, [ + { keywordId: 1, keyword: 'ebpf', segmentId: segment }, + { keywordId: 2, keyword: 'other', segmentId: other }, + ]) + + expect(outcome.restored).toEqual([original]) + expect(await liveIntegrationFor(qx, segment)).toBe(original) + expect(await findActiveMapping(qx, original, 1)).not.toBeNull() + }) + + test('rolls back every write when a mapping fails partway through', async ({ qx }) => { + const kept = await createSegment(qx) + await reconcileOctolensKeywordMappings(qx, [{ keywordId: 1, keyword: 'keep', segmentId: kept }]) + const keptIntegration = (await liveIntegrationFor(qx, kept)) as string + const added = await createSegment(qx) + + await expect( + reconcileOctolensKeywordMappings(qx, [ + { keywordId: 2, keyword: 'new', segmentId: added }, + { keywordId: 3, keyword: null as unknown as string, segmentId: added }, + ]), + ).rejects.toThrow() + + expect(await liveIntegrationFor(qx, added)).toBeNull() + expect(await findActiveMapping(qx, keptIntegration, 1)).not.toBeNull() + expect(await liveIntegrationFor(qx, kept)).toBe(keptIntegration) + }) +}) diff --git a/services/libs/data-access-layer/src/octolens/index.ts b/services/libs/data-access-layer/src/octolens/index.ts index 1032da3d6c..496d1a9284 100644 --- a/services/libs/data-access-layer/src/octolens/index.ts +++ b/services/libs/data-access-layer/src/octolens/index.ts @@ -3,46 +3,6 @@ import { IntegrationState, PlatformType } from '@crowd/types' import { QueryExecutor } from '../queryExecutor' -export interface IOctolensKeywordMapping { - id: string - integrationId: string - segmentId: string - keywordId: number - keyword: string -} - -export async function findOctolensKeywordMapping( - qx: QueryExecutor, - integrationId: string, - keywordId: number, -): Promise { - return qx.selectOneOrNone( - ` - SELECT id, "integrationId", "segmentId", "keywordId", keyword - FROM integration."octolensKeywordMappings" - WHERE "integrationId" = $(integrationId) - AND "keywordId" = $(keywordId) - AND "deactivatedAt" IS NULL - `, - { integrationId, keywordId }, - ) -} - -export async function listOctolensKeywordMappings( - qx: QueryExecutor, - integrationId: string, -): Promise { - return qx.select( - ` - SELECT id, "integrationId", "segmentId", "keywordId", keyword - FROM integration."octolensKeywordMappings" - WHERE "integrationId" = $(integrationId) AND "deactivatedAt" IS NULL - ORDER BY keyword - `, - { integrationId }, - ) -} - export interface IUpsertOctolensKeywordMapping { integrationId: string segmentId: string @@ -77,28 +37,6 @@ export async function upsertOctolensKeywordMapping( ) } -// An empty set deactivates nothing: a run that saw no mappings must never wipe them all. -export async function deactivateOctolensKeywordMappingsNotSeen( - qx: QueryExecutor, - integrationId: string, - seenKeywordIds: number[], -): Promise { - if (seenKeywordIds.length === 0) { - return 0 - } - - return qx.result( - ` - UPDATE integration."octolensKeywordMappings" - SET "deactivatedAt" = now(), "updatedAt" = now() - WHERE "integrationId" = $(integrationId) - AND "deactivatedAt" IS NULL - AND "keywordId" <> ALL($(seenKeywordIds)::int[]) - `, - { integrationId, seenKeywordIds }, - ) -} - export interface IFoundOctolensIntegration { id: string created: boolean @@ -291,3 +229,47 @@ export async function softDeleteOctolensSubprojectIntegrationsWithoutActiveMappi return rows.map((r) => r.id) }) } + +export interface IOctolensKeywordMappingInput { + keywordId: number + keyword: string + segmentId: string +} + +export interface IOctolensReconcileOutcome { + deactivated: number + created: string[] + restored: string[] + softDeleted: string[] +} + +// One transaction, so a poll never sees a moving keyword active under two subprojects, and the +// soft-delete (last) never drops an integration this run just mapped. +export async function reconcileOctolensKeywordMappings( + qx: QueryExecutor, + mappings: IOctolensKeywordMappingInput[], +): Promise { + return qx.tx(async (tx) => { + const integrationBySegment = new Map() + const seen: IOctolensKeywordMappingRef[] = [] + const created: string[] = [] + const restored: string[] = [] + for (const mapping of mappings) { + let integrationId = integrationBySegment.get(mapping.segmentId) + if (!integrationId) { + const integration = await findOrCreateOctolensSubprojectIntegration(tx, mapping.segmentId) + if (integration.created) created.push(integration.id) + if (integration.restored) restored.push(integration.id) + integrationId = integration.id + integrationBySegment.set(mapping.segmentId, integrationId) + } + await upsertOctolensKeywordMapping(tx, { integrationId, ...mapping }) + seen.push({ integrationId, keywordId: mapping.keywordId }) + } + + const deactivated = await deactivateOctolensKeywordMappingsNotSeenAcross(tx, seen) + const softDeleted = await softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings(tx) + + return { deactivated, created, restored, softDeleted } + }) +}