From f368943e0c95d920575cf900625f33460f051a6d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Uro=C5=A1=20Marolt?= <uros@marolt.me>
Date: Thu, 15 Oct 2026 18:31:16 +0200
Subject: [PATCH 1/9] feat: add data access for per-subproject octolens
 integrations
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Add find-or-create helpers for per-subproject Octolens integrations
(restoring a soft-deleted row when its keywords return) and for the
segment-less anchor integration, both serialized on an advisory lock.
Add cross-integration listing and deactivation of keyword mappings and
soft-deletion of subproject integrations left without active mappings.
upsertSyncUnits accepts insert-only emitEnabled and initialDelaySeconds;
existing callers keep the old defaults.

Refs linuxfoundation/lfx-cdp-insights-ops#748

Signed-off-by: Uroš Marolt <uros@marolt.me>
---
 .../src/connectors/syncUnits.test.ts          |  94 +++++-
 .../src/connectors/syncUnits.ts               |  17 +-
 .../data-access-layer/src/connectors/types.ts |   3 +
 .../src/octolens/index.test.ts                | 294 ++++++++++++++++++
 .../data-access-layer/src/octolens/index.ts   | 200 ++++++++++++
 5 files changed, 603 insertions(+), 5 deletions(-)

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 3a0f4b658c..11d3ef0b04 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 } from './syncUnits'
+import { listShadowDiffUnits, upsertSyncUnits } from './syncUnits'
 
 const test = withQx(base)
 
@@ -127,3 +127,95 @@ describe('listShadowDiffUnits', () => {
     expect(result).toHaveLength(0)
   })
 })
