From f87e0f201a6e2b746518df8d4fd1d0a9c40e2ea5 Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Mon, 12 Oct 2026 19:02:58 +0200 Subject: [PATCH 1/3] feat: run the proposed action in the automatic onboarding worker (CM-1841) Signed-off-by: Umberto Sgueglia --- .../src/activities/activities.ts | 18 +- .../src/activities/proposedAction.test.ts | 130 +++++++++ .../src/activities/proposedAction.ts | 72 +++++ .../project-onboarding/src/onboarder.test.ts | 208 ++++++++++++++- .../libs/project-onboarding/src/onboarder.ts | 246 ++++++++++++++---- services/libs/project-onboarding/src/types.ts | 26 ++ 6 files changed, 636 insertions(+), 64 deletions(-) create mode 100644 services/apps/automatic_onboarding_worker/src/activities/proposedAction.test.ts create mode 100644 services/apps/automatic_onboarding_worker/src/activities/proposedAction.ts diff --git a/services/apps/automatic_onboarding_worker/src/activities/activities.ts b/services/apps/automatic_onboarding_worker/src/activities/activities.ts index 0db600bdfc..946d268312 100644 --- a/services/apps/automatic_onboarding_worker/src/activities/activities.ts +++ b/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 { runProposedAction } from './proposedAction' const log = getServiceLogger() @@ -113,13 +115,15 @@ 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, + const result = await runProposedAction(project, { + findCdpSegmentByPccProject: createCdpSegmentLookup(qx), + onboardProject, + connectRepoToSegment, }).catch(async (err) => { await releaseProjectCatalogOnboardingClaim(qx, project.id) throw err diff --git a/services/apps/automatic_onboarding_worker/src/activities/proposedAction.test.ts b/services/apps/automatic_onboarding_worker/src/activities/proposedAction.test.ts new file mode 100644 index 0000000000..f67ca7bce5 --- /dev/null +++ b/services/apps/automatic_onboarding_worker/src/activities/proposedAction.test.ts @@ -0,0 +1,130 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +import { IProposedActionDeps, runProposedAction } 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() + }) +}) diff --git a/services/apps/automatic_onboarding_worker/src/activities/proposedAction.ts b/services/apps/automatic_onboarding_worker/src/activities/proposedAction.ts new file mode 100644 index 0000000000..360cabfbec --- /dev/null +++ b/services/apps/automatic_onboarding_worker/src/activities/proposedAction.ts @@ -0,0 +1,72 @@ +import { IDbProjectCatalog } from '@crowd/data-access-layer/src/project-catalog/types' +import { + CdpSegmentLookupResult, + IOnboardingInput, + IOnboardingResult, + ISegmentOnboardingInput, +} from '@crowd/project-onboarding' + +export interface IProposedActionDeps { + findCdpSegmentByPccProject: (pccProjectId: string) => Promise + onboardProject: (input: IOnboardingInput) => Promise + connectRepoToSegment: (input: ISegmentOnboardingInput) => Promise +} + +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 { + 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') + } +} diff --git a/services/libs/project-onboarding/src/onboarder.test.ts b/services/libs/project-onboarding/src/onboarder.test.ts index 56946c5d61..63754c4b98 100644 --- a/services/libs/project-onboarding/src/onboarder.test.ts +++ b/services/libs/project-onboarding/src/onboarder.test.ts @@ -1,6 +1,15 @@ -import { afterEach, describe, expect, it, vi } from 'vitest' - -import { deriveProjectNameCandidates, readErrorBody, resolveProjectSegment } from './onboarder' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +import { + buildGithubIntegrationPayload, + buildGithubOrgSettings, + connectRepoToSegment, + deriveProjectNameCandidates, + mergeGithubOrgSettings, + readErrorBody, + resolveProjectSegment, +} from './onboarder' +import { IGithubOrgSettings } from './types' interface IFakeSegment { name: string @@ -277,6 +286,199 @@ describe('resolveProjectSegment', () => { }) }) +const now = new Date('2026-10-12T10:00:00.000Z') + +function org(name: string, repoUrls: string[]): IGithubOrgSettings { + return { + name, + url: `https://github.com/${name}`, + logo: `${name}-logo`, + fullSync: false, + updatedAt: '2026-01-01T00:00:00.000Z', + repos: repoUrls.map((url) => ({ + name: url.split('/').pop() as string, + url, + forkedFrom: null, + updatedAt: '2026-01-01T00:00:00.000Z', + })), + } +} + +describe('mergeGithubOrgSettings', () => { + const incoming = buildGithubOrgSettings({ + owner: 'kubernetes', + repo: 'kops', + repoUrl: 'https://github.com/kubernetes/kops', + orgLogo: 'new-logo', + forkedFrom: null, + now, + }) + + it('appends the org when the integration does not have it yet', () => { + const existing = [org('helm', ['https://github.com/helm/helm'])] + + expect(mergeGithubOrgSettings(existing, incoming)).toEqual([...existing, incoming]) + }) + + it('adds the repo to the existing org and keeps its metadata', () => { + const existing = [org('kubernetes', ['https://github.com/kubernetes/kubernetes'])] + + const merged = mergeGithubOrgSettings(existing, incoming) + + expect(merged).toHaveLength(1) + expect(merged[0].logo).toBe('kubernetes-logo') + expect(merged[0].repos.map((repo) => repo.url)).toEqual([ + 'https://github.com/kubernetes/kubernetes', + 'https://github.com/kubernetes/kops', + ]) + }) + + it('does not duplicate a repo already connected in another url form', () => { + const existing = [org('Kubernetes', ['https://github.com/Kubernetes/Kops.git'])] + + const merged = mergeGithubOrgSettings(existing, incoming) + + expect(merged).toEqual(existing) + }) +}) + +describe('buildGithubIntegrationPayload', () => { + it('maps every repo of every org to the segment and carries the integration id', () => { + const orgs = [ + org('kubernetes', ['https://github.com/kubernetes/kubernetes']), + org('helm', ['https://github.com/helm/helm', 'https://github.com/helm/charts']), + ] + + const payload = buildGithubIntegrationPayload({ orgs, segmentId: 'seg', integrationId: 'int' }) + + expect(payload).toEqual({ + integrationId: 'int', + settings: { orgs, updateMemberAttributes: true }, + mapping: { + 'https://github.com/kubernetes/kubernetes': 'seg', + 'https://github.com/helm/helm': 'seg', + 'https://github.com/helm/charts': 'seg', + }, + segments: ['seg'], + }) + }) + + it('omits the integration id for a new integration', () => { + const payload = buildGithubIntegrationPayload({ orgs: [], segmentId: 'seg' }) + + expect(payload).not.toHaveProperty('integrationId') + }) +}) + +describe('connectRepoToSegment', () => { + const repoUrl = 'https://github.com/kubernetes/kops' + + function installConnectBackend( + existingIntegration: { id: string; orgs: IGithubOrgSettings[] } | null, + ) { + const connectPayloads: Record[] = [] + + vi.stubGlobal( + 'fetch', + vi.fn(async (url: string, init?: { body?: string }) => { + if (url.startsWith('https://api.github.com/')) { + return new Response('', { status: 404 }) + } + if (url.endsWith('/integration/query')) { + const rows = existingIntegration + ? [{ id: existingIntegration.id, settings: { orgs: existingIntegration.orgs } }] + : [] + return jsonResponse({ rows, count: rows.length }) + } + if (url.endsWith('/github-nango-connect')) { + connectPayloads.push(JSON.parse(init?.body ?? '{}')) + return jsonResponse({}) + } + return new Response('unexpected', { status: 500 }) + }), + ) + + return connectPayloads + } + + beforeEach(() => { + vi.stubEnv('CROWD_API_SERVICE_URL', 'http://api') + vi.stubEnv('CROWD_LF_AGENT_USER_TOKEN', 'token') + vi.stubEnv('CROWD_GITHUB_PERSONAL_ACCESS_TOKENS', 'gh-token') + }) + + afterEach(() => { + vi.unstubAllEnvs() + }) + + it('creates a new integration when the segment has none', async () => { + const payloads = installConnectBackend(null) + + const result = await connectRepoToSegment({ segmentId: 'seg', repoUrl }) + + expect(result).toEqual({ outcome: 'onboarded', segmentId: 'seg', error: null }) + expect(payloads).toHaveLength(1) + expect(payloads[0]).not.toHaveProperty('integrationId') + expect(payloads[0]).toMatchObject({ + mapping: { [repoUrl]: 'seg' }, + segments: ['seg'], + }) + }) + + it('sends the existing repos back together with the new one on update', async () => { + const payloads = installConnectBackend({ + id: 'int', + orgs: [org('kubernetes', ['https://github.com/kubernetes/kubernetes'])], + }) + + const result = await connectRepoToSegment({ segmentId: 'seg', repoUrl }) + + expect(result.outcome).toBe('onboarded') + expect(payloads[0]).toMatchObject({ + integrationId: 'int', + mapping: { 'https://github.com/kubernetes/kubernetes': 'seg', [repoUrl]: 'seg' }, + }) + }) + + it('returns an error result when the connect call fails', async () => { + vi.stubGlobal( + 'fetch', + vi.fn(async (url: string) => + url.endsWith('/integration/query') + ? jsonResponse({ rows: [], count: 0 }) + : new Response('boom', { status: 500 }), + ), + ) + + const result = await connectRepoToSegment({ segmentId: 'seg', repoUrl }) + + expect(result.outcome).toBe('error') + expect(result.segmentId).toBe('seg') + expect(result.error).toContain('GitHub integration returned HTTP 500') + }) + + it('returns an error result for a non GitHub url without calling the backend', async () => { + const payloads = installConnectBackend(null) + + const result = await connectRepoToSegment({ + segmentId: 'seg', + repoUrl: 'https://gitlab.com/a/b', + }) + + expect(result.outcome).toBe('error') + expect(payloads).toHaveLength(0) + }) + + it('returns an error result when the api configuration is missing', async () => { + vi.stubEnv('CROWD_API_SERVICE_URL', '') + + const result = await connectRepoToSegment({ segmentId: 'seg', repoUrl }) + + expect(result.outcome).toBe('error') + expect(result.error).toContain('Missing API configuration') + }) +}) + describe('readErrorBody', () => { it('returns the response body text', async () => { const response = new Response('{"error":"insightsProjects slug already exists"}') diff --git a/services/libs/project-onboarding/src/onboarder.ts b/services/libs/project-onboarding/src/onboarder.ts index cd55567f08..2fd17bb09d 100644 --- a/services/libs/project-onboarding/src/onboarder.ts +++ b/services/libs/project-onboarding/src/onboarder.ts @@ -1,6 +1,12 @@ import { canonicalizeRepoUrl, githubRepoPath } from '@crowd/common' -import { IOnboardingInput, IOnboardingResult } from './types' +import { + IGithubIntegrationSnapshot, + IGithubOrgSettings, + IOnboardingInput, + IOnboardingResult, + ISegmentOnboardingInput, +} from './types' const LF_OSS_INDEX_PROJECT_GROUP_SLUG = 'lf-oss-index' const LINUX_KERNEL_GITHUB_URL = 'https://github.com/torvalds/linux' @@ -12,7 +18,13 @@ interface ISegmentQueryResponse { } interface IIntegrationQueryResponse { - rows?: Array<{ settings?: { orgs?: Array<{ repos?: Array<{ url?: string }> }> } }> + rows?: Array<{ id: string; settings?: { orgs?: Array> } }> +} + +interface IOnboardingApiConfig { + apiUrl: string + apiToken: string + githubToken: string } const GITHUB_NANGO_PLATFORM = 'github-nango' @@ -20,6 +32,24 @@ const SEGMENT_QUERY_PAGE_SIZE = 20 const BACKEND_REQUEST_TIMEOUT_MS = 30_000 const GITHUB_REQUEST_TIMEOUT_MS = 10_000 const ERROR_BODY_MAX_LENGTH = 500 +const MISSING_CONFIG_ERROR = + 'Missing API configuration: CROWD_API_SERVICE_URL, CROWD_LF_AGENT_USER_TOKEN, or CROWD_GITHUB_PERSONAL_ACCESS_TOKENS' + +function readOnboardingApiConfig(): IOnboardingApiConfig | null { + const apiUrl = process.env.CROWD_API_SERVICE_URL + const apiToken = process.env.CROWD_LF_AGENT_USER_TOKEN + const githubTokens = process.env.CROWD_GITHUB_PERSONAL_ACCESS_TOKENS + + if (!apiUrl || !apiToken || !githubTokens) { + return null + } + + return { apiUrl, apiToken, githubToken: githubTokens.split(',')[0].trim() } +} + +function errorResult(error: string, segmentId: string | null = null): IOnboardingResult { + return { outcome: 'error', segmentId, error } +} export async function readErrorBody(response: Response): Promise { try { @@ -58,33 +88,62 @@ export function parseGithubUrl(repoUrl: string): { owner: string; repo: string } return { owner, repo } } -export function buildGithubIntegrationPayload(params: { +export function buildGithubOrgSettings(params: { owner: string repo: string repoUrl: string - segmentId: string orgLogo: string forkedFrom: string | null now?: Date -}): Record { - const { owner, repo, repoUrl, segmentId, orgLogo, forkedFrom, now = new Date() } = params +}): IGithubOrgSettings { + const { owner, repo, repoUrl, orgLogo, forkedFrom, now = new Date() } = params const updatedAt = now.toISOString() return { - settings: { - orgs: [ - { - name: owner, - url: repoUrl, - logo: orgLogo, - fullSync: false, - updatedAt, - repos: [{ name: repo, url: repoUrl, forkedFrom, updatedAt }], - }, - ], - updateMemberAttributes: true, - }, - mapping: { [repoUrl]: segmentId }, + name: owner, + url: repoUrl, + logo: orgLogo, + fullSync: false, + updatedAt, + repos: [{ name: repo, url: repoUrl, forkedFrom, updatedAt }], + } +} + +function hasRepo(org: IGithubOrgSettings, repoUrl: string): boolean { + const requested = comparableRepoUrl(repoUrl) + return org.repos.some((repo) => comparableRepoUrl(repo.url) === requested) +} + +export function mergeGithubOrgSettings( + existing: IGithubOrgSettings[], + incoming: IGithubOrgSettings, +): IGithubOrgSettings[] { + const incomingName = incoming.name.toLowerCase() + const current = existing.find((org) => org.name.toLowerCase() === incomingName) + if (!current) { + return [...existing, incoming] + } + + const newRepos = incoming.repos.filter((repo) => !hasRepo(current, repo.url)) + return existing.map((org) => + org === current ? { ...org, repos: [...org.repos, ...newRepos] } : org, + ) +} + +export function buildGithubIntegrationPayload(params: { + orgs: IGithubOrgSettings[] + segmentId: string + integrationId?: string +}): Record { + const { orgs, segmentId, integrationId } = params + const mapping = Object.fromEntries( + orgs.flatMap((org) => org.repos).map((repo) => [repo.url, segmentId]), + ) + + return { + ...(integrationId ? { integrationId } : {}), + settings: { orgs, updateMemberAttributes: true }, + mapping, segments: [segmentId], } } @@ -138,11 +197,22 @@ function comparableRepoUrl(url: string): string { return canonicalizeRepoUrl(url)?.url ?? url.toLowerCase() } -async function queryGithubIntegrationRepoUrls( +function toOrgSettings(org: Partial): IGithubOrgSettings { + return { + name: org.name ?? '', + url: org.url ?? '', + logo: org.logo ?? '', + fullSync: org.fullSync ?? false, + updatedAt: org.updatedAt ?? new Date().toISOString(), + repos: org.repos ?? [], + } +} + +async function queryGithubIntegration( segmentId: string, apiUrl: string, token: string, -): Promise { +): Promise { const response = await fetch(`${apiUrl}/integration/query`, { method: 'POST', headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${token}` }, @@ -166,9 +236,20 @@ async function queryGithubIntegrationRepoUrls( return null } - return (integration.settings?.orgs ?? []) - .flatMap((org) => org.repos ?? []) - .flatMap((repo) => (repo.url ? [comparableRepoUrl(repo.url)] : [])) + return { id: integration.id, orgs: (integration.settings?.orgs ?? []).map(toOrgSettings) } +} + +async function queryGithubIntegrationRepoUrls( + segmentId: string, + apiUrl: string, + token: string, +): Promise { + const integration = await queryGithubIntegration(segmentId, apiUrl, token) + if (!integration) { + return null + } + + return integration.orgs.flatMap((org) => org.repos).map((repo) => comparableRepoUrl(repo.url)) } export function deriveProjectNameCandidates(owner: string, repoName: string): string[] { @@ -272,19 +353,23 @@ async function fetchGithubForkedFrom( } } -async function createGithubIntegration( - input: { owner: string; repo: string; repoUrl: string; segmentId: string }, - apiUrl: string, - token: string, +async function fetchGithubOrgSettings( + input: { owner: string; repo: string; repoUrl: string }, githubToken: string, -): Promise { +): Promise { const [orgLogo, forkedFrom] = await Promise.all([ fetchGithubOrgLogo(input.owner, githubToken), fetchGithubForkedFrom(input.owner, input.repo, githubToken), ]) - const payload = buildGithubIntegrationPayload({ ...input, orgLogo, forkedFrom }) + return buildGithubOrgSettings({ ...input, orgLogo, forkedFrom }) +} +async function postGithubNangoConnect( + payload: Record, + apiUrl: string, + token: string, +): Promise { const response = await fetch(`${apiUrl}/github-nango-connect`, { method: 'POST', headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${token}` }, @@ -299,20 +384,44 @@ async function createGithubIntegration( } } -export async function onboardProject(input: IOnboardingInput): Promise { - const apiUrl = process.env.CROWD_API_SERVICE_URL - const apiToken = process.env.CROWD_LF_AGENT_USER_TOKEN - const githubTokens = process.env.CROWD_GITHUB_PERSONAL_ACCESS_TOKENS +async function createGithubIntegration( + input: { owner: string; repo: string; repoUrl: string; segmentId: string }, + { apiUrl, apiToken, githubToken }: IOnboardingApiConfig, +): Promise { + const org = await fetchGithubOrgSettings(input, githubToken) - if (!apiUrl || !apiToken || !githubTokens) { - return { - outcome: 'error', - segmentId: null, - error: - 'Missing API configuration: CROWD_API_SERVICE_URL, CROWD_LF_AGENT_USER_TOKEN, or CROWD_GITHUB_PERSONAL_ACCESS_TOKENS', - } + await postGithubNangoConnect( + buildGithubIntegrationPayload({ orgs: [org], segmentId: input.segmentId }), + apiUrl, + apiToken, + ) +} + +// The backend replaces settings.orgs and soft-deletes repos missing from the mapping on update, +// so the existing orgs and repos must be sent back along with the new one. +async function addRepoToGithubIntegration( + input: { owner: string; repo: string; repoUrl: string; segmentId: string }, + integration: IGithubIntegrationSnapshot, + { apiUrl, apiToken, githubToken }: IOnboardingApiConfig, +): Promise { + const org = await fetchGithubOrgSettings(input, githubToken) + + await postGithubNangoConnect( + buildGithubIntegrationPayload({ + orgs: mergeGithubOrgSettings(integration.orgs, org), + segmentId: input.segmentId, + integrationId: integration.id, + }), + apiUrl, + apiToken, + ) +} + +export async function onboardProject(input: IOnboardingInput): Promise { + const config = readOnboardingApiConfig() + if (!config) { + return errorResult(MISSING_CONFIG_ERROR) } - const githubToken = githubTokens.split(',')[0].trim() const slug = deriveProjectSlug(input.projectSlug) @@ -323,7 +432,7 @@ export async function onboardProject(input: IOnboardingInput): Promise { + const config = readOnboardingApiConfig() + if (!config) { + return errorResult(MISSING_CONFIG_ERROR, segmentId) + } + + let owner: string + let repo: string + try { + const parsed = parseGithubUrl(repoUrl) + owner = parsed.owner + repo = parsed.repo + } catch (err) { + return errorResult((err as Error).message, segmentId) + } + + try { + const integration = await queryGithubIntegration(segmentId, config.apiUrl, config.apiToken) + const input = { owner, repo, repoUrl, segmentId } + if (integration) { + await addRepoToGithubIntegration(input, integration, config) + } else { + await createGithubIntegration(input, config) + } } catch (err) { - return { outcome: 'error', segmentId, error: (err as Error).message } + return errorResult((err as Error).message, segmentId) } return { outcome: 'onboarded', segmentId, error: null } diff --git a/services/libs/project-onboarding/src/types.ts b/services/libs/project-onboarding/src/types.ts index e20108dc14..71082069ec 100644 --- a/services/libs/project-onboarding/src/types.ts +++ b/services/libs/project-onboarding/src/types.ts @@ -5,6 +5,32 @@ export interface IOnboardingInput { projectSlug: string } +export interface ISegmentOnboardingInput { + segmentId: string + repoUrl: string +} + +export interface IGithubRepoSettings { + name: string + url: string + forkedFrom: string | null + updatedAt: string +} + +export interface IGithubOrgSettings { + name: string + url: string + logo: string + fullSync: boolean + updatedAt: string + repos: IGithubRepoSettings[] +} + +export interface IGithubIntegrationSnapshot { + id: string + orgs: IGithubOrgSettings[] +} + export type OnboardingOutcome = 'onboarded' | 'error' export interface IOnboardingResult { From e9e4071163f488aff5e486c6736cb5df40ae0b5e Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Mon, 12 Oct 2026 19:30:26 +0200 Subject: [PATCH 2/3] fix: serialize segment integration updates and store canonical repo urls (CM-1841) Signed-off-by: Umberto Sgueglia --- .../src/activities/activities.ts | 8 +-- .../src/activities/proposedAction.test.ts | 55 ++++++++++++++++++- .../src/activities/proposedAction.ts | 27 +++++++++ .../src/project-catalog/projectCatalog.ts | 12 ++++ .../project-onboarding/src/onboarder.test.ts | 20 +++++++ .../libs/project-onboarding/src/onboarder.ts | 22 ++++---- 6 files changed, 127 insertions(+), 17 deletions(-) diff --git a/services/apps/automatic_onboarding_worker/src/activities/activities.ts b/services/apps/automatic_onboarding_worker/src/activities/activities.ts index 946d268312..9d20aa6fbe 100644 --- a/services/apps/automatic_onboarding_worker/src/activities/activities.ts +++ b/services/apps/automatic_onboarding_worker/src/activities/activities.ts @@ -31,7 +31,7 @@ import { isReviewAlertRequest, } from './discussionRequestAlert' import { buildInsightsProjectSkipReason } from './insightsProjectSkip' -import { runProposedAction } from './proposedAction' +import { runProposedActionSerialized } from './proposedAction' const log = getServiceLogger() @@ -120,11 +120,11 @@ export async function onboardAndUpdateProject( 'Starting onboarding.', ) - const result = await runProposedAction(project, { - findCdpSegmentByPccProject: createCdpSegmentLookup(qx), + const result = await runProposedActionSerialized(project, qx, (depsQx) => ({ + findCdpSegmentByPccProject: createCdpSegmentLookup(depsQx), onboardProject, connectRepoToSegment, - }).catch(async (err) => { + })).catch(async (err) => { await releaseProjectCatalogOnboardingClaim(qx, project.id) throw err }) diff --git a/services/apps/automatic_onboarding_worker/src/activities/proposedAction.test.ts b/services/apps/automatic_onboarding_worker/src/activities/proposedAction.test.ts index f67ca7bce5..d6e63cc6ec 100644 --- a/services/apps/automatic_onboarding_worker/src/activities/proposedAction.test.ts +++ b/services/apps/automatic_onboarding_worker/src/activities/proposedAction.test.ts @@ -1,6 +1,14 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' -import { IProposedActionDeps, runProposedAction } from './proposedAction' +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 } @@ -128,3 +136,48 @@ describe('runProposedAction', () => { expect(deps.connectRepoToSegment).not.toHaveBeenCalled() }) }) + +describe('runProposedActionSerialized', () => { + const txQx = { marker: 'tx' } as never + const qx = { tx: vi.fn(async (fn: (tx: unknown) => Promise) => 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) + }) +}) diff --git a/services/apps/automatic_onboarding_worker/src/activities/proposedAction.ts b/services/apps/automatic_onboarding_worker/src/activities/proposedAction.ts index 360cabfbec..c80d18e4d1 100644 --- a/services/apps/automatic_onboarding_worker/src/activities/proposedAction.ts +++ b/services/apps/automatic_onboarding_worker/src/activities/proposedAction.ts @@ -1,4 +1,6 @@ +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, @@ -70,3 +72,28 @@ export async function runProposedAction( 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 { + if (!writesSharedSegment(project)) { + return runProposedAction(project, buildDeps(qx)) + } + + return qx.tx(async (tx) => { + await lockProjectCatalogOnboardingTarget(tx, project.pccProjectId) + return runProposedAction(project, buildDeps(tx)) + }) +} diff --git a/services/libs/data-access-layer/src/project-catalog/projectCatalog.ts b/services/libs/data-access-layer/src/project-catalog/projectCatalog.ts index 417ab010cb..3e11b1fa1e 100644 --- a/services/libs/data-access-layer/src/project-catalog/projectCatalog.ts +++ b/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 { + 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( diff --git a/services/libs/project-onboarding/src/onboarder.test.ts b/services/libs/project-onboarding/src/onboarder.test.ts index 63754c4b98..7c3feb9a8a 100644 --- a/services/libs/project-onboarding/src/onboarder.test.ts +++ b/services/libs/project-onboarding/src/onboarder.test.ts @@ -440,6 +440,26 @@ describe('connectRepoToSegment', () => { }) }) + it('stores a mixed-case url in its canonical lowercase form', async () => { + const payloads = installConnectBackend({ + id: 'int', + orgs: [org('kubernetes', ['https://github.com/kubernetes/kubernetes'])], + }) + + const result = await connectRepoToSegment({ + segmentId: 'seg', + repoUrl: 'https://GitHub.com/Kubernetes/Kops.git', + }) + + expect(result.outcome).toBe('onboarded') + expect(payloads[0]).toMatchObject({ + integrationId: 'int', + mapping: { 'https://github.com/kubernetes/kubernetes': 'seg', [repoUrl]: 'seg' }, + }) + expect(payloads[0]).not.toHaveProperty(['mapping', 'https://GitHub.com/Kubernetes/Kops.git']) + expect(JSON.stringify(payloads[0])).not.toContain('Kops') + }) + it('returns an error result when the connect call fails', async () => { vi.stubGlobal( 'fetch', diff --git a/services/libs/project-onboarding/src/onboarder.ts b/services/libs/project-onboarding/src/onboarder.ts index 2fd17bb09d..52d1b913b4 100644 --- a/services/libs/project-onboarding/src/onboarder.ts +++ b/services/libs/project-onboarding/src/onboarder.ts @@ -1,4 +1,4 @@ -import { canonicalizeRepoUrl, githubRepoPath } from '@crowd/common' +import { canonicalizeRepoUrl } from '@crowd/common' import { IGithubIntegrationSnapshot, @@ -78,14 +78,13 @@ export function deriveProjectSlug(projectSlug: string): string { .replace(/^-|-$/g, '') } -export function parseGithubUrl(repoUrl: string): { owner: string; repo: string } { - const path = githubRepoPath(repoUrl) - if (!path) { +export function parseGithubUrl(repoUrl: string): { owner: string; repo: string; url: string } { + const canonical = canonicalizeRepoUrl(repoUrl) + if (!canonical?.isGithub) { throw new Error(`Invalid GitHub URL format: ${repoUrl}`) } - const [owner, repo] = path.split('/') - return { owner, repo } + return { owner: canonical.owner, repo: canonical.repo, url: canonical.url } } export function buildGithubOrgSettings(params: { @@ -427,10 +426,12 @@ export async function onboardProject(input: IOnboardingInput): Promise Date: Mon, 12 Oct 2026 20:06:17 +0200 Subject: [PATCH 3/3] fix: keep existing repos on their current segments when updating a github integration (CM-1841) Signed-off-by: Umberto Sgueglia --- .../project-onboarding/src/onboarder.test.ts | 92 +++++++++++++++++-- .../libs/project-onboarding/src/onboarder.ts | 49 +++++++++- 2 files changed, 130 insertions(+), 11 deletions(-) diff --git a/services/libs/project-onboarding/src/onboarder.test.ts b/services/libs/project-onboarding/src/onboarder.test.ts index 7c3feb9a8a..5b54cd3eae 100644 --- a/services/libs/project-onboarding/src/onboarder.test.ts +++ b/services/libs/project-onboarding/src/onboarder.test.ts @@ -363,6 +363,25 @@ describe('buildGithubIntegrationPayload', () => { }) }) + it('keeps every existing repo on its current segment, in any url form', () => { + const orgs = [org('helm', ['https://github.com/helm/helm', 'https://github.com/helm/charts'])] + + const payload = buildGithubIntegrationPayload({ + orgs, + segmentId: 'seg', + integrationId: 'int', + existingMapping: { 'https://github.com/Helm/Helm.git': 'subproject' }, + }) + + expect(payload).toMatchObject({ + mapping: { + 'https://github.com/helm/helm': 'subproject', + 'https://github.com/helm/charts': 'seg', + }, + segments: ['seg'], + }) + }) + it('omits the integration id for a new integration', () => { const payload = buildGithubIntegrationPayload({ orgs: [], segmentId: 'seg' }) @@ -374,9 +393,14 @@ describe('connectRepoToSegment', () => { const repoUrl = 'https://github.com/kubernetes/kops' function installConnectBackend( - existingIntegration: { id: string; orgs: IGithubOrgSettings[] } | null, + existingIntegration: { + id: string + orgs: IGithubOrgSettings[] + mapping?: Record + } | null, ) { const connectPayloads: Record[] = [] + const mappingRequests: string[] = [] vi.stubGlobal( 'fetch', @@ -390,6 +414,16 @@ describe('connectRepoToSegment', () => { : [] return jsonResponse({ rows, count: rows.length }) } + if (url.includes('/repositories?')) { + mappingRequests.push(url) + const rows = Object.entries(existingIntegration?.mapping ?? {}).map( + ([repoUrl, segmentId]) => ({ + url: repoUrl, + segment: { id: segmentId, name: segmentId }, + }), + ) + return jsonResponse(rows) + } if (url.endsWith('/github-nango-connect')) { connectPayloads.push(JSON.parse(init?.body ?? '{}')) return jsonResponse({}) @@ -398,7 +432,7 @@ describe('connectRepoToSegment', () => { }), ) - return connectPayloads + return { connectPayloads, mappingRequests } } beforeEach(() => { @@ -412,11 +446,12 @@ describe('connectRepoToSegment', () => { }) it('creates a new integration when the segment has none', async () => { - const payloads = installConnectBackend(null) + const { connectPayloads: payloads, mappingRequests } = installConnectBackend(null) const result = await connectRepoToSegment({ segmentId: 'seg', repoUrl }) expect(result).toEqual({ outcome: 'onboarded', segmentId: 'seg', error: null }) + expect(mappingRequests).toEqual([]) expect(payloads).toHaveLength(1) expect(payloads[0]).not.toHaveProperty('integrationId') expect(payloads[0]).toMatchObject({ @@ -426,22 +461,67 @@ describe('connectRepoToSegment', () => { }) it('sends the existing repos back together with the new one on update', async () => { - const payloads = installConnectBackend({ + const { connectPayloads: payloads, mappingRequests } = installConnectBackend({ id: 'int', orgs: [org('kubernetes', ['https://github.com/kubernetes/kubernetes'])], + mapping: { 'https://github.com/kubernetes/kubernetes': 'seg' }, }) const result = await connectRepoToSegment({ segmentId: 'seg', repoUrl }) expect(result.outcome).toBe('onboarded') + expect(mappingRequests).toEqual(['http://api/integration/int/repositories?segments%5B%5D=seg']) expect(payloads[0]).toMatchObject({ integrationId: 'int', mapping: { 'https://github.com/kubernetes/kubernetes': 'seg', [repoUrl]: 'seg' }, }) }) + it('leaves repos mapped to sibling subprojects on their segment', async () => { + const { connectPayloads: payloads } = installConnectBackend({ + id: 'int', + orgs: [ + org('kubernetes', [ + 'https://github.com/kubernetes/kubernetes', + 'https://github.com/kubernetes/minikube', + ]), + ], + mapping: { + 'https://github.com/kubernetes/kubernetes': 'seg', + 'https://github.com/kubernetes/minikube': 'minikube-subproject', + }, + }) + + await connectRepoToSegment({ segmentId: 'seg', repoUrl }) + + expect(payloads[0]).toMatchObject({ + mapping: { + 'https://github.com/kubernetes/kubernetes': 'seg', + 'https://github.com/kubernetes/minikube': 'minikube-subproject', + [repoUrl]: 'seg', + }, + segments: ['seg'], + }) + }) + + it('returns an error result when the current mapping cannot be read', async () => { + installConnectBackend({ id: 'int', orgs: [] }) + const fetchMock = vi.mocked(fetch) + const original = fetchMock.getMockImplementation() + fetchMock.mockImplementation(async (url, init) => + String(url).includes('/repositories?') + ? new Response('boom', { status: 500, statusText: 'Internal Server Error' }) + : original!(url, init), + ) + + const result = await connectRepoToSegment({ segmentId: 'seg', repoUrl }) + + expect(result.outcome).toBe('error') + expect(result.error).toContain('Integration repositories returned HTTP 500') + }) + it('stores a mixed-case url in its canonical lowercase form', async () => { - const payloads = installConnectBackend({ + const { connectPayloads: payloads } = installConnectBackend({ id: 'int', orgs: [org('kubernetes', ['https://github.com/kubernetes/kubernetes'])], }) @@ -478,7 +558,7 @@ describe('connectRepoToSegment', () => { }) it('returns an error result for a non GitHub url without calling the backend', async () => { - const payloads = installConnectBackend(null) + const { connectPayloads: payloads } = installConnectBackend(null) const result = await connectRepoToSegment({ segmentId: 'seg', diff --git a/services/libs/project-onboarding/src/onboarder.ts b/services/libs/project-onboarding/src/onboarder.ts index 52d1b913b4..d2bbbfb434 100644 --- a/services/libs/project-onboarding/src/onboarder.ts +++ b/services/libs/project-onboarding/src/onboarder.ts @@ -21,6 +21,11 @@ interface IIntegrationQueryResponse { rows?: Array<{ id: string; settings?: { orgs?: Array> } }> } +interface IIntegrationRepositoryMapping { + url: string + segment: { id: string } +} + interface IOnboardingApiConfig { apiUrl: string apiToken: string @@ -129,14 +134,22 @@ export function mergeGithubOrgSettings( ) } +function toComparableMapping(mapping: Record): Map { + return new Map(Object.entries(mapping).map(([url, segment]) => [comparableRepoUrl(url), segment])) +} + export function buildGithubIntegrationPayload(params: { orgs: IGithubOrgSettings[] segmentId: string integrationId?: string + existingMapping?: Record }): Record { - const { orgs, segmentId, integrationId } = params + const { orgs, segmentId, integrationId, existingMapping = {} } = params + const currentSegments = toComparableMapping(existingMapping) const mapping = Object.fromEntries( - orgs.flatMap((org) => org.repos).map((repo) => [repo.url, segmentId]), + orgs + .flatMap((org) => org.repos) + .map((repo) => [repo.url, currentSegments.get(comparableRepoUrl(repo.url)) ?? segmentId]), ) return { @@ -238,6 +251,28 @@ async function queryGithubIntegration( return { id: integration.id, orgs: (integration.settings?.orgs ?? []).map(toOrgSettings) } } +async function fetchGithubIntegrationMapping( + integrationId: string, + segmentId: string, + apiUrl: string, + token: string, +): Promise> { + const query = new URLSearchParams({ 'segments[]': segmentId }) + const response = await fetch(`${apiUrl}/integration/${integrationId}/repositories?${query}`, { + headers: { Authorization: `Bearer ${token}` }, + signal: AbortSignal.timeout(BACKEND_REQUEST_TIMEOUT_MS), + }) + + if (!response.ok) { + throw new Error( + `Integration repositories returned HTTP ${response.status}: ${response.statusText} - ${await readErrorBody(response)}`, + ) + } + + const rows = (await response.json()) as IIntegrationRepositoryMapping[] + return Object.fromEntries(rows.map((row) => [row.url, row.segment.id])) +} + async function queryGithubIntegrationRepoUrls( segmentId: string, apiUrl: string, @@ -396,20 +431,24 @@ async function createGithubIntegration( ) } -// The backend replaces settings.orgs and soft-deletes repos missing from the mapping on update, -// so the existing orgs and repos must be sent back along with the new one. +// The backend replaces settings.orgs, applies the mapping to every repo and soft-deletes repos +// missing from it, so existing repos must be sent back on their current (sub)project segments. async function addRepoToGithubIntegration( input: { owner: string; repo: string; repoUrl: string; segmentId: string }, integration: IGithubIntegrationSnapshot, { apiUrl, apiToken, githubToken }: IOnboardingApiConfig, ): Promise { - const org = await fetchGithubOrgSettings(input, githubToken) + const [org, existingMapping] = await Promise.all([ + fetchGithubOrgSettings(input, githubToken), + fetchGithubIntegrationMapping(integration.id, input.segmentId, apiUrl, apiToken), + ]) await postGithubNangoConnect( buildGithubIntegrationPayload({ orgs: mergeGithubOrgSettings(integration.orgs, org), segmentId: input.segmentId, integrationId: integration.id, + existingMapping, }), apiUrl, apiToken,