feat: add data access for per-subproject Octolens integrations

backend/src/api/integration/autoManagedGuard.test.ts

@@ -0,0 +1,133 @@
+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' }
+
+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>,
+) {
+  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],
+    ['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) => {
+    await call(integrationUpdate, { params: { id }, body: { status: 'done' } })
+
+    expect(service.update).toHaveBeenCalledWith(id, { status: 'done' })
+  })
+
+  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 }, 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]],
+    ['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 } })
+
+    expect(service.destroyAll).toHaveBeenCalledWith(ids)
+  })
+
+  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)]],
+    ['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()
+  })
+})

backend/src/api/integration/autoManagedGuard.ts

@@ -0,0 +1,59 @@
+import SequelizeRepository from '@/database/repositories/sequelizeRepository'
+import { Error400 } 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')
+  }
+}
+
+// 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('-')
+}
+
+// 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 = collectIdLeaves(ids)
+    .map(toCanonicalUuid)
+    .filter((id): id is string => id !== null)
+  const platforms = await findIntegrationPlatformsByIds(
+    SequelizeRepository.getQueryExecutor(req),
+    uuids,
+  )
+  assertNoAutoManagedPlatform(req.language, platforms)
+}

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)

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

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)

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.',

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,110 @@ 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 without a watermark 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')
+  })
+
+  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',
+    })
+  })
+})