+
+async function readUnit(
+  qx: QueryExecutor,
+  integrationId: string,
+  syncName: string,
+): Promise<{
+  emitEnabled: boolean
+  nextRunAt: Date
+  watermark: Record<string, unknown> | null
+  consecutiveFailures: number
+  channelName: string
+  delaySeconds: number
+}> {
+  return qx.selectOne(
+    `
+    SELECT "emitEnabled", "nextRunAt", watermark, "consecutiveFailures", "channelName",
+           EXTRACT(EPOCH FROM ("nextRunAt" - now()))::float8 AS "delaySeconds"
+    FROM integration.sync_units
+    WHERE "integrationId" = $(integrationId) AND "syncName" = $(syncName)
+    `,
+    { integrationId, syncName },
+  )
+}
+
+describe('upsertSyncUnits', () => {
+  const base = { platform: 'github', channelId: 'repo-1', channelName: 'org/repo' }
+
+  test('defaults to shadow mode and a 10-900s random start when new fields are omitted', async ({
+    qx,
+  }) => {
+    const integrationId = await createIntegration(qx)
+
+    await upsertSyncUnits(qx, [{ ...base, integrationId, syncName: 'issues' }])
+
+    const unit = await readUnit(qx, integrationId, 'issues')
+    expect(unit.emitEnabled).toBe(false)
+    expect(unit.delaySeconds).toBeGreaterThanOrEqual(9)
+    expect(unit.delaySeconds).toBeLessThanOrEqual(900)
+  })
+
+  test('applies emitEnabled and initialDelaySeconds on insert', async ({ qx }) => {
+    const integrationId = await createIntegration(qx)
+
+    await upsertSyncUnits(qx, [
+      { ...base, integrationId, syncName: 'issues', emitEnabled: true, initialDelaySeconds: 0 },
+      { ...base, integrationId, syncName: 'forks' },
+    ])
+
+    const issues = await readUnit(qx, integrationId, 'issues')
+    expect(issues.emitEnabled).toBe(true)
+    expect(Math.abs(issues.delaySeconds)).toBeLessThan(5)
+    const forks = await readUnit(qx, integrationId, 'forks')
+    expect(forks.emitEnabled).toBe(false)
+    expect(forks.delaySeconds).toBeGreaterThanOrEqual(9)
+  })
+
+  test('re-running leaves emitEnabled, nextRunAt, watermark and failures untouched', async ({
+    qx,
+  }) => {
+    const integrationId = await createIntegration(qx)
+    await upsertSyncUnits(qx, [{ ...base, integrationId, syncName: 'issues' }])
+    await qx.result(
+      `
+      UPDATE integration.sync_units
+      SET "emitEnabled" = true, "nextRunAt" = '2030-01-01T00:00:00Z',
+          watermark = '{"phase":"incremental"}'::jsonb, "consecutiveFailures" = 3
+      WHERE "integrationId" = $(integrationId)
+      `,
+      { integrationId },
+    )
+
+    await upsertSyncUnits(qx, [
+      {
+        ...base,
+        integrationId,
+        syncName: 'issues',
+        channelName: 'org/renamed',
+        emitEnabled: false,
+        initialDelaySeconds: 0,
+      },
+    ])
+
+    const unit = await readUnit(qx, integrationId, 'issues')
+    expect(unit).toMatchObject({
+      emitEnabled: true,
+      watermark: { phase: 'incremental' },
+      consecutiveFailures: 3,
+      channelName: 'org/renamed',
+    })
+    expect(new Date(unit.nextRunAt).toISOString()).toBe('2030-01-01T00:00:00.000Z')
+  })
+})
diff --git a/services/libs/data-access-layer/src/connectors/syncUnits.ts b/services/libs/data-access-layer/src/connectors/syncUnits.ts
index 7411da584b..1760d16548 100644
--- a/services/libs/data-access-layer/src/connectors/syncUnits.ts
+++ b/services/libs/data-access-layer/src/connectors/syncUnits.ts
@@ -20,16 +20,23 @@ export async function upsertSyncUnits(qx: QueryExecutor, units: SyncUnitUpsert[]
 
   return qx.result(
     `INSERT INTO integration.sync_units AS su
-       ("integrationId", platform, "channelId", "channelName", "syncName", watermark, "nextRunAt")
-     SELECT u.*, now() + ($(minDelaySeconds) + random() * $(delaySpanSeconds)) * interval '1 second'
+       ("integrationId", platform, "channelId", "channelName", "syncName", watermark, "emitEnabled", "nextRunAt")
+     SELECT u."integrationId", u.platform, u."channelId", u."channelName", u."syncName", u.watermark,
+            COALESCE(u."emitEnabled", false),
+            now() + COALESCE(
+              u."initialDelaySeconds",
+              $(minDelaySeconds) + random() * $(delaySpanSeconds)
+            ) * interval '1 second'
      FROM unnest(
        $(integrationIds)::uuid[],
        $(platforms)::text[],
        $(channelIds)::text[],
        $(channelNames)::text[],
        $(syncNames)::text[],
-       $(watermarks)::jsonb[]
-     ) u
+       $(watermarks)::jsonb[],
+       $(emitEnabled)::boolean[],
+       $(initialDelaySeconds)::float8[]
+     ) AS u("integrationId", platform, "channelId", "channelName", "syncName", watermark, "emitEnabled", "initialDelaySeconds")
      ON CONFLICT ("integrationId", "channelId", "syncName")
      DO UPDATE SET "channelName" = EXCLUDED."channelName",
                    watermark = COALESCE(EXCLUDED.watermark, su.watermark),
@@ -41,6 +48,8 @@ export async function upsertSyncUnits(qx: QueryExecutor, units: SyncUnitUpsert[]
       channelNames: units.map((u) => u.channelName),
       syncNames: units.map((u) => u.syncName),
       watermarks: units.map((u) => (u.watermark != null ? JSON.stringify(u.watermark) : null)),
+      emitEnabled: units.map((u) => u.emitEnabled ?? null),
+      initialDelaySeconds: units.map((u) => u.initialDelaySeconds ?? null),
       minDelaySeconds: MIN_INITIAL_DELAY_SECONDS,
       delaySpanSeconds: MAX_INITIAL_DELAY_SECONDS - MIN_INITIAL_DELAY_SECONDS,
     },
diff --git a/services/libs/data-access-layer/src/connectors/types.ts b/services/libs/data-access-layer/src/connectors/types.ts
index c0bd298624..ba841af3bc 100644
--- a/services/libs/data-access-layer/src/connectors/types.ts
+++ b/services/libs/data-access-layer/src/connectors/types.ts
@@ -35,6 +35,9 @@ export type SyncUnitUpsert = Pick<
   'integrationId' | 'platform' | 'channelId' | 'channelName' | 'syncName'
 > & {
   watermark?: Record<string, unknown> | null
+  // Insert-only: re-running an upsert never changes these on an existing unit.
+  emitEnabled?: boolean
+  initialDelaySeconds?: number
 }
 
 export type IClaimedUnit = Pick<
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 c67e6ab7c1..8b0f3607fa 100644
--- a/services/libs/data-access-layer/src/octolens/index.test.ts
+++ b/services/libs/data-access-layer/src/octolens/index.test.ts
@@ -6,8 +6,13 @@ import { withQx } from '@crowd/test-kit/db'
 
 import {
   deactivateOctolensKeywordMappingsNotSeen,
+  deactivateOctolensKeywordMappingsNotSeenAcross,
   findOctolensKeywordMapping,
+  findOrCreateOctolensAnchorIntegration,
+  findOrCreateOctolensSubprojectIntegration,
+  listActiveOctolensKeywordMappings,
   listOctolensKeywordMappings,
+  softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings,
   upsertOctolensKeywordMapping,
 } from './index'
 
@@ -263,3 +268,292 @@ describe('octolens keyword mappings', () => {
     expect(await findOctolensKeywordMapping(qx, integrationId, 1)).not.toBeNull()
   })
 })
+
+async function createPlatformIntegration(
+  qx: QueryExecutor,
+  options: {
+    segmentId: string | null
+    platform?: string
+    deletedAt?: string | null
+    updatedAt?: string
+  },
+): Promise<string> {
+  const id = generateUUIDv1()
+  await qx.result(
+    `
+    INSERT INTO public.integrations
+      (id, platform, status, "segmentId", "tenantId", "createdAt", "updatedAt", "deletedAt")
+    VALUES
+      ($(id), $(platform), 'done', $(segmentId), $(tenantId), NOW(),
+       COALESCE($(updatedAt)::timestamptz, NOW()), $(deletedAt)::timestamptz)
+    `,
+    {
+      id,
+      platform: options.platform ?? 'octolens',
+      segmentId: options.segmentId,
+      tenantId: DEFAULT_TENANT_ID,
+      deletedAt: options.deletedAt ?? null,
+      updatedAt: options.updatedAt ?? null,
+    },
+  )
+  return id
+}
+
+async function readIntegration(
+  qx: QueryExecutor,
+  id: string,
+): Promise<{ platform: string; status: string; segmentId: string | null; deletedAt: Date | null }> {
+  return qx.selectOne(
+    `SELECT platform, status, "segmentId", "deletedAt" FROM public.integrations WHERE id = $(id)`,
+    { id },
+  )
+}
+
+async function countIntegrationsForSegment(qx: QueryExecutor, segmentId: string): Promise<number> {
+  const row: { count: number } = await qx.selectOne(
+    `SELECT count(*)::int AS count FROM public.integrations WHERE "segmentId" = $(segmentId)`,
+    { segmentId },
+  )
+  return row.count
+}
+
+describe('octolens subproject integrations', () => {
+  test('findOrCreateOctolensSubprojectIntegration creates once and then finds the same row', async ({
+    qx,
+  }) => {
+    const segmentId = await createSegment(qx)
+
+    const first = await findOrCreateOctolensSubprojectIntegration(qx, segmentId)
+    const second = await findOrCreateOctolensSubprojectIntegration(qx, segmentId)
+
+    expect(first).toMatchObject({ created: true, restored: false })
+    expect(second).toEqual({ id: first.id, created: false, restored: false })
+    expect(await countIntegrationsForSegment(qx, segmentId)).toBe(1)
+    expect(await readIntegration(qx, first.id)).toMatchObject({
+      platform: 'octolens',
+      status: 'done',
+      segmentId,
+      deletedAt: null,
+    })
+  })
+
+  test('findOrCreateOctolensSubprojectIntegration ignores other platforms on the segment', async ({
+    qx,
+  }) => {
+    const segmentId = await createSegment(qx)
+    const githubId = await createPlatformIntegration(qx, { segmentId, platform: 'github' })
+
+    const result = await findOrCreateOctolensSubprojectIntegration(qx, segmentId)
+
+    expect(result.created).toBe(true)
+    expect(result.id).not.toBe(githubId)
+  })
+
+  test('findOrCreateOctolensSubprojectIntegration throws on two live rows for one segment', async ({
+    qx,
+  }) => {
+    const segmentId = await createSegment(qx)
+    await createPlatformIntegration(qx, { segmentId })
+    await createPlatformIntegration(qx, { segmentId })
+
+    await expect(findOrCreateOctolensSubprojectIntegration(qx, segmentId)).rejects.toThrow()
+  })
+
+  test('findOrCreateOctolensSubprojectIntegration restores the most recently updated soft-deleted row', async ({
+    qx,
+  }) => {
+    const segmentId = await createSegment(qx)
+    const older = await createPlatformIntegration(qx, {
+      segmentId,
+      deletedAt: '2026-01-01T00:00:00Z',
+      updatedAt: '2026-01-01T00:00:00Z',
+    })
+    const newer = await createPlatformIntegration(qx, {
+      segmentId,
+      deletedAt: '2026-02-01T00:00:00Z',
+      updatedAt: '2026-02-01T00:00:00Z',
+    })
+
+    const result = await findOrCreateOctolensSubprojectIntegration(qx, segmentId)
+
+    expect(result).toEqual({ id: newer, created: false, restored: true })
+    expect((await readIntegration(qx, newer)).deletedAt).toBeNull()
+    expect((await readIntegration(qx, older)).deletedAt).not.toBeNull()
+    expect(await countIntegrationsForSegment(qx, segmentId)).toBe(2)
+  })
+
+  test('findOrCreateOctolensAnchorIntegration creates one segment-less row and ignores subprojects', async ({
+    qx,
+  }) => {
+    const segmentId = await createSegment(qx)
+    const subproject = await findOrCreateOctolensSubprojectIntegration(qx, segmentId)
+    await createPlatformIntegration(qx, { segmentId: null, platform: 'github' })
+    const deletedAnchor = await createPlatformIntegration(qx, {
+      segmentId: null,
+      deletedAt: '2026-01-01T00:00:00Z',
+    })
+
+    const first = await findOrCreateOctolensAnchorIntegration(qx)
+    const second = await findOrCreateOctolensAnchorIntegration(qx)
+
+    expect(first.created).toBe(true)
+    expect(second).toEqual({ id: first.id, created: false })
+    expect([subproject.id, deletedAnchor]).not.toContain(first.id)
+    expect(await readIntegration(qx, first.id)).toMatchObject({
+      platform: 'octolens',
+      segmentId: null,
+      deletedAt: null,
+    })
+  })
+
+  test('findOrCreateOctolensAnchorIntegration throws on two live anchors', async ({ qx }) => {
+    await createPlatformIntegration(qx, { segmentId: null })
+    await createPlatformIntegration(qx, { segmentId: null })
+
+    await expect(findOrCreateOctolensAnchorIntegration(qx)).rejects.toThrow()
+  })
+
+  test('listActiveOctolensKeywordMappings returns active mappings of live subproject integrations only', async ({
+    qx,
+  }) => {
+    const segmentA = await createSegment(qx)
+    const segmentB = await createSegment(qx)
+    const integrationA = await createPlatformIntegration(qx, { segmentId: segmentA })
+    const integrationB = await createPlatformIntegration(qx, { segmentId: segmentB })
+    const deletedIntegration = await createPlatformIntegration(qx, {
+      segmentId: segmentA,
+      deletedAt: '2026-01-01T00:00:00Z',
+    })
+    const githubIntegration = await createPlatformIntegration(qx, {
+      segmentId: segmentA,
+      platform: 'github',
+    })
+    const anchor = await createPlatformIntegration(qx, { segmentId: null })
+
+    const upsert = (integrationId: string, segmentId: string, keywordId: number) =>
+      upsertOctolensKeywordMapping(qx, {
+        integrationId,
+        segmentId,
+        keywordId,
+        keyword: `kw-${keywordId}`,
+      })
+    await upsert(integrationA, segmentA, 1)
+    await upsert(integrationA, segmentA, 2)
+    await upsert(integrationB, segmentB, 1)
+    await upsert(integrationB, segmentB, 3)
+    await upsert(deletedIntegration, segmentA, 4)
+    await upsert(githubIntegration, segmentA, 5)
+    await upsert(anchor, segmentA, 6)
+    await deactivateOctolensKeywordMappingsNotSeen(qx, integrationB, [1])
+
+    const ours = new Set([
+      integrationA,
+      integrationB,
+      deletedIntegration,
+      githubIntegration,
+      anchor,
+    ])
+    const result = (await listActiveOctolensKeywordMappings(qx)).filter((m) =>
+      ours.has(m.integrationId),
+    )
+
+    expect(result).toHaveLength(3)
+    expect(result).toEqual(
+      expect.arrayContaining([
+        { keywordId: 1, keyword: 'kw-1', integrationId: integrationA, segmentId: segmentA },
+        { keywordId: 1, keyword: 'kw-1', integrationId: integrationB, segmentId: segmentB },
+        { keywordId: 2, keyword: 'kw-2', integrationId: integrationA, segmentId: segmentA },
+      ]),
+    )
+  })
+
+  test('deactivateOctolensKeywordMappingsNotSeenAcross keeps seen pairs and other platforms', async ({
+    qx,
+  }) => {
+    const segmentA = await createSegment(qx)
+    const segmentB = await createSegment(qx)
+    const integrationA = await createPlatformIntegration(qx, { segmentId: segmentA })
+    const integrationB = await createPlatformIntegration(qx, { segmentId: segmentB })
+    const githubIntegration = await createPlatformIntegration(qx, {
+      segmentId: segmentA,
+      platform: 'github',
+    })
+
+    for (const [integrationId, segmentId, keywordId] of [
+      [integrationA, segmentA, 1],
+      [integrationA, segmentA, 2],
+      [integrationB, segmentB, 1],
+      [githubIntegration, segmentA, 9],
+    ] as const) {
+      await upsertOctolensKeywordMapping(qx, {
+        integrationId,
+        segmentId,
+        keywordId,
+        keyword: `kw-${keywordId}`,
+      })
+    }
+
+    const deactivated = await deactivateOctolensKeywordMappingsNotSeenAcross(qx, [
+      { integrationId: integrationA, keywordId: 1 },
+      { integrationId: integrationB, keywordId: 2 },
+    ])
+
+    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()
+  })
+
+  test('deactivateOctolensKeywordMappingsNotSeenAcross with nothing seen deactivates nothing', async ({
+    qx,
+  }) => {
+    const segmentId = await createSegment(qx)
+    const integrationId = await createPlatformIntegration(qx, { segmentId })
+    await upsertOctolensKeywordMapping(qx, { integrationId, segmentId, keywordId: 1, keyword: 'k' })
+
+    expect(await deactivateOctolensKeywordMappingsNotSeenAcross(qx, [])).toBe(0)
+    expect(await findOctolensKeywordMapping(qx, integrationId, 1)).not.toBeNull()
+  })
+
+  test('softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings deletes only empty subproject integrations', async ({
+    qx,
+  }) => {
+    const segmentA = await createSegment(qx)
+    const segmentB = await createSegment(qx)
+    const withMappings = await createPlatformIntegration(qx, { segmentId: segmentA })
+    const empty = await createPlatformIntegration(qx, { segmentId: segmentB })
+    const anchor = await createPlatformIntegration(qx, { segmentId: null })
+    const github = await createPlatformIntegration(qx, { segmentId: segmentB, platform: 'github' })
+    await upsertOctolensKeywordMapping(qx, {
+      integrationId: withMappings,
+      segmentId: segmentA,
+      keywordId: 1,
+      keyword: 'live',
+    })
+    await upsertOctolensKeywordMapping(qx, {
+      integrationId: empty,
+      segmentId: segmentB,
+      keywordId: 2,
+      keyword: 'gone',
+    })
+    await deactivateOctolensKeywordMappingsNotSeenAcross(qx, [
+      { integrationId: withMappings, keywordId: 1 },
+    ])
+
+    const deleted = await softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings(qx)
+
+    expect(deleted).toContain(empty)
+    expect(deleted).not.toContain(withMappings)
+    expect(deleted).not.toContain(anchor)
+    expect(deleted).not.toContain(github)
+    expect((await readIntegration(qx, empty)).deletedAt).not.toBeNull()
+    expect((await readIntegration(qx, withMappings)).deletedAt).toBeNull()
+    expect((await readIntegration(qx, anchor)).deletedAt).toBeNull()
+    expect((await readIntegration(qx, github)).deletedAt).toBeNull()
+
+    const restored = await findOrCreateOctolensSubprojectIntegration(qx, segmentB)
+    expect(restored).toEqual({ id: empty, created: false, restored: true })
+    expect((await readIntegration(qx, empty)).deletedAt).toBeNull()
+  })
+})
diff --git a/services/libs/data-access-layer/src/octolens/index.ts b/services/libs/data-access-layer/src/octolens/index.ts
index ddc271c525..32ab352c76 100644
--- a/services/libs/data-access-layer/src/octolens/index.ts
+++ b/services/libs/data-access-layer/src/octolens/index.ts
@@ -1,3 +1,6 @@
+import { DEFAULT_TENANT_ID, generateUUIDv4 } from '@crowd/common'
+import { IntegrationState, PlatformType } from '@crowd/types'
+
 import { QueryExecutor } from '../queryExecutor'
 
 export interface IOctolensKeywordMapping {
@@ -95,3 +98,200 @@ export async function deactivateOctolensKeywordMappingsNotSeen(
     { integrationId, seenKeywordIds },
   )
 }
+
+export interface IOctolensSubprojectIntegration {
+  id: string
+  created: boolean
+  restored: boolean
+}
+
+export interface IOctolensAnchorIntegration {
+  id: string
+  created: boolean
+}
+
+export interface IActiveOctolensKeywordMapping {
+  keywordId: number
+  keyword: string
+  integrationId: string
+  segmentId: string
+}
+
+export interface IOctolensKeywordMappingRef {
+  integrationId: string
+  keywordId: number
+}
+
+// integrations has no unique key on (platform, segmentId), so every writer of
+// Octolens integration rows serializes on this lock to avoid duplicates.
+async function lockOctolensIntegrations(tx: QueryExecutor): Promise<void> {
+  await tx.select(`SELECT pg_advisory_xact_lock(hashtext('octolens-integration')::bigint)`)
+}
+
+async function insertOctolensIntegration(
+  tx: QueryExecutor,
+  segmentId: string | null,
+): Promise<string> {
+  const id = generateUUIDv4()
+  await tx.result(
+    `
+      INSERT INTO public.integrations
+        (id, platform, status, "segmentId", "tenantId", "createdAt", "updatedAt")
+      VALUES ($(id), $(platform), $(status), $(segmentId), $(tenantId), now(), now())
+    `,
+    {
+      id,
+      platform: PlatformType.OCTOLENS,
+      status: IntegrationState.DONE,
+      segmentId,
+      tenantId: DEFAULT_TENANT_ID,
+    },
+  )
+  return id
+}
+
+export async function findOrCreateOctolensSubprojectIntegration(
+  qx: QueryExecutor,
+  segmentId: string,
+): Promise<IOctolensSubprojectIntegration> {
+  return qx.tx(async (tx) => {
+    await lockOctolensIntegrations(tx)
+
+    const rows: { id: string; deletedAt: Date | null }[] = await tx.select(
+      `
+        SELECT id, "deletedAt"
+        FROM public.integrations
+        WHERE platform = $(platform) AND "segmentId" = $(segmentId)
+        ORDER BY "updatedAt" DESC, "createdAt" DESC, id
+      `,
+      { platform: PlatformType.OCTOLENS, segmentId },
+    )
+
+    const live = rows.filter((r) => r.deletedAt === null)
+    if (live.length > 1) {
+      throw new Error(`Found ${live.length} live Octolens integrations for segment ${segmentId}!`)
+    }
+    if (live.length === 1) {
+      return { id: live[0].id, created: false, restored: false }
+    }
+
+    if (rows.length > 0) {
+      const { id } = rows[0]
+      await tx.result(
+        `
+          UPDATE public.integrations
+          SET "deletedAt" = NULL, "updatedAt" = now()
+          WHERE id = $(id)
+        `,
+        { id },
+      )
+      return { id, created: false, restored: true }
+    }
+
+    const id = await insertOctolensIntegration(tx, segmentId)
+    return { id, created: true, restored: false }
+  })
+}
+
+export async function findOrCreateOctolensAnchorIntegration(
+  qx: QueryExecutor,
+): Promise<IOctolensAnchorIntegration> {
+  return qx.tx(async (tx) => {
+    await lockOctolensIntegrations(tx)
+
+    const rows: { id: string }[] = await tx.select(
+      `
+        SELECT id
+        FROM public.integrations
+        WHERE platform = $(platform) AND "segmentId" IS NULL AND "deletedAt" IS NULL
+      `,
+      { platform: PlatformType.OCTOLENS },
+    )
+
+    if (rows.length > 1) {
+      throw new Error(`Found ${rows.length} live Octolens anchor integrations!`)
+    }
+    if (rows.length === 1) {
+      return { id: rows[0].id, created: false }
+    }
+
+    const id = await insertOctolensIntegration(tx, null)
+    return { id, created: true }
+  })
+}
+
+export async function listActiveOctolensKeywordMappings(
+  qx: QueryExecutor,
+): Promise<IActiveOctolensKeywordMapping[]> {
+  return qx.select(
+    `
+      SELECT m."keywordId", m.keyword, m."integrationId", m."segmentId"
+      FROM integration."octolensKeywordMappings" m
+      JOIN public.integrations i ON i.id = m."integrationId"
+      WHERE m."deactivatedAt" IS NULL
+        AND i.platform = $(platform)
+        AND i."deletedAt" IS NULL
+        AND i."segmentId" IS NOT NULL
+      ORDER BY m."keywordId", m."integrationId"
+    `,
+    { platform: PlatformType.OCTOLENS },
+  )
+}
+
+// An empty set deactivates nothing: a run that saw no mappings must never wipe them all.
+export async function deactivateOctolensKeywordMappingsNotSeenAcross(
+  qx: QueryExecutor,
+  seen: IOctolensKeywordMappingRef[],
+): Promise<number> {
+  if (seen.length === 0) {
+    return 0
+  }
+
+  return qx.result(
+    `
+      UPDATE integration."octolensKeywordMappings" m
+      SET "deactivatedAt" = now(), "updatedAt" = now()
+      FROM public.integrations i
+      WHERE i.id = m."integrationId"
+        AND i.platform = $(platform)
+        AND i."deletedAt" IS NULL
+        AND m."deactivatedAt" IS NULL
+        AND NOT EXISTS (
+          SELECT 1
+          FROM unnest($(integrationIds)::uuid[], $(keywordIds)::int[]) AS s("integrationId", "keywordId")
+          WHERE s."integrationId" = m."integrationId" AND s."keywordId" = m."keywordId"
+        )
+    `,
+    {
+      platform: PlatformType.OCTOLENS,
+      integrationIds: seen.map((s) => s.integrationId),
+      keywordIds: seen.map((s) => s.keywordId),
+    },
+  )
+}
+
+export async function softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings(
+  qx: QueryExecutor,
+): Promise<string[]> {
+  return qx.tx(async (tx) => {
+    await lockOctolensIntegrations(tx)
+
+    const rows: { id: string }[] = await tx.select(
+      `
+        UPDATE public.integrations i
+        SET "deletedAt" = now(), "updatedAt" = now()
+        WHERE i.platform = $(platform)
+          AND i."deletedAt" IS NULL
+          AND i."segmentId" IS NOT NULL
+          AND NOT EXISTS (
+            SELECT 1
+            FROM integration."octolensKeywordMappings" m
+            WHERE m."integrationId" = i.id AND m."deactivatedAt" IS NULL
+          )
+        RETURNING i.id
+      `,
+      { platform: PlatformType.OCTOLENS },
+    )
+    return rows.map((r) => r.id)
+  })
+}

From fba0ea4cdfad1d8c8be38dc82b2d994f0f115ceb Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Uro=C5=A1=20Marolt?= <uros@marolt.me>
Date: Thu, 15 Oct 2026 18:33:02 +0200
Subject: [PATCH 2/9] fix: keep auto-managed octolens integrations out of
 global counts
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Octolens integrations are created by workers per subproject, so the
not-connected list and count no longer offer octolens for every other
subproject. The status count now joins segments like the list query, so
the segment-less anchor integration is not counted as done.

Refs linuxfoundation/lfx-cdp-insights-ops#748

Signed-off-by: Uroš Marolt <uros@marolt.me>
---
 .../integrations/globalIntegrations.test.ts   | 88 +++++++++++++++++++
 .../src/integrations/index.ts                 | 14 +--
 services/libs/types/src/enums/platforms.ts    |  3 +
 3 files changed, 100 insertions(+), 5 deletions(-)
 create mode 100644 services/libs/data-access-layer/src/integrations/globalIntegrations.test.ts

diff --git a/services/libs/data-access-layer/src/integrations/globalIntegrations.test.ts b/services/libs/data-access-layer/src/integrations/globalIntegrations.test.ts
new file mode 100644
index 0000000000..e06cef5064
--- /dev/null
+++ b/services/libs/data-access-layer/src/integrations/globalIntegrations.test.ts
@@ -0,0 +1,88 @@
+import { test as base, describe, expect } from 'vitest'
+
+import { DEFAULT_TENANT_ID, generateUUIDv1 } from '@crowd/common'
+import type { QueryExecutor } from '@crowd/database'
+import { withQx } from '@crowd/test-kit/db'
+import { createSegments } from '@crowd/test-kit/factories'
+
+import {
+  fetchGlobalIntegrations,
+  fetchGlobalIntegrationsCount,
+  fetchGlobalIntegrationsStatusCount,
+  fetchGlobalNotConnectedIntegrations,
+  fetchGlobalNotConnectedIntegrationsCount,
+} from './index'
+
+const test = withQx(base)
+
+async function createIntegration(
+  qx: QueryExecutor,
+  platform: string,
+  segmentId: string | null,
+): Promise<string> {
+  const id = generateUUIDv1()
+  await qx.result(
+    `
+    INSERT INTO public.integrations (id, platform, status, "tenantId", "segmentId", "createdAt", "updatedAt")
+    VALUES ($(id), $(platform), 'done', $(tenantId), $(segmentId), NOW(), NOW())
+    `,
+    { id, platform, tenantId: DEFAULT_TENANT_ID, segmentId },
+  )
+  return id
+}
+
+async function createTree(qx: QueryExecutor) {
+  const node = () => ({ name: generateUUIDv1(), slug: generateUUIDv1() })
+  const [group] = (
+    await createSegments(qx, [
+      { ...node(), projects: [{ ...node(), subprojects: [node(), node()] }] },
+    ])
+  ).projectGroups
+  const [connected, unconnected] = group.projects[0].subprojects
+  return { groupId: group.row.id, connectedId: connected.id, unconnectedId: unconnected.id }
+}
+
+describe('global integrations with auto-managed octolens integrations', () => {
+  test('not-connected list and count never report octolens', async ({ qx }) => {
+    const { groupId, connectedId, unconnectedId } = await createTree(qx)
+    await createIntegration(qx, 'github', connectedId)
+    await createIntegration(qx, 'octolens', connectedId)
+    await createIntegration(qx, 'octolens', null)
+
+    const rows = await fetchGlobalNotConnectedIntegrations(qx, null, '', 1000, 0, groupId)
+    const [{ count }] = await fetchGlobalNotConnectedIntegrationsCount(qx, null, '', groupId)
+
+    expect(rows.map((r) => r.platform)).not.toContain('octolens')
+    expect(rows).toContainEqual(
+      expect.objectContaining({ platform: 'github', segmentId: unconnectedId }),
+    )
+    expect(Number(count)).toBe(rows.length)
+  })
+
+  test('status count matches the list and ignores the segment-less anchor', async ({ qx }) => {
+    const { groupId, connectedId } = await createTree(qx)
+    const subprojectIntegration = await createIntegration(qx, 'octolens', connectedId)
+    const anchor = await createIntegration(qx, 'octolens', null)
+
+    const listed = await fetchGlobalIntegrations(qx, ['done'], 'octolens', '', 1000, 0, null)
+    const [{ count: listCount }] = await fetchGlobalIntegrationsCount(
+      qx,
+      ['done'],
+      'octolens',
+      '',
+      null,
+    )
+    const statusCounts = await fetchGlobalIntegrationsStatusCount(qx, 'octolens', null)
+    const doneCount = Number(statusCounts.find((s) => s.status === 'done')?.count ?? 0)
+
+    const listedIds = listed.map((i) => i.id)
+    expect(listedIds).toContain(subprojectIntegration)
+    expect(listedIds).not.toContain(anchor)
+    expect(Number(listCount)).toBe(listed.length)
+    expect(doneCount).toBe(listed.length)
+
+    const scoped = await fetchGlobalIntegrationsStatusCount(qx, 'octolens', groupId)
+    expect(scoped).toEqual([{ status: 'done', count: expect.anything() }])
+    expect(Number(scoped[0].count)).toBe(1)
+  })
+})
diff --git a/services/libs/data-access-layer/src/integrations/index.ts b/services/libs/data-access-layer/src/integrations/index.ts
index 83ac131c98..3b2b3c018b 100644
--- a/services/libs/data-access-layer/src/integrations/index.ts
+++ b/services/libs/data-access-layer/src/integrations/index.ts
@@ -1,5 +1,5 @@
 import { getServiceChildLogger } from '@crowd/logging'
-import { IIntegration, PlatformType } from '@crowd/types'
+import { AUTO_MANAGED_INTEGRATION_PLATFORMS, IIntegration, PlatformType } from '@crowd/types'
 
 import { QueryExecutor } from '../queryExecutor'
 import { getReposForGithubIntegration } from '../repositories'
@@ -171,7 +171,8 @@ export async function fetchGlobalNotConnectedIntegrations(
   return qx.select(
     `
       WITH unique_platforms AS (SELECT DISTINCT platform
-                                FROM public.integrations),
+                                FROM public.integrations
+                                WHERE platform <> ALL ($(autoManaged)::text[])),
            connected_platforms AS (SELECT i.platform, s.id as "segmentId"
                                    FROM integrations i
                                           JOIN "segments" s ON i."segmentId" = s.id
@@ -201,6 +202,7 @@ export async function fetchGlobalNotConnectedIntegrations(
       query: `%${query}%`,
       limit,
       offset,
+      autoManaged: AUTO_MANAGED_INTEGRATION_PLATFORMS,
     },
   )
 }
