diff --git a/apps/bullmq/src/jobs/pr-review-notification.test.ts b/apps/bullmq/src/jobs/pr-review-notification.test.ts index 029a9dfdf..a3fa69ab0 100644 --- a/apps/bullmq/src/jobs/pr-review-notification.test.ts +++ b/apps/bullmq/src/jobs/pr-review-notification.test.ts @@ -16,6 +16,7 @@ const { mockStickyFooterPost, mockSetPendingPrReviewAction, mockDispatchFollowUp, + mockReplayAssociation, } = vi.hoisted(() => ({ mockFindFirstTaskRun: vi.fn(), mockFindFirstTaskPullRequest: vi.fn(), @@ -32,6 +33,7 @@ const { mockStickyFooterPost: vi.fn(), mockSetPendingPrReviewAction: vi.fn(), mockDispatchFollowUp: vi.fn(), + mockReplayAssociation: vi.fn(), })); vi.mock('@roomote/db/server', () => ({ @@ -86,9 +88,19 @@ vi.mock('@roomote/sdk/server', () => ({ immediate: z.boolean().optional(), batchKind: z.enum(['human', 'roomote']).optional(), batchId: z.string().optional(), + sourceControlProvider: z.string().optional(), + }), + prReviewAssociationReplayRequestSchema: z.object({ + kind: z.literal('association_replay'), + sourceControlProvider: z.string(), + repository: z.string(), + prNumber: z.number(), + chainId: z.string(), + attempt: z.number(), }), consumePendingPrReviewActivity: mockConsumePending, requeuePendingPrReviewActivity: mockRequeuePending, + replayPrReviewNotificationAssociation: mockReplayAssociation, schedulePrReviewNotificationJob: mockSchedule, getCommunicationProviderAdapter: vi.fn( async (provider: 'slack' | 'teams' | 'telegram' | 'discord') => @@ -177,6 +189,32 @@ describe('prReviewNotificationJob', () => { }); }); + it('dispatches association replay jobs without entering delivery', async () => { + const replay = { + kind: 'association_replay', + sourceControlProvider: 'github', + repository: 'owner/repo', + prNumber: 42, + chainId: '00000000-0000-4000-8000-000000000001', + attempt: 2, + }; + + await prReviewNotificationJob({ data: replay } as never); + + expect(mockReplayAssociation).toHaveBeenCalledWith(replay); + expect(mockFindFirstTaskRun).not.toHaveBeenCalled(); + }); + + it('drains non-default provider pending activity from its provider key', async () => { + await prReviewNotificationJob( + makeJob({ sourceControlProvider: 'gitlab' }) as never, + ); + + expect(mockConsumePending).toHaveBeenCalledWith( + expect.objectContaining({ sourceControlProvider: 'gitlab' }), + ); + }); + it('posts the aggregated notification to the originating Slack thread when the task is idle', async () => { await prReviewNotificationJob(makeJob() as never); @@ -442,6 +480,7 @@ describe('prReviewNotificationJob', () => { taskId: 'task-1', repository: 'owner/repo', prNumber: 42, + sourceControlProvider: 'github', batchKind: 'roomote', batchId: 'cycle-1', immediate: true, @@ -742,6 +781,7 @@ describe('prReviewNotificationJob', () => { taskId: 'task-1', repository: 'owner/repo', prNumber: 42, + sourceControlProvider: 'github', }, events, }); @@ -759,6 +799,7 @@ describe('prReviewNotificationJob', () => { taskId: 'task-1', repository: 'owner/repo', prNumber: 42, + sourceControlProvider: 'github', }, events, }); diff --git a/apps/bullmq/src/jobs/pr-review-notification.ts b/apps/bullmq/src/jobs/pr-review-notification.ts index a15f0ebb1..e3dc6724b 100644 --- a/apps/bullmq/src/jobs/pr-review-notification.ts +++ b/apps/bullmq/src/jobs/pr-review-notification.ts @@ -17,14 +17,16 @@ import { PR_REVIEW_NOTIFICATION_MAX_DEFERRALS, attachPendingPrReviewActionMessage, getCommunicationProviderAdapter, - type PrReviewNotificationRequest, + type PrReviewNotificationQueueRequest, type PrReviewNotificationRoute, consumePendingPrReviewActivity, dispatchPrReviewFollowUp, preparePrReviewNotificationDelivery, prReviewNotificationRequestSchema, + prReviewAssociationReplayRequestSchema, recordPrReviewNotificationDeliveryBestEffort, requeuePendingPrReviewActivity, + replayPrReviewNotificationAssociation, schedulePrReviewNotificationJob, setPendingPrReviewAction, } from '@roomote/sdk/server'; @@ -39,7 +41,11 @@ import { WORKER_HEARTBEAT_STALE_MS, } from '@roomote/types'; -type PrReviewNotificationJob = Job; +type PrReviewNotificationJob = Job< + PrReviewNotificationQueueRequest, + void, + string +>; function buildPrReviewNotificationPostInput( route: PrReviewNotificationRoute, @@ -212,6 +218,13 @@ async function postPrReviewNotification({ export const prReviewNotificationJob = async ( job: PrReviewNotificationJob, ): Promise => { + const replay = prReviewAssociationReplayRequestSchema.safeParse(job.data); + + if (replay.success) { + await replayPrReviewNotificationAssociation(replay.data); + return; + } + const parsed = prReviewNotificationRequestSchema.safeParse(job.data); if (!parsed.success) { @@ -225,6 +238,7 @@ export const prReviewNotificationJob = async ( taskId: data.taskId, repository: data.repository, prNumber: data.prNumber, + sourceControlProvider: data.sourceControlProvider ?? ('github' as const), ...(data.batchKind ? { batchKind: data.batchKind } : {}), ...(data.batchId ? { batchId: data.batchId } : {}), ...(data.immediate ? { immediate: true } : {}), diff --git a/apps/bullmq/src/pr-review-notification-queue.ts b/apps/bullmq/src/pr-review-notification-queue.ts index 44e4e784d..ada7390f8 100644 --- a/apps/bullmq/src/pr-review-notification-queue.ts +++ b/apps/bullmq/src/pr-review-notification-queue.ts @@ -2,16 +2,20 @@ import { Queue, QueueEvents, Worker } from 'bullmq'; import { PR_REVIEW_NOTIFICATION_QUEUE_NAME, - type PrReviewNotificationRequest, + type PrReviewNotificationQueueRequest, } from '@roomote/sdk/server'; import { prReviewNotificationJob } from './jobs/pr-review-notification'; import { getRedis } from './redis'; +function formatJobTarget(data: PrReviewNotificationQueueRequest): string { + return `${data.repository}#${data.prNumber}`; +} + export function startPrReviewNotificationQueue() { const connection = getRedis(); - const queue = new Queue( + const queue = new Queue( PR_REVIEW_NOTIFICATION_QUEUE_NAME, { connection, @@ -24,7 +28,7 @@ export function startPrReviewNotificationQueue() { }, ); - const worker = new Worker( + const worker = new Worker( PR_REVIEW_NOTIFICATION_QUEUE_NAME, prReviewNotificationJob, { connection, concurrency: 5, autorun: true }, @@ -32,7 +36,9 @@ export function startPrReviewNotificationQueue() { worker.on('failed', (job, err) => console.error( - `[PrReviewNotificationQueue] job ${job?.id} failed for ${job?.data.repository}#${job?.data.prNumber}:`, + `[PrReviewNotificationQueue] job ${job?.id} failed for ${ + job?.data ? formatJobTarget(job.data) : 'unknown pull request' + }:`, err.message, ), ); diff --git a/apps/bullmq/src/scheduled-jobs/index.ts b/apps/bullmq/src/scheduled-jobs/index.ts index 31167ca45..78327c3b5 100644 --- a/apps/bullmq/src/scheduled-jobs/index.ts +++ b/apps/bullmq/src/scheduled-jobs/index.ts @@ -6,3 +6,4 @@ export { instancePingJob } from './instance-ping'; export { licenseUsageSyncJob } from './license-usage-sync'; export { webhookCleanupJob } from './webhook-cleanup'; export { standbyRetentionJob } from './standby-retention'; +export { prReviewAssociationRepairJob } from './pr-review-association-repair'; diff --git a/apps/bullmq/src/scheduled-jobs/pr-review-association-repair.test.ts b/apps/bullmq/src/scheduled-jobs/pr-review-association-repair.test.ts new file mode 100644 index 000000000..b63c17720 --- /dev/null +++ b/apps/bullmq/src/scheduled-jobs/pr-review-association-repair.test.ts @@ -0,0 +1,18 @@ +const mockRepair = vi.fn(); + +vi.mock('@roomote/sdk/server', () => ({ + repairOrphanPrReviewAssociationReplays: (...args: unknown[]) => + mockRepair(...args), +})); + +import { prReviewAssociationRepairJob } from './pr-review-association-repair'; + +describe('prReviewAssociationRepairJob', () => { + it('runs bounded orphan replay repair', async () => { + mockRepair.mockResolvedValue(undefined); + + await prReviewAssociationRepairJob(); + + expect(mockRepair).toHaveBeenCalledTimes(1); + }); +}); diff --git a/apps/bullmq/src/scheduled-jobs/pr-review-association-repair.ts b/apps/bullmq/src/scheduled-jobs/pr-review-association-repair.ts new file mode 100644 index 000000000..1b5b6777f --- /dev/null +++ b/apps/bullmq/src/scheduled-jobs/pr-review-association-repair.ts @@ -0,0 +1,5 @@ +import { repairOrphanPrReviewAssociationReplays } from '@roomote/sdk/server'; + +export async function prReviewAssociationRepairJob(): Promise { + await repairOrphanPrReviewAssociationReplays(); +} diff --git a/apps/bullmq/src/scheduler.ts b/apps/bullmq/src/scheduler.ts index 68070f36b..da6040e37 100644 --- a/apps/bullmq/src/scheduler.ts +++ b/apps/bullmq/src/scheduler.ts @@ -30,6 +30,7 @@ import { licenseUsageSyncJob, webhookCleanupJob, standbyRetentionJob, + prReviewAssociationRepairJob, } from './scheduled-jobs'; const QUEUE_NAME = 'scheduled-jobs'; @@ -88,6 +89,10 @@ async function createJobs(queue: Queue): Promise { { every: 60 * 1000 }, // Every 60 seconds. ); + await queue.upsertJobScheduler(ScheduledJobName.PrReviewAssociationRepair, { + every: 60 * 1000, + }); + await queue.upsertJobScheduler( ScheduledJobName.StandbyRetention, { every: 5 * 60 * 1000 }, // Every 5 minutes. @@ -216,6 +221,8 @@ const runJobs = async (job: ScheduledJob): Promise => { return webhookCleanupJob(); case ScheduledJobName.StandbyRetention: return standbyRetentionJob(); + case ScheduledJobName.PrReviewAssociationRepair: + return prReviewAssociationRepairJob(); case ScheduledJobName.CustomAutomations: await customAutomationsJob(); return; diff --git a/apps/bullmq/src/types.ts b/apps/bullmq/src/types.ts index c63a24cfd..3d9945f16 100644 --- a/apps/bullmq/src/types.ts +++ b/apps/bullmq/src/types.ts @@ -14,6 +14,7 @@ export enum ScheduledJobName { WebhookCleanup = 'WebhookCleanup', StandbyRetention = 'StandbyRetention', CustomAutomations = 'custom_automations', + PrReviewAssociationRepair = 'PrReviewAssociationRepair', } /** diff --git a/packages/sdk/src/server/index.ts b/packages/sdk/src/server/index.ts index 008d3181a..3e54acb32 100644 --- a/packages/sdk/src/server/index.ts +++ b/packages/sdk/src/server/index.ts @@ -197,6 +197,9 @@ export { PR_REVIEW_NOTIFICATION_MAX_DEFERRALS, PR_REVIEW_NOTIFICATION_QUEUE_NAME, PR_REVIEW_NOTIFICATION_ROOMOTE_FALLBACK_MS, + PR_REVIEW_ASSOCIATION_REPLAY_DELAYS_MS, + ORPHAN_REPLAY_REPAIR_DELAY_MS, + wakePrReviewNotificationAssociation, consumePendingPrReviewActivity, enqueuePrReviewNotification, enqueuePrReviewNotificationInputSchema, @@ -204,14 +207,20 @@ export { getPrReviewCompletedCycleKey, hasPrReviewNotificationThreadContext, prReviewActivityEventSchema, + prReviewAssociationReplayRequestSchema, + prReviewNotificationQueueRequestSchema, prReviewNotificationRequestSchema, requeuePendingPrReviewActivity, + replayPrReviewNotificationAssociation, + repairOrphanPrReviewAssociationReplays, resolvePrReviewNotificationRoute, schedulePrReviewNotificationJob, startPrReviewNotificationCycle, startPrReviewNotificationCycleInputSchema, type EnqueuePrReviewNotificationInput, type PrReviewActivityEvent, + type PrReviewAssociationReplayRequest, + type PrReviewNotificationQueueRequest, type PrReviewNotificationRequest, type PrReviewNotificationRoute, type StartPrReviewNotificationCycleInput, diff --git a/packages/sdk/src/server/lib/pull-requests/__tests__/source-control-pull-requests.test.ts b/packages/sdk/src/server/lib/pull-requests/__tests__/source-control-pull-requests.test.ts index 55fcea6cf..20fd534bf 100644 --- a/packages/sdk/src/server/lib/pull-requests/__tests__/source-control-pull-requests.test.ts +++ b/packages/sdk/src/server/lib/pull-requests/__tests__/source-control-pull-requests.test.ts @@ -27,6 +27,7 @@ const { mockTasksFindFirst, mockResolveLaunchTaskCommitAuthor, mockResolveRunCommitAuthor, + mockWakePrReviewNotificationAssociation, } = vi.hoisted(() => ({ mockCreateGitHubToken: vi.fn(), mockGetDeploymentPrAction: vi.fn(), @@ -47,6 +48,12 @@ const { mockTasksFindFirst: vi.fn(), mockResolveLaunchTaskCommitAuthor: vi.fn(), mockResolveRunCommitAuthor: vi.fn(), + mockWakePrReviewNotificationAssociation: vi.fn(), +})); + +vi.mock('../../task-runs/pr-review-notification', () => ({ + wakePrReviewNotificationAssociation: (...args: unknown[]) => + mockWakePrReviewNotificationAssociation(...args), })); vi.mock('@roomote/cloud-agents/server', () => ({ @@ -208,6 +215,9 @@ function attributionBody( } beforeEach(() => { + mockTaskPullRequestUpsert.mockReset(); + mockTaskPullRequestUpsert.mockResolvedValue(undefined); + mockWakePrReviewNotificationAssociation.mockResolvedValue(false); mockGetDeploymentGitHubRoomoteMentionEnabled.mockResolvedValue(true); mockResolveTelegramRuntimeCredentials.mockResolvedValue({ botUsername: 'roomote_bot', @@ -511,6 +521,113 @@ describe('platform-managed draft state', () => { expect(result.warnings).toEqual([]); }); + it('retries association persistence after the provider mutation succeeds', async () => { + makeOctokit({ + created: { + number: 9, + node_id: 'node-9', + html_url: 'https://github.com/acme/web/pull/9', + title: '[Feature] X', + draft: true, + }, + }); + mockTaskPullRequestUpsert + .mockRejectedValueOnce(new Error('database unavailable')) + .mockResolvedValueOnce(undefined); + + const result = await createOrUpdateSourceControlPullRequestForTaskRun({ + taskRun: makeTaskRun({ repo: 'acme/web' }), + input: { ...githubInput }, + }); + + expect(result).toMatchObject({ action: 'created', number: 9 }); + expect(mockTaskPullRequestUpsert).toHaveBeenCalledTimes(2); + }); + + it('wakes retained review activity after association persistence commits', async () => { + mockRepositoriesFindFirst.mockResolvedValue({ + id: 1, + sourceControlProvider: 'github', + installationId: 123, + fullName: 'owner/repo', + host: 'github.com', + htmlUrl: 'https://github.com/owner/repo', + private: false, + }); + mockCreateGitHubToken.mockResolvedValue('token'); + mockGetOctokit.mockReturnValue({ + rest: { + pulls: { + list: vi.fn().mockResolvedValue({ data: [] }), + create: vi.fn().mockResolvedValue({ + data: { + number: 42, + node_id: 'PR_42', + html_url: 'https://github.com/owner/repo/pull/42', + title: '[Fix] Preserve reviews', + draft: true, + base: { ref: 'develop' }, + }, + }), + }, + issues: { addAssignees: vi.fn(), addLabels: vi.fn() }, + }, + }); + + await createOrUpdateSourceControlPullRequestForTaskRun({ + taskRun: makeTaskRun({ repo: 'owner/repo' } as TaskRun['payload']), + input: { + action: 'create_or_update_pull_request', + repositoryFullName: 'owner/repo', + sourceBranch: 'fix/reviews', + targetBranch: 'develop', + title: '[Fix] Preserve reviews', + body: 'body', + labels: [], + assignees: [], + }, + }); + + expect(mockWakePrReviewNotificationAssociation).toHaveBeenCalledWith({ + sourceControlProvider: 'github', + repository: 'owner/repo', + prNumber: 42, + }); + }); + + it('keeps provider success after bounded association retries are exhausted', async () => { + makeOctokit({ + created: { + number: 9, + node_id: 'node-9', + html_url: 'https://github.com/acme/web/pull/9', + title: '[Feature] X', + draft: true, + }, + }); + mockTaskPullRequestUpsert.mockRejectedValue( + new Error('database unavailable'), + ); + const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined); + + const result = await createOrUpdateSourceControlPullRequestForTaskRun({ + taskRun: makeTaskRun({ repo: 'acme/web' }), + input: { ...githubInput }, + }); + + expect(result).toMatchObject({ action: 'created', number: 9 }); + expect(mockTaskPullRequestUpsert).toHaveBeenCalledTimes(3); + expect(result.warnings).toEqual([ + expect.stringContaining( + 'could not link it to this task after 3 attempts', + ), + ]); + expect(warn).toHaveBeenCalledWith( + expect.stringContaining('after 3 attempts'), + ); + warn.mockRestore(); + }); + it('uses the @roomote shorthand in attribution when enabled', async () => { mockResolveConfiguredGitHubAppSlugIfConfigured.mockResolvedValue( 'roomote-roomote', diff --git a/packages/sdk/src/server/lib/pull-requests/source-control-pull-requests.ts b/packages/sdk/src/server/lib/pull-requests/source-control-pull-requests.ts index 85a4380b9..a2765bf1c 100644 --- a/packages/sdk/src/server/lib/pull-requests/source-control-pull-requests.ts +++ b/packages/sdk/src/server/lib/pull-requests/source-control-pull-requests.ts @@ -77,8 +77,11 @@ import { type FetchImpl, type RepositoryRow, } from './source-control-pull-request-shared'; +import { wakePrReviewNotificationAssociation } from '../task-runs/pr-review-notification'; const ADO_API_VERSION = '7.1'; +const PR_ASSOCIATION_MAX_ATTEMPTS = 3; +const PR_ASSOCIATION_RETRY_DELAY_MS = 100; export const sourceControlPullRequestMutationInputSchema = z.object({ action: z.literal('create_or_update_pull_request'), @@ -391,13 +394,15 @@ export async function createOrUpdateSourceControlPullRequestForTaskRun({ } })(); - await persistSourceControlPullRequestAssociation({ + const associationWarning = await persistSourceControlPullRequestAssociation({ taskRun, result, repository, }); - return result; + return associationWarning + ? { ...result, warnings: [...result.warnings, associationWarning] } + : result; } async function resolveLiveGitHubAssigneePlan({ @@ -452,10 +457,10 @@ async function resolveEffectivePrAction(taskRun: TaskRun): Promise { * delivery path used to create this association by parsing `gh pr create` * tool output from the transcript; the server-side mutation path knows the * pull request authoritatively for every provider, so it persists the - * association directly. Association failures must not fail the mutation the - * agent already performed. + * association directly. Transient association failures are retried, but must + * not fail the provider mutation the agent already performed. */ -async function persistSourceControlPullRequestAssociation({ +export async function persistSourceControlPullRequestAssociation({ taskRun, result, repository, @@ -463,48 +468,72 @@ async function persistSourceControlPullRequestAssociation({ taskRun: TaskRun; result: SourceControlPullRequestMutationResult; repository: RepositoryRow; -}): Promise { +}): Promise { if (!taskRun.taskId) { - return; + return null; } const status = result.draft ? 'draft' : 'open'; - try { - await db - .insert(taskPullRequests) - .values({ - taskId: taskRun.taskId, - sourceControlProvider: repository.sourceControlProvider, - host: repository.host, - repositoryId: repository.id, - prUrl: result.url, - prNumber: result.number, - prTitle: result.title, - repository: result.repositoryFullName, - status, - createdByRoomote: result.action === 'created', - prBaseRef: result.targetBranch, - }) - .onConflictDoUpdate({ - target: [taskPullRequests.taskId, taskPullRequests.prUrl], - set: { + for (let attempt = 1; attempt <= PR_ASSOCIATION_MAX_ATTEMPTS; attempt += 1) { + try { + await db + .insert(taskPullRequests) + .values({ + taskId: taskRun.taskId, sourceControlProvider: repository.sourceControlProvider, host: repository.host, repositoryId: repository.id, + prUrl: result.url, + prNumber: result.number, prTitle: result.title, + repository: result.repositoryFullName, status, + createdByRoomote: result.action === 'created', prBaseRef: result.targetBranch, - updatedAt: new Date(), - }, - }); - } catch (error) { - console.warn( - `[persistSourceControlPullRequestAssociation] Failed to associate ${result.repositoryFullName}#${result.number} with task ${taskRun.taskId}: ${ - error instanceof Error ? error.message : String(error) - }`, - ); + }) + .onConflictDoUpdate({ + target: [taskPullRequests.taskId, taskPullRequests.prUrl], + set: { + sourceControlProvider: repository.sourceControlProvider, + host: repository.host, + repositoryId: repository.id, + prTitle: result.title, + status, + prBaseRef: result.targetBranch, + updatedAt: new Date(), + }, + }); + try { + await wakePrReviewNotificationAssociation({ + sourceControlProvider: repository.sourceControlProvider, + repository: result.repositoryFullName, + prNumber: result.number, + }); + } catch (error) { + console.warn( + `[persistSourceControlPullRequestAssociation] Linked ${result.repositoryFullName}#${result.number}, but could not wake retained review activity: ${error instanceof Error ? error.message : String(error)}`, + ); + } + return null; + } catch (error) { + if (attempt < PR_ASSOCIATION_MAX_ATTEMPTS) { + await new Promise((resolve) => + setTimeout(resolve, PR_ASSOCIATION_RETRY_DELAY_MS), + ); + continue; + } + + console.warn( + `[persistSourceControlPullRequestAssociation] Failed to associate ${result.repositoryFullName}#${result.number} with task ${taskRun.taskId} after ${attempt} attempts: ${ + error instanceof Error ? error.message : String(error) + }`, + ); + return `The pull request was ${result.action}, but Roomote could not link it to this task after ${attempt} attempts. Review feedback may not reach the task until create_or_update_pull_request is retried.`; + } } + + return null; } async function createOrUpdateGitHubPullRequest({ diff --git a/packages/sdk/src/server/lib/task-runs/__tests__/pr-review-association-repair.redis.test.ts b/packages/sdk/src/server/lib/task-runs/__tests__/pr-review-association-repair.redis.test.ts new file mode 100644 index 000000000..0b36a9d0f --- /dev/null +++ b/packages/sdk/src/server/lib/task-runs/__tests__/pr-review-association-repair.redis.test.ts @@ -0,0 +1,231 @@ +const mockFindManyTaskPullRequests = vi.fn(); +const mockQueueAdd = vi.fn(); +const mockTaskPullRequestUpsert = vi.fn(); + +vi.hoisted(() => { + process.env.REDIS_URL ??= 'redis://localhost:16379'; +}); + +vi.mock('@roomote/db/server', async () => { + const actual = + await vi.importActual( + '@roomote/db/server', + ); + + return { + ...actual, + db: { + query: { + taskPullRequests: { + findMany: (...args: unknown[]) => + mockFindManyTaskPullRequests(...args), + }, + }, + insert: () => ({ + values: (values: unknown) => ({ + onConflictDoUpdate: async () => { + await mockTaskPullRequestUpsert(values); + mockFindManyTaskPullRequests.mockResolvedValue([ + { taskId: 'task-1' }, + ]); + }, + }), + }), + }, + }; +}); + +vi.mock('bullmq', () => ({ + Queue: class MockQueue { + add = (...args: unknown[]) => mockQueueAdd(...args); + }, +})); + +vi.mock('../slack-task-run-routing', () => ({ + resolveSlackTaskRunRouting: vi.fn(), +})); + +import { getRedis } from '@roomote/redis'; + +import { + enqueuePrReviewNotification, + ORPHAN_REPLAY_REPAIR_DELAY_MS, + repairOrphanPrReviewAssociationReplays, + replayPrReviewNotificationAssociation, +} from '../pr-review-notification'; +import { persistSourceControlPullRequestAssociation } from '../../pull-requests/source-control-pull-requests'; + +describe('orphan PR review association repair with Redis', () => { + const redis = getRedis(); + const repository = `owner/association-repair-${process.pid}`; + const prNumber = 42; + const orphanKey = `pr-review-notification:orphan:github:${encodeURIComponent(repository)}#${prNumber}`; + const markerKey = `pr-review-notification:orphan-replay:github:${encodeURIComponent(repository)}#${prNumber}`; + const repairIndexKey = 'pr-review-notification:orphan-replay-repair'; + const repairMember = `github:${encodeURIComponent(repository)}#${prNumber}`; + const repairPayloadKey = `pr-review-notification:orphan-replay-repair-payload:${repairMember}`; + + beforeEach(async () => { + vi.clearAllMocks(); + mockFindManyTaskPullRequests.mockResolvedValue([]); + mockTaskPullRequestUpsert.mockResolvedValue(undefined); + mockQueueAdd.mockResolvedValue(undefined); + await redis.del(orphanKey, markerKey, repairPayloadKey); + await redis.zrem(repairIndexKey, repairMember); + }); + + afterAll(async () => { + await redis.del(orphanKey, markerKey, repairPayloadKey); + await redis.zrem(repairIndexKey, repairMember); + await redis.quit(); + }); + + it('recovers repeated post-association enqueue failures without another webhook and drains once', async () => { + const input = { + repository, + prNumber, + prUrl: `https://github.com/${repository}/pull/${prNumber}`, + event: { + kind: 'review' as const, + authorLogin: 'alice', + reviewState: 'changes_requested', + }, + }; + + await enqueuePrReviewNotification(input); + mockQueueAdd.mockClear(); + + mockQueueAdd + .mockRejectedValueOnce(new Error('queue down')) + .mockRejectedValueOnce(new Error('queue still down')) + .mockResolvedValueOnce(undefined) + .mockResolvedValueOnce(undefined); + + await expect( + persistSourceControlPullRequestAssociation({ + taskRun: { taskId: 'task-1' } as never, + result: { + success: true, + action: 'created', + provider: 'github', + repositoryFullName: repository, + number: prNumber, + url: `https://github.com/${repository}/pull/${prNumber}`, + title: 'Review repair', + targetBranch: 'develop', + draft: true, + warnings: [], + }, + repository: { + id: 1, + sourceControlProvider: 'github', + host: 'github.com', + } as never, + }), + ).resolves.toBeNull(); + + expect(mockTaskPullRequestUpsert).toHaveBeenCalledTimes(1); + expect(await redis.llen(orphanKey)).toBe(1); + expect(await redis.exists(markerKey)).toBe(0); + expect(await redis.exists(repairPayloadKey)).toBe(1); + expect(await redis.zscore(repairIndexKey, repairMember)).not.toBeNull(); + + const repairNow = Date.now() + ORPHAN_REPLAY_REPAIR_DELAY_MS; + await repairOrphanPrReviewAssociationReplays({ now: repairNow }); + expect(await redis.llen(orphanKey)).toBe(1); + expect(await redis.exists(markerKey)).toBe(0); + expect(await redis.exists(repairPayloadKey)).toBe(1); + + await repairOrphanPrReviewAssociationReplays({ now: repairNow }); + const replayCall = mockQueueAdd.mock.calls.at(-1); + expect(replayCall?.[0]).toBe('replay-pr-review-association'); + const replayRequest = replayCall?.[1]; + + await replayPrReviewNotificationAssociation(replayRequest); + expect(await redis.exists(orphanKey)).toBe(0); + expect(await redis.exists(markerKey)).toBe(0); + expect(await redis.exists(repairPayloadKey)).toBe(0); + expect(await redis.zscore(repairIndexKey, repairMember)).toBeNull(); + + await replayPrReviewNotificationAssociation(replayRequest); + expect( + mockQueueAdd.mock.calls.filter( + ([name]) => name === 'notify-pr-review-activity', + ), + ).toHaveLength(1); + }); + + it.each([ + { name: 'missing', stalePayload: null }, + { name: 'invalid', stalePayload: '{invalid' }, + ])( + 'preserves a newer repair intent during $name stale cleanup and recovers it', + async ({ stalePayload }) => { + const chainId = '00000000-0000-4000-8000-000000000002'; + const request = { + kind: 'association_replay' as const, + sourceControlProvider: 'github' as const, + repository, + prNumber, + chainId, + attempt: 1, + }; + const newPayload = JSON.stringify(request); + const repairNow = Date.now() + ORPHAN_REPLAY_REPAIR_DELAY_MS; + const event = JSON.stringify({ + repository, + prNumber, + prUrl: `https://github.com/${repository}/pull/${prNumber}`, + event: { + kind: 'review', + authorLogin: 'alice', + reviewState: 'changes_requested', + }, + }); + + await redis.rpush(orphanKey, event); + await redis.set(markerKey, `${chainId}:0`); + if (stalePayload !== null) { + await redis.set(repairPayloadKey, stalePayload); + } + await redis.zadd(repairIndexKey, repairNow, repairMember); + + const originalGet = redis.get.bind(redis); + const getSpy = vi + .spyOn(redis, 'get') + .mockImplementationOnce(async (key) => { + const staleValue = await originalGet(key); + await redis + .multi() + .set(repairPayloadKey, newPayload, 'EX', 15 * 60) + .zadd(repairIndexKey, repairNow, repairMember) + .exec(); + return staleValue; + }); + + await repairOrphanPrReviewAssociationReplays({ now: repairNow }); + getSpy.mockRestore(); + + expect(await redis.get(repairPayloadKey)).toBe(newPayload); + expect(await redis.zscore(repairIndexKey, repairMember)).not.toBeNull(); + + mockQueueAdd.mockRejectedValueOnce(new Error('queue still down')); + await repairOrphanPrReviewAssociationReplays({ now: repairNow }); + expect(await redis.get(repairPayloadKey)).toBe(newPayload); + expect(await redis.zscore(repairIndexKey, repairMember)).not.toBeNull(); + + await repairOrphanPrReviewAssociationReplays({ now: repairNow }); + expect(mockQueueAdd).toHaveBeenCalledWith( + 'replay-pr-review-association', + request, + { jobId: `association-replay-${chainId}-1` }, + ); + + mockFindManyTaskPullRequests.mockResolvedValue([{ taskId: 'task-1' }]); + await replayPrReviewNotificationAssociation(request); + expect(await redis.exists(orphanKey)).toBe(0); + expect(await redis.exists(repairPayloadKey)).toBe(0); + expect(await redis.zscore(repairIndexKey, repairMember)).toBeNull(); + }, + ); +}); diff --git a/packages/sdk/src/server/lib/task-runs/__tests__/pr-review-notification.test.ts b/packages/sdk/src/server/lib/task-runs/__tests__/pr-review-notification.test.ts index c6b7c5e70..5f34e2881 100644 --- a/packages/sdk/src/server/lib/task-runs/__tests__/pr-review-notification.test.ts +++ b/packages/sdk/src/server/lib/task-runs/__tests__/pr-review-notification.test.ts @@ -7,11 +7,22 @@ const mockQueueAdd = vi.fn(); const mockMultiExec = vi.fn(); const mockResolveSlackTaskRunRouting = vi.fn(); const multiCalls: Array<{ command: string; args: unknown[] }> = []; +const replayChainId = '00000000-0000-4000-8000-000000000001'; +let orphanChainId = replayChainId; +let orphanRevision = 0; function createMultiMock() { const multi: Record = {}; - for (const command of ['rpush', 'expire', 'lrange', 'del']) { + for (const command of [ + 'rpush', + 'expire', + 'lrange', + 'del', + 'set', + 'zadd', + 'zrem', + ]) { multi[command] = (...args: unknown[]) => { multiCalls.push({ command, args }); return multi; @@ -65,24 +76,53 @@ vi.mock('../slack-task-run-routing', () => ({ import { PR_REVIEW_NOTIFICATION_DEBOUNCE_MS, + PR_REVIEW_ASSOCIATION_REPLAY_DELAYS_MS, consumePendingPrReviewActivity, enqueuePrReviewNotification, formatPrReviewActivityMessage, hasPrReviewNotificationThreadContext, resolvePrReviewNotificationRoute, + replayPrReviewNotificationAssociation, startPrReviewNotificationCycle, + wakePrReviewNotificationAssociation, } from '../pr-review-notification'; describe('enqueuePrReviewNotification', () => { beforeEach(() => { vi.clearAllMocks(); multiCalls.length = 0; + orphanChainId = replayChainId; + orphanRevision = 0; mockFindManyTaskPullRequests.mockResolvedValue([{ taskId: 'task-1' }]); mockRedisSet.mockResolvedValue('OK'); mockRedisDel.mockResolvedValue(1); mockRedisGet.mockResolvedValue(null); - mockRedisEval.mockResolvedValue(1); + mockRedisEval.mockImplementation((script: string, ...args: unknown[]) => { + if (script.includes("marker = ARGV[1] .. ':0'")) { + return mockRedisGet(...args); + } + if (script.includes("redis.call('RPUSH', KEYS[1], ARGV[2])")) { + const claimed = orphanRevision === 0 ? 1 : 0; + orphanRevision += 1; + orphanChainId = claimed === 1 ? String(args[3]) : orphanChainId; + return Promise.resolve([`${orphanChainId}:${orphanRevision}`, claimed]); + } + + if (script.includes('release a quiet chain')) { + return Promise.resolve(0); + } + + if (script.includes("marker = ARGV[1] .. ':0'")) { + return mockRedisGet(...args); + } + + if (script.includes("redis.call('LLEN', KEYS[1])")) { + return Promise.resolve(0); + } + + return Promise.resolve(1); + }); mockQueueAdd.mockResolvedValue(undefined); mockMultiExec.mockResolvedValue([]); }); @@ -98,13 +138,454 @@ describe('enqueuePrReviewNotification', () => { }, }; - it('returns no_linked_tasks when no task links the PR', async () => { + it('schedules a delayed association replay before returning no_linked_tasks', async () => { mockFindManyTaskPullRequests.mockResolvedValue([]); const result = await enqueuePrReviewNotification(baseInput); + expect(result).toEqual({ notifiedTaskCount: 0, reason: 'no_linked_tasks' }); + expect(mockFindManyTaskPullRequests).toHaveBeenCalledTimes(1); + expect(mockQueueAdd).toHaveBeenCalledWith( + 'replay-pr-review-association', + { + kind: 'association_replay', + sourceControlProvider: 'github', + repository: 'owner/repo', + prNumber: 42, + chainId: expect.any(String), + attempt: 1, + }, + { + delay: PR_REVIEW_ASSOCIATION_REPLAY_DELAYS_MS[0], + jobId: expect.stringMatching(/^association-replay-[a-f0-9-]+-1$/), + }, + ); + }); + + it('atomically claims retained activity and schedules attempt 1 on association commit', async () => { + mockRedisEval.mockResolvedValueOnce(1); + + await expect( + wakePrReviewNotificationAssociation({ + sourceControlProvider: 'github', + repository: 'owner/repo', + prNumber: 42, + }), + ).resolves.toBe(true); + + expect(mockRedisEval).toHaveBeenCalledWith( + expect.stringContaining("redis.call('LLEN', KEYS[1])"), + 2, + 'pr-review-notification:orphan:github:owner%2Frepo#42', + 'pr-review-notification:orphan-replay:github:owner%2Frepo#42', + expect.any(String), + 900, + ); + expect(mockQueueAdd).toHaveBeenCalledWith( + 'replay-pr-review-association', + expect.objectContaining({ attempt: 1 }), + expect.objectContaining({ + delay: PR_REVIEW_ASSOCIATION_REPLAY_DELAYS_MS[0], + }), + ); + }); + + it('lets association commit wake the list after final replay releases it', async () => { + mockFindManyTaskPullRequests.mockResolvedValue([]); + mockRedisEval + .mockResolvedValueOnce(`${replayChainId}:1`) + .mockResolvedValueOnce(0) + .mockResolvedValueOnce(1); + + await replayPrReviewNotificationAssociation({ + kind: 'association_replay', + sourceControlProvider: 'github', + repository: 'owner/repo', + prNumber: 42, + chainId: replayChainId, + attempt: PR_REVIEW_ASSOCIATION_REPLAY_DELAYS_MS.length, + }); + await wakePrReviewNotificationAssociation({ + sourceControlProvider: 'github', + repository: 'owner/repo', + prNumber: 42, + }); + + expect(mockQueueAdd).toHaveBeenCalledTimes(1); + expect(mockQueueAdd).toHaveBeenCalledWith( + 'replay-pr-review-association', + expect.objectContaining({ attempt: 1 }), + expect.any(Object), + ); + }); + + it('preserves legacy GitHub pending and scheduled key identity', async () => { + await enqueuePrReviewNotification(baseInput); + + expect(multiCalls).toContainEqual({ + command: 'rpush', + args: [ + 'pr-review-notification:pending:task-1:owner%2Frepo#42:human', + expect.any(String), + ], + }); + expect(mockRedisSet).toHaveBeenCalledWith( + 'pr-review-notification:scheduled:task-1:owner%2Frepo#42:human:delayed', + '1', + 'EX', + expect.any(Number), + 'NX', + ); + }); + + it('namespaces non-default provider pending and scheduled keys', async () => { + await enqueuePrReviewNotification({ + ...baseInput, + sourceControlProvider: 'gitlab', + }); + + expect(multiCalls).toContainEqual({ + command: 'rpush', + args: [ + 'pr-review-notification:pending:gitlab:task-1:owner%2Frepo#42:human', + expect.any(String), + ], + }); + expect(mockRedisSet).toHaveBeenCalledWith( + 'pr-review-notification:scheduled:gitlab:task-1:owner%2Frepo#42:human:delayed', + '1', + 'EX', + expect.any(Number), + 'NX', + ); + }); + + it('coalesces an unrelated PR burst into one replay chain without dropping events', async () => { + mockFindManyTaskPullRequests.mockResolvedValue([]); + mockRedisSet.mockResolvedValue(null); + mockRedisSet.mockResolvedValueOnce('OK'); + mockRedisGet.mockResolvedValue(`${replayChainId}:50`); + + await Promise.all( + Array.from({ length: 50 }, (_, index) => + enqueuePrReviewNotification({ + ...baseInput, + event: { + kind: 'review_comment', + authorLogin: 'reviewer', + batchId: `github-review:${index}`, + url: `https://github.com/owner/repo/pull/42#discussion-${index}`, + }, + }), + ), + ); + + expect(mockQueueAdd).toHaveBeenCalledTimes(1); + expect(orphanRevision).toBe(50); + }); + + it('retains every orphan event when association persistence finishes before replay', async () => { + const orphanInputs = [ + baseInput, + { + ...baseInput, + event: { + kind: 'review_comment' as const, + authorLogin: 'bob', + batchId: 'github-review:2', + }, + }, + ]; + mockFindManyTaskPullRequests + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([{ taskId: 'task-1' }]); + + await enqueuePrReviewNotification(baseInput); + mockQueueAdd.mockClear(); + multiCalls.length = 0; + mockRedisEval.mockImplementation((script: string, ...args: unknown[]) => { + if (script.includes("marker = ARGV[1] .. ':0'")) { + return Promise.resolve(`${replayChainId}:1`); + } + if (script.includes("redis.call('RPUSH', KEYS[1], ARGV[2])")) { + return Promise.resolve([`${String(args[3])}:1`, 1]); + } + if (script.includes("redis.call('LRANGE', KEYS[1], 0, -1)")) { + return Promise.resolve( + orphanInputs.map((input) => JSON.stringify(input)), + ); + } + return Promise.resolve(1); + }); + mockQueueAdd.mockClear(); + mockRedisSet.mockResolvedValue(null); + mockRedisSet.mockResolvedValueOnce('OK'); + mockRedisGet.mockResolvedValue(`${replayChainId}:1`); + const result = await replayPrReviewNotificationAssociation({ + kind: 'association_replay', + sourceControlProvider: 'github', + repository: 'owner/repo', + prNumber: 42, + chainId: replayChainId, + attempt: 1, + }); + + expect(result).toEqual({ notifiedTaskCount: 2 }); + expect(mockFindManyTaskPullRequests).toHaveBeenCalledTimes(2); + expect( + mockQueueAdd.mock.calls.filter( + ([name]) => name === 'notify-pr-review-activity', + ), + ).toHaveLength(2); + expect(mockQueueAdd).toHaveBeenCalledWith( + 'notify-pr-review-activity', + expect.objectContaining({ taskId: 'task-1' }), + { delay: PR_REVIEW_NOTIFICATION_DEBOUNCE_MS }, + ); + const taskEventAppends = multiCalls.filter( + (call) => + call.command === 'rpush' && + String(call.args[0]).startsWith('pr-review-notification:pending:'), + ); + expect(taskEventAppends).toHaveLength(2); + expect(taskEventAppends.map((call) => String(call.args[1]))).toEqual( + expect.arrayContaining([ + expect.stringContaining('changes_requested'), + expect.stringContaining('github-review:2'), + ]), + ); + }); + + it('drains retained orphan activity exactly once after association wake', async () => { + mockFindManyTaskPullRequests.mockResolvedValue([{ taskId: 'task-1' }]); + let replayClaimed = true; + mockRedisEval.mockImplementation((script: string) => { + if (script.includes("marker = ARGV[1] .. ':0'")) { + if (!replayClaimed) { + return Promise.resolve(null); + } + replayClaimed = false; + return Promise.resolve(`${replayChainId}:0`); + } + if (script.includes("redis.call('LRANGE', KEYS[1], 0, -1)")) { + return Promise.resolve([JSON.stringify(baseInput)]); + } + return Promise.resolve(1); + }); + + const request = { + kind: 'association_replay' as const, + sourceControlProvider: 'github' as const, + repository: 'owner/repo', + prNumber: 42, + chainId: replayChainId, + attempt: 1, + }; + + expect(await replayPrReviewNotificationAssociation(request)).toEqual({ + notifiedTaskCount: 1, + }); + expect(await replayPrReviewNotificationAssociation(request)).toEqual({ + notifiedTaskCount: 0, + reason: 'stale_association_replay', + }); + expect( + mockQueueAdd.mock.calls.filter( + ([name]) => name === 'notify-pr-review-activity', + ), + ).toHaveLength(1); + }); + + it('releases only the expected marker after all five attempts', async () => { + mockFindManyTaskPullRequests.mockResolvedValue([]); + mockRedisGet.mockResolvedValue(`${replayChainId}:1`); + mockRedisEval + .mockResolvedValueOnce(`${replayChainId}:1`) + .mockResolvedValueOnce(0); + + const result = await replayPrReviewNotificationAssociation({ + kind: 'association_replay', + sourceControlProvider: 'github', + repository: 'owner/repo', + prNumber: 42, + chainId: replayChainId, + attempt: PR_REVIEW_ASSOCIATION_REPLAY_DELAYS_MS.length, + }); + expect(result).toEqual({ notifiedTaskCount: 0, reason: 'no_linked_tasks' }); expect(mockQueueAdd).not.toHaveBeenCalled(); + expect(mockRedisEval).toHaveBeenCalledWith( + expect.stringContaining('release a quiet chain'), + 2, + 'pr-review-notification:orphan:github:owner%2Frepo#42', + 'pr-review-notification:orphan-replay:github:owner%2Frepo#42', + `${replayChainId}:1`, + replayChainId, + expect.any(String), + 900, + ); + }); + + it('releases the initial claim when its replay job cannot be queued', async () => { + mockFindManyTaskPullRequests.mockResolvedValue([]); + mockQueueAdd.mockRejectedValueOnce(new Error('queue down')); + + await expect(enqueuePrReviewNotification(baseInput)).rejects.toThrow( + 'queue down', + ); + + expect(mockRedisEval).toHaveBeenCalledWith( + expect.stringContaining("redis.call('GET', KEYS[1])"), + 1, + 'pr-review-notification:orphan-replay:github:owner%2Frepo#42', + expect.any(String), + ); + }); + + it('reclaims an initial enqueue failure after a concurrent append changes the marker', async () => { + mockFindManyTaskPullRequests.mockResolvedValue([]); + mockQueueAdd + .mockRejectedValueOnce(new Error('queue down')) + .mockResolvedValueOnce(undefined); + mockRedisEval.mockImplementation((script: string, ...args: unknown[]) => { + if (script.includes("redis.call('RPUSH', KEYS[1], ARGV[2])")) { + return Promise.resolve([`${String(args[3])}:1`, 1]); + } + if (script.includes("redis.call('LLEN', KEYS[1])")) { + return Promise.resolve(1); + } + return Promise.resolve(0); + }); + + await expect(enqueuePrReviewNotification(baseInput)).rejects.toThrow( + 'queue down', + ); + + expect(mockQueueAdd).toHaveBeenCalledTimes(2); + expect(mockQueueAdd).toHaveBeenLastCalledWith( + 'replay-pr-review-association', + expect.objectContaining({ attempt: 1 }), + expect.objectContaining({ + delay: PR_REVIEW_ASSOCIATION_REPLAY_DELAYS_MS[0], + }), + ); + }); + + it('releases a terminal rotation when its wake job cannot be queued', async () => { + mockFindManyTaskPullRequests.mockResolvedValue([]); + mockRedisGet.mockResolvedValue(`${replayChainId}:2`); + mockRedisEval.mockImplementation((script: string) => + Promise.resolve( + script.includes("marker = ARGV[1] .. ':0'") ? `${replayChainId}:2` : 1, + ), + ); + mockQueueAdd.mockRejectedValueOnce(new Error('queue down')); + + await expect( + replayPrReviewNotificationAssociation({ + kind: 'association_replay', + sourceControlProvider: 'github', + repository: 'owner/repo', + prNumber: 42, + chainId: replayChainId, + attempt: PR_REVIEW_ASSOCIATION_REPLAY_DELAYS_MS.length, + }), + ).rejects.toThrow('queue down'); + + expect(mockRedisEval).toHaveBeenLastCalledWith( + expect.stringContaining("redis.call('GET', KEYS[1])"), + 1, + 'pr-review-notification:orphan-replay:github:owner%2Frepo#42', + expect.any(String), + ); + }); + + it('releases exhausted-delivery ownership when requeue scheduling fails', async () => { + mockFindManyTaskPullRequests.mockResolvedValue([{ taskId: 'task-1' }]); + mockRedisGet.mockResolvedValue(`${replayChainId}:1`); + mockRedisEval.mockImplementation((script: string) => { + if (script.includes("marker = ARGV[1] .. ':0'")) { + return Promise.resolve(`${replayChainId}:1`); + } + if (script.includes("redis.call('LRANGE', KEYS[1], 0, -1)")) { + return Promise.resolve([JSON.stringify(baseInput)]); + } + return Promise.resolve(1); + }); + mockQueueAdd + .mockRejectedValueOnce(new Error('delivery queue down')) + .mockRejectedValueOnce(new Error('replay queue down')); + + await expect( + replayPrReviewNotificationAssociation({ + kind: 'association_replay', + sourceControlProvider: 'github', + repository: 'owner/repo', + prNumber: 42, + chainId: replayChainId, + attempt: PR_REVIEW_ASSOCIATION_REPLAY_DELAYS_MS.length, + }), + ).rejects.toThrow('replay queue down'); + + expect(mockRedisEval).toHaveBeenLastCalledWith( + expect.stringContaining("redis.call('GET', KEYS[1])"), + 1, + 'pr-review-notification:orphan-replay:github:owner%2Frepo#42', + expect.any(String), + ); + }); + + it('lets a failed replay job retry reclaim its retained list', async () => { + mockFindManyTaskPullRequests.mockResolvedValue([{ taskId: 'task-1' }]); + mockRedisEval.mockImplementation((script: string) => { + if (script.includes("marker = ARGV[1] .. ':0'")) { + return Promise.resolve(`${replayChainId}:0`); + } + if (script.includes("redis.call('LRANGE', KEYS[1], 0, -1)")) { + return Promise.resolve([JSON.stringify(baseInput)]); + } + return Promise.resolve(1); + }); + + const result = await replayPrReviewNotificationAssociation({ + kind: 'association_replay', + sourceControlProvider: 'github', + repository: 'owner/repo', + prNumber: 42, + chainId: replayChainId, + attempt: 1, + }); + + expect(result).toEqual({ notifiedTaskCount: 1 }); + }); + + it('rotates activity appended during the final lookup into one fresh bounded chain', async () => { + mockFindManyTaskPullRequests.mockResolvedValue([]); + mockRedisGet.mockResolvedValue(`${replayChainId}:1`); + mockRedisEval + .mockResolvedValueOnce(`${replayChainId}:1`) + .mockResolvedValueOnce(1); + + const result = await replayPrReviewNotificationAssociation({ + kind: 'association_replay', + sourceControlProvider: 'github', + repository: 'owner/repo', + prNumber: 42, + chainId: replayChainId, + attempt: PR_REVIEW_ASSOCIATION_REPLAY_DELAYS_MS.length, + }); + + expect(result).toEqual({ notifiedTaskCount: 0, reason: 'no_linked_tasks' }); + expect(mockQueueAdd).toHaveBeenCalledWith( + 'replay-pr-review-association', + expect.objectContaining({ + chainId: expect.not.stringMatching(new RegExp(replayChainId)), + attempt: 1, + }), + expect.objectContaining({ + delay: PR_REVIEW_ASSOCIATION_REPLAY_DELAYS_MS[0], + }), + ); }); it('debounces ordinary notifications for web-only tasks without an originating conversation', async () => { @@ -121,6 +602,7 @@ describe('enqueuePrReviewNotification', () => { deferrals: 0, immediate: false, batchKind: 'human', + sourceControlProvider: 'github', }, { delay: PR_REVIEW_NOTIFICATION_DEBOUNCE_MS }, ); @@ -152,6 +634,7 @@ describe('enqueuePrReviewNotification', () => { deferrals: 0, immediate: false, batchKind: 'human', + sourceControlProvider: 'github', }, { delay: PR_REVIEW_NOTIFICATION_DEBOUNCE_MS }, ); diff --git a/packages/sdk/src/server/lib/task-runs/pr-review-notification.ts b/packages/sdk/src/server/lib/task-runs/pr-review-notification.ts index f606b21ab..952ec7ca7 100644 --- a/packages/sdk/src/server/lib/task-runs/pr-review-notification.ts +++ b/packages/sdk/src/server/lib/task-runs/pr-review-notification.ts @@ -1,3 +1,5 @@ +import { randomUUID } from 'node:crypto'; + import { Queue } from 'bullmq'; import { z } from 'zod'; @@ -50,6 +52,14 @@ export const PR_REVIEW_NOTIFICATION_MAX_DEFERRALS = 288; const PENDING_EVENTS_TTL_SECONDS = 24 * 60 * 60; const SCHEDULED_MARKER_TTL_BUFFER_SECONDS = 15 * 60; const REVIEW_CYCLE_TTL_SECONDS = 30 * 24 * 60 * 60; +const ORPHAN_ACTIVITY_TTL_SECONDS = 15 * 60; +const ORPHAN_REPLAY_REPAIR_INDEX_KEY = + 'pr-review-notification:orphan-replay-repair'; +const ORPHAN_REPLAY_REPAIR_BATCH_SIZE = 100; +export const ORPHAN_REPLAY_REPAIR_DELAY_MS = 60 * 1000; +export const PR_REVIEW_ASSOCIATION_REPLAY_DELAYS_MS = [ + 1_000, 5_000, 30_000, 120_000, 300_000, +] as const; const SET_REVIEW_CYCLE_IF_NEWER_SCRIPT = ` local current = redis.call('GET', KEYS[1]) if current then @@ -61,6 +71,93 @@ end redis.call('SET', KEYS[1], ARGV[1], 'EX', ARGV[3]) return 1 `; +const APPEND_ORPHAN_ACTIVITY_SCRIPT = ` +local marker = redis.call('GET', KEYS[2]) +local claimed = 0 +if not marker then + marker = ARGV[1] .. ':1' + claimed = 1 +else + local separator = string.match(marker, '^.*():') + local chainId = string.sub(marker, 1, separator - 1) + local revision = tonumber(string.sub(marker, separator + 1)) + 1 + marker = chainId .. ':' .. revision +end +redis.call('SET', KEYS[2], marker, 'EX', ARGV[3]) +redis.call('RPUSH', KEYS[1], ARGV[2]) +redis.call('EXPIRE', KEYS[1], ARGV[3]) +return {marker, claimed} +`; +const CONSUME_ORPHAN_ACTIVITY_SCRIPT = ` +local marker = redis.call('GET', KEYS[2]) +if not marker or string.sub(marker, 1, string.len(ARGV[1]) + 1) ~= ARGV[1] .. ':' then + return {} +end +local inputs = redis.call('LRANGE', KEYS[1], 0, -1) +redis.call('DEL', KEYS[1], KEYS[2]) +return inputs +`; +const CLAIM_ORPHAN_REPLAY_SCRIPT = ` +if redis.call('LLEN', KEYS[1]) == 0 then + return 0 +end +redis.call('SET', KEYS[2], ARGV[1] .. ':0', 'EX', ARGV[2]) +return 1 +`; +const GET_OR_RECLAIM_ORPHAN_REPLAY_SCRIPT = ` +local marker = redis.call('GET', KEYS[2]) +if marker then + return marker +end +if redis.call('LLEN', KEYS[1]) == 0 then + return false +end +marker = ARGV[1] .. ':0' +redis.call('SET', KEYS[2], marker, 'EX', ARGV[2]) +return marker +`; +const FINISH_OR_ROTATE_ORPHAN_REPLAY_SCRIPT = ` +-- release a quiet chain or rotate activity appended during final lookup +local marker = redis.call('GET', KEYS[2]) +if marker == ARGV[1] then + redis.call('DEL', KEYS[2]) + return 0 +end +if marker and string.sub(marker, 1, string.len(ARGV[2]) + 1) == ARGV[2] .. ':' then + redis.call('SET', KEYS[2], ARGV[3] .. ':0', 'EX', ARGV[4]) + return 1 +end +return 0 +`; +const RELEASE_ORPHAN_REPLAY_SCRIPT = ` +local marker = redis.call('GET', KEYS[1]) +if marker and string.sub(marker, 1, string.len(ARGV[1]) + 1) == ARGV[1] .. ':' then + return redis.call('DEL', KEYS[1]) +end +return 0 +`; +const REQUEUE_AND_CLAIM_ORPHAN_REPLAY_SCRIPT = ` +for index = 1, #ARGV - 2 do + redis.call('RPUSH', KEYS[1], ARGV[index]) +end +redis.call('EXPIRE', KEYS[1], ARGV[#ARGV]) +redis.call('SET', KEYS[2], ARGV[#ARGV - 1] .. ':0', 'EX', ARGV[#ARGV]) +return 1 +`; +const CLEAR_ORPHAN_REPLAY_REPAIR_SCRIPT = ` +if redis.call('GET', KEYS[1]) ~= ARGV[1] then + return 0 +end +redis.call('DEL', KEYS[1]) +redis.call('ZREM', KEYS[2], ARGV[2]) +return 1 +`; +const CLEAR_MISSING_ORPHAN_REPLAY_REPAIR_SCRIPT = ` +if redis.call('EXISTS', KEYS[1]) ~= 0 then + return 0 +end +return redis.call('ZREM', KEYS[2], ARGV[1]) +`; const COMPLETE_REVIEW_CYCLE_SCRIPT = ` local current = redis.call('GET', KEYS[1]) if current then @@ -145,6 +242,7 @@ type PrReviewNotificationTarget = { taskId: string; repository: string; prNumber: number; + sourceControlProvider?: z.infer; immediate?: boolean; batchKind?: z.infer; batchId?: string; @@ -174,6 +272,32 @@ export type EnqueuePrReviewNotificationInput = z.infer< typeof enqueuePrReviewNotificationInputSchema >; +export const prReviewAssociationReplayRequestSchema = z.object({ + kind: z.literal('association_replay'), + sourceControlProvider: sourceControlProviderSchema, + repository: z.string(), + prNumber: z.number().int().positive(), + chainId: z.string().uuid(), + attempt: z + .number() + .int() + .min(1) + .max(PR_REVIEW_ASSOCIATION_REPLAY_DELAYS_MS.length), +}); + +export type PrReviewAssociationReplayRequest = z.infer< + typeof prReviewAssociationReplayRequestSchema +>; + +export const prReviewNotificationQueueRequestSchema = z.union([ + prReviewAssociationReplayRequestSchema, + prReviewNotificationRequestSchema, +]); + +export type PrReviewNotificationQueueRequest = z.infer< + typeof prReviewNotificationQueueRequestSchema +>; + type EnqueuePrReviewNotificationResult = { notifiedTaskCount: number; reason?: string; @@ -195,13 +319,14 @@ export type PrReviewNotificationRoute = | { provider: 'telegram'; channelId: string; threadId: string | null } | { provider: 'discord'; channelId: string; threadId: string | null }; -let prReviewNotificationQueue: Queue | null = null; +let prReviewNotificationQueue: Queue | null = + null; -function getPrReviewNotificationQueue(): Queue { +function getPrReviewNotificationQueue(): Queue { if (!prReviewNotificationQueue) { const redis = getRedis(); - prReviewNotificationQueue = new Queue( + prReviewNotificationQueue = new Queue( PR_REVIEW_NOTIFICATION_QUEUE_NAME, { connection: redis, @@ -222,19 +347,25 @@ function buildTargetKeySuffix({ taskId, repository, prNumber, + sourceControlProvider = 'github', batchKind = 'human', batchId, }: PrReviewNotificationTarget): string { - return `${encodeURIComponent(taskId)}:${encodeURIComponent(repository)}#${prNumber}:${batchKind}${batchId ? `:${encodeURIComponent(batchId)}` : ''}`; + const providerPrefix = + sourceControlProvider === 'github' ? '' : `${sourceControlProvider}:`; + return `${providerPrefix}${encodeURIComponent(taskId)}:${encodeURIComponent(repository)}#${prNumber}:${batchKind}${batchId ? `:${encodeURIComponent(batchId)}` : ''}`; } function buildLegacyTargetKeySuffix({ taskId, repository, prNumber, + sourceControlProvider = 'github', immediate = false, }: PrReviewNotificationTarget): string { - return `${encodeURIComponent(taskId)}:${encodeURIComponent(repository)}#${prNumber}${immediate ? ':immediate' : ''}`; + const providerPrefix = + sourceControlProvider === 'github' ? '' : `${sourceControlProvider}:`; + return `${providerPrefix}${encodeURIComponent(taskId)}:${encodeURIComponent(repository)}#${prNumber}${immediate ? ':immediate' : ''}`; } function buildPendingEventsKey(target: PrReviewNotificationTarget): string { @@ -253,6 +384,331 @@ function buildScheduledMarkerKey(target: PrReviewNotificationTarget): string { return `pr-review-notification:scheduled:${buildTargetKeySuffix(target)}:${target.immediate ? 'immediate' : 'delayed'}`; } +type PrReviewAssociationTarget = Pick< + EnqueuePrReviewNotificationInput, + 'repository' | 'prNumber' +> & { sourceControlProvider: z.infer }; + +function buildOrphanActivityKey(target: PrReviewAssociationTarget): string { + return `pr-review-notification:orphan:${target.sourceControlProvider}:${encodeURIComponent(target.repository)}#${target.prNumber}`; +} + +function buildOrphanReplayMarkerKey(target: PrReviewAssociationTarget): string { + return `pr-review-notification:orphan-replay:${target.sourceControlProvider}:${encodeURIComponent(target.repository)}#${target.prNumber}`; +} + +function buildOrphanReplayRepairMember( + target: PrReviewAssociationTarget, +): string { + return `${target.sourceControlProvider}:${encodeURIComponent(target.repository)}#${target.prNumber}`; +} + +function buildOrphanReplayRepairPayloadKey( + target: PrReviewAssociationTarget, +): string { + return `pr-review-notification:orphan-replay-repair-payload:${buildOrphanReplayRepairMember(target)}`; +} + +async function recordOrphanReplayRepair( + request: PrReviewAssociationReplayRequest, +): Promise { + const target = { + sourceControlProvider: request.sourceControlProvider, + repository: request.repository, + prNumber: request.prNumber, + }; + const payload = JSON.stringify(request); + + await getRedis() + .multi() + .set( + buildOrphanReplayRepairPayloadKey(target), + payload, + 'EX', + ORPHAN_ACTIVITY_TTL_SECONDS, + ) + .zadd( + ORPHAN_REPLAY_REPAIR_INDEX_KEY, + Date.now() + ORPHAN_REPLAY_REPAIR_DELAY_MS, + buildOrphanReplayRepairMember(target), + ) + .exec(); + + return payload; +} + +async function clearOrphanReplayRepair( + request: PrReviewAssociationReplayRequest, + payload: string, +): Promise { + const target = { + sourceControlProvider: request.sourceControlProvider, + repository: request.repository, + prNumber: request.prNumber, + }; + + await getRedis().eval( + CLEAR_ORPHAN_REPLAY_REPAIR_SCRIPT, + 2, + buildOrphanReplayRepairPayloadKey(target), + ORPHAN_REPLAY_REPAIR_INDEX_KEY, + payload, + buildOrphanReplayRepairMember(target), + ); +} + +async function scheduleOrphanAssociationReplay({ + target, + chainId, + attempt, +}: { + target: PrReviewAssociationTarget; + chainId: string; + attempt: number; +}): Promise { + const delayMs = PR_REVIEW_ASSOCIATION_REPLAY_DELAYS_MS[attempt - 1]; + + if (delayMs === undefined) { + return; + } + + const request = { + kind: 'association_replay' as const, + ...target, + chainId, + attempt, + }; + const repairPayload = await recordOrphanReplayRepair(request); + + await getPrReviewNotificationQueue().add( + 'replay-pr-review-association', + request, + { + delay: delayMs, + jobId: `association-replay-${chainId}-${attempt}`, + }, + ); + await clearOrphanReplayRepair(request, repairPayload).catch(() => undefined); +} + +/** Re-enqueues orphan association replays whose Queue.add did not complete. */ +export async function repairOrphanPrReviewAssociationReplays({ + now = Date.now(), +}: { now?: number } = {}): Promise { + const redis = getRedis(); + const members = await redis.zrangebyscore( + ORPHAN_REPLAY_REPAIR_INDEX_KEY, + '-inf', + now, + 'LIMIT', + 0, + ORPHAN_REPLAY_REPAIR_BATCH_SIZE, + ); + + for (const member of members) { + const payloadKey = `pr-review-notification:orphan-replay-repair-payload:${member}`; + const rawRequest = await redis.get(payloadKey); + let request: PrReviewAssociationReplayRequest | null = null; + + if (rawRequest) { + try { + const parsed = prReviewAssociationReplayRequestSchema.safeParse( + JSON.parse(rawRequest), + ); + request = parsed.success ? parsed.data : null; + } catch { + request = null; + } + } + + if (!rawRequest) { + await redis.eval( + CLEAR_MISSING_ORPHAN_REPLAY_REPAIR_SCRIPT, + 2, + payloadKey, + ORPHAN_REPLAY_REPAIR_INDEX_KEY, + member, + ); + continue; + } + + if (!request) { + await redis.eval( + CLEAR_ORPHAN_REPLAY_REPAIR_SCRIPT, + 2, + payloadKey, + ORPHAN_REPLAY_REPAIR_INDEX_KEY, + rawRequest, + member, + ); + continue; + } + + const target = { + sourceControlProvider: request.sourceControlProvider, + repository: request.repository, + prNumber: request.prNumber, + }; + if ((await redis.llen(buildOrphanActivityKey(target))) === 0) { + await clearOrphanReplayRepair(request, rawRequest).catch(() => undefined); + continue; + } + + try { + await getPrReviewNotificationQueue().add( + 'replay-pr-review-association', + request, + { + jobId: `association-replay-${request.chainId}-${request.attempt}`, + }, + ); + await clearOrphanReplayRepair(request, rawRequest).catch(() => undefined); + } catch (error) { + console.warn( + `[repairOrphanPrReviewAssociationReplays] Failed to enqueue replay for ${request.repository}#${request.prNumber}: ${error instanceof Error ? error.message : String(error)}`, + ); + } + } +} + +async function releaseOrphanReplay( + target: PrReviewAssociationTarget, + chainId: string, +): Promise { + await getRedis() + .eval( + RELEASE_ORPHAN_REPLAY_SCRIPT, + 1, + buildOrphanReplayMarkerKey(target), + chainId, + ) + .catch(() => undefined); +} + +async function scheduleClaimedOrphanReplay({ + target, + chainId, +}: { + target: PrReviewAssociationTarget; + chainId: string; +}): Promise { + try { + await scheduleOrphanAssociationReplay({ target, chainId, attempt: 1 }); + } catch (error) { + await releaseOrphanReplay(target, chainId); + throw error; + } +} + +/** Claims retained orphan activity after an association becomes observable. */ +export async function wakePrReviewNotificationAssociation( + target: PrReviewAssociationTarget, +): Promise { + const parsedTarget = { + sourceControlProvider: sourceControlProviderSchema.parse( + target.sourceControlProvider, + ), + repository: target.repository, + prNumber: target.prNumber, + }; + const chainId = randomUUID(); + const claimed = await getRedis().eval( + CLAIM_ORPHAN_REPLAY_SCRIPT, + 2, + buildOrphanActivityKey(parsedTarget), + buildOrphanReplayMarkerKey(parsedTarget), + chainId, + ORPHAN_ACTIVITY_TTL_SECONDS, + ); + + if (claimed !== 1) { + return false; + } + + await scheduleClaimedOrphanReplay({ target: parsedTarget, chainId }); + return true; +} + +async function appendOrphanActivityAndClaimReplay({ + target, + input, +}: { + target: PrReviewAssociationTarget; + input: EnqueuePrReviewNotificationInput; +}): Promise { + const redis = getRedis(); + const chainId = randomUUID(); + const result = await redis.eval( + APPEND_ORPHAN_ACTIVITY_SCRIPT, + 2, + buildOrphanActivityKey(target), + buildOrphanReplayMarkerKey(target), + chainId, + JSON.stringify(input), + ORPHAN_ACTIVITY_TTL_SECONDS, + ); + + return Array.isArray(result) && result[1] === 1 + ? (String(result[0]).split(':')[0] ?? null) + : null; +} + +async function consumeOrphanActivity( + target: PrReviewAssociationTarget, + chainId: string, +): Promise { + const rawInputs = await getRedis().eval( + CONSUME_ORPHAN_ACTIVITY_SCRIPT, + 2, + buildOrphanActivityKey(target), + buildOrphanReplayMarkerKey(target), + chainId, + ); + + if (!Array.isArray(rawInputs)) { + return []; + } + + return rawInputs.flatMap((raw) => { + if (typeof raw !== 'string') { + return []; + } + + try { + const parsed = enqueuePrReviewNotificationInputSchema.safeParse( + JSON.parse(raw), + ); + return parsed.success ? [parsed.data] : []; + } catch { + return []; + } + }); +} + +async function requeueOrphanActivity({ + target, + chainId, + inputs, +}: { + target: PrReviewAssociationTarget; + chainId: string; + inputs: EnqueuePrReviewNotificationInput[]; +}): Promise { + if (inputs.length === 0) { + return; + } + + await getRedis().eval( + REQUEUE_AND_CLAIM_ORPHAN_REPLAY_SCRIPT, + 2, + buildOrphanActivityKey(target), + buildOrphanReplayMarkerKey(target), + ...inputs.map((input) => JSON.stringify(input)), + chainId, + ORPHAN_ACTIVITY_TTL_SECONDS, + ); +} + function getPrReviewCycleStateKey({ repository, prNumber, @@ -664,24 +1120,165 @@ export async function enqueuePrReviewNotification( ): Promise { const parsedInput = enqueuePrReviewNotificationInputSchema.parse(input); - const prTaskLinks = await db.query.taskPullRequests.findMany({ + return enqueuePrReviewNotificationAttempt(parsedInput); +} + +export async function replayPrReviewNotificationAssociation( + request: PrReviewAssociationReplayRequest, +): Promise { + const parsed = prReviewAssociationReplayRequestSchema.parse(request); + const target = { + sourceControlProvider: parsed.sourceControlProvider, + repository: parsed.repository, + prNumber: parsed.prNumber, + }; + const activeMarker = await getRedis().eval( + GET_OR_RECLAIM_ORPHAN_REPLAY_SCRIPT, + 2, + buildOrphanActivityKey(target), + buildOrphanReplayMarkerKey(target), + parsed.chainId, + ORPHAN_ACTIVITY_TTL_SECONDS, + ); + + if ( + typeof activeMarker !== 'string' || + !activeMarker.startsWith(`${parsed.chainId}:`) + ) { + return { notifiedTaskCount: 0, reason: 'stale_association_replay' }; + } + + const prTaskLinks = await findPrReviewTaskLinks(target); + + if (prTaskLinks.length === 0) { + const nextAttempt = parsed.attempt + 1; + + if (nextAttempt <= PR_REVIEW_ASSOCIATION_REPLAY_DELAYS_MS.length) { + await scheduleOrphanAssociationReplay({ + target, + chainId: parsed.chainId, + attempt: nextAttempt, + }); + } else { + const nextChainId = randomUUID(); + const rotated = await getRedis().eval( + FINISH_OR_ROTATE_ORPHAN_REPLAY_SCRIPT, + 2, + buildOrphanActivityKey(target), + buildOrphanReplayMarkerKey(target), + activeMarker, + parsed.chainId, + nextChainId, + ORPHAN_ACTIVITY_TTL_SECONDS, + ); + + if (rotated === 1) { + await scheduleClaimedOrphanReplay({ target, chainId: nextChainId }); + } + } + + return { notifiedTaskCount: 0, reason: 'no_linked_tasks' }; + } + + const inputs = await consumeOrphanActivity(target, parsed.chainId); + let notifiedTaskCount = 0; + + for (const [index, input] of inputs.entries()) { + try { + const result = await enqueueLinkedPrReviewNotification( + input, + prTaskLinks, + ); + notifiedTaskCount += result.notifiedTaskCount; + } catch (error) { + const nextAttempt = parsed.attempt + 1; + const requeueChainId = + nextAttempt <= PR_REVIEW_ASSOCIATION_REPLAY_DELAYS_MS.length + ? parsed.chainId + : randomUUID(); + await requeueOrphanActivity({ + target, + chainId: requeueChainId, + inputs: inputs.slice(index), + }); + + try { + await scheduleOrphanAssociationReplay({ + target, + chainId: requeueChainId, + attempt: + nextAttempt <= PR_REVIEW_ASSOCIATION_REPLAY_DELAYS_MS.length + ? nextAttempt + : 1, + }); + } catch (scheduleError) { + await releaseOrphanReplay(target, requeueChainId); + throw scheduleError; + } + + throw error; + } + } + + return { notifiedTaskCount }; +} + +async function enqueuePrReviewNotificationAttempt( + parsedInput: EnqueuePrReviewNotificationInput, +): Promise { + const target = { + sourceControlProvider: parsedInput.sourceControlProvider ?? 'github', + repository: parsedInput.repository, + prNumber: parsedInput.prNumber, + }; + const prTaskLinks = await findPrReviewTaskLinks(target); + + if (prTaskLinks.length === 0) { + const chainId = await appendOrphanActivityAndClaimReplay({ + target, + input: parsedInput, + }); + + if (chainId) { + try { + await scheduleOrphanAssociationReplay({ + target, + chainId, + attempt: 1, + }); + } catch (error) { + await releaseOrphanReplay(target, chainId); + await wakePrReviewNotificationAssociation(target).catch(() => false); + throw error; + } + } + + return { notifiedTaskCount: 0, reason: 'no_linked_tasks' }; + } + + await wakePrReviewNotificationAssociation(target); + return enqueueLinkedPrReviewNotification(parsedInput, prTaskLinks); +} + +async function findPrReviewTaskLinks( + target: PrReviewAssociationTarget, +): Promise> { + return db.query.taskPullRequests.findMany({ where: and( - eq( - taskPullRequests.sourceControlProvider, - parsedInput.sourceControlProvider ?? 'github', - ), - eq(taskPullRequests.repository, parsedInput.repository), - eq(taskPullRequests.prNumber, parsedInput.prNumber), + eq(taskPullRequests.sourceControlProvider, target.sourceControlProvider), + eq(taskPullRequests.repository, target.repository), + eq(taskPullRequests.prNumber, target.prNumber), ), columns: { taskId: true }, }); +} +async function enqueueLinkedPrReviewNotification( + parsedInput: EnqueuePrReviewNotificationInput, + prTaskLinks: Array<{ taskId: string }>, +): Promise { const taskIds = Array.from(new Set(prTaskLinks.map((link) => link.taskId))); - if (taskIds.length === 0) { - return { notifiedTaskCount: 0, reason: 'no_linked_tasks' }; - } - const isRoomoteEvent = parsedInput.event.roomoteAuthored === true; const isRoomoteSummary = isRoomoteEvent && parsedInput.event.kind === 'review_summary'; @@ -798,6 +1395,8 @@ export async function enqueuePrReviewNotification( taskId, repository: parsedInput.repository, prNumber: parsedInput.prNumber, + sourceControlProvider: + parsedInput.sourceControlProvider ?? ('github' as const), immediate, batchKind: isRoomoteEvent ? ('roomote' as const) : ('human' as const), ...(event.batchId ? { batchId: event.batchId } : {}), @@ -818,7 +1417,7 @@ export async function enqueuePrReviewNotification( prUrl: parsedInput.prUrl, deferrals: 0, immediate, - sourceControlProvider: parsedInput.sourceControlProvider, + sourceControlProvider: target.sourceControlProvider, }, { delay: notificationDelayMs }, ); diff --git a/turbo.json b/turbo.json index 3dff58f92..ff8c0cd22 100644 --- a/turbo.json +++ b/turbo.json @@ -13,6 +13,7 @@ "globalPassThroughEnv": [ "APP_ENV", "DATABASE_URL", + "REDIS_URL", "MISE_DATA_DIR", "MISE_CACHE_DIR", "SKIP_ENV_VALIDATION",