feat: run the proposed action in the automatic onboarding worker (CM-1841)

services/apps/automatic_onboarding_worker/src/activities/activities.ts

@@ -17,7 +17,8 @@ import { IPipelineRunFinish } from '@crowd/data-access-layer/src/project-catalog
 import { IDbProjectCatalog } from '@crowd/data-access-layer/src/project-catalog/types'
 import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor'
 import { getServiceLogger } from '@crowd/logging'
-import { deriveProjectSlug, onboardProject } from '@crowd/project-onboarding'
+import { connectRepoToSegment, deriveProjectSlug, onboardProject } from '@crowd/project-onboarding'
+import { createCdpSegmentLookup } from '@crowd/project-onboarding/src/cdpSegmentLookup'
 import { SlackChannel, SlackPersona, sendSlackNotificationAsync } from '@crowd/slack'
 
 import { svc } from '../main'
@@ -30,6 +31,7 @@ import {
   isReviewAlertRequest,
 } from './discussionRequestAlert'
 import { buildInsightsProjectSkipReason } from './insightsProjectSkip'
+import { runProposedActionSerialized } from './proposedAction'
 
 const log = getServiceLogger()
 
@@ -113,14 +115,16 @@ export async function onboardAndUpdateProject(
     return resolveUnclaimedOutcome(qx, project)
   }
 
-  log.info({ id: project.id, repoUrl: project.repoUrl }, 'Starting onboarding.')
+  log.info(
+    { id: project.id, repoUrl: project.repoUrl, proposedAction: project.proposedAction },
+    'Starting onboarding.',
+  )
 