@@ -222,7 +224,8 @@ export async function fetchGlobalNotConnectedIntegrationsCount(
   return qx.select(
     `
       WITH unique_platforms AS (SELECT DISTINCT platform
-                                FROM public.integrations),
+                                FROM public.integrations
+                                WHERE platform <> ALL ($(autoManaged)::text[])),
            connected_platforms AS (SELECT i.platform, s.id as "segmentId"
                                    FROM integrations i
                                           JOIN "segments" s ON i."segmentId" = s.id
@@ -243,6 +246,7 @@ export async function fetchGlobalNotConnectedIntegrationsCount(
       platform,
       segmentId,
       query: `%${query}%`,
+      autoManaged: AUTO_MANAGED_INTEGRATION_PLATFORMS,
     },
   )
 }
@@ -264,10 +268,10 @@ export async function fetchGlobalIntegrationsStatusCount(
         SELECT i.status,
                COUNT(*) AS count
         FROM "integrations" i
+               JOIN segments s ON i."segmentId" = s.id
         WHERE i."deletedAt" IS NULL
           AND ($(platform) IS NULL OR i."platform" = $(platform))
-          AND ($(segmentId) IS NULL OR i."segmentId" = $(segmentId) OR 
-               EXISTS (SELECT 1 FROM segments s WHERE s.id = i."segmentId" AND (s."parentId" = $(segmentId) OR s."grandparentId" = $(segmentId))))
+          AND ($(segmentId) IS NULL OR s.id = $(segmentId) OR s."parentId" = $(segmentId) OR s."grandparentId" = $(segmentId))
         GROUP BY i.status
     `,
     {
diff --git a/services/libs/types/src/enums/platforms.ts b/services/libs/types/src/enums/platforms.ts
index dc936f3dc1..06962c5fed 100644
--- a/services/libs/types/src/enums/platforms.ts
+++ b/services/libs/types/src/enums/platforms.ts
@@ -44,6 +44,9 @@ export type CodePlatform =
 
 export const ALL_PLATFORM_TYPES = Object.keys(PlatformType) as PlatformType[]
 
+// Integrations of these platforms are created by workers, never connected by a user.
+export const AUTO_MANAGED_INTEGRATION_PLATFORMS: PlatformType[] = [PlatformType.OCTOLENS]
+
 export enum IntegrationType {
   DEVTO = 'devto',
   SLACK = 'slack',

From f19a34d6011255a1ebda3679cfb0a421b730fc9f Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Uro=C5=A1=20Marolt?= <uros@marolt.me>
Date: Thu, 15 Oct 2026 18:34:17 +0200
Subject: [PATCH 3/9] refactor: drop redundant comment on auto-managed platform
 list
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 <uros@marolt.me>
---
 services/libs/types/src/enums/platforms.ts | 1 -
 1 file changed, 1 deletion(-)

diff --git a/services/libs/types/src/enums/platforms.ts b/services/libs/types/src/enums/platforms.ts
index 06962c5fed..27b98ee51c 100644
--- a/services/libs/types/src/enums/platforms.ts
+++ b/services/libs/types/src/enums/platforms.ts
@@ -44,7 +44,6 @@ export type CodePlatform =
 
 export const ALL_PLATFORM_TYPES = Object.keys(PlatformType) as PlatformType[]
 
-// Integrations of these platforms are created by workers, never connected by a user.
 export const AUTO_MANAGED_INTEGRATION_PLATFORMS: PlatformType[] = [PlatformType.OCTOLENS]
 
 export enum IntegrationType {

From 6b1bf48606c5c6b0661132108108308e6ca1f472 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Uro=C5=A1=20Marolt?= <uros@marolt.me>
Date: Thu, 15 Oct 2026 18:50:06 +0200
Subject: [PATCH 4/9] fix: harden octolens integration lifecycle queries
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 <uros@marolt.me>
---
 .../src/octolens/index.test.ts                | 74 +++++++++++++++++--
 .../data-access-layer/src/octolens/index.ts   | 63 +++++++---------
 2 files changed, 94 insertions(+), 43 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 8b0f3607fa..748e18ecb5 100644
--- a/services/libs/data-access-layer/src/octolens/index.test.ts
+++ b/services/libs/data-access-layer/src/octolens/index.test.ts
@@ -387,18 +387,14 @@ describe('octolens subproject integrations', () => {
   }) => {
     const segmentId = await createSegment(qx)
     const subproject = await findOrCreateOctolensSubprojectIntegration(qx, segmentId)
-    await createPlatformIntegration(qx, { segmentId: null, platform: 'github' })
-    const deletedAnchor = await createPlatformIntegration(qx, {
-      segmentId: null,
-      deletedAt: '2026-01-01T00:00:00Z',
-    })
+    const github = await createPlatformIntegration(qx, { segmentId: null, platform: 'github' })
 
     const first = await findOrCreateOctolensAnchorIntegration(qx)
     const second = await findOrCreateOctolensAnchorIntegration(qx)
 
-    expect(first.created).toBe(true)
-    expect(second).toEqual({ id: first.id, created: false })
-    expect([subproject.id, deletedAnchor]).not.toContain(first.id)
+    expect(first).toMatchObject({ created: true, restored: false })
+    expect(second).toEqual({ id: first.id, created: false, restored: false })
+    expect([subproject.id, github]).not.toContain(first.id)
     expect(await readIntegration(qx, first.id)).toMatchObject({
       platform: 'octolens',
       segmentId: null,
@@ -406,6 +402,29 @@ describe('octolens subproject integrations', () => {
     })
   })
 
+  test('findOrCreateOctolensAnchorIntegration restores the most recently updated soft-deleted anchor', async ({
+    qx,
+  }) => {
+    const segmentId = await createSegment(qx)
+    await createPlatformIntegration(qx, { segmentId, deletedAt: '2026-03-01T00:00:00Z' })
+    const older = await createPlatformIntegration(qx, {
+      segmentId: null,
+      deletedAt: '2026-01-01T00:00:00Z',
+      updatedAt: '2026-01-01T00:00:00Z',
+    })
+    const newer = await createPlatformIntegration(qx, {
+      segmentId: null,
+      deletedAt: '2026-02-01T00:00:00Z',
+      updatedAt: '2026-02-01T00:00:00Z',
+    })
+
+    const result = await findOrCreateOctolensAnchorIntegration(qx)
+
+    expect(result).toEqual({ id: newer, created: false, restored: true })
+    expect((await readIntegration(qx, newer)).deletedAt).toBeNull()
+    expect((await readIntegration(qx, older)).deletedAt).not.toBeNull()
+  })
+
   test('findOrCreateOctolensAnchorIntegration throws on two live anchors', async ({ qx }) => {
     await createPlatformIntegration(qx, { segmentId: null })
     await createPlatformIntegration(qx, { segmentId: null })
@@ -556,4 +575,43 @@ describe('octolens subproject integrations', () => {
     expect(restored).toEqual({ id: empty, created: false, restored: true })
     expect((await readIntegration(qx, empty)).deletedAt).toBeNull()
   })
+
+  test('integrations owning an active sync unit are never treated as subproject integrations', async ({
+    qx,
+  }) => {
+    const segmentId = await createSegment(qx)
+    const owner = await createPlatformIntegration(qx, { segmentId })
+    await qx.result(
+      `
+      INSERT INTO integration.sync_units
+        ("integrationId", platform, "channelId", "channelName", "syncName", status)
+      VALUES ($(owner), 'octolens', 'octolens', 'octolens', 'mentions', 'active')
+      `,
+      { owner },
+    )
+    for (const keywordId of [1, 2]) {
+      await upsertOctolensKeywordMapping(qx, {
+        integrationId: owner,
+        segmentId,
+        keywordId,
+        keyword: `kw-${keywordId}`,
+      })
+    }
+    await deactivateOctolensKeywordMappingsNotSeen(qx, owner, [2])
+    const other = await createPlatformIntegration(qx, { segmentId: await createSegment(qx) })
+
+    const listed = await listActiveOctolensKeywordMappings(qx)
+    await deactivateOctolensKeywordMappingsNotSeenAcross(qx, [
+      { integrationId: other, keywordId: 1 },
+    ])
+
+    expect(listed.map((m) => m.integrationId)).not.toContain(owner)
+    expect(await findOctolensKeywordMapping(qx, owner, 2)).not.toBeNull()
+
+    await deactivateOctolensKeywordMappingsNotSeen(qx, owner, [99])
+    const deleted = await softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings(qx)
+
+    expect(deleted).not.toContain(owner)
+    expect((await readIntegration(qx, owner)).deletedAt).toBeNull()
+  })
 })
diff --git a/services/libs/data-access-layer/src/octolens/index.ts b/services/libs/data-access-layer/src/octolens/index.ts
index 32ab352c76..696375bdb1 100644
--- a/services/libs/data-access-layer/src/octolens/index.ts
+++ b/services/libs/data-access-layer/src/octolens/index.ts
@@ -99,17 +99,12 @@ export async function deactivateOctolensKeywordMappingsNotSeen(
   )
 }
 
-export interface IOctolensSubprojectIntegration {
+export interface IFoundOctolensIntegration {
   id: string
   created: boolean
   restored: boolean
 }
 
-export interface IOctolensAnchorIntegration {
-  id: string
-  created: boolean
-}
-
 export interface IActiveOctolensKeywordMapping {
   keywordId: number
   keyword: string
@@ -122,10 +117,15 @@ export interface IOctolensKeywordMappingRef {
   keywordId: number
 }
 
+// An integration owning active sync units (the anchor, or a legacy one) is never a subproject integration.
+const OWNS_NO_ACTIVE_SYNC_UNITS = `NOT EXISTS (
+  SELECT 1 FROM integration.sync_units su WHERE su."integrationId" = i.id AND su.status = 'active'
+)`
+
 // integrations has no unique key on (platform, segmentId), so every writer of
 // Octolens integration rows serializes on this lock to avoid duplicates.
 async function lockOctolensIntegrations(tx: QueryExecutor): Promise<void> {
-  await tx.select(`SELECT pg_advisory_xact_lock(hashtext('octolens-integration')::bigint)`)
+  await tx.select(`SELECT pg_advisory_xact_lock(hashtextextended('octolensIntegration', 0))`)
 }
 
 async function insertOctolensIntegration(
@@ -150,10 +150,10 @@ async function insertOctolensIntegration(
   return id
 }
 
-export async function findOrCreateOctolensSubprojectIntegration(
+async function findOrCreateOctolensIntegration(
   qx: QueryExecutor,
-  segmentId: string,
-): Promise<IOctolensSubprojectIntegration> {
+  segmentId: string | null,
+): Promise<IFoundOctolensIntegration> {
   return qx.tx(async (tx) => {
     await lockOctolensIntegrations(tx)
 
@@ -161,7 +161,7 @@ export async function findOrCreateOctolensSubprojectIntegration(
       `
         SELECT id, "deletedAt"
         FROM public.integrations
-        WHERE platform = $(platform) AND "segmentId" = $(segmentId)
+        WHERE platform = $(platform) AND "segmentId" IS NOT DISTINCT FROM $(segmentId)::uuid
         ORDER BY "updatedAt" DESC, "createdAt" DESC, id
       `,
       { platform: PlatformType.OCTOLENS, segmentId },
@@ -169,7 +169,9 @@ export async function findOrCreateOctolensSubprojectIntegration(
 
     const live = rows.filter((r) => r.deletedAt === null)
     if (live.length > 1) {
-      throw new Error(`Found ${live.length} live Octolens integrations for segment ${segmentId}!`)
+      throw new Error(
+        `Found ${live.length} live Octolens integrations for segment ${segmentId ?? 'NULL'}!`,
+      )
     }
     if (live.length === 1) {
       return { id: live[0].id, created: false, restored: false }
@@ -193,31 +195,17 @@ export async function findOrCreateOctolensSubprojectIntegration(
   })
 }
 
-export async function findOrCreateOctolensAnchorIntegration(
+export async function findOrCreateOctolensSubprojectIntegration(
   qx: QueryExecutor,
-): Promise<IOctolensAnchorIntegration> {
-  return qx.tx(async (tx) => {
-    await lockOctolensIntegrations(tx)
-
-    const rows: { id: string }[] = await tx.select(
-      `
-        SELECT id
-        FROM public.integrations
-        WHERE platform = $(platform) AND "segmentId" IS NULL AND "deletedAt" IS NULL
-      `,
-      { platform: PlatformType.OCTOLENS },
-    )
-
-    if (rows.length > 1) {
-      throw new Error(`Found ${rows.length} live Octolens anchor integrations!`)
-    }
-    if (rows.length === 1) {
-      return { id: rows[0].id, created: false }
-    }
+  segmentId: string,
+): Promise<IFoundOctolensIntegration> {
+  return findOrCreateOctolensIntegration(qx, segmentId)
+}
 
-    const id = await insertOctolensIntegration(tx, null)
-    return { id, created: true }
-  })
+export async function findOrCreateOctolensAnchorIntegration(
+  qx: QueryExecutor,
+): Promise<IFoundOctolensIntegration> {
+  return findOrCreateOctolensIntegration(qx, null)
 }
 
 export async function listActiveOctolensKeywordMappings(
@@ -232,6 +220,7 @@ export async function listActiveOctolensKeywordMappings(
         AND i.platform = $(platform)
         AND i."deletedAt" IS NULL
         AND i."segmentId" IS NOT NULL
+        AND ${OWNS_NO_ACTIVE_SYNC_UNITS}
       ORDER BY m."keywordId", m."integrationId"
     `,
     { platform: PlatformType.OCTOLENS },
@@ -255,6 +244,7 @@ export async function deactivateOctolensKeywordMappingsNotSeenAcross(
       WHERE i.id = m."integrationId"
         AND i.platform = $(platform)
         AND i."deletedAt" IS NULL
+        AND ${OWNS_NO_ACTIVE_SYNC_UNITS}
         AND m."deactivatedAt" IS NULL
         AND NOT EXISTS (
           SELECT 1
@@ -270,6 +260,8 @@ export async function deactivateOctolensKeywordMappingsNotSeenAcross(
   )
 }
 
+// Run only after the same discovery run's mapping upserts and deactivation, and never
+// concurrently with another discovery run, or it deletes integrations that are still mapped.
 export async function softDeleteOctolensSubprojectIntegrationsWithoutActiveMappings(
   qx: QueryExecutor,
 ): Promise<string[]> {
@@ -283,6 +275,7 @@ export async function softDeleteOctolensSubprojectIntegrationsWithoutActiveMappi
         WHERE i.platform = $(platform)
           AND i."deletedAt" IS NULL
           AND i."segmentId" IS NOT NULL
+          AND ${OWNS_NO_ACTIVE_SYNC_UNITS}
           AND NOT EXISTS (
             SELECT 1
             FROM integration."octolensKeywordMappings" m

From 24ce5eeddea473b0bdb2ccbe0b35d9aae4d17d88 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Uro=C5=A1=20Marolt?= <uros@marolt.me>
Date: Thu, 15 Oct 2026 19:16:22 +0200
Subject: [PATCH 5/9] fix: limit the status count change to the octolens anchor
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 <uros@marolt.me>
---
 .../integrations/globalIntegrations.test.ts   | 41 +++++++++++--------
 .../src/integrations/index.ts                 |  6 ++-
 2 files changed, 29 insertions(+), 18 deletions(-)

diff --git a/services/libs/data-access-layer/src/integrations/globalIntegrations.test.ts b/services/libs/data-access-layer/src/integrations/globalIntegrations.test.ts
index e06cef5064..16010cc543 100644
--- a/services/libs/data-access-layer/src/integrations/globalIntegrations.test.ts
+++ b/services/libs/data-access-layer/src/integrations/globalIntegrations.test.ts
@@ -59,30 +59,39 @@ describe('global integrations with auto-managed octolens integrations', () => {
     expect(Number(count)).toBe(rows.length)
   })
 
-  test('status count matches the list and ignores the segment-less anchor', async ({ qx }) => {
+  test('status count skips only the segment-less octolens anchor', async ({ qx }) => {
     const { groupId, connectedId } = await createTree(qx)
+    const doneCount = async (platform: string) => {
+      const counts = await fetchGlobalIntegrationsStatusCount(qx, platform, null)
+      return Number(counts.find((s) => s.status === 'done')?.count ?? 0)
+    }
+    const githubBefore = await doneCount('github')
+    const octolensBefore = await doneCount('octolens')
+
+    await createIntegration(qx, 'github', null)
+    await createIntegration(qx, 'github', connectedId)
+    await createIntegration(qx, 'octolens', null)
+    await createIntegration(qx, 'octolens', connectedId)
+
+    expect(await doneCount('github')).toBe(githubBefore + 2)
+    expect(await doneCount('octolens')).toBe(octolensBefore + 1)
+
+    const scoped = await fetchGlobalIntegrationsStatusCount(qx, 'octolens', groupId)
+    expect(scoped).toEqual([{ status: 'done', count: expect.anything() }])
+    expect(Number(scoped[0].count)).toBe(1)
+  })
+
+  test('list and count never include the segment-less anchor', async ({ qx }) => {
+    const { connectedId } = await createTree(qx)
     const subprojectIntegration = await createIntegration(qx, 'octolens', connectedId)
     const anchor = await createIntegration(qx, 'octolens', null)
 
     const listed = await fetchGlobalIntegrations(qx, ['done'], 'octolens', '', 1000, 0, null)
-    const [{ count: listCount }] = await fetchGlobalIntegrationsCount(
-      qx,
-      ['done'],
-      'octolens',
-      '',
-      null,
-    )
-    const statusCounts = await fetchGlobalIntegrationsStatusCount(qx, 'octolens', null)
-    const doneCount = Number(statusCounts.find((s) => s.status === 'done')?.count ?? 0)
+    const [{ count }] = await fetchGlobalIntegrationsCount(qx, ['done'], 'octolens', '', null)
 
     const listedIds = listed.map((i) => i.id)
     expect(listedIds).toContain(subprojectIntegration)
     expect(listedIds).not.toContain(anchor)
-    expect(Number(listCount)).toBe(listed.length)
-    expect(doneCount).toBe(listed.length)
-
-    const scoped = await fetchGlobalIntegrationsStatusCount(qx, 'octolens', groupId)
-    expect(scoped).toEqual([{ status: 'done', count: expect.anything() }])
-    expect(Number(scoped[0].count)).toBe(1)
+    expect(Number(count)).toBe(listed.length)
   })
 })
diff --git a/services/libs/data-access-layer/src/integrations/index.ts b/services/libs/data-access-layer/src/integrations/index.ts
index 3b2b3c018b..f6b30c3c97 100644
--- a/services/libs/data-access-layer/src/integrations/index.ts
+++ b/services/libs/data-access-layer/src/integrations/index.ts
@@ -268,15 +268,17 @@ export async function fetchGlobalIntegrationsStatusCount(
         SELECT i.status,
                COUNT(*) AS count
         FROM "integrations" i
-               JOIN segments s ON i."segmentId" = s.id
         WHERE i."deletedAt" IS NULL
           AND ($(platform) IS NULL OR i."platform" = $(platform))
-          AND ($(segmentId) IS NULL OR s.id = $(segmentId) OR s."parentId" = $(segmentId) OR s."grandparentId" = $(segmentId))
+          AND ($(segmentId) IS NULL OR i."segmentId" = $(segmentId) OR
+               EXISTS (SELECT 1 FROM segments s WHERE s.id = i."segmentId" AND (s."parentId" = $(segmentId) OR s."grandparentId" = $(segmentId))))
+          AND NOT (i.platform = ANY ($(autoManaged)::text[]) AND i."segmentId" IS NULL)
         GROUP BY i.status
     `,
     {
       platform,
       segmentId,
+      autoManaged: AUTO_MANAGED_INTEGRATION_PLATFORMS,
     },
   )
 }

From 74f4bbf713e20358f74b92f2ff9eb624744bf317 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Uro=C5=A1=20Marolt?= <uros@marolt.me>
Date: Thu, 15 Oct 2026 19:18:26 +0200
Subject: [PATCH 6/9] fix: refuse manual create, update and delete of
 auto-managed integrations
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 <uros@marolt.me>
---
 .../api/integration/autoManagedGuard.test.ts  | 116 ++++++++++++++++++
 .../src/api/integration/autoManagedGuard.ts   |  26 ++++
 .../src/api/integration/integrationCreate.ts  |   3 +
 .../src/api/integration/integrationDestroy.ts |   3 +
 .../src/api/integration/integrationUpdate.ts  |   4 +
 services/libs/common/src/i18n/en.ts           |   2 +
 .../integrations/globalIntegrations.test.ts   |  18 +++
 .../src/integrations/index.ts                 |  15 +++
 8 files changed, 187 insertions(+)
 create mode 100644 backend/src/api/integration/autoManagedGuard.test.ts
 create mode 100644 backend/src/api/integration/autoManagedGuard.ts

diff --git a/backend/src/api/integration/autoManagedGuard.test.ts b/backend/src/api/integration/autoManagedGuard.test.ts
new file mode 100644
index 0000000000..1f1f37b115
--- /dev/null
+++ b/backend/src/api/integration/autoManagedGuard.test.ts
@@ -0,0 +1,116 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+import { Error400 } from '@crowd/common'
+
+const service = vi.hoisted(() => ({
+  create: vi.fn(),
+  update: vi.fn(),
+  destroyAll: vi.fn(),
+}))
+const findIntegrationPlatformsByIds = vi.hoisted(() => vi.fn())
+
+vi.mock('../../services/integrationService', () => ({
+  default: class {
+    create = service.create
+    update = service.update
+    destroyAll = service.destroyAll
+  },
+}))
+
+vi.mock('../../services/user/permissionChecker', () => ({
+  default: class {
+    validateHas = vi.fn()
+  },
+}))
+
+vi.mock('@/database/repositories/sequelizeRepository', () => ({
+  default: { getQueryExecutor: vi.fn(() => ({})) },
+}))
+
+vi.mock('@crowd/data-access-layer/src/integrations', () => ({ findIntegrationPlatformsByIds }))
+
+import integrationCreate from './integrationCreate'
+import integrationDestroy from './integrationDestroy'
+import integrationUpdate from './integrationUpdate'
+
+const GITHUB_ID = '6f1c2a5e-8f4b-4d3a-9b2c-1a2b3c4d5e6f'
+const OCTOLENS_ID = '0c9d8e7f-6a5b-4c3d-8e2f-9a8b7c6d5e4f'
+const platformsById: Record<string, string> = { [GITHUB_ID]: 'github', [OCTOLENS_ID]: 'octolens' }
+
+async function call(
+  handler: (req: unknown, res: unknown) => Promise<void>,
+  req: Record<string, unknown>,
+) {
+  const success = vi.fn()
+  await handler({ language: 'en', responseHandler: { success }, ...req }, {})
+  return success
+}
+
+describe('auto-managed integration guard', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    findIntegrationPlatformsByIds.mockImplementation(async (_qx, ids: string[]) =>
+      ids.map((id) => platformsById[id]).filter(Boolean),
+    )
+  })
+
+  it.each([
+    ['another platform', { platform: 'github' }],
+    ['no platform', {}],
+    ['no body', undefined],
+  ])('creates as before for %s', async (_name, body) => {
+    const success = await call(integrationCreate, { body })
+
+    expect(service.create).toHaveBeenCalledWith(body)
+    expect(success).toHaveBeenCalled()
+  })
+
+  it('refuses to create an octolens integration', async () => {
+    await expect(call(integrationCreate, { body: { platform: 'octolens' } })).rejects.toThrow(
+      /managed automatically/,
+    )
+    expect(service.create).not.toHaveBeenCalled()
+  })
+
+  it.each([
+    ['another platform', GITHUB_ID],
+    ['an unknown id', '11111111-2222-4333-8444-555555555555'],
+    ['a malformed id', 'not-a-uuid'],
+  ])('updates as before for %s', async (_name, id) => {
+    await call(integrationUpdate, { params: { id }, body: { status: 'done' } })
+
+    expect(service.update).toHaveBeenCalledWith(id, { status: 'done' })
+  })
+
+  it('refuses to update an octolens integration', async () => {
+    await expect(
+      call(integrationUpdate, { params: { id: OCTOLENS_ID }, body: { status: 'done' } }),
+    ).rejects.toBeInstanceOf(Error400)
+    expect(service.update).not.toHaveBeenCalled()
+  })
+
+  it('refuses to change an integration platform to octolens', async () => {
+    await expect(
+      call(integrationUpdate, { params: { id: GITHUB_ID }, body: { platform: 'octolens' } }),
+    ).rejects.toBeInstanceOf(Error400)
+    expect(service.update).not.toHaveBeenCalled()
+  })
+
+  it.each([
+    ['other platforms', [GITHUB_ID]],
+    ['malformed ids', ['not-a-uuid']],
+    ['no ids', undefined],
+  ])('destroys as before for %s', async (_name, ids) => {
+    await call(integrationDestroy, { query: { ids } })
+
+    expect(service.destroyAll).toHaveBeenCalledWith(ids)
+  })
+
+  it.each([
+    ['one of several ids', [GITHUB_ID, OCTOLENS_ID]],
+    ['a single id', OCTOLENS_ID],
+  ])('refuses to destroy when %s is an octolens integration', async (_name, ids) => {
+    await expect(call(integrationDestroy, { query: { ids } })).rejects.toBeInstanceOf(Error400)
+    expect(service.destroyAll).not.toHaveBeenCalled()
+  })
+})
diff --git a/backend/src/api/integration/autoManagedGuard.ts b/backend/src/api/integration/autoManagedGuard.ts
new file mode 100644
index 0000000000..cce483e6f7
--- /dev/null
+++ b/backend/src/api/integration/autoManagedGuard.ts
@@ -0,0 +1,26 @@
+import SequelizeRepository from '@/database/repositories/sequelizeRepository'
+import { Error400, validateUUID } from '@crowd/common'
+import { findIntegrationPlatformsByIds } from '@crowd/data-access-layer/src/integrations'
+import { AUTO_MANAGED_INTEGRATION_PLATFORMS } from '@crowd/types'
+
+export function isAutoManagedPlatform(platform: unknown): boolean {
+  return AUTO_MANAGED_INTEGRATION_PLATFORMS.some((p) => p === platform)
+}
+
+export function assertNoAutoManagedPlatform(language: string, platforms: unknown[]): void {
+  if (platforms.some(isAutoManagedPlatform)) {
+    throw new Error400(language, 'errors.integrations.autoManaged')
+  }
+}
+
+// Malformed or unknown ids are skipped, so requests for other platforms fail exactly as before.
+export async function assertNoAutoManagedIntegration(req, ids: unknown): Promise<void> {
+  const uuids = [ids]
+    .flat()
+    .filter((id): id is string => typeof id === 'string' && validateUUID(id))
+  const platforms = await findIntegrationPlatformsByIds(
+    SequelizeRepository.getQueryExecutor(req),
+    uuids,
+  )
+  assertNoAutoManagedPlatform(req.language, platforms)
+}
diff --git a/backend/src/api/integration/integrationCreate.ts b/backend/src/api/integration/integrationCreate.ts
index dfe0ce7b6b..33b77c1b7f 100644
--- a/backend/src/api/integration/integrationCreate.ts
+++ b/backend/src/api/integration/integrationCreate.ts
@@ -1,10 +1,13 @@
 import Permissions from '../../security/permissions'
 import IntegrationService from '../../services/integrationService'
 import PermissionChecker from '../../services/user/permissionChecker'
+import { assertNoAutoManagedPlatform } from './autoManagedGuard'
 
 export default async (req, res) => {
   new PermissionChecker(req).validateHas(Permissions.values.integrationCreate)
 
+  assertNoAutoManagedPlatform(req.language, [req.body?.platform])
+
   const payload = await new IntegrationService(req).create(req.body)
 
   await req.responseHandler.success(req, res, payload)
diff --git a/backend/src/api/integration/integrationDestroy.ts b/backend/src/api/integration/integrationDestroy.ts
index d28076e986..a86895eb1c 100644
--- a/backend/src/api/integration/integrationDestroy.ts
+++ b/backend/src/api/integration/integrationDestroy.ts
@@ -1,10 +1,13 @@
 import Permissions from '../../security/permissions'
 import IntegrationService from '../../services/integrationService'
 import PermissionChecker from '../../services/user/permissionChecker'
+import { assertNoAutoManagedIntegration } from './autoManagedGuard'
 
 export default async (req, res) => {
   new PermissionChecker(req).validateHas(Permissions.values.integrationDestroy)
 
+  await assertNoAutoManagedIntegration(req, req.query.ids)
+
   await new IntegrationService(req).destroyAll(req.query.ids)
 
   const payload = true
diff --git a/backend/src/api/integration/integrationUpdate.ts b/backend/src/api/integration/integrationUpdate.ts
index 621aec29fc..c76c731640 100644
--- a/backend/src/api/integration/integrationUpdate.ts
+++ b/backend/src/api/integration/integrationUpdate.ts
@@ -1,10 +1,14 @@
 import Permissions from '../../security/permissions'
 import IntegrationService from '../../services/integrationService'
 import PermissionChecker from '../../services/user/permissionChecker'
+import { assertNoAutoManagedIntegration, assertNoAutoManagedPlatform } from './autoManagedGuard'
 
 export default async (req, res) => {
   new PermissionChecker(req).validateHas(Permissions.values.integrationEdit)
 
+  assertNoAutoManagedPlatform(req.language, [req.body?.platform])
+  await assertNoAutoManagedIntegration(req, req.params.id)
+
   const payload = await new IntegrationService(req).update(req.params.id, req.body)
 
   await req.responseHandler.success(req, res, payload)
diff --git a/services/libs/common/src/i18n/en.ts b/services/libs/common/src/i18n/en.ts
index ba22fc4ab6..926ac89b95 100644
--- a/services/libs/common/src/i18n/en.ts
+++ b/services/libs/common/src/i18n/en.ts
@@ -137,6 +137,8 @@ const en = {
       badEndpoint: 'Bad endpoint: {0}',
       repoAlreadyMapped:
         'Trying to update repo {0} mapping with integrationId {1} but it is already mapped to integration {2}!',
+      autoManaged:
+        "Octolens integrations are managed automatically and can't be connected, changed or disconnected manually",
     },
     sentiment: {
       mood: 'Invalid sentiment data. The {1} property must exist and be a number.',
diff --git a/services/libs/data-access-layer/src/integrations/globalIntegrations.test.ts b/services/libs/data-access-layer/src/integrations/globalIntegrations.test.ts
index 16010cc543..7a36ab3c62 100644
--- a/services/libs/data-access-layer/src/integrations/globalIntegrations.test.ts
+++ b/services/libs/data-access-layer/src/integrations/globalIntegrations.test.ts
@@ -11,6 +11,7 @@ import {
   fetchGlobalIntegrationsStatusCount,
   fetchGlobalNotConnectedIntegrations,
   fetchGlobalNotConnectedIntegrationsCount,
+  findIntegrationPlatformsByIds,
 } from './index'
 
 const test = withQx(base)
@@ -95,3 +96,20 @@ describe('global integrations with auto-managed octolens integrations', () => {
     expect(Number(count)).toBe(listed.length)
   })
 })
+
+describe('findIntegrationPlatformsByIds', () => {
+  test('returns the distinct platforms of the given ids only', async ({ qx }) => {
+    const { connectedId } = await createTree(qx)
+    const github = await createIntegration(qx, 'github', connectedId)
+    const octolens = await createIntegration(qx, 'octolens', null)
+    await createIntegration(qx, 'gitlab', connectedId)
+
+    expect(await findIntegrationPlatformsByIds(qx, [])).toEqual([])
+    expect(await findIntegrationPlatformsByIds(qx, [github, github])).toEqual(['github'])
+    expect((await findIntegrationPlatformsByIds(qx, [github, octolens])).sort()).toEqual([
+      'github',
+      'octolens',
+    ])
+    expect(await findIntegrationPlatformsByIds(qx, [generateUUIDv1()])).toEqual([])
+  })
+})
diff --git a/services/libs/data-access-layer/src/integrations/index.ts b/services/libs/data-access-layer/src/integrations/index.ts
index f6b30c3c97..2bcde0a256 100644
--- a/services/libs/data-access-layer/src/integrations/index.ts
+++ b/services/libs/data-access-layer/src/integrations/index.ts
@@ -291,6 +291,21 @@ export interface INangoIntegrationData {
   createdAt: string
 }
 
+export async function findIntegrationPlatformsByIds(
+  qx: QueryExecutor,
+  ids: string[],
+): Promise<string[]> {
+  if (ids.length === 0) {
+    return []
+  }
+
+  const rows: { platform: string }[] = await qx.select(
+    `SELECT DISTINCT platform FROM integrations WHERE id = ANY ($(ids)::uuid[])`,
+    { ids },
+  )
+  return rows.map((r) => r.platform)
+}
+
 export async function fetchIntegrationById(
   qx: QueryExecutor,
   id: string,

From 6441beb2457e0bdab1c5db38ad8ca8ff115f4013 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Uro=C5=A1=20Marolt?= <uros@marolt.me>
Date: Thu, 15 Oct 2026 19:19:58 +0200
Subject: [PATCH 7/9] fix: keep null-platform integrations in global
 integration counts
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

integrations.platform is nullable; the auto-managed exclusions now treat
a NULL platform as not auto-managed, as main does.

Refs linuxfoundation/lfx-cdp-insights-ops#748

Signed-off-by: Uroš Marolt <uros@marolt.me>
---
 .../src/integrations/globalIntegrations.test.ts       | 11 +++++++++--
 .../libs/data-access-layer/src/integrations/index.ts  |  6 +++---
 2 files changed, 12 insertions(+), 5 deletions(-)

diff --git a/services/libs/data-access-layer/src/integrations/globalIntegrations.test.ts b/services/libs/data-access-layer/src/integrations/globalIntegrations.test.ts
index 7a36ab3c62..198185131a 100644
--- a/services/libs/data-access-layer/src/integrations/globalIntegrations.test.ts
+++ b/services/libs/data-access-layer/src/integrations/globalIntegrations.test.ts
@@ -18,7 +18,7 @@ const test = withQx(base)
 
 async function createIntegration(
   qx: QueryExecutor,
-  platform: string,
+  platform: string | null,
   segmentId: string | null,
 ): Promise<string> {
   const id = generateUUIDv1()
@@ -49,6 +49,7 @@ describe('global integrations with auto-managed octolens integrations', () => {
     await createIntegration(qx, 'github', connectedId)
     await createIntegration(qx, 'octolens', connectedId)
     await createIntegration(qx, 'octolens', null)
+    await createIntegration(qx, null, null)
 
     const rows = await fetchGlobalNotConnectedIntegrations(qx, null, '', 1000, 0, groupId)
     const [{ count }] = await fetchGlobalNotConnectedIntegrationsCount(qx, null, '', groupId)
@@ -57,15 +58,19 @@ describe('global integrations with auto-managed octolens integrations', () => {
     expect(rows).toContainEqual(
       expect.objectContaining({ platform: 'github', segmentId: unconnectedId }),
     )
+    expect(rows).toContainEqual(
+      expect.objectContaining({ platform: null, segmentId: unconnectedId }),
+    )
     expect(Number(count)).toBe(rows.length)
   })
 
   test('status count skips only the segment-less octolens anchor', async ({ qx }) => {
     const { groupId, connectedId } = await createTree(qx)
-    const doneCount = async (platform: string) => {
+    const doneCount = async (platform: string | null) => {
       const counts = await fetchGlobalIntegrationsStatusCount(qx, platform, null)
       return Number(counts.find((s) => s.status === 'done')?.count ?? 0)
     }
+    const allBefore = await doneCount(null)
     const githubBefore = await doneCount('github')
     const octolensBefore = await doneCount('octolens')
 
@@ -73,9 +78,11 @@ describe('global integrations with auto-managed octolens integrations', () => {
     await createIntegration(qx, 'github', connectedId)
     await createIntegration(qx, 'octolens', null)
     await createIntegration(qx, 'octolens', connectedId)
+    await createIntegration(qx, null, null)
 
     expect(await doneCount('github')).toBe(githubBefore + 2)
     expect(await doneCount('octolens')).toBe(octolensBefore + 1)
+    expect(await doneCount(null)).toBe(allBefore + 4)
 
     const scoped = await fetchGlobalIntegrationsStatusCount(qx, 'octolens', groupId)
     expect(scoped).toEqual([{ status: 'done', count: expect.anything() }])
diff --git a/services/libs/data-access-layer/src/integrations/index.ts b/services/libs/data-access-layer/src/integrations/index.ts
index 2bcde0a256..5cc6810a04 100644
--- a/services/libs/data-access-layer/src/integrations/index.ts
+++ b/services/libs/data-access-layer/src/integrations/index.ts
@@ -172,7 +172,7 @@ export async function fetchGlobalNotConnectedIntegrations(
     `
       WITH unique_platforms AS (SELECT DISTINCT platform
                                 FROM public.integrations
-                                WHERE platform <> ALL ($(autoManaged)::text[])),
+                                WHERE NOT COALESCE(platform = ANY ($(autoManaged)::text[]), false)),
            connected_platforms AS (SELECT i.platform, s.id as "segmentId"
                                    FROM integrations i
                                           JOIN "segments" s ON i."segmentId" = s.id
@@ -225,7 +225,7 @@ export async function fetchGlobalNotConnectedIntegrationsCount(
     `
       WITH unique_platforms AS (SELECT DISTINCT platform
                                 FROM public.integrations
-                                WHERE platform <> ALL ($(autoManaged)::text[])),
+                                WHERE NOT COALESCE(platform = ANY ($(autoManaged)::text[]), false)),
            connected_platforms AS (SELECT i.platform, s.id as "segmentId"
                                    FROM integrations i
                                           JOIN "segments" s ON i."segmentId" = s.id
@@ -272,7 +272,7 @@ export async function fetchGlobalIntegrationsStatusCount(
           AND ($(platform) IS NULL OR i."platform" = $(platform))
           AND ($(segmentId) IS NULL OR i."segmentId" = $(segmentId) OR
                EXISTS (SELECT 1 FROM segments s WHERE s.id = i."segmentId" AND (s."parentId" = $(segmentId) OR s."grandparentId" = $(segmentId))))
-          AND NOT (i.platform = ANY ($(autoManaged)::text[]) AND i."segmentId" IS NULL)
+          AND NOT (COALESCE(i.platform = ANY ($(autoManaged)::text[]), false) AND i."segmentId" IS NULL)
         GROUP BY i.status
     `,
     {

From 7914bdb6dce2d35954911012036d78a508047496 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Uro=C5=A1=20Marolt?= <uros@marolt.me>
Date: Thu, 15 Oct 2026 20:52:09 +0200
Subject: [PATCH 8/9] fix: address copilot review on octolens integration
 access
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 <uros@marolt.me>
---
 .../api/integration/autoManagedGuard.test.ts  | 16 ++++++-
 .../src/api/integration/autoManagedGuard.ts   | 27 ++++++++++-
 .../src/connectors/syncUnits.test.ts          | 17 ++++++-
 .../src/octolens/index.test.ts                | 45 +++++++++++++++----
 .../data-access-layer/src/octolens/index.ts   | 11 +++--
 5 files changed, 99 insertions(+), 17 deletions(-)

diff --git a/backend/src/api/integration/autoManagedGuard.test.ts b/backend/src/api/integration/autoManagedGuard.test.ts
index 1f1f37b115..2ced7d1d8d 100644
--- a/backend/src/api/integration/autoManagedGuard.test.ts
+++ b/backend/src/api/integration/autoManagedGuard.test.ts
@@ -37,6 +37,9 @@ const GITHUB_ID = '6f1c2a5e-8f4b-4d3a-9b2c-1a2b3c4d5e6f'
 const OCTOLENS_ID = '0c9d8e7f-6a5b-4c3d-8e2f-9a8b7c6d5e4f'
 const platformsById: Record<string, string> = { [GITHUB_ID]: 'github', [OCTOLENS_ID]: 'octolens' }
 
+const noHyphens = (id: string) => id.replace(/-/g, '')
+const braced = (id: string) => `{${id.toUpperCase()}}`
+
 async function call(
   handler: (req: unknown, res: unknown) => Promise<void>,
   req: Record<string, unknown>,
@@ -74,6 +77,8 @@ describe('auto-managed integration guard', () => {
 
   it.each([
     ['another platform', GITHUB_ID],
+    ['another platform without hyphens', noHyphens(GITHUB_ID)],
+    ['another platform in braces', braced(GITHUB_ID)],
     ['an unknown id', '11111111-2222-4333-8444-555555555555'],
     ['a malformed id', 'not-a-uuid'],
   ])('updates as before for %s', async (_name, id) => {
@@ -82,9 +87,13 @@ describe('auto-managed integration guard', () => {
     expect(service.update).toHaveBeenCalledWith(id, { status: 'done' })
   })
 
-  it('refuses to update an octolens integration', async () => {
+  it.each([
+    ['canonical', OCTOLENS_ID],
+    ['without hyphens', noHyphens(OCTOLENS_ID)],
+    ['in braces', braced(OCTOLENS_ID)],
+  ])('refuses to update an octolens integration given a %s id', async (_name, id) => {
     await expect(
-      call(integrationUpdate, { params: { id: OCTOLENS_ID }, body: { status: 'done' } }),
+      call(integrationUpdate, { params: { id }, body: { status: 'done' } }),
     ).rejects.toBeInstanceOf(Error400)
     expect(service.update).not.toHaveBeenCalled()
   })
@@ -98,6 +107,7 @@ describe('auto-managed integration guard', () => {
 
   it.each([
     ['other platforms', [GITHUB_ID]],
+    ['other platforms without hyphens or in braces', [noHyphens(GITHUB_ID), braced(GITHUB_ID)]],
     ['malformed ids', ['not-a-uuid']],
     ['no ids', undefined],
   ])('destroys as before for %s', async (_name, ids) => {
@@ -109,6 +119,8 @@ describe('auto-managed integration guard', () => {
   it.each([
     ['one of several ids', [GITHUB_ID, OCTOLENS_ID]],
     ['a single id', OCTOLENS_ID],
+    ['an id without hyphens', [noHyphens(OCTOLENS_ID)]],
+    ['an id in braces', [braced(OCTOLENS_ID)]],
   ])('refuses to destroy when %s is an octolens integration', async (_name, ids) => {
     await expect(call(integrationDestroy, { query: { ids } })).rejects.toBeInstanceOf(Error400)
     expect(service.destroyAll).not.toHaveBeenCalled()
diff --git a/backend/src/api/integration/autoManagedGuard.ts b/backend/src/api/integration/autoManagedGuard.ts
index cce483e6f7..0dcfee9a35 100644
--- a/backend/src/api/integration/autoManagedGuard.ts
+++ b/backend/src/api/integration/autoManagedGuard.ts
@@ -1,5 +1,5 @@
 import SequelizeRepository from '@/database/repositories/sequelizeRepository'
-import { Error400, validateUUID } from '@crowd/common'
+import { Error400 } from '@crowd/common'
 import { findIntegrationPlatformsByIds } from '@crowd/data-access-layer/src/integrations'
 import { AUTO_MANAGED_INTEGRATION_PLATFORMS } from '@crowd/types'
 
@@ -13,11 +13,34 @@ export function assertNoAutoManagedPlatform(language: string, platforms: unknown
   }
 }
 
+// Postgres also accepts uuids without hyphens or wrapped in braces, so normalize before the lookup.
+export function toCanonicalUuid(id: unknown): string | null {
+  if (typeof id !== 'string') {
+    return null
+  }
+  const hex = id
+    .trim()
+    .replace(/^\{(.*)\}$/, '$1')
+    .replace(/-/g, '')
+    .toLowerCase()
+  if (!/^[0-9a-f]{32}$/.test(hex)) {
+    return null
+  }
+  return [
+    hex.slice(0, 8),
+    hex.slice(8, 12),
+    hex.slice(12, 16),
+    hex.slice(16, 20),
+    hex.slice(20),
+  ].join('-')
+}
+
 // Malformed or unknown ids are skipped, so requests for other platforms fail exactly as before.
 export async function assertNoAutoManagedIntegration(req, ids: unknown): Promise<void> {
   const uuids = [ids]
     .flat()
-    .filter((id): id is string => typeof id === 'string' && validateUUID(id))
+    .map(toCanonicalUuid)
+    .filter((id): id is string => id !== null)
   const platforms = await findIntegrationPlatformsByIds(
     SequelizeRepository.getQueryExecutor(req),
     uuids,
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 11d3ef0b04..3b9b9acd9e 100644
--- a/services/libs/data-access-layer/src/connectors/syncUnits.test.ts
+++ b/services/libs/data-access-layer/src/connectors/syncUnits.test.ts
@@ -183,7 +183,7 @@ describe('upsertSyncUnits', () => {
     expect(forks.delaySeconds).toBeGreaterThanOrEqual(9)
   })
 
-  test('re-running leaves emitEnabled, nextRunAt, watermark and failures untouched', async ({
+  test('re-running without a watermark leaves emitEnabled, nextRunAt, watermark and failures untouched', async ({
     qx,
   }) => {
     const integrationId = await createIntegration(qx)
@@ -218,4 +218,19 @@ describe('upsertSyncUnits', () => {
     })
     expect(new Date(unit.nextRunAt).toISOString()).toBe('2030-01-01T00:00:00.000Z')
   })
+
+  test('re-running with a watermark replaces the stored watermark', async ({ qx }) => {
+    const integrationId = await createIntegration(qx)
+    await upsertSyncUnits(qx, [
+      { ...base, integrationId, syncName: 'issues', watermark: { phase: 'backfill' } },
+    ])
+
+    await upsertSyncUnits(qx, [
+      { ...base, integrationId, syncName: 'issues', watermark: { phase: 'incremental' } },
+    ])
+
+    expect((await readUnit(qx, integrationId, 'issues')).watermark).toEqual({
+      phase: 'incremental',
+    })
+  })
 })
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 748e18ecb5..c3eb95fb6c 100644
--- a/services/libs/data-access-layer/src/octolens/index.test.ts
+++ b/services/libs/data-access-layer/src/octolens/index.test.ts
@@ -497,12 +497,14 @@ describe('octolens subproject integrations', () => {
       segmentId: segmentA,
       platform: 'github',
     })
+    const anchor = await createPlatformIntegration(qx, { segmentId: null })
 
     for (const [integrationId, segmentId, keywordId] of [
       [integrationA, segmentA, 1],
       [integrationA, segmentA, 2],
       [integrationB, segmentB, 1],
       [githubIntegration, segmentA, 9],
+      [anchor, segmentA, 7],
     ] as const) {
       await upsertOctolensKeywordMapping(qx, {
         integrationId,
@@ -522,6 +524,7 @@ describe('octolens subproject integrations', () => {
     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()
   })
 
   test('deactivateOctolensKeywordMappingsNotSeenAcross with nothing seen deactivates nothing', async ({
@@ -579,16 +582,42 @@ describe('octolens subproject integrations', () => {
   test('integrations owning an active sync unit are never treated as subproject integrations', async ({
     qx,
   }) => {
+    const createActiveSyncUnit = (integrationId: string) =>
+      qx.result(
+        `
+        INSERT INTO integration.sync_units
+          ("integrationId", platform, "channelId", "channelName", "syncName", status)
+        VALUES ($(integrationId), 'octolens', 'octolens', 'octolens', 'mentions', 'active')
+        `,
+        { integrationId },
+      )
     const segmentId = await createSegment(qx)
     const owner = await createPlatformIntegration(qx, { segmentId })
-    await qx.result(
-      `
-      INSERT INTO integration.sync_units
-        ("integrationId", platform, "channelId", "channelName", "syncName", status)
-      VALUES ($(owner), 'octolens', 'octolens', 'octolens', 'mentions', 'active')
-      `,
-      { owner },
-    )
+    await createActiveSyncUnit(owner)
+
+    const found = await findOrCreateOctolensSubprojectIntegration(qx, segmentId)
+    expect(found.created).toBe(true)
+    expect(found.id).not.toBe(owner)
+
+    const deletedSegmentId = await createSegment(qx)
+    const deletedOwner = await createPlatformIntegration(qx, {
+      segmentId: deletedSegmentId,
+      deletedAt: '2026-01-01T00:00:00Z',
+    })
+    await createActiveSyncUnit(deletedOwner)
+    const notRestored = await findOrCreateOctolensSubprojectIntegration(qx, deletedSegmentId)
+    expect(notRestored.created).toBe(true)
+    expect(notRestored.id).not.toBe(deletedOwner)
+    expect((await readIntegration(qx, deletedOwner)).deletedAt).not.toBeNull()
+
+    const anchor = await createPlatformIntegration(qx, { segmentId: null })
+    await createActiveSyncUnit(anchor)
+    expect(await findOrCreateOctolensAnchorIntegration(qx)).toEqual({
+      id: anchor,
+      created: false,
+      restored: false,
+    })
+
     for (const keywordId of [1, 2]) {
       await upsertOctolensKeywordMapping(qx, {
         integrationId: owner,
diff --git a/services/libs/data-access-layer/src/octolens/index.ts b/services/libs/data-access-layer/src/octolens/index.ts
index 696375bdb1..1032da3d6c 100644
--- a/services/libs/data-access-layer/src/octolens/index.ts
+++ b/services/libs/data-access-layer/src/octolens/index.ts
@@ -159,10 +159,12 @@ async function findOrCreateOctolensIntegration(
 
     const rows: { id: string; deletedAt: Date | null }[] = await tx.select(
       `
-        SELECT id, "deletedAt"
-        FROM public.integrations
-        WHERE platform = $(platform) AND "segmentId" IS NOT DISTINCT FROM $(segmentId)::uuid
-        ORDER BY "updatedAt" DESC, "createdAt" DESC, id
+        SELECT i.id, i."deletedAt"
+        FROM public.integrations i
+        WHERE i.platform = $(platform)
+          AND i."segmentId" IS NOT DISTINCT FROM $(segmentId)::uuid
+          AND ($(segmentId)::uuid IS NULL OR ${OWNS_NO_ACTIVE_SYNC_UNITS})
+        ORDER BY i."updatedAt" DESC, i."createdAt" DESC, i.id
       `,
       { platform: PlatformType.OCTOLENS, segmentId },
     )
@@ -244,6 +246,7 @@ export async function deactivateOctolensKeywordMappingsNotSeenAcross(
       WHERE i.id = m."integrationId"
         AND i.platform = $(platform)
         AND i."deletedAt" IS NULL
+        AND i."segmentId" IS NOT NULL
         AND ${OWNS_NO_ACTIVE_SYNC_UNITS}
         AND m."deactivatedAt" IS NULL
         AND NOT EXISTS (

From 692c8d914a8b714d40ce2895b21878eb7d4c3eb4 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Uro=C5=A1=20Marolt?= <uros@marolt.me>
Date: Thu, 15 Oct 2026 20:59:44 +0200
Subject: [PATCH 9/9] fix: check nested ids in the auto-managed integration
 guard
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 <uros@marolt.me>
---
 .../src/api/integration/autoManagedGuard.test.ts   |  5 +++++
 backend/src/api/integration/autoManagedGuard.ts    | 14 ++++++++++++--
 2 files changed, 17 insertions(+), 2 deletions(-)

diff --git a/backend/src/api/integration/autoManagedGuard.test.ts b/backend/src/api/integration/autoManagedGuard.test.ts
index 2ced7d1d8d..fe712a33da 100644
--- a/backend/src/api/integration/autoManagedGuard.test.ts
+++ b/backend/src/api/integration/autoManagedGuard.test.ts
@@ -110,6 +110,7 @@ describe('auto-managed integration guard', () => {
     ['other platforms without hyphens or in braces', [noHyphens(GITHUB_ID), braced(GITHUB_ID)]],
     ['malformed ids', ['not-a-uuid']],
     ['no ids', undefined],
+    ['other platforms in nested arrays', [[GITHUB_ID]]],
   ])('destroys as before for %s', async (_name, ids) => {
     await call(integrationDestroy, { query: { ids } })
 
@@ -121,6 +122,10 @@ describe('auto-managed integration guard', () => {
     ['a single id', OCTOLENS_ID],
     ['an id without hyphens', [noHyphens(OCTOLENS_ID)]],
     ['an id in braces', [braced(OCTOLENS_ID)]],
+    ['an id in a nested array', [[OCTOLENS_ID]]],
+    ['an id deeply nested in arrays', [GITHUB_ID, [[OCTOLENS_ID]]]],
+    ['an id nested in an object', { a: OCTOLENS_ID }],
+    ['an id nested in an array of objects', [{ x: [OCTOLENS_ID] }]],
   ])('refuses to destroy when %s is an octolens integration', async (_name, ids) => {
     await expect(call(integrationDestroy, { query: { ids } })).rejects.toBeInstanceOf(Error400)
     expect(service.destroyAll).not.toHaveBeenCalled()
diff --git a/backend/src/api/integration/autoManagedGuard.ts b/backend/src/api/integration/autoManagedGuard.ts
index 0dcfee9a35..5207330304 100644
--- a/backend/src/api/integration/autoManagedGuard.ts
+++ b/backend/src/api/integration/autoManagedGuard.ts
@@ -35,10 +35,20 @@ export function toCanonicalUuid(id: unknown): string | null {
   ].join('-')
 }
 
+// Query parsing can nest ids in arrays or objects (e.g. ids[0][0]=…), so collect every leaf.
+export function collectIdLeaves(ids: unknown): unknown[] {
+  if (Array.isArray(ids)) {
+    return ids.flatMap(collectIdLeaves)
+  }
+  if (ids !== null && typeof ids === 'object') {
+    return Object.values(ids).flatMap(collectIdLeaves)
+  }
+  return [ids]
+}
+
 // Malformed or unknown ids are skipped, so requests for other platforms fail exactly as before.
 export async function assertNoAutoManagedIntegration(req, ids: unknown): Promise<void> {
-  const uuids = [ids]
-    .flat()
+  const uuids = collectIdLeaves(ids)
     .map(toCanonicalUuid)
     .filter((id): id is string => id !== null)
   const platforms = await findIntegrationPlatformsByIds(