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