diff --git a/backend/src/api/index.ts b/backend/src/api/index.ts index 19a8df1e36..4fa11e008b 100644 --- a/backend/src/api/index.ts +++ b/backend/src/api/index.ts @@ -113,7 +113,7 @@ setImmediate(async () => { const temporal = TEMPORAL_CONFIG.serverUrl ? await getTemporalClient(TEMPORAL_CONFIG) : null // Mounted before DB/Redis/OpenSearch middleware to protect Slack's 3s ack window. - mountInteractivityRoute(app) + mountInteractivityRoute(app, { temporal }) mountEventsRoute(app, { redis, temporal }) // Initializes and adds the database middleware. diff --git a/backend/src/api/slack/index.ts b/backend/src/api/slack/index.ts index 692b63cccd..cdf153ea1b 100644 --- a/backend/src/api/slack/index.ts +++ b/backend/src/api/slack/index.ts @@ -12,7 +12,10 @@ import { claimSlackEvent } from './eventDeduplication' // Mounted ahead of the shared rate limiter and tenant/segment middleware // to protect Slack's 3-second acknowledgement window; keeps its own limiter. -export function mountInteractivityRoute(app: Application): void { +export function mountInteractivityRoute( + app: Application, + { temporal }: { temporal: TemporalClient | null }, +): void { if (!getSlackBotConfig().signingSecret) { return } @@ -37,7 +40,7 @@ export function mountInteractivityRoute(app: Application): void { interactivityRateLimiter, bodyParser.urlencoded({ limit: '5mb', extended: true, verify: captureRawBody }), handleParserError, - require('./interactivity').default, + require('./interactivity').createInteractivityHandler(temporal), ) } diff --git a/backend/src/api/slack/interactivity.test.ts b/backend/src/api/slack/interactivity.test.ts index d0479145e8..cdfe1f09fb 100644 --- a/backend/src/api/slack/interactivity.test.ts +++ b/backend/src/api/slack/interactivity.test.ts @@ -5,15 +5,26 @@ vi.mock('@/services/slack/forceOnboardingCommand', () => ({ runForceOnboardingCommand: vi.fn(async () => undefined), replaceWithForceOnboardingFailure: vi.fn(async () => undefined), })) +vi.mock('@/services/slack/reviewDecisionCommand', () => ({ + runReviewDecisionCommand: vi.fn(async () => undefined), + postReviewDecisionFailure: vi.fn(async () => undefined), +})) import { replaceWithForceOnboardingFailure, runForceOnboardingCommand, } from '@/services/slack/forceOnboardingCommand' +import { + postReviewDecisionFailure, + runReviewDecisionCommand, +} from '@/services/slack/reviewDecisionCommand' -import interactivity from './interactivity' +import { createInteractivityHandler } from './interactivity' import { verifySlackSignature } from './verifySignature' +const temporal = { workflow: { start: vi.fn() } } as any +const interactivity = createInteractivityHandler(temporal) + function call(payload: object) { const req = { body: { payload: JSON.stringify(payload) }, @@ -30,6 +41,11 @@ const forcePayload = { actions: [{ action_id: 'force_onboarding', value: 'catalog-1' }], } +const approvePayload = { + ...forcePayload, + actions: [{ action_id: 'approve_review', value: 'catalog-2' }], +} + describe('interactivity', () => { beforeEach(() => { vi.clearAllMocks() @@ -62,11 +78,53 @@ describe('interactivity', () => { ) }) + it.each([ + ['approve_review', 'approve'], + ['reject_review', 'reject'], + ])('runs the review decision for %s', async (actionId, decision) => { + const { res } = await call({ + ...approvePayload, + actions: [{ action_id: actionId, value: 'catalog-2' }], + }) + + expect(res.sendStatus).toHaveBeenCalledWith(200) + expect(runReviewDecisionCommand).toHaveBeenCalledWith( + expect.objectContaining({ + catalogId: 'catalog-2', + decision, + responseUrl: 'https://hooks.slack.com/response', + actorId: 'U123', + temporal, + }), + ) + expect(runForceOnboardingCommand).not.toHaveBeenCalled() + }) + + it('replaces the review message when the decision flow rejects', async () => { + vi.mocked(runReviewDecisionCommand).mockRejectedValueOnce(new Error('db down')) + + const { req } = await call(approvePayload) + await new Promise((resolve) => setImmediate(resolve)) + + expect(postReviewDecisionFailure).toHaveBeenCalledWith( + 'https://hooks.slack.com/response', + req.log, + ) + }) + + it('ignores a review click without a user', async () => { + const { req } = await call({ ...approvePayload, user: undefined }) + + expect(runReviewDecisionCommand).not.toHaveBeenCalled() + expect(req.log.warn).toHaveBeenCalledWith('Unhandled Slack block action.') + }) + it('ignores unknown actions', async () => { const { res } = await call({ ...forcePayload, actions: [{ action_id: 'other' }] }) expect(res.sendStatus).toHaveBeenCalledWith(200) expect(runForceOnboardingCommand).not.toHaveBeenCalled() + expect(runReviewDecisionCommand).not.toHaveBeenCalled() }) it('does nothing for an unverified request', async () => { diff --git a/backend/src/api/slack/interactivity.ts b/backend/src/api/slack/interactivity.ts index 5687af6196..3ddf5252b7 100644 --- a/backend/src/api/slack/interactivity.ts +++ b/backend/src/api/slack/interactivity.ts @@ -5,7 +5,16 @@ import { replaceWithForceOnboardingFailure, runForceOnboardingCommand, } from '@/services/slack/forceOnboardingCommand' -import { FORCE_ONBOARDING_ACTION_ID } from '@/services/slack/slackActionIds' +import type { OnboardingRequestTemporal } from '@/services/slack/onboardingRequestBot' +import { + postReviewDecisionFailure, + runReviewDecisionCommand, +} from '@/services/slack/reviewDecisionCommand' +import { + APPROVE_REVIEW_ACTION_ID, + FORCE_ONBOARDING_ACTION_ID, + REJECT_REVIEW_ACTION_ID, +} from '@/services/slack/slackActionIds' import { validateOrThrow } from '@/utils/validation' import { verifySlackSignature } from './verifySignature' @@ -23,51 +32,95 @@ const payloadSchema = z.object({ type InteractivityPayload = z.infer -function dispatchForceOnboarding(payload: InteractivityPayload, req: Request) { - const action = payload.actions?.find((a) => a.action_id === FORCE_ONBOARDING_ACTION_ID) +interface IActionClick { + value: string + responseUrl: string + actorId: string +} + +function findActionClick(payload: InteractivityPayload, actionId: string): IActionClick | null { + const action = payload.actions?.find((a) => a.action_id === actionId) if (!action || !payload.user || !payload.response_url || !action.value) { + return null + } + + return { value: action.value, responseUrl: payload.response_url, actorId: payload.user.id } +} + +function dispatchForceOnboarding(payload: InteractivityPayload, req: Request) { + const click = findActionClick(payload, FORCE_ONBOARDING_ACTION_ID) + if (!click) { return false } - const responseUrl = payload.response_url runForceOnboardingCommand({ - catalogId: action.value, - responseUrl, - actorId: payload.user.id, + catalogId: click.value, + responseUrl: click.responseUrl, + actorId: click.actorId, log: req.log, }).catch((err) => { req.log.error(err, 'Force onboarding failed unexpectedly!') - replaceWithForceOnboardingFailure(responseUrl, req.log) + replaceWithForceOnboardingFailure(click.responseUrl, req.log) + }) + return true +} + +function dispatchReviewDecision( + payload: InteractivityPayload, + req: Request, + temporal: OnboardingRequestTemporal | null, +) { + const approve = findActionClick(payload, APPROVE_REVIEW_ACTION_ID) + const click = approve ?? findActionClick(payload, REJECT_REVIEW_ACTION_ID) + if (!click) { + return false + } + + runReviewDecisionCommand({ + catalogId: click.value, + decision: approve ? 'approve' : 'reject', + responseUrl: click.responseUrl, + actorId: click.actorId, + temporal, + log: req.log, + }).catch((err) => { + req.log.error(err, 'Review decision failed unexpectedly!') + postReviewDecisionFailure(click.responseUrl, req.log) }) return true } // Mounted ahead of responseHandlerMiddleware, so errors are handled here // directly instead of via the global errorMiddleware. -export default async (req: Request, res: Response) => { - if (!verifySlackSignature(req)) { - req.log.warn('Received unverified Slack interactivity payload!') - res.sendStatus(200) - return - } +export function createInteractivityHandler(temporal: OnboardingRequestTemporal | null) { + return async (req: Request, res: Response) => { + if (!verifySlackSignature(req)) { + req.log.warn('Received unverified Slack interactivity payload!') + res.sendStatus(200) + return + } - try { - const { payload: rawPayload } = validateOrThrow(bodySchema, req.body) - const payload = validateOrThrow(payloadSchema, JSON.parse(rawPayload)) + try { + const { payload: rawPayload } = validateOrThrow(bodySchema, req.body) + const payload = validateOrThrow(payloadSchema, JSON.parse(rawPayload)) - req.log.info( - { type: payload.type, actionIds: payload.actions?.map((a) => a.action_id) }, - 'Received Slack interactivity payload.', - ) + req.log.info( + { type: payload.type, actionIds: payload.actions?.map((a) => a.action_id) }, + 'Received Slack interactivity payload.', + ) - res.sendStatus(200) + res.sendStatus(200) - if (payload.type === 'block_actions' && !dispatchForceOnboarding(payload, req)) { - // TODO(CM-1791): wire up the claim-button interactive handler. - req.log.warn('Unhandled Slack block action.') + if ( + payload.type === 'block_actions' && + !dispatchForceOnboarding(payload, req) && + !dispatchReviewDecision(payload, req, temporal) + ) { + req.log.warn('Unhandled Slack block action.') + } + } catch (err) { + req.log.error(err, 'Error processing Slack interactivity payload!') + res.sendStatus(200) } - } catch (err) { - req.log.error(err, 'Error processing Slack interactivity payload!') - res.sendStatus(200) } } diff --git a/backend/src/services/slack/onboardProjectCommand.ts b/backend/src/services/slack/onboardProjectCommand.ts index 1b9410a782..0c6bcbe268 100644 --- a/backend/src/services/slack/onboardProjectCommand.ts +++ b/backend/src/services/slack/onboardProjectCommand.ts @@ -20,7 +20,7 @@ import { } from '@crowd/data-access-layer' import type { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' import { onboardProject } from '@crowd/project-onboarding' -import { getSlackPermalink, postSlackMessage } from '@crowd/slack' +import { IReviewDecisionMessage, getSlackPermalink, postSlackMessage } from '@crowd/slack' import { createDailyProjectCatalogCap } from '../../api/public/v1/projectCatalog/dailyRequestCap' import { createDailyLlmCap } from '../../api/public/v1/projectEvaluation/dailyLlmCap' @@ -64,9 +64,11 @@ function forceOnboardingPrompt( .buildToObject() } +export type ResponseUrlMessage = SlackMessageDto | IReviewDecisionMessage + export async function postToResponseUrl( responseUrl: string | undefined, - message: SlackMessageDto, + message: ResponseUrlMessage, log: IServiceOptions['log'], ): Promise { if (!responseUrl) { diff --git a/backend/src/services/slack/reviewDecisionCommand.test.ts b/backend/src/services/slack/reviewDecisionCommand.test.ts new file mode 100644 index 0000000000..27b96020db --- /dev/null +++ b/backend/src/services/slack/reviewDecisionCommand.test.ts @@ -0,0 +1,161 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +import { WorkflowIdReusePolicy } from '@crowd/temporal' + +vi.mock('@crowd/data-access-layer', () => ({ + resolveProjectCatalogReview: vi.fn(), +})) +vi.mock('./onboardProjectCommand', () => ({ + postToResponseUrl: vi.fn(async () => undefined), + textMessage: vi.fn((text: string) => ({ text })), +})) +vi.mock('./slackBackground', () => ({ getBgQx: vi.fn(async () => ({})) })) + +import { resolveProjectCatalogReview } from '@crowd/data-access-layer' + +import { postToResponseUrl } from './onboardProjectCommand' +import { postReviewDecisionFailure, runReviewDecisionCommand } from './reviewDecisionCommand' + +const CATALOG_ID = '8f1c2a44-5b7e-4d2f-9c1a-3e6f7a8b9c0d' +const RESPONSE_URL = 'https://hooks.slack.com/response' + +const log = { info: vi.fn(), warn: vi.fn(), error: vi.fn() } as any +const startWorkflow = vi.fn(async () => undefined) +const temporal = { workflow: { start: startWorkflow } } as never + +const reviewedRow = { + id: CATALOG_ID, + repoName: 'kops', + repoUrl: 'https://github.com/kubernetes/kops', + projectName: 'Kubernetes', + classificationKind: 'lf_in_cdp', + proposedAction: 'update_integration', + sourceUrl: null, + reviewedAt: '2026-10-13T07:00:00.000Z', +} as any + +function run(overrides: Record = {}) { + return runReviewDecisionCommand({ + catalogId: CATALOG_ID, + decision: 'approve', + responseUrl: RESPONSE_URL, + actorId: 'U123', + temporal, + log, + ...overrides, + }) +} + +function replacedMessage() { + return vi.mocked(postToResponseUrl).mock.calls[0][1] +} + +describe('runReviewDecisionCommand', () => { + beforeEach(() => { + vi.clearAllMocks() + vi.mocked(resolveProjectCatalogReview).mockResolvedValue(reviewedRow) + }) + + it('approves the row, starts its onboarding right away and replaces the card', async () => { + await run() + + expect(resolveProjectCatalogReview).toHaveBeenCalledWith(expect.anything(), CATALOG_ID, { + decision: 'approve', + reviewedBy: 'U123', + }) + expect(startWorkflow).toHaveBeenCalledWith('onboardProjects', { + taskQueue: 'automatic-onboarding', + workflowId: `onboard-review-${CATALOG_ID}-${Date.parse('2026-10-13T07:00:00.000Z')}`, + workflowIdReusePolicy: WorkflowIdReusePolicy.REJECT_DUPLICATE, + args: [{ catalogIds: [CATALOG_ID] }], + }) + expect(replacedMessage()).toMatchObject({ text: 'Approved — kops', replace_original: true }) + expect(JSON.stringify(replacedMessage())).toContain('onboarding started.') + }) + + it('rejects the row without starting any workflow', async () => { + await run({ decision: 'reject' }) + + expect(resolveProjectCatalogReview).toHaveBeenCalledWith(expect.anything(), CATALOG_ID, { + decision: 'reject', + reviewedBy: 'U123', + }) + expect(startWorkflow).not.toHaveBeenCalled() + expect(replacedMessage()).toMatchObject({ text: 'Rejected — kops', replace_original: true }) + }) + + it('starts a new workflow for a later review cycle of the same row', async () => { + vi.mocked(resolveProjectCatalogReview).mockResolvedValue({ + ...reviewedRow, + reviewedAt: '2026-10-14T07:00:00.000Z', + }) + + await run() + + expect(startWorkflow).toHaveBeenCalledWith( + 'onboardProjects', + expect.objectContaining({ + workflowId: `onboard-review-${CATALOG_ID}-${Date.parse('2026-10-14T07:00:00.000Z')}`, + }), + ) + }) + + it('tolerates an onboarding workflow that was already started', async () => { + const alreadyStarted = new Error('already started') + alreadyStarted.name = 'WorkflowExecutionAlreadyStartedError' + startWorkflow.mockRejectedValueOnce(alreadyStarted) + + await run() + + expect(replacedMessage()).toMatchObject({ text: 'Approved — kops' }) + }) + + it('leaves the approved row to the schedule when temporal is not configured', async () => { + await run({ temporal: null }) + + expect(startWorkflow).not.toHaveBeenCalled() + expect(log.warn).toHaveBeenCalled() + expect(replacedMessage()).toMatchObject({ text: 'Approved — kops' }) + expect(JSON.stringify(replacedMessage())).toContain('queued for the next onboarding run.') + }) + + it('keeps the approval when starting the onboarding workflow fails', async () => { + startWorkflow.mockRejectedValueOnce(new Error('temporal down')) + + await run() + + expect(log.error).toHaveBeenCalled() + expect(replacedMessage()).toMatchObject({ text: 'Approved — kops' }) + expect(JSON.stringify(replacedMessage())).toContain('queued for the next onboarding run.') + }) + + it('tells the reviewer when the row is no longer pending review', async () => { + vi.mocked(resolveProjectCatalogReview).mockResolvedValue(null) + + await run() + + expect(startWorkflow).not.toHaveBeenCalled() + expect(replacedMessage()).toMatchObject({ + text: 'This request was already reviewed or is no longer pending review.', + replace_original: true, + }) + }) + + it('rejects an invalid catalog id before touching the database', async () => { + await run({ catalogId: 'not-a-uuid' }) + + expect(resolveProjectCatalogReview).not.toHaveBeenCalled() + expect(replacedMessage()).toMatchObject({ text: ':no_entry: This request is no longer valid.' }) + }) +}) + +describe('postReviewDecisionFailure', () => { + it('keeps the review card in place so the reviewer can retry', async () => { + vi.clearAllMocks() + await postReviewDecisionFailure(RESPONSE_URL, log) + + const message = vi.mocked(postToResponseUrl).mock.calls[0][1] + expect(message).not.toHaveProperty('replace_original') + expect(JSON.stringify(message)).toContain('Please try again') + }) +}) diff --git a/backend/src/services/slack/reviewDecisionCommand.ts b/backend/src/services/slack/reviewDecisionCommand.ts new file mode 100644 index 0000000000..96d0ceeadd --- /dev/null +++ b/backend/src/services/slack/reviewDecisionCommand.ts @@ -0,0 +1,122 @@ +import { z } from 'zod' + +import { resolveProjectCatalogReview } from '@crowd/data-access-layer' +import { ReviewDecision, buildReviewDecisionMessage } from '@crowd/slack' +import { WorkflowIdReusePolicy } from '@crowd/temporal' + +import { IServiceOptions } from '../IServiceOptions' +import type { OnboardingRequestTemporal } from './onboardingRequestBot' +import { ResponseUrlMessage, postToResponseUrl, textMessage } from './onboardProjectCommand' +import { getBgQx } from './slackBackground' + +const catalogIdSchema = z.string().uuid() + +export interface IReviewDecisionCommand { + catalogId: string + decision: ReviewDecision + responseUrl: string + actorId: string + temporal: OnboardingRequestTemporal | null + log: IServiceOptions['log'] +} + +function replaceOriginal( + responseUrl: string, + message: ResponseUrlMessage, + log: IServiceOptions['log'], +) { + const replaced = { ...message, replace_original: true } + return postToResponseUrl(responseUrl, replaced, log) +} + +export function postReviewDecisionFailure(responseUrl: string, log: IServiceOptions['log']) { + return postToResponseUrl( + responseUrl, + textMessage(':no_entry: Recording the review decision failed. Please try again.'), + log, + ) +} + +function reviewCycleWorkflowId(catalogId: string, reviewedAt: string | null): string { + const cycle = reviewedAt ? Date.parse(reviewedAt) : Date.now() + return `onboard-review-${catalogId}-${cycle}` +} + +async function startApprovedOnboarding( + temporal: OnboardingRequestTemporal | null, + catalogId: string, + reviewedAt: string | null, + log: IServiceOptions['log'], +): Promise { + if (!temporal) { + log.warn({ catalogId }, 'Temporal not configured, the approved row waits for the schedule.') + return false + } + + try { + await temporal.workflow.start('onboardProjects', { + taskQueue: 'automatic-onboarding', + workflowId: reviewCycleWorkflowId(catalogId, reviewedAt), + workflowIdReusePolicy: WorkflowIdReusePolicy.REJECT_DUPLICATE, + args: [{ catalogIds: [catalogId] }], + }) + return true + } catch (err) { + if (err instanceof Error && err.name === 'WorkflowExecutionAlreadyStartedError') { + return true + } + log.error( + { catalogId, err }, + 'Starting the approved onboarding failed, the row waits for the schedule.', + ) + return false + } +} + +export async function runReviewDecisionCommand({ + catalogId: rawCatalogId, + decision, + responseUrl, + actorId, + temporal, + log, +}: IReviewDecisionCommand): Promise { + const parsedId = catalogIdSchema.safeParse(rawCatalogId) + if (!parsedId.success) { + log.warn({ rawCatalogId }, 'Review decision click carried an invalid catalog id.') + await replaceOriginal( + responseUrl, + textMessage(':no_entry: This request is no longer valid.'), + log, + ) + return + } + const catalogId = parsedId.data + + const qx = await getBgQx() + const resolved = await resolveProjectCatalogReview(qx, catalogId, { + decision, + reviewedBy: actorId, + }) + if (!resolved) { + log.info({ catalogId, decision, actorId }, 'Review decision for a row no longer in review.') + await replaceOriginal( + responseUrl, + textMessage('This request was already reviewed or is no longer pending review.'), + log, + ) + return + } + + log.info({ catalogId, repoUrl: resolved.repoUrl, decision, actorId }, 'Review decision recorded.') + + const onboardingStarted = + decision === 'approve' && + (await startApprovedOnboarding(temporal, catalogId, resolved.reviewedAt, log)) + + await replaceOriginal( + responseUrl, + buildReviewDecisionMessage(resolved, decision, actorId, { onboardingStarted }), + log, + ) +} diff --git a/backend/src/services/slack/slackActionIds.ts b/backend/src/services/slack/slackActionIds.ts index ede613d9aa..816e6cfce4 100644 --- a/backend/src/services/slack/slackActionIds.ts +++ b/backend/src/services/slack/slackActionIds.ts @@ -1 +1,3 @@ +export { APPROVE_REVIEW_ACTION_ID, REJECT_REVIEW_ACTION_ID } from '@crowd/slack' + export const FORCE_ONBOARDING_ACTION_ID = 'force_onboarding' diff --git a/services/apps/projects_evaluation_worker/src/activities/activities.ts b/services/apps/projects_evaluation_worker/src/activities/activities.ts index 224e550a09..43f1ddf82e 100644 --- a/services/apps/projects_evaluation_worker/src/activities/activities.ts +++ b/services/apps/projects_evaluation_worker/src/activities/activities.ts @@ -7,10 +7,13 @@ import { findProjectCatalogById, findProjectCatalogPendingEvaluation, findProjectCatalogPendingOnboarding, + findProjectCatalogReviewWithoutMessage, finishPipelineRun, + lockProjectCatalogReviewRequest, markProjectCatalogClassified, promoteProjectCatalogIdsToEvaluate, promoteProjectsToEvaluate, + setProjectCatalogReviewMessageTs, startPipelineRun, } from '@crowd/data-access-layer' import { IPipelineRunFinish } from '@crowd/data-access-layer/src/project-catalog-pipeline-runs/types' @@ -25,7 +28,12 @@ import { loadBatchRepoLookups, withCatalogRowClassifierDeps, } from '@crowd/project-onboarding/src/requestClassifierDeps' -import { SlackChannel, SlackPersona, sendSlackNotificationAsync } from '@crowd/slack' +import { + SlackChannel, + SlackPersona, + postReviewRequest, + sendSlackNotificationAsync, +} from '@crowd/slack' import { WorkflowIdReusePolicy } from '@crowd/temporal' import { estimateLlmCostUsd } from '@crowd/types' @@ -234,6 +242,48 @@ export async function notifySkippedHumanRequest( } } +const REVIEW_REQUEST_LIMIT = 50 + +export async function fetchReviewRequests(catalogIds?: string[]): Promise { + const qx = pgpQx(svc.postgres.writer.connection()) + + const projects = await findProjectCatalogReviewWithoutMessage(qx, { + limit: catalogIds ? undefined : REVIEW_REQUEST_LIMIT, + ids: catalogIds, + }) + + log.info({ count: projects.length, catalogIds }, 'Fetched review rows without a Slack request.') + + return projects +} + +export async function notifyReviewRequest(project: IDbProjectCatalog): Promise { + const qx = pgpQx(svc.postgres.writer.connection()) + + await qx.tx(async (tx) => { + const locked = await lockProjectCatalogReviewRequest(tx, project.id) + if (!locked) { + log.info( + { id: project.id, repoUrl: project.repoUrl }, + 'Review request already posted by another run or the row left review.', + ) + return + } + + const result = await postReviewRequest(project) + if (!result.ok || !result.ts) { + log.warn( + { id: project.id, repoUrl: project.repoUrl, error: result.error }, + 'Review request Slack message was not sent.', + ) + return + } + + await setProjectCatalogReviewMessageTs(tx, project.id, result.ts) + log.info({ id: project.id, repoUrl: project.repoUrl, ts: result.ts }, 'Review request posted.') + }) +} + async function startImmediateOnboarding(catalogIds: string[], workflowId: string): Promise { try { await svc.temporal.workflow.start('onboardProjects', { diff --git a/services/apps/projects_evaluation_worker/src/workflows/evaluateProjects.ts b/services/apps/projects_evaluation_worker/src/workflows/evaluateProjects.ts index 852cdc7011..bbbd4c5fa7 100644 --- a/services/apps/projects_evaluation_worker/src/workflows/evaluateProjects.ts +++ b/services/apps/projects_evaluation_worker/src/workflows/evaluateProjects.ts @@ -148,6 +148,18 @@ export async function evaluateProjects(input: IEvaluateProjectsInput = {}): Prom } } + const reviewRequests = await fetchActivities.fetchReviewRequests(catalogIds) + for (const project of reviewRequests) { + try { + await notifyActivities.notifyReviewRequest(project) + } catch (notifyErr) { + // The row keeps a null reviewMessageTs and is fetched again next run; the batch must go on. + log.error( + `Failed to post the review request for project id=${project.id}: ${String(notifyErr)}`, + ) + } + } + if (remainingProjects.length > 0) { log.info(`Evaluating ${remainingProjects.length} project(s) (batch size: ${batchSize}).`) diff --git a/services/libs/data-access-layer/src/project-catalog/projectCatalog.test.ts b/services/libs/data-access-layer/src/project-catalog/projectCatalog.test.ts index ce2a2df526..4a4e360754 100644 --- a/services/libs/data-access-layer/src/project-catalog/projectCatalog.test.ts +++ b/services/libs/data-access-layer/src/project-catalog/projectCatalog.test.ts @@ -13,6 +13,7 @@ import { findProjectCatalogPendingEvaluation, findProjectCatalogPendingOnboarding, findProjectCatalogPendingReview, + findProjectCatalogReviewWithoutMessage, insertProjectCatalog, markProjectCatalogClassified, markProjectCatalogOnboardingFailed, @@ -21,6 +22,7 @@ import { promoteProjectCatalogIdsToEvaluate, promoteProjectCatalogProvenance, releaseProjectCatalogOnboardingClaim, + lockProjectCatalogReviewRequest, resolveProjectCatalogReview, setProjectCatalogReviewMessageTs, setProjectCatalogSourceUrl, @@ -722,6 +724,28 @@ describe('resolveProjectCatalogReview', () => { }) }) +describe('lockProjectCatalogReviewRequest', () => { + test('locks a review row without a message inside a transaction', async ({ qx }) => { + const inserted = await insertProjectCatalog(qx, catalogRow({ action: 'review' })) + + const locked = await qx.tx((tx) => lockProjectCatalogReviewRequest(tx, inserted.id)) + + expect(locked).toBe(true) + }) + + test('does not lock a row that already has a message or left review', async ({ qx }) => { + const withMessage = await insertProjectCatalog(qx, catalogRow({ action: 'review' })) + await setProjectCatalogReviewMessageTs(qx, withMessage.id, '1700000000.000100') + const onboard = await insertProjectCatalog( + qx, + catalogRow({ action: 'onboard', repoUrl: 'https://github.com/gerritcodereview/plugins' }), + ) + + expect(await qx.tx((tx) => lockProjectCatalogReviewRequest(tx, withMessage.id))).toBe(false) + expect(await qx.tx((tx) => lockProjectCatalogReviewRequest(tx, onboard.id))).toBe(false) + }) +}) + describe('setProjectCatalogReviewMessageTs', () => { test('stores the Slack message ts on a row in review', async ({ qx }) => { const inserted = await insertProjectCatalog(qx, catalogRow({ action: 'review' })) @@ -742,6 +766,49 @@ describe('setProjectCatalogReviewMessageTs', () => { expect(updatedRows).toBe(0) expect((await findProjectCatalogById(qx, inserted.id))?.reviewMessageTs).toBeNull() }) + + test('keeps the first message ts when a second request is posted', async ({ qx }) => { + const inserted = await insertProjectCatalog(qx, catalogRow({ action: 'review' })) + await setProjectCatalogReviewMessageTs(qx, inserted.id, '1700000000.000100') + + const updatedRows = await setProjectCatalogReviewMessageTs(qx, inserted.id, '1700000000.000200') + + expect(updatedRows).toBe(0) + expect((await findProjectCatalogById(qx, inserted.id))?.reviewMessageTs).toBe( + '1700000000.000100', + ) + }) +}) + +describe('findProjectCatalogReviewWithoutMessage', () => { + test('returns only rows in review that have no Slack message yet', async ({ qx }) => { + const pending = await insertProjectCatalog(qx, catalogRow({ action: 'review' })) + const posted = await insertProjectCatalog( + qx, + catalogRow({ action: 'review', repoUrl: 'https://github.com/foo/posted' }), + ) + await setProjectCatalogReviewMessageTs(qx, posted.id, '1700000000.000100') + await insertProjectCatalog( + qx, + catalogRow({ action: 'evaluate', repoUrl: 'https://github.com/foo/evaluate' }), + ) + + const rows = await findProjectCatalogReviewWithoutMessage(qx) + + expect(rows.map((row) => row.id)).toEqual([pending.id]) + }) + + test('restricts the rows to the requested ids', async ({ qx }) => { + const requested = await insertProjectCatalog(qx, catalogRow({ action: 'review' })) + await insertProjectCatalog( + qx, + catalogRow({ action: 'review', repoUrl: 'https://github.com/foo/other' }), + ) + + const rows = await findProjectCatalogReviewWithoutMessage(qx, { ids: [requested.id] }) + + expect(rows.map((row) => row.id)).toEqual([requested.id]) + }) }) describe('upsertProjectCatalogManualAction', () => { 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 3e11b1fa1e..1a050e07f2 100644 --- a/services/libs/data-access-layer/src/project-catalog/projectCatalog.ts +++ b/services/libs/data-access-layer/src/project-catalog/projectCatalog.ts @@ -176,6 +176,25 @@ export async function findProjectCatalogPendingReview( ) } +export async function findProjectCatalogReviewWithoutMessage( + qx: QueryExecutor, + options: { limit?: number; ids?: string[] } = {}, +): Promise { + const { limit, ids } = options + + return qx.select( + ` + SELECT ${prepareSelectColumns(PROJECT_CATALOG_COLUMNS)} + FROM "projectCatalog" + WHERE action = 'review' AND "reviewMessageTs" IS NULL + ${ids !== undefined ? 'AND id = ANY($(ids)::uuid[])' : ''} + ORDER BY "classifiedAt" ASC NULLS LAST, "createdAt" ASC, id ASC + ${limit !== undefined ? 'LIMIT $(limit)' : ''} + `, + { limit, ids }, + ) +} + export async function findExistingProjectCatalogRepoUrls( qx: QueryExecutor, repoUrls: string[], @@ -967,6 +986,24 @@ export async function resolveProjectCatalogReview( ) } +// Row lock held until the caller's transaction ends: concurrent scheduled and bot-started +// runs serialize on it, so only one of them posts the Slack card for the row. +export async function lockProjectCatalogReviewRequest( + qx: QueryExecutor, + id: string, +): Promise { + const row = await qx.selectOneOrNone( + ` + SELECT id + FROM "projectCatalog" + WHERE id = $(id) AND "action" = 'review' AND "reviewMessageTs" IS NULL + FOR UPDATE + `, + { id }, + ) + return row !== null +} + export async function setProjectCatalogReviewMessageTs( qx: QueryExecutor, id: string, @@ -976,7 +1013,7 @@ export async function setProjectCatalogReviewMessageTs( ` UPDATE "projectCatalog" SET "reviewMessageTs" = $(reviewMessageTs), "updatedAt" = NOW() - WHERE id = $(id) AND "action" = 'review' + WHERE id = $(id) AND "action" = 'review' AND "reviewMessageTs" IS NULL `, { id, reviewMessageTs }, ) diff --git a/services/libs/project-onboarding/src/requestClassificationAlert.ts b/services/libs/project-onboarding/src/requestClassificationAlert.ts index b08964e6ec..7748f04a5b 100644 --- a/services/libs/project-onboarding/src/requestClassificationAlert.ts +++ b/services/libs/project-onboarding/src/requestClassificationAlert.ts @@ -1,4 +1,4 @@ -import { SlackMessageSection } from '@crowd/slack' +import { SlackMessageSection, escapeSlackText } from '@crowd/slack' import { IPccCandidate, OnboardingResolution } from './requestResolver' @@ -26,10 +26,6 @@ const INTEGRATION_ACTION_LABELS = { human_review: 'Human review: GitHub v1 integration, migration to v2 needed', } as const -export function escapeSlackText(value: string): string { - return value.replace(/&/g, '&').replace(//g, '>') -} - function formatCandidate(candidate: IPccCandidate): string { const level = candidate.isLeaf ? 'project' : 'parent group' return `• ${escapeSlackText(candidate.name)} (${escapeSlackText(candidate.slug)}), score ${candidate.score.toFixed(2)}, ${level}` diff --git a/services/libs/slack/src/botConfig.ts b/services/libs/slack/src/botConfig.ts index 7658e12ba4..d4a4202636 100644 --- a/services/libs/slack/src/botConfig.ts +++ b/services/libs/slack/src/botConfig.ts @@ -7,6 +7,7 @@ const log = getServiceLogger() export function getSlackBotConfig(): SlackBotConfig { const botToken = process.env.CROWD_SLACK_BOT_TOKEN const signingSecret = process.env.CROWD_SLACK_SIGNING_SECRET + const reviewChannelId = process.env.CROWD_SLACK_REVIEW_CHANNEL_ID if (!botToken || !signingSecret) { log.warn( @@ -14,5 +15,5 @@ export function getSlackBotConfig(): SlackBotConfig { ) } - return { botToken, signingSecret } + return { botToken, signingSecret, reviewChannelId } } diff --git a/services/libs/slack/src/index.ts b/services/libs/slack/src/index.ts index 96d7452298..611204b30c 100644 --- a/services/libs/slack/src/index.ts +++ b/services/libs/slack/src/index.ts @@ -4,3 +4,5 @@ export * from './channels' export * from './notify' export * from './botConfig' export * from './chatClient' +export * from './mrkdwn' +export * from './reviewRequest' diff --git a/services/libs/slack/src/mrkdwn.ts b/services/libs/slack/src/mrkdwn.ts new file mode 100644 index 0000000000..4b2b9fca13 --- /dev/null +++ b/services/libs/slack/src/mrkdwn.ts @@ -0,0 +1,3 @@ +export function escapeSlackText(value: string): string { + return value.replace(/&/g, '&').replace(//g, '>') +} diff --git a/services/libs/slack/src/reviewRequest.test.ts b/services/libs/slack/src/reviewRequest.test.ts new file mode 100644 index 0000000000..330cc98ae7 --- /dev/null +++ b/services/libs/slack/src/reviewRequest.test.ts @@ -0,0 +1,146 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { postSlackMessage, getSlackBotConfig } = vi.hoisted(() => ({ + postSlackMessage: vi.fn(), + getSlackBotConfig: vi.fn(), +})) +vi.mock('./chatClient', () => ({ postSlackMessage })) +vi.mock('./botConfig', () => ({ getSlackBotConfig })) + +import { + APPROVE_REVIEW_ACTION_ID, + IReviewRequestCard, + REJECT_REVIEW_ACTION_ID, + buildReviewDecisionMessage, + buildReviewRequestBlocks, + postReviewRequest, +} from './reviewRequest' + +const card: IReviewRequestCard = { + id: 'catalog-1', + repoName: 'kops', + repoUrl: 'https://github.com/kubernetes/kops', + projectName: 'Kubernetes', + classificationKind: 'lf_in_cdp', + proposedAction: 'update_integration', + sourceUrl: 'https://lfx.slack.com/archives/C123/p1700000000000100', +} + +describe('buildReviewRequestBlocks', () => { + it('renders the summary, the classification and both buttons carrying the catalog id', () => { + const blocks = buildReviewRequestBlocks(card) + + expect(JSON.stringify(blocks)).toContain('*kops*') + expect(JSON.stringify(blocks)).toContain( + '', + ) + expect(JSON.stringify(blocks)).toContain('`lf_in_cdp`') + expect(JSON.stringify(blocks)).toContain('`update_integration`') + + const actions = blocks.find((block) => block.type === 'actions') + expect(actions).toMatchObject({ + elements: [ + { action_id: APPROVE_REVIEW_ACTION_ID, value: 'catalog-1' }, + { action_id: REJECT_REVIEW_ACTION_ID, value: 'catalog-1' }, + ], + }) + }) + + it('links a non-Slack origin as is and marks missing fields', () => { + const blocks = buildReviewRequestBlocks({ + ...card, + projectName: null, + proposedAction: null, + sourceUrl: 'https://github.com/org/repo/discussions/1', + }) + + const rendered = JSON.stringify(blocks) + expect(rendered).toContain( + '', + ) + expect(rendered).toContain('Project: _not provided_') + expect(rendered).toContain('*Proposed action*\\n_unknown_') + }) +}) + +describe('buildReviewRequestBlocks escaping', () => { + it('encodes mrkdwn control characters in untrusted fields', () => { + const rendered = JSON.stringify( + buildReviewRequestBlocks({ + ...card, + repoName: 'ac', + projectName: ' R&D', + sourceUrl: 'https://example.com/?a=1&b=<2>', + }), + ) + + expect(rendered).toContain('*a<b>c*') + expect(rendered).toContain('Project: <!channel> R&D') + expect(rendered).toContain( + '|https://example.com/?a=1&b=<2>>', + ) + expect(rendered).not.toContain('') + }) +}) + +describe('buildReviewDecisionMessage', () => { + it('drops the buttons and records who decided', () => { + const message = buildReviewDecisionMessage(card, 'approve', 'U123', { + onboardingStarted: true, + }) + + expect(message.text).toBe('Approved — kops') + expect(message.blocks.some((block) => block.type === 'actions')).toBe(false) + expect(JSON.stringify(message.blocks)).toContain('Approved by <@U123>, onboarding started.') + }) + + it('says the approved row is queued when onboarding did not start', () => { + const message = buildReviewDecisionMessage(card, 'approve', 'U123', { + onboardingStarted: false, + }) + + expect(JSON.stringify(message.blocks)).toContain( + 'Approved by <@U123>, queued for the next onboarding run.', + ) + }) + + it('renders a rejection', () => { + const message = buildReviewDecisionMessage(card, 'reject', 'U123', { + onboardingStarted: false, + }) + + expect(message.text).toBe('Rejected — kops') + expect(JSON.stringify(message.blocks)).toContain('Rejected by <@U123>') + }) +}) + +describe('postReviewRequest', () => { + beforeEach(() => { + postSlackMessage.mockReset() + getSlackBotConfig.mockReset() + }) + + it('fails without a review channel and never calls Slack', async () => { + getSlackBotConfig.mockReturnValue({ reviewChannelId: undefined }) + + const result = await postReviewRequest(card) + + expect(result).toEqual({ ok: false, error: 'Slack review channel not configured' }) + expect(postSlackMessage).not.toHaveBeenCalled() + }) + + it('posts the card to the review channel and returns the message ts', async () => { + getSlackBotConfig.mockReturnValue({ reviewChannelId: 'C-REVIEW' }) + postSlackMessage.mockResolvedValue({ ok: true, ts: '1.2' }) + + const result = await postReviewRequest(card) + + expect(result).toEqual({ ok: true, ts: '1.2' }) + expect(postSlackMessage).toHaveBeenCalledWith( + expect.objectContaining({ + channel: 'C-REVIEW', + text: 'Review needed — kops (update_integration)', + }), + ) + }) +}) diff --git a/services/libs/slack/src/reviewRequest.ts b/services/libs/slack/src/reviewRequest.ts new file mode 100644 index 0000000000..3302e182bd --- /dev/null +++ b/services/libs/slack/src/reviewRequest.ts @@ -0,0 +1,154 @@ +import type { KnownBlock } from '@slack/web-api' + +import { getSlackBotConfig } from './botConfig' +import { postSlackMessage } from './chatClient' +import { escapeSlackText } from './mrkdwn' + +export const APPROVE_REVIEW_ACTION_ID = 'approve_review' +export const REJECT_REVIEW_ACTION_ID = 'reject_review' + +export type ReviewDecision = 'approve' | 'reject' + +export interface IReviewRequestCard { + id: string + repoName: string + repoUrl: string + projectName: string | null + classificationKind: string | null + proposedAction: string | null + sourceUrl: string | null +} + +export interface IReviewDecisionMessage { + text: string + blocks: KnownBlock[] +} + +const SLACK_PERMALINK_PATTERN = /^https:\/\/[^/]+\.slack\.com\/archives\// + +function formatRequestedIn(sourceUrl: string | null): string { + if (!sourceUrl) { + return '_source not recorded_' + } + + return SLACK_PERMALINK_PATTERN.test(sourceUrl) + ? `<${sourceUrl}|Slack request>` + : `<${sourceUrl}|${escapeSlackText(sourceUrl)}>` +} + +function formatValue(value: string | null): string { + return value ? `\`${escapeSlackText(value)}\`` : '_unknown_' +} + +function summaryBlocks(card: IReviewRequestCard): KnownBlock[] { + return [ + { + type: 'section', + text: { + type: 'mrkdwn', + text: [ + `*${escapeSlackText(card.repoName)}*`, + escapeSlackText(card.repoUrl), + `Project: ${card.projectName ? escapeSlackText(card.projectName) : '_not provided_'}`, + `Requested in: ${formatRequestedIn(card.sourceUrl)}`, + ].join('\n'), + }, + }, + { + type: 'section', + fields: [ + { type: 'mrkdwn', text: `*Classification*\n${formatValue(card.classificationKind)}` }, + { type: 'mrkdwn', text: `*Proposed action*\n${formatValue(card.proposedAction)}` }, + ], + }, + ] +} + +export function buildReviewRequestBlocks(card: IReviewRequestCard): KnownBlock[] { + return [ + ...summaryBlocks(card), + { + type: 'actions', + elements: [ + { + type: 'button', + action_id: APPROVE_REVIEW_ACTION_ID, + value: card.id, + style: 'primary', + text: { type: 'plain_text', text: 'Approve' }, + confirm: { + title: { type: 'plain_text', text: 'Approve onboarding?' }, + text: { + type: 'mrkdwn', + text: `${formatValue(card.repoUrl)} will be onboarded with the proposed action ${formatValue(card.proposedAction)}.`, + }, + confirm: { type: 'plain_text', text: 'Approve' }, + deny: { type: 'plain_text', text: 'Cancel' }, + }, + }, + { + type: 'button', + action_id: REJECT_REVIEW_ACTION_ID, + value: card.id, + style: 'danger', + text: { type: 'plain_text', text: 'Reject' }, + confirm: { + title: { type: 'plain_text', text: 'Reject onboarding?' }, + text: { type: 'mrkdwn', text: `${formatValue(card.repoUrl)} will be skipped.` }, + confirm: { type: 'plain_text', text: 'Reject' }, + deny: { type: 'plain_text', text: 'Cancel' }, + }, + }, + ], + }, + ] +} + +export function buildReviewRequestText(card: IReviewRequestCard): string { + return `Review needed — ${escapeSlackText(card.repoName)} (${card.proposedAction ?? 'unknown action'})` +} + +function decisionOutcome( + decision: ReviewDecision, + reviewedBy: string, + onboardingStarted: boolean, +): string { + if (decision === 'reject') { + return `:no_entry_sign: Rejected by <@${reviewedBy}>, the repository was skipped.` + } + return onboardingStarted + ? `:white_check_mark: Approved by <@${reviewedBy}>, onboarding started.` + : `:white_check_mark: Approved by <@${reviewedBy}>, queued for the next onboarding run.` +} + +export function buildReviewDecisionMessage( + card: IReviewRequestCard, + decision: ReviewDecision, + reviewedBy: string, + options: { onboardingStarted: boolean }, +): IReviewDecisionMessage { + const outcome = decisionOutcome(decision, reviewedBy, options.onboardingStarted) + + return { + text: `${decision === 'approve' ? 'Approved' : 'Rejected'} — ${escapeSlackText(card.repoName)}`, + blocks: [ + ...summaryBlocks(card), + { type: 'context', elements: [{ type: 'mrkdwn', text: outcome }] }, + ], + } +} + +export async function postReviewRequest( + card: IReviewRequestCard, +): Promise<{ ok: boolean; ts?: string; error?: string }> { + const { reviewChannelId } = getSlackBotConfig() + if (!reviewChannelId) { + return { ok: false, error: 'Slack review channel not configured' } + } + + return postSlackMessage({ + channel: reviewChannelId, + text: buildReviewRequestText(card), + blocks: buildReviewRequestBlocks(card), + }) +} diff --git a/services/libs/slack/src/types.ts b/services/libs/slack/src/types.ts index e3eed1e1ff..7038e8e9fe 100644 --- a/services/libs/slack/src/types.ts +++ b/services/libs/slack/src/types.ts @@ -46,4 +46,5 @@ export interface SlackChannelConfig { export interface SlackBotConfig { botToken: string | undefined signingSecret: string | undefined + reviewChannelId: string | undefined }