-  const result = await onboardProject({
-    id: project.id,
-    repoUrl: project.repoUrl,
-    repoName: project.repoName,
-    projectSlug: project.projectSlug,
-  }).catch(async (err) => {
+  const result = await runProposedActionSerialized(project, qx, (depsQx) => ({
+    findCdpSegmentByPccProject: createCdpSegmentLookup(depsQx),
+    onboardProject,
+    connectRepoToSegment,
+  })).catch(async (err) => {
     await releaseProjectCatalogOnboardingClaim(qx, project.id)
     throw err
   })

services/apps/automatic_onboarding_worker/src/activities/proposedAction.test.ts

@@ -0,0 +1,183 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+vi.mock('@crowd/data-access-layer', () => ({ lockProjectCatalogOnboardingTarget: vi.fn() }))
+
+import { lockProjectCatalogOnboardingTarget } from '@crowd/data-access-layer'
+
+import {
+  IProposedActionDeps,
+  runProposedAction,
+  runProposedActionSerialized,
+} from './proposedAction'
+
+const onboarded = { outcome: 'onboarded' as const, segmentId: 'seg', error: null }
+
+const deps: IProposedActionDeps = {
+  findCdpSegmentByPccProject: vi.fn(),
+  onboardProject: vi.fn(async () => onboarded),
+  connectRepoToSegment: vi.fn(async () => onboarded),
+}
+
+const project = {
+  id: 'row-1',
+  repoUrl: 'https://github.com/kubernetes/kops',
+  repoName: 'kops',
+  projectSlug: 'kubernetes-kops',
+  proposedAction: null,
+  pccProjectId: null,
+}
+
+describe('runProposedAction', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    vi.mocked(deps.findCdpSegmentByPccProject).mockResolvedValue({
+      segmentId: 'seg',
+      name: 'Kubernetes',
+      integration: 'none',
+    })
+  })
+
+  it.each([null, 'create_segment' as const])(
+    'creates a non-LF segment and integration for proposedAction %s',
+    async (proposedAction) => {
+      const result = await runProposedAction({ ...project, proposedAction }, deps)
+
+      expect(result).toEqual(onboarded)
+      expect(deps.onboardProject).toHaveBeenCalledWith({
+        id: 'row-1',
+        repoUrl: project.repoUrl,
+        repoName: 'kops',
+        projectSlug: 'kubernetes-kops',
+      })
+      expect(deps.connectRepoToSegment).not.toHaveBeenCalled()
+    },
+  )
+
+  it.each(['create_integration' as const, 'update_integration' as const])(
+    'connects the repo to the CDP segment of the PCC project for proposedAction %s',
+    async (proposedAction) => {
+      const result = await runProposedAction(
+        { ...project, proposedAction, pccProjectId: 'pcc-1' },
+        deps,
+      )
+
+      expect(result).toEqual(onboarded)
+      expect(deps.findCdpSegmentByPccProject).toHaveBeenCalledWith('pcc-1')
+      expect(deps.connectRepoToSegment).toHaveBeenCalledWith({
+        segmentId: 'seg',
+        repoUrl: project.repoUrl,
+      })
+      expect(deps.onboardProject).not.toHaveBeenCalled()
+    },
+  )
+
+  it('fails when an LF action has no PCC project id', async () => {
+    const result = await runProposedAction(
+      { ...project, proposedAction: 'create_integration' },
+      deps,
+    )
+
+    expect(result.outcome).toBe('error')
+    expect(result.error).toContain('requires a PCC project id')
+    expect(deps.connectRepoToSegment).not.toHaveBeenCalled()
+  })
+
+  it('fails when the PCC project has no CDP segment', async () => {
+    vi.mocked(deps.findCdpSegmentByPccProject).mockResolvedValue(null)
+
+    const result = await runProposedAction(
+      { ...project, proposedAction: 'create_integration', pccProjectId: 'pcc-1' },
+      deps,
+    )
+
+    expect(result.outcome).toBe('error')
+    expect(result.error).toContain('No CDP segment found')
+  })
+
+  it('fails when several CDP segments share the PCC project', async () => {
+    vi.mocked(deps.findCdpSegmentByPccProject).mockResolvedValue({
+      duplicates: [
+        { segmentId: 'a', name: 'A', integration: 'none' },
+        { segmentId: 'b', name: 'B', integration: 'none' },
+      ],
+    })
+
+    const result = await runProposedAction(
+      { ...project, proposedAction: 'update_integration', pccProjectId: 'pcc-1' },
+      deps,
+    )
+
+    expect(result.outcome).toBe('error')
+    expect(result.error).toContain('Several CDP segments')
+  })
+
+  it('fails when the segment has a legacy GitHub integration', async () => {
+    vi.mocked(deps.findCdpSegmentByPccProject).mockResolvedValue({
+      segmentId: 'seg',
+      name: 'Kubernetes',
+      integration: 'github-v1',
+    })
+
+    const result = await runProposedAction(
+      { ...project, proposedAction: 'update_integration', pccProjectId: 'pcc-1' },
+      deps,
+    )
+
+    expect(result.outcome).toBe('error')
+    expect(result.error).toContain('legacy GitHub integration')
+    expect(deps.connectRepoToSegment).not.toHaveBeenCalled()
+  })
+
+  it('never onboards a human_review row', async () => {
+    const result = await runProposedAction({ ...project, proposedAction: 'human_review' }, deps)
+
+    expect(result.outcome).toBe('error')
+    expect(deps.onboardProject).not.toHaveBeenCalled()
+    expect(deps.connectRepoToSegment).not.toHaveBeenCalled()
+  })
+})
+
+describe('runProposedActionSerialized', () => {
+  const txQx = { marker: 'tx' } as never
+  const qx = { tx: vi.fn(async (fn: (tx: unknown) => Promise<unknown>) => fn(txQx)) } as never
+  const buildDeps = vi.fn(() => deps)
+
+  beforeEach(() => {
+    vi.clearAllMocks()
+    vi.mocked(deps.findCdpSegmentByPccProject).mockResolvedValue({
+      segmentId: 'seg',
+      name: 'Kubernetes',
+      integration: 'none',
+    })
+  })
+
+  it.each(['create_integration' as const, 'update_integration' as const])(
+    'locks the PCC project inside a transaction for proposedAction %s',
+    async (proposedAction) => {
+      const result = await runProposedActionSerialized(
+        { ...project, proposedAction, pccProjectId: 'pcc-1' },
+        qx,
+        buildDeps,
+      )
+
+      expect(result).toEqual(onboarded)
+      expect(lockProjectCatalogOnboardingTarget).toHaveBeenCalledWith(txQx, 'pcc-1')
+      expect(lockProjectCatalogOnboardingTarget).toHaveBeenCalledBefore(
+        vi.mocked(deps.connectRepoToSegment),
+      )
+      expect(buildDeps).toHaveBeenCalledWith(txQx)
+    },
+  )
+
+  it('runs a segment creation without any lock', async () => {
+    const result = await runProposedActionSerialized(
+      { ...project, proposedAction: 'create_segment' },
+      qx,
+      buildDeps,
+    )
+
+    expect(result).toEqual(onboarded)
+    expect(lockProjectCatalogOnboardingTarget).not.toHaveBeenCalled()
+    expect(buildDeps).toHaveBeenCalledWith(qx)
+  })
+})

services/apps/automatic_onboarding_worker/src/activities/proposedAction.ts

