From 4c0377e5420960c14d085531b2deda5aa69532b5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Thu, 15 Oct 2026 22:01:02 +0200 Subject: [PATCH 1/9] feat: emit octolens mentions under the keyword's subproject integration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The mention poll now resolves each keyword's active mapping across every Octolens subproject integration and emits under that integration and its segment, instead of under the poll's own integration. A keyword mapped under more than one integration is treated as unmapped. Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- .../src/connectors/octolens/deps.test.ts | 56 +++++++++++++------ .../src/connectors/octolens/deps.ts | 32 ++++++++--- .../octolens/syncs/mappingDiscovery.test.ts | 2 +- .../octolens/syncs/mentionPoll.test.ts | 55 ++++++++++-------- .../connectors/octolens/syncs/mentionPoll.ts | 28 ++++++---- 5 files changed, 112 insertions(+), 61 deletions(-) diff --git a/services/libs/connectors/src/connectors/octolens/deps.test.ts b/services/libs/connectors/src/connectors/octolens/deps.test.ts index 6a584bf81c..afaedc4618 100644 --- a/services/libs/connectors/src/connectors/octolens/deps.test.ts +++ b/services/libs/connectors/src/connectors/octolens/deps.test.ts @@ -3,7 +3,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' import { queryInsightsProjects } from '@crowd/data-access-layer/src/collections' import { deactivateOctolensKeywordMappingsNotSeen, - listOctolensKeywordMappings, + listActiveOctolensKeywordMappings, upsertOctolensKeywordMapping, } from '@crowd/data-access-layer/src/octolens' import type { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' @@ -12,7 +12,7 @@ import { findSegmentById } from '@crowd/data-access-layer/src/segments' import { createOctolensDeps } from './deps' vi.mock('@crowd/data-access-layer/src/octolens', () => ({ - listOctolensKeywordMappings: vi.fn(), + listActiveOctolensKeywordMappings: vi.fn(), upsertOctolensKeywordMapping: vi.fn(), deactivateOctolensKeywordMappingsNotSeen: vi.fn(), })) @@ -25,7 +25,7 @@ vi.mock('@crowd/data-access-layer/src/segments', async (importOriginal) => ({ findSegmentById: vi.fn(), })) -const mockedListMappings = vi.mocked(listOctolensKeywordMappings) +const mockedListMappings = vi.mocked(listActiveOctolensKeywordMappings) const mockedQueryProjects = vi.mocked(queryInsightsProjects) const mockedFindSegment = vi.mocked(findSegmentById) const qx = {} as QueryExecutor @@ -35,34 +35,56 @@ type Segment = NonNullable>> beforeEach(() => { vi.clearAllMocks() mockedListMappings.mockResolvedValue([ - { integrationId: 'integration-1', segmentId: 'segment-a', keywordId: 1 }, - { integrationId: 'integration-1', segmentId: 'segment-b', keywordId: 2 }, - ] as Awaited>) + { integrationId: 'integration-a', segmentId: 'segment-a', keywordId: 1, keyword: 'ebpf' }, + { integrationId: 'integration-b', segmentId: 'segment-b', keywordId: 2, keyword: 'cilium' }, + ]) }) describe('createOctolensDeps', () => { - it('resolves keywordIds to segments from the integration mappings', async () => { - const deps = createOctolensDeps({ qx, integrationId: 'integration-1' }) + it("resolves keywordIds to their subproject integration, not the poll's own", async () => { + const deps = createOctolensDeps({ qx, integrationId: 'anchor' }) - expect(await deps.resolveSegmentId(1)).toBe('segment-a') - expect(await deps.resolveSegmentId(2)).toBe('segment-b') - expect(await deps.resolveSegmentId(3)).toBeNull() - expect(mockedListMappings).toHaveBeenCalledWith(qx, 'integration-1') + expect(await deps.resolveKeywordTarget(1)).toEqual({ + segmentId: 'segment-a', + integrationId: 'integration-a', + }) + expect(await deps.resolveKeywordTarget(2)).toEqual({ + segmentId: 'segment-b', + integrationId: 'integration-b', + }) + expect(await deps.resolveKeywordTarget(3)).toBeNull() + expect(mockedListMappings).toHaveBeenCalledWith(qx) + }) + + it('treats a keyword mapped under two integrations as unmapped', async () => { + mockedListMappings.mockResolvedValue([ + { integrationId: 'integration-a', segmentId: 'segment-a', keywordId: 1, keyword: 'ebpf' }, + { integrationId: 'integration-b', segmentId: 'segment-b', keywordId: 1, keyword: 'ebpf' }, + { integrationId: 'integration-c', segmentId: 'segment-c', keywordId: 1, keyword: 'ebpf' }, + { integrationId: 'integration-b', segmentId: 'segment-b', keywordId: 2, keyword: 'cilium' }, + ]) + const deps = createOctolensDeps({ qx, integrationId: 'anchor' }) + + expect(await deps.resolveKeywordTarget(1)).toBeNull() + expect(await deps.resolveKeywordTarget(2)).toEqual({ + segmentId: 'segment-b', + integrationId: 'integration-b', + }) }) it('loads the mappings once per deps instance, not once per lookup', async () => { const deps = createOctolensDeps({ qx, integrationId: 'integration-1' }) - await deps.resolveSegmentId(1) - await deps.resolveSegmentId(2) - await deps.resolveSegmentId(1) + await deps.resolveKeywordTarget(1) + await deps.resolveKeywordTarget(2) + await deps.resolveKeywordTarget(1) expect(mockedListMappings).toHaveBeenCalledTimes(1) }) it('reloads for a new deps instance, so each sync run sees fresh mappings', async () => { - await createOctolensDeps({ qx, integrationId: 'integration-1' }).resolveSegmentId(1) - await createOctolensDeps({ qx, integrationId: 'integration-1' }).resolveSegmentId(1) + await createOctolensDeps({ qx, integrationId: 'integration-1' }).resolveKeywordTarget(1) + await createOctolensDeps({ qx, integrationId: 'integration-1' }).resolveKeywordTarget(1) expect(mockedListMappings).toHaveBeenCalledTimes(2) }) diff --git a/services/libs/connectors/src/connectors/octolens/deps.ts b/services/libs/connectors/src/connectors/octolens/deps.ts index be6bceab3c..6c98a19d19 100644 --- a/services/libs/connectors/src/connectors/octolens/deps.ts +++ b/services/libs/connectors/src/connectors/octolens/deps.ts @@ -4,7 +4,7 @@ import { } from '@crowd/data-access-layer/src/collections' import { deactivateOctolensKeywordMappingsNotSeen, - listOctolensKeywordMappings, + listActiveOctolensKeywordMappings, upsertOctolensKeywordMapping, } from '@crowd/data-access-layer/src/octolens' import { findSegmentById, isSegmentSubproject } from '@crowd/data-access-layer/src/segments' @@ -17,8 +17,14 @@ export interface OctolensKeywordMappingInput { segmentId: string } +export interface OctolensKeywordTarget { + segmentId: string + integrationId: string +} + export interface OctolensDeps { - resolveSegmentId(keywordId: number): Promise + // 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 @@ -26,15 +32,25 @@ export interface OctolensDeps { } export function createOctolensDeps({ qx, integrationId }: DepsFactoryInput): OctolensDeps { - let keywordSegments: Map | null = null + let keywordTargets: Map | null = null return { - async resolveSegmentId(keywordId) { - if (!keywordSegments) { - const mappings = await listOctolensKeywordMappings(qx, integrationId) - keywordSegments = new Map(mappings.map((m) => [m.keywordId, m.segmentId])) + async resolveKeywordTarget(keywordId) { + if (!keywordTargets) { + const targets = new Map() + for (const m of await listActiveOctolensKeywordMappings(qx)) { + // Discovery never maps one keyword to two subprojects; if it ever did, routing to either + // would misroute the other's mentions, so the keyword counts as unmapped. + targets.set( + m.keywordId, + targets.has(m.keywordId) + ? null + : { segmentId: m.segmentId, integrationId: m.integrationId }, + ) + } + keywordTargets = targets } - return keywordSegments.get(keywordId) ?? null + return keywordTargets.get(keywordId) ?? null }, async resolveProjectSegmentId(slug) { diff --git a/services/libs/connectors/src/connectors/octolens/syncs/mappingDiscovery.test.ts b/services/libs/connectors/src/connectors/octolens/syncs/mappingDiscovery.test.ts index 7859fad30d..3b40129731 100644 --- a/services/libs/connectors/src/connectors/octolens/syncs/mappingDiscovery.test.ts +++ b/services/libs/connectors/src/connectors/octolens/syncs/mappingDiscovery.test.ts @@ -65,7 +65,7 @@ function makeCtx() { const deactivateCalls: number[][] = [] const log = { info: vi.fn(), warn: vi.fn(), error: vi.fn() } const deps: OctolensDeps = { - resolveSegmentId: async () => null, + resolveKeywordTarget: async () => null, resolveProjectSegmentId: vi.fn(async (slug: string) => SEGMENTS[slug] ?? null), upsertMapping: async (mapping) => { upserts.push(mapping) 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 12f4115d91..6ad6637cfe 100644 --- a/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts +++ b/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts @@ -53,6 +53,10 @@ function mention(overrides: Partial = {}): OctolensMention { } } +function target(segmentId: string) { + return { segmentId, integrationId: `integration-for-${segmentId}` } +} + function page(data: OctolensMention[], nextCursor: string | null = null): OctolensMentionsPage { return { data, nextCursor } } @@ -67,7 +71,7 @@ function makeCtx( const emitted: { records: unknown[]; segmentId?: string; integrationId?: string }[] = [] const watermarks: Record[] = [] - const { resolveSegmentId, ...ctxOverrides } = overrides + const { resolveKeywordTarget, ...ctxOverrides } = overrides const ctx: OctolensSyncContext = { channel: { channelId: 'octolens-global', channelName: 'octolens-global' }, integrationId: 'integration-1', @@ -87,7 +91,8 @@ function makeCtx( log: LOG, getCredential: async () => OCTOLENS_CREDENTIAL, deps: { - resolveSegmentId: resolveSegmentId ?? (async (keywordId) => `segment-${keywordId}`), + resolveKeywordTarget: + resolveKeywordTarget ?? (async (keywordId) => target(`segment-${keywordId}`)), resolveProjectSegmentId: async () => null, upsertMapping: async () => undefined, deactivateMappingsNotSeen: async () => 0, @@ -107,7 +112,7 @@ beforeEach(() => { }) describe('runMentionPoll', () => { - it('emits mapped mentions grouped by resolved segment and commits the watermark', async () => { + it('emits mapped mentions grouped by subproject integration and commits the watermark', async () => { mockedFetchMentionsPage.mockResolvedValue( page([mention({ keywordId: 1 }), mention({ keywordId: 2, sourceId: 'src-2' })]), ) @@ -117,8 +122,12 @@ describe('runMentionPoll', () => { expect(outcome).toEqual({ complete: true }) expect(emitted).toHaveLength(2) - expect(emitted.map((e) => e.segmentId).sort()).toEqual(['segment-1', 'segment-2']) - expect(emitted.every((e) => e.integrationId === 'integration-1')).toBe(true) + expect(emitted.map(({ segmentId, integrationId }) => ({ segmentId, integrationId }))).toEqual( + expect.arrayContaining([ + { segmentId: 'segment-1', integrationId: 'integration-for-segment-1' }, + { segmentId: 'segment-2', integrationId: 'integration-for-segment-2' }, + ]), + ) expect(watermarks.map(pollFields)).toEqual([ { startDate: '2026-10-12T08:00:00.000Z', cursor: null, pendingMaxTimestamp: null }, ]) @@ -167,24 +176,24 @@ describe('runMentionPoll', () => { it('skips a mention whose keywordId has no mapping', async () => { mockedFetchMentionsPage.mockResolvedValue(page([mention()])) - const { ctx, emitted } = makeCtx({ resolveSegmentId: async () => null }) + const { ctx, emitted } = makeCtx({ resolveKeywordTarget: async () => null }) await runMentionPoll(ctx) expect(emitted).toHaveLength(0) }) - it('skips a mention with no keywordId without calling resolveSegmentId', async () => { + it('skips a mention with no keywordId without calling resolveKeywordTarget', async () => { const { keywordId, ...withoutKeywordId } = mention() void keywordId mockedFetchMentionsPage.mockResolvedValue(page([withoutKeywordId as OctolensMention])) - const resolveSegmentId = vi.fn(async () => 'segment-1') - const { ctx, emitted } = makeCtx({ resolveSegmentId }) + const resolveKeywordTarget = vi.fn(async () => target('segment-1')) + const { ctx, emitted } = makeCtx({ resolveKeywordTarget }) await runMentionPoll(ctx) expect(emitted).toHaveLength(0) - expect(resolveSegmentId).not.toHaveBeenCalled() + expect(resolveKeywordTarget).not.toHaveBeenCalled() }) it('skips a mention that fails to map (e.g. a malformed field) instead of crashing the page', async () => { @@ -250,10 +259,10 @@ describe('runMentionPoll', () => { expect((emitted[0].records[0] as { sourceId: string }).sourceId).toBe('ok') }) - it('propagates a resolveSegmentId failure instead of swallowing it as a skip', async () => { + it('propagates a resolveKeywordTarget failure instead of swallowing it as a skip', async () => { mockedFetchMentionsPage.mockResolvedValue(page([mention()])) const { ctx, watermarks } = makeCtx({ - resolveSegmentId: async () => { + resolveKeywordTarget: async () => { throw new Error('db connection lost') }, }) @@ -393,11 +402,11 @@ describe('runMentionPoll', () => { }) describe('runMentionPoll keyword backfill', () => { - const MAPPED = async (keywordId: number) => (keywordId === 7 ? 'segment-7' : null) + const MAPPED = async (keywordId: number) => (keywordId === 7 ? target('segment-7') : null) it('records a keyword whose mentions were skipped for lack of a mapping', async () => { mockedFetchMentionsPage.mockResolvedValue(page([mention({ keywordId: 7 })])) - const { ctx, emitted, watermarks } = makeCtx({ resolveSegmentId: async () => null }) + const { ctx, emitted, watermarks } = makeCtx({ resolveKeywordTarget: async () => null }) const outcome = await runMentionPoll(ctx) @@ -411,7 +420,7 @@ describe('runMentionPoll keyword backfill', () => { .mockResolvedValueOnce(page([])) .mockResolvedValueOnce(page([mention({ keywordId: 7, sourceId: 'old-1' })])) const { ctx, emitted, watermarks } = makeCtx({ - resolveSegmentId: MAPPED, + resolveKeywordTarget: MAPPED, watermark: { startDate: '2026-10-12T08:00:00.000Z', unmappedKeywords: [7, 8] }, }) @@ -438,7 +447,7 @@ describe('runMentionPoll keyword backfill', () => { .mockResolvedValueOnce(page([])) .mockResolvedValueOnce(page([mention({ keywordId: 7 })], 'backfill-cursor')) const { ctx, watermarks } = makeCtx({ - resolveSegmentId: MAPPED, + resolveKeywordTarget: MAPPED, watermark: { startDate: '2026-10-12T08:00:00.000Z', backfill: [{ keywordId: 7, cursor: null }], @@ -456,7 +465,7 @@ describe('runMentionPoll keyword backfill', () => { mockedFetchMentionsPage.mockReset() mockedFetchMentionsPage.mockResolvedValueOnce(page([])).mockResolvedValueOnce(page([])) const next = makeCtx({ - resolveSegmentId: MAPPED, + resolveKeywordTarget: MAPPED, watermark: watermarks[watermarks.length - 1] ?? null, }) @@ -472,7 +481,7 @@ describe('runMentionPoll keyword backfill', () => { let calls = 0 mockedFetchMentionsPage.mockResolvedValue(page([], 'poll-cursor')) const { ctx, watermarks } = makeCtx({ - resolveSegmentId: MAPPED, + resolveKeywordTarget: MAPPED, watermark: { startDate: '2026-10-12T08:00:00.000Z', unmappedKeywords: [7] }, hasRunBudget: () => calls++ === 0, }) @@ -489,7 +498,7 @@ describe('runMentionPoll keyword backfill', () => { it('drops a backfill whose keyword lost its mapping, so it restarts when mapped again', async () => { mockedFetchMentionsPage.mockResolvedValue(page([])) const { ctx, watermarks } = makeCtx({ - resolveSegmentId: async () => null, + resolveKeywordTarget: async () => null, watermark: { startDate: '2026-10-12T08:00:00.000Z', backfill: [{ keywordId: 7, cursor: 'half-way' }], @@ -510,7 +519,7 @@ describe('runMentionPoll keyword backfill', () => { mockedFetchMentionsPage.mockResolvedValueOnce(page([], `backfill-${i}`)) } const { ctx, watermarks } = makeCtx({ - resolveSegmentId: async () => 'segment-7', + resolveKeywordTarget: async () => target('segment-7'), watermark: { startDate: '2026-10-12T08:00:00.000Z', backfill: [{ keywordId: 7, cursor: null }], @@ -527,7 +536,7 @@ describe('runMentionPoll keyword backfill', () => { it('ends a backfill whose cursor stops advancing instead of looping forever', async () => { mockedFetchMentionsPage.mockResolvedValueOnce(page([])).mockResolvedValueOnce(page([], 'stuck')) const { ctx, watermarks } = makeCtx({ - resolveSegmentId: async () => 'segment-7', + resolveKeywordTarget: async () => target('segment-7'), watermark: { startDate: '2026-10-12T08:00:00.000Z', backfill: [{ keywordId: 7, cursor: 'stuck' }], @@ -545,7 +554,7 @@ describe('runMentionPoll keyword backfill', () => { .mockResolvedValueOnce(page([], 'next')) .mockResolvedValueOnce(page([])) const { ctx, watermarks } = makeCtx({ - resolveSegmentId: async () => 'segment-7', + resolveKeywordTarget: async () => target('segment-7'), watermark: { startDate: '2026-10-12T08:00:00.000Z', backfill: [{ keywordId: 7, cursor: null }], @@ -564,7 +573,7 @@ describe('runMentionPoll keyword backfill', () => { it('ignores malformed keyword tracking in a stored watermark', async () => { mockedFetchMentionsPage.mockResolvedValue(page([])) const { ctx, watermarks } = makeCtx({ - resolveSegmentId: async () => null, + resolveKeywordTarget: async () => null, watermark: { startDate: '2026-10-12T08:00:00.000Z', unmappedKeywords: ['7', null], diff --git a/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.ts b/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.ts index f2851b7051..cc6208956b 100644 --- a/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.ts +++ b/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.ts @@ -1,6 +1,6 @@ import { assertCredentialKind } from '../../../types' import type { SyncContext, SyncOutcome } from '../../../types' -import type { OctolensDeps } from '../deps' +import type { OctolensDeps, OctolensKeywordTarget } from '../deps' import { mapMentionToActivity, octolensActivitySchema, toIsoTimestamp } from '../mapper' import type { MappedOctolensMention } from '../mapper' import { fetchMentionsPage } from '../mentionsClient' @@ -57,11 +57,11 @@ export async function runMentionPoll(ctx: SyncContext): Promise item.keywordId === keywordId)) { backfill.push({ keywordId, cursor: null }) @@ -91,7 +91,10 @@ export async function runMentionPoll(ctx: SyncContext): Promise void, ) => { - const bySegment = new Map[]>() + const byTarget = new Map< + string, + { target: OctolensKeywordTarget; activities: Record[] } + >() for (const mention of mentions) { let mapped: MappedOctolensMention | null // A malformed field must not crash the whole page, so wrap mapping in try/catch — @@ -130,23 +133,24 @@ export async function runMentionPoll(ctx: SyncContext): Promise Date: Thu, 15 Oct 2026 22:04:32 +0200 Subject: [PATCH 2/9] feat: run octolens discovery on a schedule with an integration per subproject MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Keyword-to-project mapping discovery moves out of the connector's sync units into an hourly connectors_worker Temporal schedule (overlap skip). Each run ensures the segment-less anchor integration and its single mention-poll sync unit, created with emitEnabled so mentions become activities, then writes each mapping under its subproject's Octolens integration, deactivates the mappings it no longer sees and soft-deletes subproject integrations left without keywords. An incomplete Octolens snapshot still writes nothing. Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- .../apps/connectors_worker/src/activities.ts | 2 + .../octolensDiscoveryActivities.test.ts | 177 ++++++++++++++++++ .../activities/octolensDiscoveryActivities.ts | 103 ++++++++++ services/apps/connectors_worker/src/main.ts | 2 + .../src/schedules/octolensDiscovery.ts | 35 ++++ .../apps/connectors_worker/src/workflows.ts | 3 +- .../src/workflows/octolensDiscovery.ts | 15 ++ .../src/connectors/octolens/deps.test.ts | 37 +--- .../src/connectors/octolens/deps.ts | 55 ++---- .../src/connectors/octolens/index.ts | 11 +- .../{syncs => }/mappingDiscovery.test.ts | 40 ++-- .../octolens/{syncs => }/mappingDiscovery.ts | 48 +++-- .../octolens/syncs/mentionPoll.test.ts | 3 - 13 files changed, 409 insertions(+), 122 deletions(-) create mode 100644 services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts create mode 100644 services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts create mode 100644 services/apps/connectors_worker/src/schedules/octolensDiscovery.ts create mode 100644 services/apps/connectors_worker/src/workflows/octolensDiscovery.ts rename services/libs/connectors/src/connectors/octolens/{syncs => }/mappingDiscovery.test.ts (90%) rename services/libs/connectors/src/connectors/octolens/{syncs => }/mappingDiscovery.ts (77%) 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..8fc08bf441 --- /dev/null +++ b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts @@ -0,0 +1,177 @@ +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(), + findOrCreateOctolensSubprojectIntegration: vi.fn(), + upsertOctolensKeywordMapping: vi.fn(), + deactivateOctolensKeywordMappingsNotSeenAcross: vi.fn(), + softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings: vi.fn(), + upsertSyncUnits: vi.fn(), + getCredential: vi.fn(), + discover: vi.fn(), + runMappingDiscovery: 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, + findOrCreateOctolensSubprojectIntegration: mocks.findOrCreateOctolensSubprojectIntegration, + upsertOctolensKeywordMapping: mocks.upsertOctolensKeywordMapping, + deactivateOctolensKeywordMappingsNotSeenAcross: + mocks.deactivateOctolensKeywordMappingsNotSeenAcross, + softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings: + mocks.softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings, +})) + +vi.mock('@crowd/data-access-layer/src/connectors', () => ({ + upsertSyncUnits: mocks.upsertSyncUnits, +})) + +vi.mock('@crowd/connectors', async (importOriginal) => ({ + ...(await importOriginal()), + getCredential: mocks.getCredential, + getManifest: () => ({ discover: mocks.discover }), +})) + +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: 2, keyword: 'cilium', segmentId: 'segment-ebpf' }, + { keywordId: 3, keyword: 'valkey', segmentId: 'segment-valkey' }, +] + +beforeEach(() => { + vi.clearAllMocks() + calls.length = 0 + 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.findOrCreateOctolensSubprojectIntegration.mockImplementation( + async (_qx: unknown, segmentId: string) => { + calls.push(`findOrCreateSubproject ${segmentId}`) + return { id: `integration-${segmentId}`, created: false, restored: false } + }, + ) + mocks.upsertOctolensKeywordMapping.mockImplementation( + async (_qx: unknown, mapping: { keywordId: number }) => { + calls.push(`upsertMapping ${mapping.keywordId}`) + }, + ) + mocks.deactivateOctolensKeywordMappingsNotSeenAcross.mockImplementation(record('deactivate', 4)) + mocks.softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings.mockImplementation( + record('softDelete', []), + ) + 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('writes each mapping under its subproject integration, one integration per segment', async () => { + const result = await runOctolensDiscovery() + + expect(mocks.findOrCreateOctolensSubprojectIntegration).toHaveBeenCalledTimes(2) + expect(mocks.upsertOctolensKeywordMapping).toHaveBeenCalledWith(expect.anything(), { + integrationId: 'integration-segment-ebpf', + keywordId: 2, + keyword: 'cilium', + segmentId: 'segment-ebpf', + }) + expect(mocks.deactivateOctolensKeywordMappingsNotSeenAcross).toHaveBeenCalledWith( + expect.anything(), + [ + { integrationId: 'integration-segment-ebpf', keywordId: 1 }, + { integrationId: 'integration-segment-ebpf', keywordId: 2 }, + { integrationId: 'integration-segment-valkey', keywordId: 3 }, + ], + ) + expect(result).toEqual({ complete: true, mappings: 3, deactivated: 4 }) + }) + + it('soft-deletes emptied integrations only after every upsert and the deactivation', async () => { + await runOctolensDiscovery() + + expect(calls).toEqual([ + 'anchor', + 'upsertSyncUnits [[{"integrationId":"anchor","platform":"octolens","channelId":"octolens-global","channelName":"octolens-global","syncName":"mentionPoll","emitEnabled":true}]]', + 'findOrCreateSubproject segment-ebpf', + 'upsertMapping 1', + 'upsertMapping 2', + 'findOrCreateSubproject segment-valkey', + 'upsertMapping 3', + expect.stringMatching(/^deactivate /), + 'softDelete', + ]) + }) + + it('writes nothing when discovery does not reconcile (incomplete snapshot)', async () => { + mocks.runMappingDiscovery.mockResolvedValue({ complete: false, mappings: 0, deactivated: 0 }) + + await runOctolensDiscovery() + + expect(mocks.findOrCreateOctolensSubprojectIntegration).not.toHaveBeenCalled() + expect(mocks.deactivateOctolensKeywordMappingsNotSeenAcross).not.toHaveBeenCalled() + expect( + mocks.softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings, + ).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..3b616fa8a3 --- /dev/null +++ b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts @@ -0,0 +1,103 @@ +import { assertCredentialKind, getCredential, getManifest } 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 { upsertSyncUnits } from '@crowd/data-access-layer/src/connectors' +import { + deactivateOctolensKeywordMappingsNotSeenAcross, + findOrCreateOctolensAnchorIntegration, + findOrCreateOctolensSubprojectIntegration, + softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings, + upsertOctolensKeywordMapping, +} 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' + +// Order matters: integrations and mappings first, then deactivation, then soft-delete, which +// would otherwise delete the integrations this run just created. The schedule never overlaps runs. +async function reconcileOctolensMappings( + qx: QueryExecutor, + mappings: OctolensKeywordMappingInput[], +): Promise { + const integrationBySegment = new Map() + for (const mapping of mappings) { + let integrationId = integrationBySegment.get(mapping.segmentId) + if (!integrationId) { + const integration = await findOrCreateOctolensSubprojectIntegration(qx, mapping.segmentId) + if (integration.created || integration.restored) { + svc.log.info( + { + integrationId: integration.id, + segmentId: mapping.segmentId, + created: integration.created, + restored: integration.restored, + }, + 'octolens discovery: subproject integration created or restored', + ) + } + integrationId = integration.id + integrationBySegment.set(mapping.segmentId, integrationId) + } + await upsertOctolensKeywordMapping(qx, { integrationId, ...mapping }) + } + + const deactivated = await deactivateOctolensKeywordMappingsNotSeenAcross( + qx, + mappings.map((m) => ({ + integrationId: integrationBySegment.get(m.segmentId) as string, + keywordId: m.keywordId, + })), + ) + + const softDeleted = await softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings(qx) + if (softDeleted.length > 0) { + svc.log.info( + { integrationIds: softDeleted }, + 'octolens discovery: soft-deleted subproject integrations with no active keywords', + ) + } + + return deactivated +} + +export async function runOctolensDiscovery(): Promise { + 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 created = 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, syncUnits: created }, + 'octolens discovery: ensured the mention poll sync unit (emitEnabled on insert)', + ) + + return runMappingDiscovery({ + token: credential.data.token, + log: svc.log, + resolveProjectSegmentId: (slug) => resolveOctolensProjectSegmentId(qx, slug), + reconcileMappings: (mappings) => reconcileOctolensMappings(qx, mappings), + }) +} 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..e0b6b9202a --- /dev/null +++ b/services/apps/connectors_worker/src/schedules/octolensDiscovery.ts @@ -0,0 +1,35 @@ +import { ScheduleAlreadyRunning, ScheduleOverlapPolicy } from '@temporalio/client' + +import { svc } from '../main' +import { octolensDiscovery } from '../workflows/octolensDiscovery' + +// Hourly: new projects and webhooks appear rarely, and each run costs 3 requests of the shared +// Octolens budget. SKIP overlap is required: two concurrent runs could soft-delete each other's +// freshly created 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..1c91dec30e --- /dev/null +++ b/services/apps/connectors_worker/src/workflows/octolensDiscovery.ts @@ -0,0 +1,15 @@ +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 }, +}) + +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..85812a4237 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' }, @@ -89,9 +83,7 @@ describe('createOctolensDeps', () => { expect(mockedListMappings).toHaveBeenCalledTimes(2) }) - describe('resolveProjectSegmentId', () => { - const deps = () => createOctolensDeps({ qx, integrationId: 'integration-1' }) - + describe('resolveOctolensProjectSegmentId', () => { it('returns the insights project segment when it is a subproject', async () => { mockedQueryProjects.mockResolvedValue([{ segmentId: 'segment-sub' }] as never) mockedFindSegment.mockResolvedValue({ @@ -101,7 +93,7 @@ describe('createOctolensDeps', () => { grandparentSlug: 'lf', } as Segment) - expect(await deps().resolveProjectSegmentId('ebpf')).toBe('segment-sub') + expect(await resolveOctolensProjectSegmentId(qx, 'ebpf')).toBe('segment-sub') expect(mockedQueryProjects).toHaveBeenCalledWith( qx, expect.objectContaining({ filter: { slug: { eq: 'ebpf' } } }), @@ -117,31 +109,14 @@ describe('createOctolensDeps', () => { grandparentSlug: null, } as Segment) - expect(await deps().resolveProjectSegmentId('lf')).toBeNull() + expect(await resolveOctolensProjectSegmentId(qx, 'lf')).toBeNull() }) it('returns null when no insights project has the slug', async () => { mockedQueryProjects.mockResolvedValue([]) - expect(await deps().resolveProjectSegmentId('missing')).toBeNull() + expect(await resolveOctolensProjectSegmentId(qx, 'missing')).toBeNull() expect(mockedFindSegment).not.toHaveBeenCalled() }) }) - - 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]) - - 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) - }) }) 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..308cf538a0 100644 --- a/services/libs/connectors/src/connectors/octolens/index.ts +++ b/services/libs/connectors/src/connectors/octolens/index.ts @@ -3,13 +3,14 @@ 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 = { platform: 'octolens', // Octolens shares one org-wide bearer token (no per-installation rotation), so unlike // GitHub this manifest sets no preparePool/mintToken/probeBudget. + // Keyword→project mapping discovery runs outside the sync units, on its own connectors_worker + // schedule, because it creates the integrations the poll emits under. syncs: [ { name: 'mentionPoll', @@ -17,14 +18,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 77% rename from services/libs/connectors/src/connectors/octolens/syncs/mappingDiscovery.ts rename to services/libs/connectors/src/connectors/octolens/mappingDiscovery.ts index 3d9821c07c..433a7eb4a1 100644 --- a/services/libs/connectors/src/connectors/octolens/syncs/mappingDiscovery.ts +++ b/services/libs/connectors/src/connectors/octolens/mappingDiscovery.ts @@ -1,13 +1,34 @@ -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 + // Activities only land on subproject segments, so a slug resolving to anything else is unusable. + resolveProjectSegmentId(slug: string): Promise + // Writes the complete set of mappings seen this run and deactivates every other one. + // Returns how many were deactivated. + 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 +97,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 +137,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 +152,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, } From 6313b857e116b35c401ec065b7e72e7a28f83de4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Thu, 15 Oct 2026 22:11:59 +0200 Subject: [PATCH 3/9] fix: skip octolens discovery where no octolens token is configured MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Without CROWD_OCTOLENS_API_TOKEN the hourly run would create the anchor integration and then fail on every run. It now logs and skips before any write. Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- .../activities/octolensDiscoveryActivities.test.ts | 14 ++++++++++++++ .../src/activities/octolensDiscoveryActivities.ts | 13 ++++++++++++- services/libs/connectors/src/credentials.ts | 5 +++++ 3 files changed, 31 insertions(+), 1 deletion(-) diff --git a/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts index 8fc08bf441..8247bfa654 100644 --- a/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts +++ b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts @@ -20,6 +20,7 @@ const mocks = vi.hoisted(() => ({ getCredential: vi.fn(), discover: vi.fn(), runMappingDiscovery: vi.fn(), + isOctolensCredentialConfigured: vi.fn(), })) vi.mock('../main', () => ({ @@ -51,6 +52,7 @@ 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', () => ({ @@ -70,6 +72,7 @@ const MAPPINGS = [ beforeEach(() => { vi.clearAllMocks() calls.length = 0 + mocks.isOctolensCredentialConfigured.mockReturnValue(true) mocks.findOrCreateOctolensAnchorIntegration.mockImplementation( record('anchor', { id: 'anchor', created: true, restored: false }), ) @@ -163,6 +166,17 @@ describe('runOctolensDiscovery', () => { ]) }) + 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('writes nothing when discovery does not reconcile (incomplete snapshot)', async () => { mocks.runMappingDiscovery.mockResolvedValue({ complete: false, mappings: 0, deactivated: 0 }) diff --git a/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts index 3b616fa8a3..b260522cba 100644 --- a/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts +++ b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts @@ -1,4 +1,9 @@ -import { assertCredentialKind, getCredential, getManifest } from '@crowd/connectors' +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 { @@ -69,6 +74,12 @@ async function reconcileOctolensMappings( } export async function runOctolensDiscovery(): Promise { + // Checked before any write, so an environment without a token never gets an anchor integration. + if (!isOctolensCredentialConfigured()) { + svc.log.warn('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. 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 { From 2a62bd049aeed72708ccd95282f566ac9d766558 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Thu, 15 Oct 2026 22:15:39 +0200 Subject: [PATCH 4/9] fix: reconcile octolens mappings in one transaction MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A keyword moving between subprojects was briefly active under both, so a poll in that window treated it as unmapped and queued a needless full backfill. The upserts, deactivation and soft-delete now commit together. Also trims comments to the repo's comment rules. Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- .../octolensDiscoveryActivities.test.ts | 11 ++++++++-- .../activities/octolensDiscoveryActivities.ts | 22 ++++++++++--------- .../src/schedules/octolensDiscovery.ts | 4 +--- .../src/connectors/octolens/index.ts | 2 -- .../connectors/octolens/mappingDiscovery.ts | 3 --- 5 files changed, 22 insertions(+), 20 deletions(-) diff --git a/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts index 8247bfa654..f3d45046de 100644 --- a/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts +++ b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts @@ -3,6 +3,12 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' import { runOctolensDiscovery } from './octolensDiscoveryActivities' const calls: string[] = [] +const qx = { + tx: async (fn: (tx: unknown) => Promise) => { + calls.push('tx') + return fn(qx) + }, +} const record = (name: string, value: T) => async (...args: unknown[]) => { @@ -31,7 +37,7 @@ vi.mock('../main', () => ({ })) vi.mock('@crowd/data-access-layer/src/queryExecutor', () => ({ - dbStoreQx: vi.fn(() => ({})), + dbStoreQx: vi.fn(() => qx), })) vi.mock('@crowd/data-access-layer/src/octolens', () => ({ @@ -150,12 +156,13 @@ describe('runOctolensDiscovery', () => { expect(result).toEqual({ complete: true, mappings: 3, deactivated: 4 }) }) - it('soft-deletes emptied integrations only after every upsert and the deactivation', async () => { + it('reconciles in one transaction, soft-deleting emptied integrations last', async () => { await runOctolensDiscovery() expect(calls).toEqual([ 'anchor', 'upsertSyncUnits [[{"integrationId":"anchor","platform":"octolens","channelId":"octolens-global","channelName":"octolens-global","syncName":"mentionPoll","emitEnabled":true}]]', + 'tx', 'findOrCreateSubproject segment-ebpf', 'upsertMapping 1', 'upsertMapping 2', diff --git a/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts index b260522cba..592611b302 100644 --- a/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts +++ b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts @@ -26,13 +26,21 @@ import { svc } from '../main' const MENTION_POLL_SYNC_NAME = 'mentionPoll' -// Order matters: integrations and mappings first, then deactivation, then soft-delete, which -// would otherwise delete the integrations this run just created. The schedule never overlaps runs. +// 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. async function reconcileOctolensMappings( qx: QueryExecutor, mappings: OctolensKeywordMappingInput[], +): Promise { + return qx.tx((tx) => reconcileInTransaction(tx, mappings)) +} + +async function reconcileInTransaction( + qx: QueryExecutor, + mappings: OctolensKeywordMappingInput[], ): Promise { const integrationBySegment = new Map() + const seen: { integrationId: string; keywordId: number }[] = [] for (const mapping of mappings) { let integrationId = integrationBySegment.get(mapping.segmentId) if (!integrationId) { @@ -52,15 +60,10 @@ async function reconcileOctolensMappings( integrationBySegment.set(mapping.segmentId, integrationId) } await upsertOctolensKeywordMapping(qx, { integrationId, ...mapping }) + seen.push({ integrationId, keywordId: mapping.keywordId }) } - const deactivated = await deactivateOctolensKeywordMappingsNotSeenAcross( - qx, - mappings.map((m) => ({ - integrationId: integrationBySegment.get(m.segmentId) as string, - keywordId: m.keywordId, - })), - ) + const deactivated = await deactivateOctolensKeywordMappingsNotSeenAcross(qx, seen) const softDeleted = await softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings(qx) if (softDeleted.length > 0) { @@ -74,7 +77,6 @@ async function reconcileOctolensMappings( } export async function runOctolensDiscovery(): Promise { - // Checked before any write, so an environment without a token never gets an anchor integration. if (!isOctolensCredentialConfigured()) { svc.log.warn('octolens discovery skipped: CROWD_OCTOLENS_API_TOKEN is not set') return { complete: false, mappings: 0, deactivated: 0 } diff --git a/services/apps/connectors_worker/src/schedules/octolensDiscovery.ts b/services/apps/connectors_worker/src/schedules/octolensDiscovery.ts index e0b6b9202a..ec029f7d81 100644 --- a/services/apps/connectors_worker/src/schedules/octolensDiscovery.ts +++ b/services/apps/connectors_worker/src/schedules/octolensDiscovery.ts @@ -3,9 +3,7 @@ import { ScheduleAlreadyRunning, ScheduleOverlapPolicy } from '@temporalio/clien import { svc } from '../main' import { octolensDiscovery } from '../workflows/octolensDiscovery' -// Hourly: new projects and webhooks appear rarely, and each run costs 3 requests of the shared -// Octolens budget. SKIP overlap is required: two concurrent runs could soft-delete each other's -// freshly created integrations. +// 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({ diff --git a/services/libs/connectors/src/connectors/octolens/index.ts b/services/libs/connectors/src/connectors/octolens/index.ts index 308cf538a0..152c5cb850 100644 --- a/services/libs/connectors/src/connectors/octolens/index.ts +++ b/services/libs/connectors/src/connectors/octolens/index.ts @@ -9,8 +9,6 @@ export const octolensConnector: Manifest = { platform: 'octolens', // Octolens shares one org-wide bearer token (no per-installation rotation), so unlike // GitHub this manifest sets no preparePool/mintToken/probeBudget. - // Keyword→project mapping discovery runs outside the sync units, on its own connectors_worker - // schedule, because it creates the integrations the poll emits under. syncs: [ { name: 'mentionPoll', diff --git a/services/libs/connectors/src/connectors/octolens/mappingDiscovery.ts b/services/libs/connectors/src/connectors/octolens/mappingDiscovery.ts index 433a7eb4a1..be6c861756 100644 --- a/services/libs/connectors/src/connectors/octolens/mappingDiscovery.ts +++ b/services/libs/connectors/src/connectors/octolens/mappingDiscovery.ts @@ -12,10 +12,7 @@ export interface OctolensKeywordMappingInput { export interface OctolensMappingDiscoveryInput { token: string log: Logger - // Activities only land on subproject segments, so a slug resolving to anything else is unusable. resolveProjectSegmentId(slug: string): Promise - // Writes the complete set of mappings seen this run and deactivates every other one. - // Returns how many were deactivated. reconcileMappings(mappings: OctolensKeywordMappingInput[]): Promise } From 5b0fe0df8813e7a6902c3cd4ac07e753884f38e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Thu, 15 Oct 2026 22:44:17 +0200 Subject: [PATCH 5/9] fix: revive the dead-lettered octolens mention poll once the token works again MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Only provider.auth failures dead-letter a sync unit, and nothing brought the anchor's mention poll back after the token was fixed. Discovery now revives it right after its own Octolens calls succeed with that token. Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- .../octolensDiscoveryActivities.test.ts | 18 ++++++ .../activities/octolensDiscoveryActivities.ts | 18 +++++- .../src/connectors/syncUnits.test.ts | 58 ++++++++++++++++++- .../src/connectors/syncUnits.ts | 20 +++++++ 4 files changed, 111 insertions(+), 3 deletions(-) diff --git a/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts index f3d45046de..b13ea940d7 100644 --- a/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts +++ b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts @@ -23,6 +23,7 @@ const mocks = vi.hoisted(() => ({ deactivateOctolensKeywordMappingsNotSeenAcross: vi.fn(), softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings: vi.fn(), upsertSyncUnits: vi.fn(), + reviveDeadLetteredSyncUnits: vi.fn(), getCredential: vi.fn(), discover: vi.fn(), runMappingDiscovery: vi.fn(), @@ -52,6 +53,7 @@ vi.mock('@crowd/data-access-layer/src/octolens', () => ({ vi.mock('@crowd/data-access-layer/src/connectors', () => ({ upsertSyncUnits: mocks.upsertSyncUnits, + reviveDeadLetteredSyncUnits: mocks.reviveDeadLetteredSyncUnits, })) vi.mock('@crowd/connectors', async (importOriginal) => ({ @@ -91,6 +93,7 @@ beforeEach(() => { { channelId: 'octolens-global', channelName: 'octolens-global' }, ]) mocks.upsertSyncUnits.mockImplementation(record('upsertSyncUnits', 1)) + mocks.reviveDeadLetteredSyncUnits.mockImplementation(record('revive', 0)) mocks.findOrCreateOctolensSubprojectIntegration.mockImplementation( async (_qx: unknown, segmentId: string) => { calls.push(`findOrCreateSubproject ${segmentId}`) @@ -170,6 +173,7 @@ describe('runOctolensDiscovery', () => { 'upsertMapping 3', expect.stringMatching(/^deactivate /), 'softDelete', + 'revive ["anchor","mentionPoll"]', ]) }) @@ -184,6 +188,20 @@ describe('runOctolensDiscovery', () => { expect(mocks.runMappingDiscovery).not.toHaveBeenCalled() }) + it('revives the dead-lettered mention poll only after Octolens answered with this token', async () => { + await runOctolensDiscovery() + expect(mocks.reviveDeadLetteredSyncUnits).toHaveBeenCalledWith( + expect.anything(), + 'anchor', + 'mentionPoll', + ) + + mocks.reviveDeadLetteredSyncUnits.mockClear() + mocks.runMappingDiscovery.mockRejectedValue(new Error('provider authentication failed')) + await expect(runOctolensDiscovery()).rejects.toThrow('provider authentication failed') + expect(mocks.reviveDeadLetteredSyncUnits).not.toHaveBeenCalled() + }) + it('writes nothing when discovery does not reconcile (incomplete snapshot)', async () => { mocks.runMappingDiscovery.mockResolvedValue({ complete: false, mappings: 0, deactivated: 0 }) diff --git a/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts index 592611b302..23dde7a55f 100644 --- a/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts +++ b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts @@ -10,7 +10,10 @@ import type { OctolensKeywordMappingInput, OctolensMappingDiscoveryResult, } from '@crowd/connectors/src/connectors/octolens/mappingDiscovery' -import { upsertSyncUnits } from '@crowd/data-access-layer/src/connectors' +import { + reviveDeadLetteredSyncUnits, + upsertSyncUnits, +} from '@crowd/data-access-layer/src/connectors' import { deactivateOctolensKeywordMappingsNotSeenAcross, findOrCreateOctolensAnchorIntegration, @@ -107,10 +110,21 @@ export async function runOctolensDiscovery(): Promise 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/libs/data-access-layer/src/connectors/syncUnits.test.ts b/services/libs/data-access-layer/src/connectors/syncUnits.test.ts index 3b9b9acd9e..617492bf92 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,59 @@ describe('upsertSyncUnits', () => { }) }) }) + +describe('reviveDeadLetteredSyncUnits', () => { + async function readUnit(qx: QueryExecutor, id: string) { + return qx.selectOne( + `SELECT status, "consecutiveFailures", 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 WHERE id = $(dead)`, + { dead }, + ) + + expect(await reviveDeadLetteredSyncUnits(qx, integrationId, 'mentionPoll')).toBe(1) + + expect(await readUnit(qx, dead)).toEqual({ + status: 'active', + consecutiveFailures: 0, + 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..df8975eb8c 100644 --- a/services/libs/data-access-layer/src/connectors/syncUnits.ts +++ b/services/libs/data-access-layer/src/connectors/syncUnits.ts @@ -210,6 +210,26 @@ 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, + "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 * From f2ca6ff8797b99022b47c8cd5a570b3b3e96c2a5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Thu, 15 Oct 2026 22:44:18 +0200 Subject: [PATCH 6/9] refactor: remove per-integration octolens mapping queries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Discovery and the mention poll now work across all subproject integrations, so the per-integration find, list and deactivate queries have no callers left. Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- .../src/octolens/index.test.ts | 177 ++++++------------ .../data-access-layer/src/octolens/index.ts | 62 ------ 2 files changed, 53 insertions(+), 186 deletions(-) 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..7597bef322 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,10 @@ import type { QueryExecutor } from '@crowd/database' import { withQx } from '@crowd/test-kit/db' import { - deactivateOctolensKeywordMappingsNotSeen, deactivateOctolensKeywordMappingsNotSeenAcross, - findOctolensKeywordMapping, findOrCreateOctolensAnchorIntegration, findOrCreateOctolensSubprojectIntegration, listActiveOctolensKeywordMappings, - listOctolensKeywordMappings, softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings, upsertOctolensKeywordMapping, } from './index' @@ -64,6 +61,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 +110,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 +133,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 +149,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 +186,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 +392,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 +449,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 +464,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 +555,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,9 +564,9 @@ 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) diff --git a/services/libs/data-access-layer/src/octolens/index.ts b/services/libs/data-access-layer/src/octolens/index.ts index 1032da3d6c..774a576adc 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 From 22f24cb54b9306e5eb70c5db92a59eb3e664320a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Fri, 16 Oct 2026 08:57:16 +0200 Subject: [PATCH 7/9] fix: drop the octolens keyword target interface duplicated by the main merge MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- services/libs/connectors/src/connectors/octolens/deps.ts | 5 ----- 1 file changed, 5 deletions(-) diff --git a/services/libs/connectors/src/connectors/octolens/deps.ts b/services/libs/connectors/src/connectors/octolens/deps.ts index dc06ae440a..85021a44ee 100644 --- a/services/libs/connectors/src/connectors/octolens/deps.ts +++ b/services/libs/connectors/src/connectors/octolens/deps.ts @@ -13,11 +13,6 @@ export interface OctolensKeywordTarget { integrationId: string } -export interface OctolensKeywordTarget { - segmentId: string - integrationId: string -} - export interface OctolensDeps { // Where a keyword's mentions land: its subproject's Octolens integration, never the poll's own. resolveKeywordTarget(keywordId: number): Promise From c7bf8ecfc82486276b05a283dc0916427240b77f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Fri, 16 Oct 2026 09:02:59 +0200 Subject: [PATCH 8/9] fix: tighten octolens discovery logging, retries and dead-letter revive MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Integration changes are logged only after the reconcile transaction commits, an auth failure is no longer retried within a run, the revive clears the stale auth error, and a missing token logs at info level. Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- .../octolensDiscoveryActivities.test.ts | 2 +- .../activities/octolensDiscoveryActivities.ts | 49 ++++++++------- .../src/workflows/octolensDiscovery.ts | 6 +- .../src/connectors/octolens/deps.test.ts | 62 +++++++++---------- .../src/connectors/syncUnits.test.ts | 5 +- .../src/connectors/syncUnits.ts | 5 +- 6 files changed, 69 insertions(+), 60 deletions(-) diff --git a/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts index b13ea940d7..c9e5c020e1 100644 --- a/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts +++ b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts @@ -202,7 +202,7 @@ describe('runOctolensDiscovery', () => { expect(mocks.reviveDeadLetteredSyncUnits).not.toHaveBeenCalled() }) - it('writes nothing when discovery does not reconcile (incomplete snapshot)', async () => { + it('makes no mapping writes when discovery does not reconcile (incomplete snapshot)', async () => { mocks.runMappingDiscovery.mockResolvedValue({ complete: false, mappings: 0, deactivated: 0 }) await runOctolensDiscovery() diff --git a/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts index 23dde7a55f..29324c925f 100644 --- a/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts +++ b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts @@ -35,30 +35,38 @@ async function reconcileOctolensMappings( qx: QueryExecutor, mappings: OctolensKeywordMappingInput[], ): Promise { - return qx.tx((tx) => reconcileInTransaction(tx, mappings)) + const outcome = await qx.tx((tx) => reconcileInTransaction(tx, 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 } async function reconcileInTransaction( qx: QueryExecutor, mappings: OctolensKeywordMappingInput[], -): Promise { +): Promise<{ deactivated: number; created: string[]; restored: string[]; softDeleted: string[] }> { const integrationBySegment = new Map() const seen: { integrationId: string; keywordId: number }[] = [] + const created: string[] = [] + const restored: string[] = [] for (const mapping of mappings) { let integrationId = integrationBySegment.get(mapping.segmentId) if (!integrationId) { const integration = await findOrCreateOctolensSubprojectIntegration(qx, mapping.segmentId) - if (integration.created || integration.restored) { - svc.log.info( - { - integrationId: integration.id, - segmentId: mapping.segmentId, - created: integration.created, - restored: integration.restored, - }, - 'octolens discovery: subproject integration created or restored', - ) - } + if (integration.created) created.push(integration.id) + if (integration.restored) restored.push(integration.id) integrationId = integration.id integrationBySegment.set(mapping.segmentId, integrationId) } @@ -67,21 +75,14 @@ async function reconcileInTransaction( } const deactivated = await deactivateOctolensKeywordMappingsNotSeenAcross(qx, seen) - const softDeleted = await softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings(qx) - if (softDeleted.length > 0) { - svc.log.info( - { integrationIds: softDeleted }, - 'octolens discovery: soft-deleted subproject integrations with no active keywords', - ) - } - return deactivated + return { deactivated, created, restored, softDeleted } } export async function runOctolensDiscovery(): Promise { if (!isOctolensCredentialConfigured()) { - svc.log.warn('octolens discovery skipped: CROWD_OCTOLENS_API_TOKEN is not set') + svc.log.info('octolens discovery skipped: CROWD_OCTOLENS_API_TOKEN is not set') return { complete: false, mappings: 0, deactivated: 0 } } @@ -94,7 +95,7 @@ export async function runOctolensDiscovery(): Promise ({ integrationId: anchor.id, @@ -106,7 +107,7 @@ export async function runOctolensDiscovery(): Promise({ startToCloseTimeout: '10 minutes', - retry: { maximumAttempts: 2, backoffCoefficient: 2 }, + retry: { + maximumAttempts: 2, + backoffCoefficient: 2, + nonRetryableErrorTypes: ['ProviderAuthError'], + }, }) export async function octolensDiscovery(): Promise { diff --git a/services/libs/connectors/src/connectors/octolens/deps.test.ts b/services/libs/connectors/src/connectors/octolens/deps.test.ts index 85812a4237..dd902098ad 100644 --- a/services/libs/connectors/src/connectors/octolens/deps.test.ts +++ b/services/libs/connectors/src/connectors/octolens/deps.test.ts @@ -82,41 +82,41 @@ describe('createOctolensDeps', () => { expect(mockedListMappings).toHaveBeenCalledTimes(2) }) +}) - 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' } } }), - ) - }) +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 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) + 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 resolveOctolensProjectSegmentId(qx, 'lf')).toBeNull() - }) + expect(await resolveOctolensProjectSegmentId(qx, 'lf')).toBeNull() + }) - it('returns null when no insights project has the slug', async () => { - mockedQueryProjects.mockResolvedValue([]) + it('returns null when no insights project has the slug', async () => { + mockedQueryProjects.mockResolvedValue([]) - expect(await resolveOctolensProjectSegmentId(qx, 'missing')).toBeNull() - expect(mockedFindSegment).not.toHaveBeenCalled() - }) + expect(await resolveOctolensProjectSegmentId(qx, 'missing')).toBeNull() + expect(mockedFindSegment).not.toHaveBeenCalled() }) }) 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 617492bf92..655797dca7 100644 --- a/services/libs/data-access-layer/src/connectors/syncUnits.test.ts +++ b/services/libs/data-access-layer/src/connectors/syncUnits.test.ts @@ -238,7 +238,7 @@ describe('upsertSyncUnits', () => { describe('reviveDeadLetteredSyncUnits', () => { async function readUnit(qx: QueryExecutor, id: string) { return qx.selectOne( - `SELECT status, "consecutiveFailures", watermark FROM integration.sync_units WHERE id = $(id)`, + `SELECT status, "consecutiveFailures", "lastErrorClass", watermark FROM integration.sync_units WHERE id = $(id)`, { id }, ) } @@ -265,7 +265,7 @@ describe('reviveDeadLetteredSyncUnits', () => { status: 'dead_letter', }) await qx.result( - `UPDATE integration.sync_units SET "consecutiveFailures" = 5 WHERE id = $(dead)`, + `UPDATE integration.sync_units SET "consecutiveFailures" = 5, "lastErrorClass" = 'provider.auth' WHERE id = $(dead)`, { dead }, ) @@ -274,6 +274,7 @@ describe('reviveDeadLetteredSyncUnits', () => { 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') diff --git a/services/libs/data-access-layer/src/connectors/syncUnits.ts b/services/libs/data-access-layer/src/connectors/syncUnits.ts index df8975eb8c..3cf5485b6b 100644 --- a/services/libs/data-access-layer/src/connectors/syncUnits.ts +++ b/services/libs/data-access-layer/src/connectors/syncUnits.ts @@ -210,7 +210,8 @@ export async function recordRunFailure( ) } -// Only provider.auth failures dead-letter a unit, so callers revive it once that credential works again. +// 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, @@ -220,6 +221,8 @@ export async function reviveDeadLetteredSyncUnits( `UPDATE integration.sync_units SET status = 'active', "consecutiveFailures" = 0, + "lastErrorClass" = NULL, + "lastErrorMessage" = NULL, "nextRunAt" = now(), "lockedAt" = NULL, "updatedAt" = now() From b37bdefe08c9b8c6db89aac83c7f92ebf55f48c3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Fri, 16 Oct 2026 09:28:47 +0200 Subject: [PATCH 9/9] refactor: move octolens mapping reconcile into the data-access layer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The reconcile is now reconcileOctolensKeywordMappings and is tested against the real test database: keyword moves, integration restore and soft-delete, and rollback of every write when a mapping fails partway. The discovery activity only orchestrates and logs. Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- .../octolensDiscoveryActivities.test.ts | 87 ++------------- .../activities/octolensDiscoveryActivities.ts | 36 +------ .../src/octolens/index.test.ts | 101 ++++++++++++++++++ .../data-access-layer/src/octolens/index.ts | 44 ++++++++ 4 files changed, 158 insertions(+), 110 deletions(-) diff --git a/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts index c9e5c020e1..3f596c4623 100644 --- a/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts +++ b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.test.ts @@ -3,12 +3,6 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' import { runOctolensDiscovery } from './octolensDiscoveryActivities' const calls: string[] = [] -const qx = { - tx: async (fn: (tx: unknown) => Promise) => { - calls.push('tx') - return fn(qx) - }, -} const record = (name: string, value: T) => async (...args: unknown[]) => { @@ -18,10 +12,7 @@ const record = const mocks = vi.hoisted(() => ({ findOrCreateOctolensAnchorIntegration: vi.fn(), - findOrCreateOctolensSubprojectIntegration: vi.fn(), - upsertOctolensKeywordMapping: vi.fn(), - deactivateOctolensKeywordMappingsNotSeenAcross: vi.fn(), - softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings: vi.fn(), + reconcileOctolensKeywordMappings: vi.fn(), upsertSyncUnits: vi.fn(), reviveDeadLetteredSyncUnits: vi.fn(), getCredential: vi.fn(), @@ -38,17 +29,12 @@ vi.mock('../main', () => ({ })) vi.mock('@crowd/data-access-layer/src/queryExecutor', () => ({ - dbStoreQx: vi.fn(() => qx), + dbStoreQx: vi.fn(() => ({})), })) vi.mock('@crowd/data-access-layer/src/octolens', () => ({ findOrCreateOctolensAnchorIntegration: mocks.findOrCreateOctolensAnchorIntegration, - findOrCreateOctolensSubprojectIntegration: mocks.findOrCreateOctolensSubprojectIntegration, - upsertOctolensKeywordMapping: mocks.upsertOctolensKeywordMapping, - deactivateOctolensKeywordMappingsNotSeenAcross: - mocks.deactivateOctolensKeywordMappingsNotSeenAcross, - softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings: - mocks.softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings, + reconcileOctolensKeywordMappings: mocks.reconcileOctolensKeywordMappings, })) vi.mock('@crowd/data-access-layer/src/connectors', () => ({ @@ -73,7 +59,6 @@ vi.mock('@crowd/connectors/src/connectors/octolens/mappingDiscovery', () => ({ const MAPPINGS = [ { keywordId: 1, keyword: 'ebpf', segmentId: 'segment-ebpf' }, - { keywordId: 2, keyword: 'cilium', segmentId: 'segment-ebpf' }, { keywordId: 3, keyword: 'valkey', segmentId: 'segment-valkey' }, ] @@ -94,20 +79,8 @@ beforeEach(() => { ]) mocks.upsertSyncUnits.mockImplementation(record('upsertSyncUnits', 1)) mocks.reviveDeadLetteredSyncUnits.mockImplementation(record('revive', 0)) - mocks.findOrCreateOctolensSubprojectIntegration.mockImplementation( - async (_qx: unknown, segmentId: string) => { - calls.push(`findOrCreateSubproject ${segmentId}`) - return { id: `integration-${segmentId}`, created: false, restored: false } - }, - ) - mocks.upsertOctolensKeywordMapping.mockImplementation( - async (_qx: unknown, mapping: { keywordId: number }) => { - calls.push(`upsertMapping ${mapping.keywordId}`) - }, - ) - mocks.deactivateOctolensKeywordMappingsNotSeenAcross.mockImplementation(record('deactivate', 4)) - mocks.softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings.mockImplementation( - record('softDelete', []), + mocks.reconcileOctolensKeywordMappings.mockImplementation( + record('reconcile', { deactivated: 4, created: [], restored: [], softDeleted: [] }), ) mocks.runMappingDiscovery.mockImplementation( async ({ @@ -138,43 +111,16 @@ describe('runOctolensDiscovery', () => { ]) }) - it('writes each mapping under its subproject integration, one integration per segment', async () => { + it('reconciles the discovered mappings, then revives the mention poll', async () => { const result = await runOctolensDiscovery() - expect(mocks.findOrCreateOctolensSubprojectIntegration).toHaveBeenCalledTimes(2) - expect(mocks.upsertOctolensKeywordMapping).toHaveBeenCalledWith(expect.anything(), { - integrationId: 'integration-segment-ebpf', - keywordId: 2, - keyword: 'cilium', - segmentId: 'segment-ebpf', - }) - expect(mocks.deactivateOctolensKeywordMappingsNotSeenAcross).toHaveBeenCalledWith( - expect.anything(), - [ - { integrationId: 'integration-segment-ebpf', keywordId: 1 }, - { integrationId: 'integration-segment-ebpf', keywordId: 2 }, - { integrationId: 'integration-segment-valkey', keywordId: 3 }, - ], - ) - expect(result).toEqual({ complete: true, mappings: 3, deactivated: 4 }) - }) - - it('reconciles in one transaction, soft-deleting emptied integrations last', async () => { - await runOctolensDiscovery() - expect(calls).toEqual([ 'anchor', 'upsertSyncUnits [[{"integrationId":"anchor","platform":"octolens","channelId":"octolens-global","channelName":"octolens-global","syncName":"mentionPoll","emitEnabled":true}]]', - 'tx', - 'findOrCreateSubproject segment-ebpf', - 'upsertMapping 1', - 'upsertMapping 2', - 'findOrCreateSubproject segment-valkey', - 'upsertMapping 3', - expect.stringMatching(/^deactivate /), - 'softDelete', + `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 () => { @@ -188,16 +134,9 @@ describe('runOctolensDiscovery', () => { expect(mocks.runMappingDiscovery).not.toHaveBeenCalled() }) - it('revives the dead-lettered mention poll only after Octolens answered with this token', async () => { - await runOctolensDiscovery() - expect(mocks.reviveDeadLetteredSyncUnits).toHaveBeenCalledWith( - expect.anything(), - 'anchor', - 'mentionPoll', - ) - - mocks.reviveDeadLetteredSyncUnits.mockClear() + 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() }) @@ -207,10 +146,6 @@ describe('runOctolensDiscovery', () => { await runOctolensDiscovery() - expect(mocks.findOrCreateOctolensSubprojectIntegration).not.toHaveBeenCalled() - expect(mocks.deactivateOctolensKeywordMappingsNotSeenAcross).not.toHaveBeenCalled() - expect( - mocks.softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings, - ).not.toHaveBeenCalled() + 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 index 29324c925f..eb438d58b5 100644 --- a/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts +++ b/services/apps/connectors_worker/src/activities/octolensDiscoveryActivities.ts @@ -15,11 +15,8 @@ import { upsertSyncUnits, } from '@crowd/data-access-layer/src/connectors' import { - deactivateOctolensKeywordMappingsNotSeenAcross, findOrCreateOctolensAnchorIntegration, - findOrCreateOctolensSubprojectIntegration, - softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings, - upsertOctolensKeywordMapping, + 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' @@ -29,13 +26,11 @@ import { svc } from '../main' const MENTION_POLL_SYNC_NAME = 'mentionPoll' -// 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. async function reconcileOctolensMappings( qx: QueryExecutor, mappings: OctolensKeywordMappingInput[], ): Promise { - const outcome = await qx.tx((tx) => reconcileInTransaction(tx, mappings)) + const outcome = await reconcileOctolensKeywordMappings(qx, mappings) if (outcome.created.length > 0 || outcome.restored.length > 0) { svc.log.info( @@ -53,33 +48,6 @@ async function reconcileOctolensMappings( return outcome.deactivated } -async function reconcileInTransaction( - qx: QueryExecutor, - mappings: OctolensKeywordMappingInput[], -): Promise<{ deactivated: number; created: string[]; restored: string[]; softDeleted: string[] }> { - const integrationBySegment = new Map() - const seen: { integrationId: string; keywordId: number }[] = [] - const created: string[] = [] - const restored: string[] = [] - for (const mapping of mappings) { - let integrationId = integrationBySegment.get(mapping.segmentId) - if (!integrationId) { - const integration = await findOrCreateOctolensSubprojectIntegration(qx, 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(qx, { integrationId, ...mapping }) - seen.push({ integrationId, keywordId: mapping.keywordId }) - } - - const deactivated = await deactivateOctolensKeywordMappingsNotSeenAcross(qx, seen) - const softDeleted = await softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings(qx) - - return { deactivated, created, restored, softDeleted } -} - export async function runOctolensDiscovery(): Promise { if (!isOctolensCredentialConfigured()) { svc.log.info('octolens discovery skipped: CROWD_OCTOLENS_API_TOKEN is not set') 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 7597bef322..390809e40c 100644 --- a/services/libs/data-access-layer/src/octolens/index.test.ts +++ b/services/libs/data-access-layer/src/octolens/index.test.ts @@ -9,6 +9,7 @@ import { findOrCreateOctolensAnchorIntegration, findOrCreateOctolensSubprojectIntegration, listActiveOctolensKeywordMappings, + reconcileOctolensKeywordMappings, softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings, upsertOctolensKeywordMapping, } from './index' @@ -573,3 +574,103 @@ describe('octolens subproject integrations', () => { 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 774a576adc..496d1a9284 100644 --- a/services/libs/data-access-layer/src/octolens/index.ts +++ b/services/libs/data-access-layer/src/octolens/index.ts @@ -229,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 } + }) +}