feat: emit octolens mentions under the keyword's subproject integration

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<Awaited<ReturnType<typeof findSegmentById>>>
 beforeEach(() => {
   vi.clearAllMocks()
   mockedListMappings.mockResolvedValue([
-    { integrationId: 'integration-1', segmentId: 'segment-a', keywordId: 1 },
-    { integrationId: 'integration-1', segmentId: 'segment-b', keywordId: 2 },
-  ] as Awaited<ReturnType<typeof listOctolensKeywordMappings>>)
+    { 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)
   })

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,24 +17,40 @@ export interface OctolensKeywordMappingInput {
   segmentId: string
 }
 
+export interface OctolensKeywordTarget {
+  segmentId: string
+  integrationId: string
+}
+
 export interface OctolensDeps {
-  resolveSegmentId(keywordId: number): Promise<string | null>
+  // Where a keyword's mentions land: its subproject's Octolens integration, never the poll's own.
+  resolveKeywordTarget(keywordId: number): Promise<OctolensKeywordTarget | null>
   // Activities only land on subproject segments, so a slug resolving to anything else is unusable.
   resolveProjectSegmentId(slug: string): Promise<string | null>
   upsertMapping(mapping: OctolensKeywordMappingInput): Promise<void>
   deactivateMappingsNotSeen(seenKeywordIds: number[]): Promise<number>
 }
 
 export function createOctolensDeps({ qx, integrationId }: DepsFactoryInput): OctolensDeps {
-  let keywordSegments: Map<number, string> | null = null
+  let keywordTargets: Map<number, OctolensKeywordTarget | null> | 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<number, OctolensKeywordTarget | null>()
+        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) {

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)

services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts

@@ -53,6 +53,10 @@ function mention(overrides: Partial<OctolensMention> = {}): 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<string, unknown>[] = []
 
-  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],