@@ -0,0 +1,99 @@
+import { lockProjectCatalogOnboardingTarget } from '@crowd/data-access-layer'
+import { IDbProjectCatalog } from '@crowd/data-access-layer/src/project-catalog/types'
+import { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
+import {
+  CdpSegmentLookupResult,
+  IOnboardingInput,
+  IOnboardingResult,
+  ISegmentOnboardingInput,
+} from '@crowd/project-onboarding'
+
+export interface IProposedActionDeps {
+  findCdpSegmentByPccProject: (pccProjectId: string) => Promise<CdpSegmentLookupResult>
+  onboardProject: (input: IOnboardingInput) => Promise<IOnboardingResult>
+  connectRepoToSegment: (input: ISegmentOnboardingInput) => Promise<IOnboardingResult>
+}
+
+type ProposedActionProject = Pick<
+  IDbProjectCatalog,
+  'id' | 'repoUrl' | 'repoName' | 'projectSlug' | 'proposedAction' | 'pccProjectId'
+>
+
+function errorResult(error: string): IOnboardingResult {
+  return { outcome: 'error', segmentId: null, error }
+}
+
+async function resolveLfSegmentId(
+  project: ProposedActionProject,
+  deps: IProposedActionDeps,
+): Promise<{ segmentId: string } | { error: string }> {
+  if (!project.pccProjectId) {
+    return { error: `Proposed action ${project.proposedAction} requires a PCC project id` }
+  }
+
+  const segment = await deps.findCdpSegmentByPccProject(project.pccProjectId)
+  if (!segment) {
+    return { error: `No CDP segment found for PCC project ${project.pccProjectId}` }
+  }
+  if ('duplicates' in segment) {
+    return { error: `Several CDP segments share PCC project ${project.pccProjectId}` }
+  }
+  if (segment.integration === 'github-v1') {
+    return {
+      error: `Segment ${segment.segmentId} has a legacy GitHub integration, needs human review`,
+    }
+  }
+
+  return { segmentId: segment.segmentId }
+}
+
+export async function runProposedAction(
+  project: ProposedActionProject,
+  deps: IProposedActionDeps,
+): Promise<IOnboardingResult> {
+  switch (project.proposedAction) {
+    case null:
+    case 'create_segment':
+      return deps.onboardProject({
+        id: project.id,
+        repoUrl: project.repoUrl,
+        repoName: project.repoName,
+        projectSlug: project.projectSlug,
+      })
+    case 'create_integration':
+    case 'update_integration': {
+      const resolved = await resolveLfSegmentId(project, deps)
+      if ('error' in resolved) {
+        return errorResult(resolved.error)
+      }
+      return deps.connectRepoToSegment({ segmentId: resolved.segmentId, repoUrl: project.repoUrl })
+    }
+    case 'human_review':
+      return errorResult('Proposed action human_review cannot be onboarded automatically')
+  }
+}
+
+function writesSharedSegment(project: ProposedActionProject): project is ProposedActionProject & {
+  pccProjectId: string
+} {
+  return (
+    project.pccProjectId !== null &&
+    (project.proposedAction === 'create_integration' ||
+      project.proposedAction === 'update_integration')
+  )
+}
+
+export async function runProposedActionSerialized(
+  project: ProposedActionProject,
+  qx: QueryExecutor,
+  buildDeps: (qx: QueryExecutor) => IProposedActionDeps,
+): Promise<IOnboardingResult> {
+  if (!writesSharedSegment(project)) {
+    return runProposedAction(project, buildDeps(qx))
+  }
+
+  return qx.tx(async (tx) => {
+    await lockProjectCatalogOnboardingTarget(tx, project.pccProjectId)
+    return runProposedAction(project, buildDeps(tx))
+  })
+}

services/libs/data-access-layer/src/project-catalog/projectCatalog.ts

@@ -687,6 +687,18 @@ export async function claimProjectCatalogForSlackEvaluation(
   )
 }
 
+// Held until the transaction ends; serializes the read-modify-write of one segment's GitHub
+// integration across the scheduled and the bot-started onboarding workflows.
+export async function lockProjectCatalogOnboardingTarget(
+  qx: QueryExecutor,
+  pccProjectId: string,
+): Promise<void> {
+  await qx.result(
+    `SELECT pg_advisory_xact_lock(hashtextextended('projectCatalogOnboarding:' || $(pccProjectId)::text, 0))`,
+    { pccProjectId },
+  )
+}
+
 // Claims a row queued for onboarding by stamping onboardedAt before the external call —
 // the nightly worker's onboardedAt-truthy guard then treats it as already handled.
 export async function claimProjectCatalogForOnboarding(