diff --git a/.github/workflows/preview_sweep.yml b/.github/workflows/preview_sweep.yml index 41fdda7..87bc613 100644 --- a/.github/workflows/preview_sweep.yml +++ b/.github/workflows/preview_sweep.yml @@ -54,7 +54,7 @@ jobs: QUEUE_PR_NUMBERS=$(curl -fsS -H "Authorization: Bearer ${CLOUDFLARE_API_TOKEN}" \ "https://api.cloudflare.com/client/v4/accounts/${CLOUDFLARE_ACCOUNT_ID}/queues" \ | jq -r '.result[].queue_name // empty' \ - | sed -n 's/^tipbot-pending-tip-pr\([0-9][0-9]*\)\(-dlq\)\?$/\1/p' || true) + | sed -n 's/^tipbot-\(pending-tip\|slack-reaction\)-pr\([0-9][0-9]*\)\(-dlq\)\?$/\2/p' || true) COMMENT_PR_NUMBERS=$(gh api "repos/${{ github.repository }}/issues/comments" --paginate \ --jq '.[] | select(.body | contains("tipbot-preview")) | .issue_url | split("/")[-1]' \ 2>/dev/null || true) diff --git a/.github/workflows/production.yml b/.github/workflows/production.yml index 3acb2ff..28efeca 100644 --- a/.github/workflows/production.yml +++ b/.github/workflows/production.yml @@ -32,6 +32,28 @@ jobs: - name: Setup pnpm uses: ./.github/actions/setup-pnpm + - name: Setup production Queues + run: | + QUEUES=$(node -e ' + const text = require("node:fs").readFileSync("wrangler.jsonc", "utf8") + .replace(/(?<=^[^"]*(?:"[^"]*"[^"]*)*)\/\/.*$/gm, "") + .replace(/,(\s*[\]}])/g, "$1") + const queues = JSON.parse(text).env.production.queues + const names = new Set([ + ...queues.producers.map((producer) => producer.queue), + ...queues.consumers.flatMap((consumer) => + consumer.dead_letter_queue ? [consumer.queue, consumer.dead_letter_queue] : [consumer.queue], + ), + ]) + for (const name of names) console.log(name) + ') + for QUEUE in $QUEUES; do + pnpm exec wrangler queues create "$QUEUE" 2>/dev/null || true + done + env: + CLOUDFLARE_ACCOUNT_ID: ${{ secrets.CLOUDFLARE_ACCOUNT_ID }} + CLOUDFLARE_API_TOKEN: ${{ secrets.CLOUDFLARE_API_TOKEN }} + - name: Run migrations run: pnpm exec wrangler d1 migrations apply tipbot --env production --remote env: diff --git a/src/api.ts b/src/api.ts index ff9eb06..d8c5c17 100644 --- a/src/api.ts +++ b/src/api.ts @@ -1071,6 +1071,45 @@ export const api = new Hono<{ ) return new Response('Invalid signature', { status: 401 }) + const reaction = (() => { + let payload: unknown + try { + payload = JSON.parse(body) + } catch { + return null + } + const parsed = z + .object({ + authorizations: Slack.reactionEventSchema.shape.authorizations, + event: Slack.reactionEventSchema.omit({ + authorizations: true, + event_id: true, + team_id: true, + }), + event_id: z.string().min(1), + team_id: z.string().min(1), + type: z.literal('event_callback'), + }) + .safeParse(payload) + if (!parsed.success) return null + return Slack.reactionEventSchema.parse({ + ...parsed.data.event, + authorizations: parsed.data.authorizations, + event_id: parsed.data.event_id, + team_id: parsed.data.team_id, + }) + })() + if (reaction) { + if (reaction.type === 'reaction_removed') return new Response('', { status: 200 }) + try { + await c.env.SLACK_REACTION_QUEUE.send(reaction) + return new Response('', { status: 200 }) + } catch (error) { + console.error('Failed to enqueue signed Slack reaction event:', error) + return new Response('Queue unavailable', { status: 503 }) + } + } + const params = request.headers .get('content-type') ?.includes('application/x-www-form-urlencoded') @@ -1124,30 +1163,7 @@ export const api = new Hono<{ actions ?? '', ].join(':') })() - if (interaction) return interaction - if (params) return null - - // Dedupe reaction Events API retries by Slack event_id. Chat SDK already - // dedupes message events, so keep this scoped to reactions only. - let payload: unknown - try { - payload = JSON.parse(body) - } catch { - return null - } - const parsed = z - .object({ - event: z.looseObject({ type: z.string().min(1) }).optional(), - event_id: z.string().min(1).optional(), - type: z.string().min(1).optional(), - }) - .safeParse(payload) - if (!parsed.success) return null - if (parsed.data.type !== 'event_callback') return null - if (!parsed.data.event_id) return null - if (!['reaction_added', 'reaction_removed'].includes(parsed.data.event?.type ?? '')) - return null - return `slack:webhook:${parsed.data.event_id}` + return interaction })() if (duplicateKey) { await Chat.getChat().initialize() diff --git a/src/api.workers.test.ts b/src/api.workers.test.ts index d27eb74..796b8b9 100644 --- a/src/api.workers.test.ts +++ b/src/api.workers.test.ts @@ -65,7 +65,9 @@ beforeEach(async () => { function isExpectedApiWorkerLog(args: unknown[]) { const message = typeof args[0] === 'string' ? args[0] : '' return ( - message.startsWith('Twitter webhook ') || message.startsWith('Twitter OAuth callback failed:') + message.startsWith('Failed to enqueue signed Slack reaction event:') || + message.startsWith('Twitter webhook ') || + message.startsWith('Twitter OAuth callback failed:') ) } @@ -1324,6 +1326,81 @@ describe('/api/chat/slack', () => { expect(response.status).toBe(401) }) + test('durably enqueues signed Slack reaction additions before acknowledging', async () => { + const sendSpy = vi.spyOn(env.SLACK_REACTION_QUEUE, 'send').mockResolvedValue({ + metadata: { metrics: { backlogBytes: 0, backlogCount: 0 } }, + }) + const body = slackReactionBody('reaction_added') + + const response = await client.api.chat.slack.$post( + {}, + { + headers: { + ...(await createSlackHeaders(body, env.SLACK_SIGNING_SECRET)), + 'content-type': 'application/json', + }, + init: { body }, + }, + ) + + expect(response.status).toBe(200) + expect(sendSpy).toHaveBeenCalledWith({ + event_id: 'EvReactionQueue', + event_ts: '1700000000.000002', + item: { + channel: Constants.slack.channelId, + ts: '1700000000.000001', + type: 'message', + }, + item_user: Constants.slack.memberUserId, + reaction: 'money_with_wings', + team_id: Constants.slack.teamId, + type: 'reaction_added', + user: Constants.slack.adminUserId, + }) + expect(executionCtx.waitUntil).not.toHaveBeenCalled() + }) + + test('acknowledges Slack reaction removals without enqueueing', async () => { + const sendSpy = vi.spyOn(env.SLACK_REACTION_QUEUE, 'send') + const body = slackReactionBody('reaction_removed') + + const response = await client.api.chat.slack.$post( + {}, + { + headers: { + ...(await createSlackHeaders(body, env.SLACK_SIGNING_SECRET)), + 'content-type': 'application/json', + }, + init: { body }, + }, + ) + + expect(response.status).toBe(200) + expect(sendSpy).not.toHaveBeenCalled() + expect(executionCtx.waitUntil).not.toHaveBeenCalled() + }) + + test('asks Slack to retry when reaction enqueueing fails', async () => { + vi.spyOn(env.SLACK_REACTION_QUEUE, 'send').mockRejectedValue(new Error('Queue unavailable')) + const body = slackReactionBody('reaction_added') + + const response = await client.api.chat.slack.$post( + {}, + { + headers: { + ...(await createSlackHeaders(body, env.SLACK_SIGNING_SECRET)), + 'content-type': 'application/json', + }, + init: { body }, + }, + ) + + expect(response.status).toBe(503) + expect(await response.text()).toBe('Queue unavailable') + expect(executionCtx.waitUntil).not.toHaveBeenCalled() + }) + test('publishes workspace missing Home tab on app_home_opened', async () => { const providerId = `T${Nanoid.generate()}` await Chat.getChat().initialize() @@ -3077,6 +3154,26 @@ function slackFetchBodyParams(body: BodyInit | null | undefined) { return new URLSearchParams() } +function slackReactionBody(type: 'reaction_added' | 'reaction_removed') { + return JSON.stringify({ + event: { + event_ts: '1700000000.000002', + item: { + channel: Constants.slack.channelId, + ts: '1700000000.000001', + type: 'message', + }, + item_user: Constants.slack.memberUserId, + reaction: 'money_with_wings', + type, + user: Constants.slack.adminUserId, + }, + event_id: 'EvReactionQueue', + team_id: Constants.slack.teamId, + type: 'event_callback', + }) +} + async function slackFetchBodyJson( call: Parameters, ): Promise> { diff --git a/src/chat.ts b/src/chat.ts index 34a5209..850c54c 100644 --- a/src/chat.ts +++ b/src/chat.ts @@ -279,86 +279,6 @@ export function getChat() { }, }) }) - bot.onReaction(async (event) => { - if (event.adapter !== getSlack()) throw new Error('Provider not implemented yet.') - - const reaction = z.parse(Slack.reactionEventSchema, event.raw) - if (reaction.type !== 'reaction_added') return - if (reaction.item.type !== 'message') return - if (reaction.item.channel.startsWith('D')) return - - const context = await (async () => { - const db = DB.create(env.DB) - const existingReceiptWorkspace = isReceiptBoostReaction(reaction.reaction) - ? await db - .selectFrom('tip_receipt_message') - .innerJoin('workspace', 'workspace.id', 'tip_receipt_message.workspace_id') - .select([ - 'workspace.chain_id', - 'workspace.created_at', - 'workspace.default_amount', - 'workspace.default_token_address', - 'workspace.id', - 'workspace.installed_at', - 'workspace.name', - 'workspace.provider', - 'workspace.provider_id', - 'workspace.uninstalled_at', - 'workspace.updated_at', - ]) - .where('tip_receipt_message.channel_id', '=', reaction.item.channel) - .where('tip_receipt_message.message_ts', '=', reaction.item.ts) - .executeTakeFirst() - : null - const providerId = - existingReceiptWorkspace?.provider_id ?? - reaction.authorizations?.find((authorization) => authorization.team_id)?.team_id ?? - reaction.team_id - const workspace = await db - .selectFrom('workspace') - .select([ - 'workspace.chain_id', - 'workspace.created_at', - 'workspace.default_amount', - 'workspace.default_token_address', - 'workspace.id', - 'workspace.installed_at', - 'workspace.name', - 'workspace.provider', - 'workspace.provider_id', - 'workspace.uninstalled_at', - 'workspace.updated_at', - ]) - .where('workspace.provider', '=', 'slack') - .where('workspace.provider_id', '=', providerId) - .executeTakeFirst() - if (!workspace) return - const reactionTipConfigs = await db - .selectFrom('reaction_tip_config') - .select(['amount', 'emoji']) - .where('workspace_id', '=', workspace.id) - .execute() - const reactionTipConfig = ( - reactionTipConfigs.length ? reactionTipConfigs : Tip.defaultReactionTipConfigs - ).find((config) => config.emoji === reaction.reaction) - if (!reactionTipConfig && !isReceiptBoostReaction(reaction.reaction)) return - return { - db, - provider: { id: providerId, type: 'slack' }, - ...(reactionTipConfig - ? { - reactionTipConfig: { - amount: reactionTipConfig.amount, - emoji: reactionTipConfig.emoji, - }, - } - : {}), - workspace, - } satisfies ReactionHandlerContext - })() - if (!context) return - await handleSlackReactionTip(reaction, context) - }) bot.onSlashCommand(getSlackCommand(env.HOST), async (event) => { if (event.adapter !== getSlack()) throw new Error('Provider not implemented yet.') @@ -3094,6 +3014,51 @@ async function resolveSlackConnectRecipient( } } +export async function processSlackReaction(reaction: Slack.ReactionEvent) { + if (reaction.type !== 'reaction_added') return + if (reaction.item.type !== 'message') return + if (reaction.item.channel.startsWith('D')) return + + const db = DB.create(env.DB) + const existingReceiptWorkspace = isReceiptBoostReaction(reaction.reaction) + ? await db + .selectFrom('tip_receipt_message') + .innerJoin('workspace', 'workspace.id', 'tip_receipt_message.workspace_id') + .selectAll('workspace') + .where('tip_receipt_message.channel_id', '=', reaction.item.channel) + .where('tip_receipt_message.message_ts', '=', reaction.item.ts) + .executeTakeFirst() + : null + const providerId = + existingReceiptWorkspace?.provider_id ?? + reaction.authorizations?.find((authorization) => authorization.team_id)?.team_id ?? + reaction.team_id + const workspace = await db + .selectFrom('workspace') + .selectAll() + .where('provider', '=', 'slack') + .where('provider_id', '=', providerId) + .executeTakeFirst() + if (!workspace) return + const reactionTipConfigs = await db + .selectFrom('reaction_tip_config') + .select(['amount', 'emoji']) + .where('workspace_id', '=', workspace.id) + .execute() + const reactionTipConfig = ( + reactionTipConfigs.length ? reactionTipConfigs : Tip.defaultReactionTipConfigs + ).find((config) => config.emoji === reaction.reaction) + if (!reactionTipConfig && !isReceiptBoostReaction(reaction.reaction)) return + await handleSlackReactionTip(reaction, { + db, + provider: { id: providerId, type: 'slack' }, + ...(reactionTipConfig ? { reactionTipConfig } : {}), + workspace, + }) +} + +export class RetrySlackReactionError extends Error {} + async function handleSlackReactionTip(event: Slack.ReactionEvent, context: ReactionHandlerContext) { const { db, provider, reactionTipConfig } = context let workspace = context.workspace @@ -3492,14 +3457,7 @@ async function handleSlackReactionTip(event: Slack.ReactionEvent, context: React source: 'reaction', tokenAddress: rows[0].token_address, workspaceProviderId: rows[0].receipt_provider_id, - }).catch( - (error) => - ({ - code: 'failed', - message: error instanceof Error ? error.message : 'Boost failed.', - ok: false, - }) satisfies Tip.TipBatchResult, - ) + }) if (result.ok) { if (result.status === 'sent') { @@ -3554,6 +3512,7 @@ async function handleSlackReactionTip(event: Slack.ReactionEvent, context: React return } + if (result.code === 'pending') throw new RetrySlackReactionError('Boost still sending.') await postSlackEphemeral( provider.id, event.item.channel, @@ -3561,7 +3520,6 @@ async function handleSlackReactionTip(event: Slack.ReactionEvent, context: React (() => { if (result.code === 'insufficient_funds') return 'Boost not sent. Your wallet has insufficient funds.' - if (result.code === 'pending') return 'Boost still sending.' if (result.code === 'recipient_unconnected') return 'Boost not sent. A recipient needs to connect Tipbot before receiving payments.' if (result.code === 'self_tip') return 'Boost not sent. Cannot send a payment to yourself.' @@ -3634,7 +3592,7 @@ async function handleSlackReactionTip(event: Slack.ReactionEvent, context: React const recipient = resolvedRecipient.value ? await getConnectedSlackRecipient(db, provider.type, workspace, resolvedRecipient.value) : null - const idempotencyKey = [ + let idempotencyKey = [ reactionTipIdempotencyPrefix, workspace.id, event.item.channel, @@ -3661,14 +3619,7 @@ async function handleSlackReactionTip(event: Slack.ReactionEvent, context: React settingsProviderId: provider.id, source: 'reaction', workspaceProviderId: workspace.provider_id, - }).catch( - (error) => - ({ - code: 'failed', - message: error instanceof Error ? error.message : 'Reaction tip failed.', - ok: false, - }) satisfies Tip.TipResult, - ) + }) if (result.ok && result.status === 'queued') { const messageTs = await postSlackQueuedTipMessage( { db, provider, text: '', threadTs: message.thread_ts ?? event.item.ts }, @@ -3695,42 +3646,52 @@ async function handleSlackReactionTip(event: Slack.ReactionEvent, context: React const existing = await db .selectFrom('reaction_tip') - .select('id') + .select(['id', 'idempotency_key', 'tip_id']) .where('workspace_id', '=', workspace.id) .where('channel_id', '=', event.item.channel) .where('message_ts', '=', event.item.ts) .where('reaction', '=', event.reaction) .where('sender_member_id', '=', sender.memberId) .executeTakeFirst() - if (existing) return + if (existing?.tip_id) { + await updateReactionTipAggregate(provider.id, { + channelId: event.item.channel, + threadTs: message.thread_ts ?? event.item.ts, + workspaceId: workspace.id, + }) + return + } + if (existing) idempotencyKey = existing.idempotency_key - const inserted = await (async () => { - const now = new Date().toISOString() - try { - await db - .insertInto('reaction_tip') - .values({ - channel_id: event.item.channel, - created_at: now, - id: Nanoid.generate(), - idempotency_key: idempotencyKey, - message_ts: event.item.ts, - reaction: event.reaction, - recipient_member_id: recipient.memberId, - sender_member_id: sender.memberId, - thread_ts: message.thread_ts ?? event.item.ts, - tip_id: null, - updated_at: now, - workspace_id: workspace.id, - }) - .execute() - return true - } catch (error) { - if (isUniqueConstraintError(error)) return false - throw error - } - })() - if (!inserted) return + const inserted = existing + ? true + : await (async () => { + const now = new Date().toISOString() + try { + await db + .insertInto('reaction_tip') + .values({ + channel_id: event.item.channel, + created_at: now, + id: Nanoid.generate(), + idempotency_key: idempotencyKey, + message_ts: event.item.ts, + reaction: event.reaction, + recipient_member_id: recipient.memberId, + sender_member_id: sender.memberId, + thread_ts: message.thread_ts ?? event.item.ts, + tip_id: null, + updated_at: now, + workspace_id: workspace.id, + }) + .execute() + return true + } catch (error) { + if (isUniqueConstraintError(error)) return false + throw error + } + })() + if (!inserted) throw new RetrySlackReactionError('Reaction marker insert raced.') const result = await Tip.handleTipBatchRequest(env, { amount: reactionTipConfig.amount, @@ -3744,14 +3705,7 @@ async function handleSlackReactionTip(event: Slack.ReactionEvent, context: React settingsProviderId: provider.id, source: 'reaction', workspaceProviderId: workspace.provider_id, - }).catch( - (error) => - ({ - code: 'failed', - message: error instanceof Error ? error.message : 'Reaction tip failed.', - ok: false, - }) satisfies Tip.TipResult, - ) + }) if (result.ok) { const tip = await db @@ -3759,7 +3713,7 @@ async function handleSlackReactionTip(event: Slack.ReactionEvent, context: React .select('id') .where('idempotency_key', '=', idempotencyKey) .executeTakeFirst() - if (!tip) return + if (!tip) throw new RetrySlackReactionError('Reaction payment has no tip yet.') await db .updateTable('reaction_tip') @@ -3770,8 +3724,6 @@ async function handleSlackReactionTip(event: Slack.ReactionEvent, context: React channelId: event.item.channel, threadTs: message.thread_ts ?? event.item.ts, workspaceId: workspace.id, - }).catch((error) => { - console.error('Failed to update Slack reaction tip aggregate:', error) }) return } @@ -3811,6 +3763,7 @@ async function handleSlackReactionTip(event: Slack.ReactionEvent, context: React return } + if (result.code === 'pending') throw new RetrySlackReactionError('Payment still sending.') await db.deleteFrom('reaction_tip').where('idempotency_key', '=', idempotencyKey).execute() await postSlackEphemeral( provider.id, @@ -3819,7 +3772,6 @@ async function handleSlackReactionTip(event: Slack.ReactionEvent, context: React (() => { if (result.code === 'insufficient_funds') return 'Payment not sent. Your wallet has insufficient funds. Add funds and try again.' - if (result.code === 'pending') return 'Payment still sending.' return 'Payment failed.' })(), ) diff --git a/src/chat.workers.test.ts b/src/chat.workers.test.ts index 2b98f85..e766423 100644 --- a/src/chat.workers.test.ts +++ b/src/chat.workers.test.ts @@ -6,9 +6,11 @@ import * as Chat from '#/chat.ts' import { closeExpired } from '#/crons/close.ts' import entryServer from '#/entry-server.ts' import * as Nanoid from '#/lib/nanoid.ts' +import * as Slack from '#/lib/slack.ts' import * as Tempo from '#/lib/tempo.ts' import * as Tip from '#/lib/tip.ts' import { processPendingTipMessage } from '#/queues/pendingTip.ts' +import { processSlackReactionMessage } from '#/queues/slackReaction.ts' import { WebClient } from '@slack/web-api' import { createExecutionContext, @@ -5745,6 +5747,171 @@ test('reaction tipping ignores duplicate signed Slack event deliveries', async ( expect(tips).toHaveLength(1) }) +test('reaction tipping resumes an orphaned marker with its stored idempotency key', async () => { + const connected = await connectTipAccounts() + if (!connected.recipientMember) throw new Error('Expected connected recipient.') + const channelId = await createSlackTestChannel('rt') + const message = await memberSlack.chat.postMessage({ + channel: channelId, + text: 'resume orphaned reaction tip', + }) + if (!message.ts) throw new Error('Expected Slack message timestamp.') + const storedIdempotencyKey = `${Chat.reactionTipIdempotencyPrefix}stored-${Nanoid.generate()}` + await factory.reaction_tip.insert({ + channel_id: channelId, + idempotency_key: storedIdempotencyKey, + message_ts: message.ts, + reaction: 'money_with_wings', + recipient_member_id: connected.recipientMember.id, + sender_member_id: connected.senderMember.id, + thread_ts: message.ts, + tip_id: null, + workspace_id: connected.workspace.id, + }) + + const response = await postSlackReaction({ + channelId, + eventTs: `${message.ts}-new-reaction`, + messageTs: message.ts, + reaction: 'money_with_wings', + userId: Constants.slack.adminUserId, + }) + const reactionTip = await db + .selectFrom('reaction_tip') + .select(['idempotency_key', 'tip_id']) + .where('workspace_id', '=', connected.workspace.id) + .where('channel_id', '=', channelId) + .where('message_ts', '=', message.ts) + .executeTakeFirstOrThrow() + const tips = await db + .selectFrom('tip') + .select(['confirmed_at', 'idempotency_key']) + .where('workspace_id', '=', connected.workspace.id) + .where('idempotency_key', '=', storedIdempotencyKey) + .execute() + + expect(response.status).toBe(200) + expect(reactionTip).toEqual({ idempotency_key: storedIdempotencyKey, tip_id: expect.any(String) }) + expect(tips).toEqual([ + { confirmed_at: expect.any(String), idempotency_key: storedIdempotencyKey }, + ]) +}) + +test('reaction queue delays retry and preserves a pending marker', async () => { + const connected = await connectTipAccounts() + if (!connected.recipientMember) throw new Error('Expected connected recipient.') + const channelId = await createSlackTestChannel('rt') + const message = await memberSlack.chat.postMessage({ + channel: channelId, + text: 'retry pending reaction tip', + }) + if (!message.ts) throw new Error('Expected Slack message timestamp.') + const storedIdempotencyKey = `${Chat.reactionTipIdempotencyPrefix}pending-${Nanoid.generate()}` + const reactionTip = await factory.reaction_tip.insert({ + channel_id: channelId, + idempotency_key: storedIdempotencyKey, + message_ts: message.ts, + reaction: 'money_with_wings', + recipient_member_id: connected.recipientMember.id, + sender_member_id: connected.senderMember.id, + thread_ts: message.ts, + tip_id: null, + workspace_id: connected.workspace.id, + }) + await factory.tip_batch.insert({ + amount_each: 1000, + idempotency_key: storedIdempotencyKey, + provider: 'slack', + provider_channel_id: channelId, + provider_id: providerId, + recipient_count: 1, + sender_member_id: connected.senderMember.id, + source: 'reaction', + status: 'pending', + token_address: Tempo.addressLookup.pathUsd, + total_amount: 1000, + workspace_id: connected.workspace.id, + }) + const batch = createMessageBatch( + processSlackReactionMessage.queueName, + [ + { + attempts: 1, + body: { + event_ts: `${message.ts}-new-reaction`, + item: { channel: channelId, ts: message.ts, type: 'message' }, + item_user: Constants.slack.memberUserId, + reaction: 'money_with_wings', + team_id: providerId, + type: 'reaction_added', + user: Constants.slack.adminUserId, + }, + id: crypto.randomUUID(), + timestamp: new Date(), + }, + ], + ) + const ackSpy = vi.spyOn(batch.messages[0]!, 'ack') + const retrySpy = vi.spyOn(batch.messages[0]!, 'retry') + vi.spyOn(console, 'error').mockImplementation(() => {}) + + await entryServer.queue?.(batch, env) + + expect(ackSpy).not.toHaveBeenCalled() + expect(retrySpy).toHaveBeenCalledWith({ delaySeconds: 45 }) + await expect( + db + .selectFrom('reaction_tip') + .select(['id', 'tip_id']) + .where('id', '=', reactionTip.id) + .executeTakeFirstOrThrow(), + ).resolves.toEqual({ id: reactionTip.id, tip_id: null }) +}) + +test('reaction queue retries unexpected payment errors without deleting its marker', async () => { + const connected = await connectTipAccounts() + const channelId = await createSlackTestChannel('rt') + const message = await memberSlack.chat.postMessage({ + channel: channelId, + text: 'retry failed reaction processing', + }) + if (!message.ts) throw new Error('Expected Slack message timestamp.') + vi.spyOn(Tip, 'handleTipBatchRequest').mockRejectedValueOnce(new Error('D1 unavailable')) + const batch = createMessageBatch( + processSlackReactionMessage.queueName, + [ + { + attempts: 1, + body: { + event_ts: `${message.ts}-reaction`, + item: { channel: channelId, ts: message.ts, type: 'message' }, + item_user: Constants.slack.memberUserId, + reaction: 'money_with_wings', + team_id: providerId, + type: 'reaction_added', + user: Constants.slack.adminUserId, + }, + id: crypto.randomUUID(), + timestamp: new Date(), + }, + ], + ) + const ackSpy = vi.spyOn(batch.messages[0]!, 'ack') + const retrySpy = vi.spyOn(batch.messages[0]!, 'retry') + vi.spyOn(console, 'error').mockImplementation(() => {}) + + await entryServer.queue?.(batch, env) + const reactionTips = await db + .selectFrom('reaction_tip') + .select(['tip_id', 'workspace_id']) + .where('workspace_id', '=', connected.workspace.id) + .execute() + + expect(ackSpy).not.toHaveBeenCalled() + expect(retrySpy).toHaveBeenCalledWith({}) + expect(reactionTips).toEqual([{ tip_id: null, workspace_id: connected.workspace.id }]) +}) + test('reaction tipping reports unconnected sender', async () => { const fetchSpy = vi.spyOn(globalThis, 'fetch') await connectTipAccounts() @@ -6342,7 +6509,8 @@ test('reaction tipping reports approval required', async () => { .execute() expect(response.status).toBe(200) - expect(reactionTips).toHaveLength(0) + expect(reactionTips).toHaveLength(1) + expect(reactionTips[0]).toMatchObject({ tip_id: null }) expect(tips).toHaveLength(0) await expectSlackPostEphemeralCall(fetchSpy, 'Tipbot needs your approval to send this payment.') await expectSlackPostEphemeralCall(fetchSpy, '"action_id":"confirm_cancel"') @@ -9138,6 +9306,10 @@ async function postSlackReaction(options: { teamId?: string userId: string }) { + const send = vi.spyOn(env.SLACK_REACTION_QUEUE, 'send').mockImplementation(async (reaction) => { + await processSlackReactionMessage({ body: reaction } as Message) + return { metadata: { metrics: { backlogBytes: 0, backlogCount: 0 } } } + }) const body = JSON.stringify({ ...(options.authorizations ? { authorizations: options.authorizations } : {}), event: { @@ -9166,6 +9338,7 @@ async function postSlackReaction(options: { init: { body }, }, ) + send.mockRestore() await drainWaitUntil() return response } diff --git a/src/entry-server.ts b/src/entry-server.ts index 81db6d3..9ec7da0 100644 --- a/src/entry-server.ts +++ b/src/entry-server.ts @@ -1,9 +1,11 @@ import serverEntry from '@tanstack/react-start/server-entry' import { Kv as AccountsKv } from 'accounts/server' import { api } from '#/api.ts' +import * as Chat from '#/chat.ts' import { closeExpired } from '#/crons/close.ts' import { rpc } from '#/lib/rpc.ts' import { processPendingTipMessage } from '#/queues/pendingTip.ts' +import { processSlackReactionMessage } from '#/queues/slackReaction.ts' import { z } from 'zod' export default { @@ -23,17 +25,25 @@ export default { if (previewApex) return batch.queue.replace(`-${previewApex}`, '') return batch.queue })() - const queue = z.parse(z.enum([processPendingTipMessage.queueName]), queueName) - const handler = { [processPendingTipMessage.queueName]: processPendingTipMessage }[queue] + const queue = z.parse( + z.enum([processPendingTipMessage.queueName, processSlackReactionMessage.queueName]), + queueName, + ) + const handler = { + [processPendingTipMessage.queueName]: processPendingTipMessage, + [processSlackReactionMessage.queueName]: processSlackReactionMessage, + }[queue] for (const message of batch.messages) { try { await handler(message as never) message.ack() } catch (error) { - if (queue === processPendingTipMessage.queueName && message.attempts >= 3) - console.error('Pending tip queue message reached DLQ threshold:', message.body) + if (message.attempts >= 3) + console.error(`${queue} queue message reached DLQ threshold:`, message.id) console.error(`Queue message ${message.id} failed:`, error) - message.retry() + message.retry( + error instanceof Chat.RetrySlackReactionError ? { delaySeconds: 45 } : {}, // 45 seconds + ) } } }, @@ -46,7 +56,7 @@ export default { const task = crons[controller.cron as keyof typeof crons] if (task) ctx.waitUntil(task(env, ctx)) }, -} satisfies ExportedHandler +} satisfies ExportedHandler declare module '@tanstack/react-start' { interface Register { diff --git a/src/queues/slackReaction.ts b/src/queues/slackReaction.ts new file mode 100644 index 0000000..d5988ad --- /dev/null +++ b/src/queues/slackReaction.ts @@ -0,0 +1,15 @@ +import * as Chat from '#/chat.ts' +import * as Slack from '#/lib/slack.ts' + +export async function processSlackReactionMessage( + message: Message, +) { + await Chat.getChat().initialize() + await Chat.processSlackReaction(message.body) +} + +processSlackReactionMessage.queueName = 'tipbot-slack-reaction' as const + +export namespace processSlackReactionMessage { + export type Body = Slack.ReactionEvent +} diff --git a/src/worker-configuration.d.ts b/src/worker-configuration.d.ts index b5f34b2..a5e3f9c 100644 --- a/src/worker-configuration.d.ts +++ b/src/worker-configuration.d.ts @@ -1,5 +1,5 @@ /* eslint-disable */ -// Generated by Wrangler by running `wrangler types src/worker-configuration.d.ts` (hash: 8a894d1cdc6ecbdbdab717f77330ae9f) +// Generated by Wrangler by running `wrangler types src/worker-configuration.d.ts` (hash: 80957e85b1f7a70b8525c280ec39eff0) // Runtime types generated with workerd@1.20260511.1 2026-05-07 nodejs_compat declare namespace Cloudflare { interface GlobalProps { @@ -9,6 +9,7 @@ declare namespace Cloudflare { interface PreviewEnv { DB: D1Database; PENDING_TIP_QUEUE: Queue; + SLACK_REACTION_QUEUE: Queue; AI: Ai; CF_VERSION_METADATA: WorkerVersionMetadata; HOST: "__PREVIEW_APEX__.tip.bot"; @@ -28,6 +29,7 @@ declare namespace Cloudflare { interface ProductionEnv { DB: D1Database; PENDING_TIP_QUEUE: Queue; + SLACK_REACTION_QUEUE: Queue; AI: Ai; CF_VERSION_METADATA: WorkerVersionMetadata; HOST: "tip.bot"; @@ -54,6 +56,7 @@ declare namespace Cloudflare { interface Env { DB: D1Database; PENDING_TIP_QUEUE: Queue; + SLACK_REACTION_QUEUE: Queue; AI: Ai; CF_VERSION_METADATA: WorkerVersionMetadata; HOST: "__PREVIEW_APEX__.tip.bot" | "tip.bot" | "tipbot.localhost"; diff --git a/wrangler.jsonc b/wrangler.jsonc index 6cc32cf..8d1cb3c 100644 --- a/wrangler.jsonc +++ b/wrangler.jsonc @@ -33,7 +33,10 @@ ], }, "queues": { - "producers": [{ "binding": "PENDING_TIP_QUEUE", "queue": "tipbot-pending-tip" }], + "producers": [ + { "binding": "PENDING_TIP_QUEUE", "queue": "tipbot-pending-tip" }, + { "binding": "SLACK_REACTION_QUEUE", "queue": "tipbot-slack-reaction" }, + ], "consumers": [ { "queue": "tipbot-pending-tip", @@ -42,6 +45,14 @@ "max_retries": 3, "dead_letter_queue": "tipbot-pending-tip-dlq", }, + { + "queue": "tipbot-slack-reaction", + "max_batch_size": 1, + "max_batch_timeout": 5, + "max_retries": 3, + "retry_delay": 30, // 30 seconds + "dead_letter_queue": "tipbot-slack-reaction-dlq", + }, ], }, "triggers": { @@ -120,6 +131,7 @@ "queues": { "producers": [ { "binding": "PENDING_TIP_QUEUE", "queue": "tipbot-pending-tip-__PREVIEW_APEX__" }, + { "binding": "SLACK_REACTION_QUEUE", "queue": "tipbot-slack-reaction-__PREVIEW_APEX__" }, ], "consumers": [ { @@ -129,6 +141,14 @@ "max_retries": 3, "dead_letter_queue": "tipbot-pending-tip-__PREVIEW_APEX__-dlq", }, + { + "queue": "tipbot-slack-reaction-__PREVIEW_APEX__", + "max_batch_size": 1, + "max_batch_timeout": 5, + "max_retries": 3, + "retry_delay": 30, // 30 seconds + "dead_letter_queue": "tipbot-slack-reaction-__PREVIEW_APEX__-dlq", + }, ], }, "triggers": { @@ -208,7 +228,10 @@ "binding": "AI", }, "queues": { - "producers": [{ "binding": "PENDING_TIP_QUEUE", "queue": "tipbot-pending-tip" }], + "producers": [ + { "binding": "PENDING_TIP_QUEUE", "queue": "tipbot-pending-tip" }, + { "binding": "SLACK_REACTION_QUEUE", "queue": "tipbot-slack-reaction" }, + ], "consumers": [ { "queue": "tipbot-pending-tip", @@ -217,6 +240,14 @@ "max_retries": 3, "dead_letter_queue": "tipbot-pending-tip-dlq", }, + { + "queue": "tipbot-slack-reaction", + "max_batch_size": 1, + "max_batch_timeout": 5, + "max_retries": 3, + "retry_delay": 30, // 30 seconds + "dead_letter_queue": "tipbot-slack-reaction-dlq", + }, ], }, "triggers": {