From 1d78124c9d58a2bf777f3161b2e3302ea1e62b51 Mon Sep 17 00:00:00 2001 From: "@mrubens" <2600+mrubens@users.noreply.github.com> Date: Wed, 9 Sep 2026 19:56:48 +0000 Subject: [PATCH 1/2] fix: keep one Slack review action set per thread --- .../pr-review-notification-units.test.ts | 171 ++++++++++++++++++ .../src/lib/pr-review-notification-units.ts | 145 ++++++++++++--- .../__tests__/pr-review-action.test.ts | 51 ++++-- .../server/lib/task-runs/pr-review-action.ts | 61 +++++-- 4 files changed, 378 insertions(+), 50 deletions(-) diff --git a/packages/db/src/lib/__tests__/pr-review-notification-units.test.ts b/packages/db/src/lib/__tests__/pr-review-notification-units.test.ts index 57976a504..a2760962b 100644 --- a/packages/db/src/lib/__tests__/pr-review-notification-units.test.ts +++ b/packages/db/src/lib/__tests__/pr-review-notification-units.test.ts @@ -3,6 +3,7 @@ import { randomUUID } from 'node:crypto'; import { and, attachCanonicalPrReviewActionMessage, + attachCanonicalPrReviewActionMessageWithRetirement, claimCanonicalPrReviewAction, claimDueCanonicalPrReviewDeliveries, completeCanonicalPrReviewActionDispatch, @@ -1119,6 +1120,50 @@ describe('canonical PR review notification ownership', () => { return claim; }; + const setUpSlackTaskDelivery = async (input: { + prNumber: number; + workspaceId: string; + channelId: string; + threadId: string; + }) => { + const task = await taskFactory.create(); + const repository = `owner/slack-thread-${task.id}`; + await associate(task.id, repository, input.prNumber); + await persistPrReviewEvent( + eventInput({ + repository, + prNumber: input.prNumber, + eventKey: `slack-thread-${task.id}`, + }), + ); + const claim = (await claimForRepository(repository)).find( + ({ repository: claimedRepository }) => claimedRepository === repository, + ); + if (!claim || claim.ownershipVersion !== 'canonical') { + throw new Error('expected canonical claim'); + } + await transitionCanonicalPrReviewDelivery({ + deliveryId: claim.deliveryId, + leaseToken: claim.leaseToken, + expected: 'claimed', + status: 'prepared', + }); + await transitionCanonicalPrReviewDelivery({ + deliveryId: claim.deliveryId, + leaseToken: claim.leaseToken, + expected: 'prepared', + status: 'prompt_posting', + values: { + followUpPrompt: `Resolve feedback on ${repository}#${input.prNumber}.`, + routeProvider: 'slack', + routeWorkspaceId: input.workspaceId, + routeChannelId: input.channelId, + routeThreadId: input.threadId, + }, + }); + return { claim, repository, taskId: task.id }; + }; + const postAction = async (repository: string) => { const claim = await claimToPromptPosting(repository); await expect( @@ -1205,6 +1250,87 @@ describe('canonical PR review notification ownership', () => { }); }); + it('keeps only the newest Slack action across repositories and isolates workspaces and threads', async () => { + const mainRoute = { + workspaceId: 'T-thread-wide', + channelId: 'C-shared', + threadId: '111.222', + }; + const [older, newest, otherWorkspace, otherThread] = await Promise.all([ + setUpSlackTaskDelivery({ ...mainRoute, prNumber: 41 }), + setUpSlackTaskDelivery({ ...mainRoute, prNumber: 42 }), + setUpSlackTaskDelivery({ + ...mainRoute, + workspaceId: 'T-other', + prNumber: 43, + }), + setUpSlackTaskDelivery({ + ...mainRoute, + threadId: '333.444', + prNumber: 44, + }), + ]); + + await attachCanonicalPrReviewActionMessage( + older.claim.deliveryId, + '100.000001', + older.claim.leaseToken, + ); + await attachCanonicalPrReviewActionMessage( + otherWorkspace.claim.deliveryId, + 'message-workspace', + otherWorkspace.claim.leaseToken, + ); + await attachCanonicalPrReviewActionMessage( + otherThread.claim.deliveryId, + 'message-thread', + otherThread.claim.leaseToken, + ); + await expect( + attachCanonicalPrReviewActionMessageWithRetirement( + newest.claim.deliveryId, + '100.000002', + newest.claim.leaseToken, + ), + ).resolves.toEqual({ + attached: true, + superseded: [ + expect.objectContaining({ + deliveryId: older.claim.deliveryId, + provider: 'slack', + slackTeamId: mainRoute.workspaceId, + channelId: mainRoute.channelId, + threadId: mainRoute.threadId, + messageId: '100.000001', + }), + ], + }); + + await expect(deliveryStatusOf(older.claim.deliveryId)).resolves.toBe( + 'dismissed', + ); + await expect(deliveryStatusOf(newest.claim.deliveryId)).resolves.toBe( + 'awaiting_user_action', + ); + await expect( + deliveryStatusOf(otherWorkspace.claim.deliveryId), + ).resolves.toBe('awaiting_user_action'); + await expect(deliveryStatusOf(otherThread.claim.deliveryId)).resolves.toBe( + 'awaiting_user_action', + ); + await expect( + claimCanonicalPrReviewAction({ + deliveryId: newest.claim.deliveryId, + choice: 'yes', + expectedSlackTeamId: mainRoute.workspaceId, + }), + ).resolves.toMatchObject({ + taskId: newest.taskId, + repository: newest.repository, + prNumber: 42, + }); + }); + it('retires only awaiting offers from older PR heads after a new commit', async () => { const task = await taskFactory.create(); const run = await runFactory.create({ taskId: task.id }); @@ -1756,4 +1882,49 @@ describe('canonical PR review notification ownership', () => { ); expect(statuses.filter((s) => s === 'dismissed')).toHaveLength(1); }); + + it('keeps exactly one awaiting Slack offer when different task destinations attach concurrently', async () => { + const route = { + workspaceId: 'T-race', + channelId: 'C-race', + threadId: '555.666', + }; + const [first, second] = await Promise.all([ + setUpSlackTaskDelivery({ ...route, prNumber: 51 }), + setUpSlackTaskDelivery({ ...route, prNumber: 52 }), + ]); + + const results = await Promise.all([ + attachCanonicalPrReviewActionMessageWithRetirement( + first.claim.deliveryId, + '100.000001', + first.claim.leaseToken, + ), + attachCanonicalPrReviewActionMessageWithRetirement( + second.claim.deliveryId, + '100.000002', + second.claim.leaseToken, + ), + ]); + expect(results).toEqual([ + expect.objectContaining({ attached: true }), + expect.objectContaining({ attached: true }), + ]); + expect(results.flatMap(({ superseded }) => superseded)).toEqual([ + expect.objectContaining({ + provider: 'slack', + slackTeamId: route.workspaceId, + channelId: route.channelId, + threadId: route.threadId, + messageId: '100.000001', + }), + ]); + + await expect(deliveryStatusOf(first.claim.deliveryId)).resolves.toBe( + 'dismissed', + ); + await expect(deliveryStatusOf(second.claim.deliveryId)).resolves.toBe( + 'awaiting_user_action', + ); + }); }); diff --git a/packages/db/src/lib/pr-review-notification-units.ts b/packages/db/src/lib/pr-review-notification-units.ts index cb5064721..6e67dcb1e 100644 --- a/packages/db/src/lib/pr-review-notification-units.ts +++ b/packages/db/src/lib/pr-review-notification-units.ts @@ -1341,27 +1341,59 @@ export async function getCanonicalPrReviewAction(deliveryId: string): Promise<{ /** * Marks the posted canonical action as awaiting the user and retires every - * older awaiting offer in the same destination conversation, mirroring the - * legacy Redis path's claim-older-offers-on-attach semantics. Only the newest - * offer in a conversation stays actionable; earlier ones would otherwise - * accumulate as a stack of pending cards. Attach and retirement run in one - * transaction serialized per destination (claims are only serialized per - * repository/PR, so concurrent attaches into the same conversation would - * otherwise dismiss each other), and retired offers' cached transcript - * payloads are dismissed so already-rendered session cards deactivate. + * older awaiting offer in the same Slack thread or logical destination, + * mirroring the legacy Redis path's claim-older-offers-on-attach semantics. + * Only the newest offer in a conversation stays actionable; earlier ones + * would otherwise accumulate as a stack of pending cards. Attach and + * retirement run in one transaction serialized per retirement scope (claims + * are only serialized per repository/PR, so concurrent attaches into the same + * conversation would otherwise dismiss each other), and retired offers' + * cached transcript payloads are dismissed so already-rendered session cards + * deactivate. */ export async function attachCanonicalPrReviewActionMessage( deliveryId: string, messageId: string, leaseToken: string, ): Promise { + return ( + await attachCanonicalPrReviewActionMessageWithRetirement( + deliveryId, + messageId, + leaseToken, + ) + ).attached; +} + +export async function attachCanonicalPrReviewActionMessageWithRetirement( + deliveryId: string, + messageId: string, + leaseToken: string, +): Promise<{ + attached: boolean; + superseded: Array<{ + deliveryId: string; + provider: 'slack' | 'teams' | 'telegram' | 'discord' | null; + slackTeamId: string | null; + channelId: string | null; + threadId: string | null; + messageId: string | null; + }>; +}> { return db.transaction(async (tx) => { const delivery = await tx.query.prReviewNotificationDeliveries.findFirst({ where: eq(prReviewNotificationDeliveries.id, deliveryId), - columns: { destinationKind: true, destinationKey: true }, + columns: { + destinationKind: true, + destinationKey: true, + routeProvider: true, + routeWorkspaceId: true, + routeChannelId: true, + routeThreadId: true, + }, }); if (!delivery) { - return false; + return { attached: false, superseded: [] }; } const destination = @@ -1371,9 +1403,25 @@ export async function attachCanonicalPrReviewActionMessage( key: delivery.destinationKey, } : null; - if (destination) { + const slackThread = + delivery.routeProvider === 'slack' && + delivery.routeWorkspaceId && + delivery.routeChannelId && + delivery.routeThreadId + ? { + workspaceId: delivery.routeWorkspaceId, + channelId: delivery.routeChannelId, + threadId: delivery.routeThreadId, + } + : null; + const retirementLock = slackThread + ? `pr-review-slack-thread:${slackThread.workspaceId}:${slackThread.channelId}:${slackThread.threadId}` + : destination + ? `pr-review-destination:${destination.kind}:${destination.key}` + : null; + if (retirementLock) { await tx.execute( - sql`select pg_advisory_xact_lock(hashtextextended(${`pr-review-destination:${destination.kind}:${destination.key}`}, 0))`, + sql`select pg_advisory_xact_lock(hashtextextended(${retirementLock}, 0))`, ); } @@ -1395,11 +1443,60 @@ export async function attachCanonicalPrReviewActionMessage( ) .returning({ id: prReviewNotificationDeliveries.id }); if (rows.length !== 1) { - return false; + return { attached: false, superseded: [] }; } - if (!destination) { - return true; + const retirementScope = slackThread + ? and( + eq(prReviewNotificationDeliveries.routeProvider, 'slack'), + eq( + prReviewNotificationDeliveries.routeWorkspaceId, + slackThread.workspaceId, + ), + eq( + prReviewNotificationDeliveries.routeChannelId, + slackThread.channelId, + ), + eq( + prReviewNotificationDeliveries.routeThreadId, + slackThread.threadId, + ), + ) + : destination + ? and( + eq( + prReviewNotificationDeliveries.destinationKind, + destination.kind, + ), + eq(prReviewNotificationDeliveries.destinationKey, destination.key), + ) + : null; + if (!retirementScope) { + return { attached: true, superseded: [] }; + } + + let retainedDeliveryId = deliveryId; + if (slackThread) { + const awaiting = await tx + .select({ + deliveryId: prReviewNotificationDeliveries.id, + messageId: prReviewNotificationDeliveries.providerMessageId, + }) + .from(prReviewNotificationDeliveries) + .where( + and( + eq(prReviewNotificationDeliveries.status, 'awaiting_user_action'), + retirementScope, + ), + ); + retainedDeliveryId = awaiting.reduce( + (newest, candidate) => + candidate.messageId && + (!newest.messageId || candidate.messageId > newest.messageId) + ? candidate + : newest, + awaiting[0]!, + ).deliveryId; } const retired = await tx @@ -1413,12 +1510,18 @@ export async function attachCanonicalPrReviewActionMessage( .where( and( eq(prReviewNotificationDeliveries.status, 'awaiting_user_action'), - eq(prReviewNotificationDeliveries.destinationKind, destination.kind), - eq(prReviewNotificationDeliveries.destinationKey, destination.key), - ne(prReviewNotificationDeliveries.id, deliveryId), + retirementScope, + ne(prReviewNotificationDeliveries.id, retainedDeliveryId), ), ) - .returning({ id: prReviewNotificationDeliveries.id }); + .returning({ + deliveryId: prReviewNotificationDeliveries.id, + provider: prReviewNotificationDeliveries.routeProvider, + slackTeamId: prReviewNotificationDeliveries.routeWorkspaceId, + channelId: prReviewNotificationDeliveries.routeChannelId, + threadId: prReviewNotificationDeliveries.routeThreadId, + messageId: prReviewNotificationDeliveries.providerMessageId, + }); if (retired.length > 0) { // Session cards render from the cached message payload, so retiring @@ -1432,12 +1535,12 @@ export async function attachCanonicalPrReviewActionMessage( .where( inArray( sql`${fastAgentMessages.payload} -> 'prReviewAction' ->> 'deliveryId'`, - retired.map(({ id }) => id), + retired.map(({ deliveryId }) => deliveryId), ), ); } - return true; + return { attached: true, superseded: retired }; }); } diff --git a/packages/sdk/src/server/lib/task-runs/__tests__/pr-review-action.test.ts b/packages/sdk/src/server/lib/task-runs/__tests__/pr-review-action.test.ts index 6cc3ed645..8294c6bf0 100644 --- a/packages/sdk/src/server/lib/task-runs/__tests__/pr-review-action.test.ts +++ b/packages/sdk/src/server/lib/task-runs/__tests__/pr-review-action.test.ts @@ -68,7 +68,7 @@ vi.mock('@roomote/db/server', async () => { return { ...actual, - attachCanonicalPrReviewActionMessage: (...args: unknown[]) => + attachCanonicalPrReviewActionMessageWithRetirement: (...args: unknown[]) => mockAttachCanonical(...args), claimCanonicalPrReviewAction: vi.fn().mockResolvedValue(null), retireCanonicalPrReviewActionsForDestination: (...args: unknown[]) => @@ -119,7 +119,10 @@ describe('PR review action state', () => { mockFindPreference.mockResolvedValue(null); mockRetireCanonical.mockResolvedValue([]); mockRetireCanonicalForPullRequest.mockResolvedValue([]); - mockAttachCanonical.mockResolvedValue(false); + mockAttachCanonical.mockResolvedValue({ + attached: false, + superseded: [], + }); mockGetCommunicationProviderAdapter.mockResolvedValue(null); }); @@ -270,7 +273,7 @@ describe('PR review action state', () => { expect(mockEval.mock.calls[0]?.[0]).toContain('prior.retired = true'); }); - it('returns and de-indexes the prior offer for the same PR context', async () => { + it('returns and de-indexes a prior Slack offer for another PR in the same thread', async () => { mockGet.mockResolvedValue( JSON.stringify({ nonce: 'nonce-new', @@ -278,8 +281,8 @@ describe('PR review action state', () => { slackTeamId: 'T1', channelId: 'C1', threadId: '111.222', - repository: 'owner/repo', - prNumber: 42, + repository: 'other/repository', + prNumber: 99, }), ); mockEval.mockResolvedValue([ @@ -315,7 +318,13 @@ describe('PR review action state', () => { 'pr-review-action:thread:slack:T1:C1:111.222', ); expect(mockEval.mock.calls[0]?.[0]).toContain( - 'prior.prNumber == pending.prNumber', + "if pending.provider == 'slack' then", + ); + expect(mockEval.mock.calls[0]?.[0]).toContain( + "return prior.provider == 'slack' and sameSlackTeam", + ); + expect(mockEval.mock.calls[0]?.[0]).toContain( + 'prior.repository == pending.repository', ); }); @@ -353,11 +362,12 @@ describe('PR review action state', () => { }); }); - it('retires legacy offers after a canonical attachment succeeds', async () => { + it('returns canonical and legacy Slack offers from other PRs in the same thread', async () => { const context = { nonce: '00000000-0000-4000-8000-000000000001', canonicalDeliveryId: '00000000-0000-4000-8000-000000000001', - provider: 'discord' as const, + provider: 'slack' as const, + slackTeamId: 'T1', taskId: 'task-1', repository: 'owner/repo', prNumber: 42, @@ -370,9 +380,23 @@ describe('PR review action state', () => { ...context, nonce: 'legacy-nonce', canonicalDeliveryId: undefined, + repository: 'legacy/repository', + prNumber: 7, messageId: 'legacy-message', }; - mockAttachCanonical.mockResolvedValue(true); + mockAttachCanonical.mockResolvedValue({ + attached: true, + superseded: [ + { + deliveryId: '00000000-0000-4000-8000-000000000002', + provider: 'slack', + slackTeamId: 'T1', + channelId: 'channel-1', + threadId: 'thread-1', + messageId: 'canonical-old-message', + }, + ], + }); mockEval.mockResolvedValue([JSON.stringify(legacy)]); await expect( @@ -383,14 +407,17 @@ describe('PR review action state', () => { ), ).resolves.toEqual({ attached: true, - superseded: [expect.objectContaining({ nonce: 'legacy-nonce' })], + superseded: [ + expect.objectContaining({ messageId: 'canonical-old-message' }), + expect.objectContaining({ nonce: 'legacy-nonce' }), + ], }); expect(mockEval.mock.calls[0]?.[0]).toContain( - 'pending.repository == context.repository', + "if context.provider == 'slack' then", ); expect(mockEval.mock.calls[0]?.[2]).toBe( - 'pr-review-action:thread:discord:channel-1:thread-1', + 'pr-review-action:thread:slack:T1:channel-1:thread-1', ); }); diff --git a/packages/sdk/src/server/lib/task-runs/pr-review-action.ts b/packages/sdk/src/server/lib/task-runs/pr-review-action.ts index 40eac52b0..34e4a374f 100644 --- a/packages/sdk/src/server/lib/task-runs/pr-review-action.ts +++ b/packages/sdk/src/server/lib/task-runs/pr-review-action.ts @@ -1,6 +1,6 @@ import { and, - attachCanonicalPrReviewActionMessage, + attachCanonicalPrReviewActionMessageWithRetirement as attachCanonicalPrReviewActionMessage, claimCanonicalPrReviewAction, completeCanonicalPrReviewActionDispatch, db, @@ -127,9 +127,12 @@ local nonces = redis.call('smembers', KEYS[2]) local function sameContext(prior) local sameSlackTeam = (prior.slackTeamId == pending.slackTeamId) or (not prior.slackTeamId and not pending.slackTeamId) - return prior.repository == pending.repository + if pending.provider == 'slack' then + return prior.provider == 'slack' and sameSlackTeam + end + return prior.provider == pending.provider + and prior.repository == pending.repository and prior.prNumber == pending.prNumber - and sameSlackTeam end for _, nonce in ipairs(nonces) do if nonce ~= pending.nonce then @@ -206,9 +209,15 @@ for _, nonce in ipairs(nonces) do local pending = cjson.decode(val) local sameSlackTeam = (pending.slackTeamId == context.slackTeamId) or (not pending.slackTeamId and not context.slackTeamId) - if pending.repository == context.repository - and pending.prNumber == context.prNumber - and sameSlackTeam then + local sameContext = false + if context.provider == 'slack' then + sameContext = pending.provider == 'slack' and sameSlackTeam + else + sameContext = pending.provider == context.provider + and pending.repository == context.repository + and pending.prNumber == context.prNumber + end + if sameContext then redis.call('srem', KEYS[1], nonce) if pending.messageId then redis.call('del', actionKey) @@ -265,8 +274,8 @@ export async function setPendingPrReviewAction( /** * Records the posted notification message id on an already-stored pending * offer so retirement can edit the message later. This also atomically claims - * every older offer for the same PR conversation. No-op when the new offer was - * already claimed. + * every older offer for the same Slack thread or provider-specific PR + * conversation. No-op when the new offer was already claimed. */ export async function attachPendingPrReviewActionMessage( nonce: string, @@ -287,21 +296,39 @@ export async function attachPendingPrReviewActionMessageWithRetirement( options: { leaseToken?: string; context?: PendingPrReviewAction } = {}, ): Promise<{ attached: boolean; - superseded: PendingPrReviewAction[]; + superseded: RetirablePrReviewActionMessage[]; }> { - if ( - isUuid(nonce) && - options.leaseToken && - (await attachCanonicalPrReviewActionMessage( + if (isUuid(nonce) && options.leaseToken) { + const canonicalResult = await attachCanonicalPrReviewActionMessage( nonce, messageId, options.leaseToken, - )) - ) { - const superseded = options.context + ); + if (!canonicalResult.attached) { + return { attached: false, superseded: [] }; + } + const canonicalSuperseded = canonicalResult.superseded.flatMap((action) => + action.provider && action.provider !== 'teams' && action.channelId + ? [ + { + provider: action.provider, + ...(action.provider === 'slack' && action.slackTeamId + ? { slackTeamId: action.slackTeamId } + : {}), + channelId: action.channelId, + threadId: action.threadId, + messageId: action.messageId, + } satisfies RetirablePrReviewActionMessage, + ] + : [], + ); + const legacySuperseded = options.context ? await retireLegacyPrReviewActionsForContext(options.context) : []; - return { attached: true, superseded }; + return { + attached: true, + superseded: [...canonicalSuperseded, ...legacySuperseded], + }; } const redis = getRedis(); From f67d0ed1823e1ba47c94b67b7abc6393b860d2b6 Mon Sep 17 00:00:00 2001 From: "@mrubens" <2600+mrubens@users.noreply.github.com> Date: Wed, 9 Sep 2026 20:33:19 +0000 Subject: [PATCH 2/2] fix: serialize Slack review action retirement --- .../pr-review-notification-units.test.ts | 137 ++++++++ .../src/lib/pr-review-notification-units.ts | 127 +++++++- .../__tests__/pr-review-action.test.ts | 179 +++++++++-- .../server/lib/task-runs/pr-review-action.ts | 296 +++++++++++++++--- 4 files changed, 676 insertions(+), 63 deletions(-) diff --git a/packages/db/src/lib/__tests__/pr-review-notification-units.test.ts b/packages/db/src/lib/__tests__/pr-review-notification-units.test.ts index a2760962b..0732edbf1 100644 --- a/packages/db/src/lib/__tests__/pr-review-notification-units.test.ts +++ b/packages/db/src/lib/__tests__/pr-review-notification-units.test.ts @@ -32,6 +32,7 @@ import { upsertPrReviewAutoPreference, userFactory, withCanonicalPrReviewAutoDispatchFence, + withCanonicalPrReviewSlackThreadActionFence, } from '../../server'; import { RunStatus } from '@roomote/types'; @@ -1331,6 +1332,142 @@ describe('canonical PR review notification ownership', () => { }); }); + it('rolls back a canonical Slack attachment when shared-store arbitration fails', async () => { + const delivery = await setUpSlackTaskDelivery({ + workspaceId: 'T-rollback', + channelId: 'C-rollback', + threadId: '111.333', + prNumber: 45, + }); + + await expect( + attachCanonicalPrReviewActionMessageWithRetirement( + delivery.claim.deliveryId, + '100.000001', + delivery.claim.leaseToken, + { + arbitrateSlackThread: async () => { + throw new Error('Redis unavailable'); + }, + }, + ), + ).rejects.toThrow('Redis unavailable'); + await expect( + db.query.prReviewNotificationDeliveries.findFirst({ + where: eq(prReviewNotificationDeliveries.id, delivery.claim.deliveryId), + columns: { + status: true, + leaseToken: true, + providerMessageId: true, + }, + }), + ).resolves.toEqual({ + status: 'prompt_posting', + leaseToken: delivery.claim.leaseToken, + providerMessageId: null, + }); + }); + + it('retires canonical Slack controls when a newer legacy message wins the shared fence', async () => { + const canonical = await setUpSlackTaskDelivery({ + workspaceId: 'T-shared-fence', + channelId: 'C-shared-fence', + threadId: '222.333', + prNumber: 46, + }); + await attachCanonicalPrReviewActionMessage( + canonical.claim.deliveryId, + '100.000001', + canonical.claim.leaseToken, + ); + + await expect( + withCanonicalPrReviewSlackThreadActionFence( + { + slackTeamId: 'T-shared-fence', + channelId: 'C-shared-fence', + threadId: '222.333', + }, + async (newestCanonicalMessageId) => ({ + newestMessageId: '100.000002', + result: newestCanonicalMessageId, + }), + ), + ).resolves.toEqual({ + result: '100.000001', + superseded: [ + expect.objectContaining({ + deliveryId: canonical.claim.deliveryId, + messageId: '100.000001', + }), + ], + }); + await expect(deliveryStatusOf(canonical.claim.deliveryId)).resolves.toBe( + 'dismissed', + ); + }); + + it('retires a newly attached canonical action when arbitration finds a newer legacy message', async () => { + const canonical = await setUpSlackTaskDelivery({ + workspaceId: 'T-canonical-loses', + channelId: 'C-canonical-loses', + threadId: '333.444', + prNumber: 47, + }); + + await expect( + attachCanonicalPrReviewActionMessageWithRetirement( + canonical.claim.deliveryId, + '100.000001', + canonical.claim.leaseToken, + { arbitrateSlackThread: async () => '100.000002' }, + ), + ).resolves.toEqual({ + attached: true, + superseded: [ + expect.objectContaining({ + deliveryId: canonical.claim.deliveryId, + messageId: '100.000001', + }), + ], + }); + await expect(deliveryStatusOf(canonical.claim.deliveryId)).resolves.toBe( + 'dismissed', + ); + }); + + it('does not retire canonical actions when a legacy attachment was already claimed', async () => { + const canonical = await setUpSlackTaskDelivery({ + workspaceId: 'T-claimed-legacy', + channelId: 'C-claimed-legacy', + threadId: '444.555', + prNumber: 48, + }); + await attachCanonicalPrReviewActionMessage( + canonical.claim.deliveryId, + '100.000001', + canonical.claim.leaseToken, + ); + + await expect( + withCanonicalPrReviewSlackThreadActionFence( + { + slackTeamId: 'T-claimed-legacy', + channelId: 'C-claimed-legacy', + threadId: '444.555', + }, + async (newestCanonicalMessageId) => ({ + newestMessageId: newestCanonicalMessageId!, + retireCanonical: false, + result: 'not-attached', + }), + ), + ).resolves.toEqual({ result: 'not-attached', superseded: [] }); + await expect(deliveryStatusOf(canonical.claim.deliveryId)).resolves.toBe( + 'awaiting_user_action', + ); + }); + it('retires only awaiting offers from older PR heads after a new commit', async () => { const task = await taskFactory.create(); const run = await runFactory.create({ taskId: task.id }); diff --git a/packages/db/src/lib/pr-review-notification-units.ts b/packages/db/src/lib/pr-review-notification-units.ts index 6e67dcb1e..83962ba4f 100644 --- a/packages/db/src/lib/pr-review-notification-units.ts +++ b/packages/db/src/lib/pr-review-notification-units.ts @@ -1369,6 +1369,11 @@ export async function attachCanonicalPrReviewActionMessageWithRetirement( deliveryId: string, messageId: string, leaseToken: string, + options: { + arbitrateSlackThread?: ( + newestCanonicalMessageId: string, + ) => Promise; + } = {}, ): Promise<{ attached: boolean; superseded: Array<{ @@ -1475,7 +1480,6 @@ export async function attachCanonicalPrReviewActionMessageWithRetirement( return { attached: true, superseded: [] }; } - let retainedDeliveryId = deliveryId; if (slackThread) { const awaiting = await tx .select({ @@ -1489,14 +1493,23 @@ export async function attachCanonicalPrReviewActionMessageWithRetirement( retirementScope, ), ); - retainedDeliveryId = awaiting.reduce( + const newestCanonical = awaiting.reduce( (newest, candidate) => candidate.messageId && (!newest.messageId || candidate.messageId > newest.messageId) ? candidate : newest, awaiting[0]!, - ).deliveryId; + ); + const retainedMessageId = options.arbitrateSlackThread + ? await options.arbitrateSlackThread(newestCanonical.messageId!) + : newestCanonical.messageId!; + const superseded = await retireCanonicalPrReviewSlackActionsExcept( + tx, + retirementScope, + retainedMessageId, + ); + return { attached: true, superseded }; } const retired = await tx @@ -1511,7 +1524,7 @@ export async function attachCanonicalPrReviewActionMessageWithRetirement( and( eq(prReviewNotificationDeliveries.status, 'awaiting_user_action'), retirementScope, - ne(prReviewNotificationDeliveries.id, retainedDeliveryId), + ne(prReviewNotificationDeliveries.id, deliveryId), ), ) .returning({ @@ -1544,6 +1557,112 @@ export async function attachCanonicalPrReviewActionMessageWithRetirement( }); } +/** Serializes a legacy Slack attachment with canonical thread retirement. */ +export async function withCanonicalPrReviewSlackThreadActionFence( + input: { + slackTeamId: string; + channelId: string; + threadId: string; + }, + arbitrate: (newestCanonicalMessageId: string | null) => Promise<{ + newestMessageId: string; + result: T; + retireCanonical?: boolean; + }>, +): Promise<{ result: T; superseded: CanonicalPrReviewActionMessage[] }> { + return db.transaction(async (tx) => { + await tx.execute( + sql`select pg_advisory_xact_lock(hashtextextended(${`pr-review-slack-thread:${input.slackTeamId}:${input.channelId}:${input.threadId}`}, 0))`, + ); + const retirementScope = and( + eq(prReviewNotificationDeliveries.routeProvider, 'slack'), + eq(prReviewNotificationDeliveries.routeWorkspaceId, input.slackTeamId), + eq(prReviewNotificationDeliveries.routeChannelId, input.channelId), + eq(prReviewNotificationDeliveries.routeThreadId, input.threadId), + ); + const awaiting = await tx + .select({ messageId: prReviewNotificationDeliveries.providerMessageId }) + .from(prReviewNotificationDeliveries) + .where( + and( + eq(prReviewNotificationDeliveries.status, 'awaiting_user_action'), + retirementScope, + ), + ); + const newestCanonicalMessageId = awaiting.reduce( + (newest, { messageId: candidate }) => + candidate && (!newest || candidate > newest) ? candidate : newest, + null, + ); + const arbitration = await arbitrate(newestCanonicalMessageId); + const superseded = + arbitration.retireCanonical === false + ? [] + : await retireCanonicalPrReviewSlackActionsExcept( + tx, + retirementScope, + arbitration.newestMessageId, + ); + return { result: arbitration.result, superseded }; + }); +} + +type CanonicalPrReviewActionMessage = { + deliveryId: string; + provider: 'slack' | 'teams' | 'telegram' | 'discord' | null; + slackTeamId: string | null; + channelId: string | null; + threadId: string | null; + messageId: string | null; +}; + +async function retireCanonicalPrReviewSlackActionsExcept( + tx: DatabaseOrTransaction, + retirementScope: ReturnType, + retainedMessageId: string, +): Promise { + const retired = await tx + .update(prReviewNotificationDeliveries) + .set({ + status: 'dismissed', + actionClaimedAt: new Date(), + completedAt: new Date(), + updatedAt: new Date(), + }) + .where( + and( + eq(prReviewNotificationDeliveries.status, 'awaiting_user_action'), + retirementScope, + ne(prReviewNotificationDeliveries.providerMessageId, retainedMessageId), + ), + ) + .returning({ + deliveryId: prReviewNotificationDeliveries.id, + provider: prReviewNotificationDeliveries.routeProvider, + slackTeamId: prReviewNotificationDeliveries.routeWorkspaceId, + channelId: prReviewNotificationDeliveries.routeChannelId, + threadId: prReviewNotificationDeliveries.routeThreadId, + messageId: prReviewNotificationDeliveries.providerMessageId, + }); + + if (retired.length > 0) { + await tx + .update(fastAgentMessages) + .set({ + payload: sql`jsonb_set(coalesce(${fastAgentMessages.payload}, '{}'::jsonb), '{prReviewAction,status}', to_jsonb('dismissed'::text), true)`, + updatedAt: sql`now()`, + }) + .where( + inArray( + sql`${fastAgentMessages.payload} -> 'prReviewAction' ->> 'deliveryId'`, + retired.map(({ deliveryId: id }) => id), + ), + ); + } + + return retired; +} + export async function claimCanonicalPrReviewAction(input: { deliveryId: string; choice: 'yes' | 'auto' | 'dismiss'; diff --git a/packages/sdk/src/server/lib/task-runs/__tests__/pr-review-action.test.ts b/packages/sdk/src/server/lib/task-runs/__tests__/pr-review-action.test.ts index 8294c6bf0..3c479beda 100644 --- a/packages/sdk/src/server/lib/task-runs/__tests__/pr-review-action.test.ts +++ b/packages/sdk/src/server/lib/task-runs/__tests__/pr-review-action.test.ts @@ -11,6 +11,7 @@ const { mockRetireCanonical, mockRetireCanonicalForPullRequest, mockAttachCanonical, + mockSlackThreadFence, mockGetCommunicationProviderAdapter, mockSlackInstallation, mockSlackBlocks, @@ -32,6 +33,7 @@ const { mockRetireCanonical: vi.fn(), mockRetireCanonicalForPullRequest: vi.fn(), mockAttachCanonical: vi.fn(), + mockSlackThreadFence: vi.fn(), mockGetCommunicationProviderAdapter: vi.fn(), mockSlackInstallation: vi.fn(), mockSlackBlocks: vi.fn(), @@ -70,6 +72,8 @@ vi.mock('@roomote/db/server', async () => { ...actual, attachCanonicalPrReviewActionMessageWithRetirement: (...args: unknown[]) => mockAttachCanonical(...args), + withCanonicalPrReviewSlackThreadActionFence: (...args: unknown[]) => + mockSlackThreadFence(...args), claimCanonicalPrReviewAction: vi.fn().mockResolvedValue(null), retireCanonicalPrReviewActionsForDestination: (...args: unknown[]) => mockRetireCanonical(...args), @@ -123,6 +127,15 @@ describe('PR review action state', () => { attached: false, superseded: [], }); + mockSlackThreadFence.mockImplementation( + async ( + _input, + arbitrate: (messageId: string | null) => Promise, + ) => { + const arbitration = (await arbitrate(null)) as { result: unknown }; + return { result: arbitration.result, superseded: [] }; + }, + ); mockGetCommunicationProviderAdapter.mockResolvedValue(null); }); @@ -247,7 +260,7 @@ describe('PR review action state', () => { prNumber: 42, }), ); - mockEval.mockResolvedValue([1]); + mockEval.mockResolvedValue([1, 'message-1']); await expect( attachPendingPrReviewActionMessageWithRetirement('nonce-1', 'message-1'), @@ -260,6 +273,7 @@ describe('PR review action state', () => { 'pr-review-action:thread:discord:channel-1:thread-1', 'message-1', 'pr-review-action:', + '', ); expect(mockEval.mock.calls[0]?.[0]).toContain("'KEEPTTL'"); expect(mockEval.mock.calls[0]?.[0]).toContain( @@ -287,6 +301,7 @@ describe('PR review action state', () => { ); mockEval.mockResolvedValue([ 1, + '200.000002', JSON.stringify({ nonce: 'nonce-old', provider: 'slack', @@ -295,21 +310,21 @@ describe('PR review action state', () => { threadId: '111.222', repository: 'owner/repo', prNumber: 42, - messageId: 'message-old', + messageId: '200.000001', }), ]); await expect( attachPendingPrReviewActionMessageWithRetirement( 'nonce-new', - 'message-new', + '200.000002', ), ).resolves.toEqual({ attached: true, superseded: [ expect.objectContaining({ nonce: 'nonce-old', - messageId: 'message-old', + messageId: '200.000001', }), ], }); @@ -323,9 +338,73 @@ describe('PR review action state', () => { expect(mockEval.mock.calls[0]?.[0]).toContain( "return prior.provider == 'slack' and sameSlackTeam", ); + expect(mockEval.mock.calls[0]?.[0]).toContain( + 'prior.messageId and prior.messageId > winner', + ); expect(mockEval.mock.calls[0]?.[0]).toContain( 'prior.repository == pending.repository', ); + expect(mockSlackThreadFence).toHaveBeenCalledWith( + { + slackTeamId: 'T1', + channelId: 'C1', + threadId: '111.222', + }, + expect.any(Function), + ); + }); + + it('retires older canonical controls when a later-posted legacy offer wins', async () => { + const pending = { + nonce: 'legacy-newer', + provider: 'slack' as const, + slackTeamId: 'T1', + taskId: 'legacy-task', + repository: 'legacy/repository', + prNumber: 99, + prUrl: 'https://github.com/legacy/repository/pull/99', + channelId: 'C1', + threadId: '111.222', + followUpPrompt: 'Address the newer feedback.', + }; + mockGet.mockResolvedValue(JSON.stringify(pending)); + mockEval.mockResolvedValue([1, '200.000002']); + mockSlackThreadFence.mockImplementation(async (_input, arbitrate) => { + const arbitration = await arbitrate('200.000001'); + return { + result: arbitration.result, + superseded: [ + { + deliveryId: '00000000-0000-4000-8000-000000000001', + provider: 'slack', + slackTeamId: 'T1', + channelId: 'C1', + threadId: '111.222', + messageId: '200.000001', + }, + ], + }; + }); + + await expect( + attachPendingPrReviewActionMessageWithRetirement( + pending.nonce, + '200.000002', + ), + ).resolves.toEqual({ + attached: true, + superseded: [expect.objectContaining({ messageId: '200.000001' })], + }); + + expect(mockEval).toHaveBeenCalledWith( + expect.any(String), + 2, + 'pr-review-action:legacy-newer', + 'pr-review-action:thread:slack:T1:C1:111.222', + '200.000002', + 'pr-review-action:', + '200.000001', + ); }); it('returns a late-posting offer so its own stale controls are retired', async () => { @@ -342,6 +421,7 @@ describe('PR review action state', () => { mockGet.mockResolvedValue(JSON.stringify(lateOffer)); mockEval.mockResolvedValue([ 1, + 'message-old', JSON.stringify({ ...lateOffer, messageId: 'message-old' }), ]); @@ -384,43 +464,100 @@ describe('PR review action state', () => { prNumber: 7, messageId: 'legacy-message', }; - mockAttachCanonical.mockResolvedValue({ - attached: true, - superseded: [ - { - deliveryId: '00000000-0000-4000-8000-000000000002', - provider: 'slack', - slackTeamId: 'T1', - channelId: 'channel-1', - threadId: 'thread-1', - messageId: 'canonical-old-message', - }, - ], - }); - mockEval.mockResolvedValue([JSON.stringify(legacy)]); + mockAttachCanonical.mockImplementation( + async (_nonce, _messageId, _leaseToken, options) => { + await options.arbitrateSlackThread('200.000002'); + return { + attached: true, + superseded: [ + { + deliveryId: '00000000-0000-4000-8000-000000000002', + provider: 'slack', + slackTeamId: 'T1', + channelId: 'channel-1', + threadId: 'thread-1', + messageId: '200.000001', + }, + ], + }; + }, + ); + mockEval.mockResolvedValue([ + '200.000002', + JSON.stringify({ ...legacy, messageId: '100.000001' }), + ]); await expect( attachPendingPrReviewActionMessageWithRetirement( context.nonce, - 'canonical-message', + '200.000002', { leaseToken: 'lease-token', context }, ), ).resolves.toEqual({ attached: true, superseded: [ - expect.objectContaining({ messageId: 'canonical-old-message' }), + expect.objectContaining({ messageId: '200.000001' }), expect.objectContaining({ nonce: 'legacy-nonce' }), ], }); expect(mockEval.mock.calls[0]?.[0]).toContain( - "if context.provider == 'slack' then", + 'pending.messageId and pending.messageId > winner', ); expect(mockEval.mock.calls[0]?.[2]).toBe( 'pr-review-action:thread:slack:T1:channel-1:thread-1', ); }); + it('visually retires Redis losers when canonical attachment rolls back after arbitration', async () => { + const context = { + nonce: '00000000-0000-4000-8000-000000000003', + canonicalDeliveryId: '00000000-0000-4000-8000-000000000003', + provider: 'slack' as const, + slackTeamId: 'T1', + taskId: 'task-1', + repository: 'owner/repo', + prNumber: 42, + prUrl: 'https://github.com/owner/repo/pull/42', + channelId: 'C1', + threadId: '111.222', + followUpPrompt: 'Address the feedback.', + }; + const legacy = { + ...context, + nonce: 'legacy-loser', + canonicalDeliveryId: undefined, + messageId: '100.000001', + }; + mockEval.mockResolvedValue(['100.000002', JSON.stringify(legacy)]); + mockAttachCanonical.mockImplementation( + async (_nonce, _messageId, _leaseToken, options) => { + await options.arbitrateSlackThread('100.000002'); + throw new Error('database commit failed'); + }, + ); + mockSlackInstallation.mockResolvedValue({ botAccessToken: 'xoxb-test' }); + mockSlackBlocks.mockResolvedValue([ + { type: 'markdown', text: 'Review text remains.' }, + { type: 'actions', block_id: 'pr_review_action', elements: [] }, + ]); + + await expect( + attachPendingPrReviewActionMessageWithRetirement( + context.nonce, + '100.000002', + { leaseToken: 'lease-token', context }, + ), + ).rejects.toThrow('database commit failed'); + expect(mockSlackUpdate).toHaveBeenCalledWith({ + channel: 'C1', + ts: '100.000001', + message: { + blocks: [{ type: 'markdown', text: 'Review text remains.' }], + }, + }); + }); + it('claims every indexed offer through one atomic script', async () => { mockEval.mockResolvedValue([ JSON.stringify({ nonce: 'nonce-1', messageId: 'message-1' }), diff --git a/packages/sdk/src/server/lib/task-runs/pr-review-action.ts b/packages/sdk/src/server/lib/task-runs/pr-review-action.ts index 34e4a374f..9eae0adff 100644 --- a/packages/sdk/src/server/lib/task-runs/pr-review-action.ts +++ b/packages/sdk/src/server/lib/task-runs/pr-review-action.ts @@ -10,6 +10,7 @@ import { retireCanonicalPrReviewActionsForPullRequest, slackInstallations, upsertPrReviewAutoPreference, + withCanonicalPrReviewSlackThreadActionFence, } from '@roomote/db/server'; import { getRedis } from '@roomote/redis'; import { @@ -56,7 +57,7 @@ function isUuid(value: string): boolean { */ export interface PendingPrReviewAction { nonce: string; - /** Monotonic creation order used when concurrent offers finish out of order. */ + /** Monotonic creation order used by non-Slack attachment arbitration. */ createdOrder?: number; provider: PrReviewActionProvider; /** Slack workspace identity. Absent only on legacy pending records. */ @@ -114,13 +115,14 @@ return val // click claimed after the notification was posted. const ATTACH_PR_REVIEW_ACTION_MESSAGE_LUA = ` local val = redis.call('get', KEYS[1]) -if not val then return {0} end +if not val then return {0, ''} end local pending = cjson.decode(val) pending.messageId = ARGV[1] if pending.retired then redis.call('del', KEYS[1]) redis.call('srem', KEYS[2], pending.nonce) - return {1, cjson.encode(pending)} + local winner = ARGV[3] ~= '' and ARGV[3] or pending.messageId + return {1, winner, cjson.encode(pending)} end redis.call('set', KEYS[1], cjson.encode(pending), 'KEEPTTL') local nonces = redis.call('smembers', KEYS[2]) @@ -134,6 +136,44 @@ local function sameContext(prior) and prior.repository == pending.repository and prior.prNumber == pending.prNumber end +local claimed = {} +if pending.provider == 'slack' then + local winner = pending.messageId + if ARGV[3] ~= '' and ARGV[3] > winner then winner = ARGV[3] end + for _, nonce in ipairs(nonces) do + if nonce ~= pending.nonce then + local previous = redis.call('get', ARGV[2] .. nonce) + if previous then + local prior = cjson.decode(previous) + if sameContext(prior) and prior.messageId and prior.messageId > winner then + winner = prior.messageId + end + end + end + end + for _, nonce in ipairs(nonces) do + if nonce ~= pending.nonce then + local previousKey = ARGV[2] .. nonce + local previous = redis.call('get', previousKey) + if previous then + local prior = cjson.decode(previous) + if sameContext(prior) and prior.messageId and prior.messageId < winner then + redis.call('srem', KEYS[2], nonce) + redis.call('del', previousKey) + table.insert(claimed, previous) + end + end + end + end + if pending.messageId < winner then + redis.call('del', KEYS[1]) + redis.call('srem', KEYS[2], pending.nonce) + table.insert(claimed, cjson.encode(pending)) + end + table.insert(claimed, 1, winner) + table.insert(claimed, 1, 1) + return claimed +end for _, nonce in ipairs(nonces) do if nonce ~= pending.nonce then local previous = redis.call('get', ARGV[2] .. nonce) @@ -145,12 +185,11 @@ for _, nonce in ipairs(nonces) do and priorCreatedOrder > pendingCreatedOrder then redis.call('del', KEYS[1]) redis.call('srem', KEYS[2], pending.nonce) - return {1, cjson.encode(pending)} + return {1, pending.messageId, cjson.encode(pending)} end end end end -local claimed = {} for _, nonce in ipairs(nonces) do if nonce ~= pending.nonce then local previousKey = ARGV[2] .. nonce @@ -170,6 +209,7 @@ for _, nonce in ipairs(nonces) do end end end +table.insert(claimed, 1, pending.messageId) table.insert(claimed, 1, 1) return claimed `; @@ -232,6 +272,42 @@ end return retired `; +const ARBITRATE_LEGACY_SLACK_PR_REVIEW_ACTIONS_LUA = ` +local context = cjson.decode(ARGV[2]) +local winner = ARGV[3] +local nonces = redis.call('smembers', KEYS[1]) +for _, nonce in ipairs(nonces) do + local val = redis.call('get', ARGV[1] .. nonce) + if val then + local pending = cjson.decode(val) + local sameSlackTeam = (pending.slackTeamId == context.slackTeamId) + or (not pending.slackTeamId and not context.slackTeamId) + if pending.provider == 'slack' and sameSlackTeam + and pending.messageId and pending.messageId > winner then + winner = pending.messageId + end + end +end +local retired = {} +for _, nonce in ipairs(nonces) do + local actionKey = ARGV[1] .. nonce + local val = redis.call('get', actionKey) + if val then + local pending = cjson.decode(val) + local sameSlackTeam = (pending.slackTeamId == context.slackTeamId) + or (not pending.slackTeamId and not context.slackTeamId) + if pending.provider == 'slack' and sameSlackTeam + and pending.messageId and pending.messageId < winner then + redis.call('srem', KEYS[1], nonce) + redis.call('del', actionKey) + table.insert(retired, val) + end + end +end +table.insert(retired, 1, winner) +return retired +`; + function getPrReviewActionKey(nonce: string): string { return `${PR_REVIEW_ACTION_PREFIX}${nonce}`; } @@ -299,35 +375,60 @@ export async function attachPendingPrReviewActionMessageWithRetirement( superseded: RetirablePrReviewActionMessage[]; }> { if (isUuid(nonce) && options.leaseToken) { - const canonicalResult = await attachCanonicalPrReviewActionMessage( - nonce, - messageId, - options.leaseToken, - ); + const slackContext = + options.context?.provider === 'slack' && + options.context.slackTeamId && + options.context.threadId + ? { + ...options.context, + provider: 'slack' as const, + slackTeamId: options.context.slackTeamId, + threadId: options.context.threadId, + } + : null; + let legacySuperseded: PendingPrReviewAction[] = []; + let canonicalResult: Awaited< + ReturnType + >; + try { + canonicalResult = await attachCanonicalPrReviewActionMessage( + nonce, + messageId, + options.leaseToken, + slackContext + ? { + arbitrateSlackThread: async (newestCanonicalMessageId) => { + const result = await arbitrateLegacySlackPrReviewActions( + slackContext, + newestCanonicalMessageId, + ); + legacySuperseded = result.superseded; + return result.newestMessageId; + }, + } + : {}, + ); + } catch (error) { + await retirePrReviewActionMessagesBestEffort(legacySuperseded); + throw error; + } if (!canonicalResult.attached) { return { attached: false, superseded: [] }; } - const canonicalSuperseded = canonicalResult.superseded.flatMap((action) => - action.provider && action.provider !== 'teams' && action.channelId - ? [ - { - provider: action.provider, - ...(action.provider === 'slack' && action.slackTeamId - ? { slackTeamId: action.slackTeamId } - : {}), - channelId: action.channelId, - threadId: action.threadId, - messageId: action.messageId, - } satisfies RetirablePrReviewActionMessage, - ] - : [], + const canonicalSuperseded = toRetirableCanonicalMessages( + canonicalResult.superseded, ); - const legacySuperseded = options.context - ? await retireLegacyPrReviewActionsForContext(options.context) - : []; + const nonSlackLegacySuperseded = + options.context && !slackContext + ? await retireLegacyPrReviewActionsForContext(options.context) + : []; return { attached: true, - superseded: [...canonicalSuperseded, ...legacySuperseded], + superseded: [ + ...canonicalSuperseded, + ...legacySuperseded, + ...nonSlackLegacySuperseded, + ], }; } @@ -342,17 +443,130 @@ export async function attachPendingPrReviewActionMessageWithRetirement( return { attached: false, superseded: [] }; } - const rawClaims = await redis.eval( - ATTACH_PR_REVIEW_ACTION_MESSAGE_LUA, - 2, - getPrReviewActionKey(nonce), - getPrReviewActionThreadKey(pending), - messageId, - PR_REVIEW_ACTION_PREFIX, + const attachRedis = async (newestCanonicalMessageId: string | null) => { + const rawClaims = await redis.eval( + ATTACH_PR_REVIEW_ACTION_MESSAGE_LUA, + 2, + getPrReviewActionKey(nonce), + getPrReviewActionThreadKey(pending), + messageId, + PR_REVIEW_ACTION_PREFIX, + newestCanonicalMessageId ?? '', + ); + return parseRedisPrReviewActionRetirement(rawClaims); + }; + + if (pending.provider === 'slack' && pending.slackTeamId && pending.threadId) { + let redisSuperseded: PendingPrReviewAction[] = []; + let fenced: { + result: Awaited>; + superseded: Parameters[0]; + }; + try { + fenced = await withCanonicalPrReviewSlackThreadActionFence( + { + slackTeamId: pending.slackTeamId, + channelId: pending.channelId, + threadId: pending.threadId, + }, + async (newestCanonicalMessageId) => { + const result = await attachRedis(newestCanonicalMessageId); + redisSuperseded = result.superseded; + return { + newestMessageId: + result.newestMessageId ?? newestCanonicalMessageId ?? messageId, + retireCanonical: result.attached, + result, + }; + }, + ); + } catch (error) { + await retirePrReviewActionMessagesBestEffort(redisSuperseded); + throw error; + } + return { + attached: fenced.result.attached, + superseded: [ + ...toRetirableCanonicalMessages(fenced.superseded), + ...fenced.result.superseded, + ], + }; + } + + const result = await attachRedis(null); + return { attached: result.attached, superseded: result.superseded }; +} + +function parseRedisPrReviewActionRetirement(rawResult: unknown): { + attached: boolean; + newestMessageId: string | null; + superseded: PendingPrReviewAction[]; +} { + const values = Array.isArray(rawResult) ? rawResult : []; + const superseded: PendingPrReviewAction[] = []; + for (const raw of values.slice(2)) { + if (typeof raw !== 'string') continue; + try { + superseded.push(JSON.parse(raw) as PendingPrReviewAction); + } catch { + // Malformed record; skip. + } + } + return { + attached: values[0] === 1, + newestMessageId: + typeof values[1] === 'string' && values[1] ? values[1] : null, + superseded, + }; +} + +function toRetirableCanonicalMessages( + actions: Array<{ + provider: 'slack' | 'teams' | 'telegram' | 'discord' | null; + slackTeamId: string | null; + channelId: string | null; + threadId: string | null; + messageId: string | null; + }>, +): RetirablePrReviewActionMessage[] { + return actions.flatMap((action) => + action.provider && action.provider !== 'teams' && action.channelId + ? [ + { + provider: action.provider, + ...(action.provider === 'slack' && action.slackTeamId + ? { slackTeamId: action.slackTeamId } + : {}), + channelId: action.channelId, + threadId: action.threadId, + messageId: action.messageId, + } satisfies RetirablePrReviewActionMessage, + ] + : [], ); +} - const values = Array.isArray(rawClaims) ? rawClaims : []; - const attached = values[0] === 1; +async function arbitrateLegacySlackPrReviewActions( + context: PendingPrReviewAction & { + provider: 'slack'; + slackTeamId: string; + threadId: string; + }, + newestCanonicalMessageId: string, +): Promise<{ + newestMessageId: string; + superseded: PendingPrReviewAction[]; +}> { + const redis = getRedis(); + const rawResult = await redis.eval( + ARBITRATE_LEGACY_SLACK_PR_REVIEW_ACTIONS_LUA, + 1, + getPrReviewActionThreadKey(context), + PR_REVIEW_ACTION_PREFIX, + JSON.stringify(context), + newestCanonicalMessageId, + ); + const values = Array.isArray(rawResult) ? rawResult : []; const superseded: PendingPrReviewAction[] = []; for (const raw of values.slice(1)) { if (typeof raw !== 'string') continue; @@ -362,7 +576,13 @@ export async function attachPendingPrReviewActionMessageWithRetirement( // Malformed record; skip. } } - return { attached, superseded }; + return { + newestMessageId: + typeof values[0] === 'string' && values[0] + ? values[0] + : newestCanonicalMessageId, + superseded, + }; } async function retireLegacyPrReviewActionsForContext(