diff --git a/apps/service/src/__tests__/workflow.test.ts b/apps/service/src/__tests__/workflow.test.ts index f03c63f39..a631fe992 100644 --- a/apps/service/src/__tests__/workflow.test.ts +++ b/apps/service/src/__tests__/workflow.test.ts @@ -131,7 +131,7 @@ describe('pipelineWorkflow (unit — stubbed activities)', () => { }) }) - it('processes stripe_event signals as optimistic updates', async () => { + it('processes source_input signals as optimistic updates', async () => { const syncCalls: { pipelineId: string; input?: SourceInput[] }[] = [] const worker = await Worker.create({ diff --git a/apps/service/src/api/app.test.ts b/apps/service/src/api/app.test.ts index b6fb0fdb2..e58635efa 100644 --- a/apps/service/src/api/app.test.ts +++ b/apps/service/src/api/app.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it, beforeAll, afterAll, vi } from 'vitest' import type { WorkflowClient } from '@temporalio/client' import { TestWorkflowEnvironment } from '@temporalio/testing' import { Worker } from '@temporalio/worker' +import { createHmac } from 'node:crypto' import { createServer } from 'node:http' import type { AddressInfo } from 'node:net' import path from 'node:path' @@ -776,3 +777,138 @@ describe('pipeline CRUD', () => { await a.request(`/pipelines/${created.id}`, { method: 'DELETE' }) }) }) + +// --------------------------------------------------------------------------- +// Webhook ingress +// --------------------------------------------------------------------------- + +describe('POST /webhooks/{pipeline_id}', () => { + const webhookSecret = 'whsec_test_secret' + + function signPayload( + payload: string, + secret = webhookSecret, + timestamp = Math.floor(Date.now() / 1000) + ) { + const signature = createHmac('sha256', secret).update(`${timestamp}.${payload}`).digest('hex') + return `t=${timestamp},v1=${signature}` + } + + function stripeEventPayload(overrides: Record = {}) { + return JSON.stringify({ + id: 'evt_test_1', + object: 'event', + api_version: '2025-03-31.basil', + created: Math.floor(Date.now() / 1000), + data: { object: { id: 'cus_test_1', object: 'customer' } }, + livemode: false, + pending_webhooks: 0, + request: { id: null, idempotency_key: null }, + type: 'customer.created', + ...overrides, + }) + } + + // Unlike mockTemporalClient() above, keeps a stable `signal` mock across + // getHandle() calls so assertions can target it directly. + function mockTemporalClientWithHandle() { + const signal = vi.fn(async () => undefined) + const client = { + start: vi.fn(async () => undefined), + getHandle: vi.fn(() => ({ + signal, + query: vi.fn(async () => ({})), + terminate: vi.fn(async () => undefined), + })), + list: vi.fn(async function* () {}), + } as unknown as WorkflowClient + return { client, signal } + } + + async function seedStripePipeline(pipelineStore: PipelineStore, id: string) { + await pipelineStore.set(id, { + id, + source: { + type: 'stripe', + stripe: { + api_key: 'sk_test_123', + api_version: '2025-03-31.basil', + webhook_secret: webhookSecret, + }, + }, + destination: { type: 'test', test: {} }, + desired_status: 'active', + status: 'ready', + } as Pipeline) + } + + it('signals the workflow with source_input on a validly signed event', async () => { + const pipelineStore = memoryPipelineStore() + await seedStripePipeline(pipelineStore, 'pipe_webhook_1') + const { client: temporalClient, signal } = mockTemporalClientWithHandle() + const webhookApp = createApp({ + temporal: { client: temporalClient, taskQueue: 'test-webhooks' }, + resolver, + pipelineStore, + }) + + const payload = stripeEventPayload() + const res = await webhookApp.request('/webhooks/pipe_webhook_1', { + method: 'POST', + headers: { 'stripe-signature': signPayload(payload) }, + body: payload, + }) + + expect(res.status).toBe(200) + // This is the regression this test guards against: the route used to signal + // 'stripe_event', which the workflow has no handler for, so events were + // silently dropped even though Stripe saw a 200. + expect(signal).toHaveBeenCalledWith('source_input', { + type: 'source_input', + source_input: expect.objectContaining({ id: 'evt_test_1', type: 'customer.created' }), + }) + }) + + it('rejects an incorrectly signed event without signaling the workflow', async () => { + const pipelineStore = memoryPipelineStore() + await seedStripePipeline(pipelineStore, 'pipe_webhook_2') + const { client: temporalClient, signal } = mockTemporalClientWithHandle() + const webhookApp = createApp({ + temporal: { client: temporalClient, taskQueue: 'test-webhooks' }, + resolver, + pipelineStore, + }) + + const payload = stripeEventPayload() + const res = await webhookApp.request('/webhooks/pipe_webhook_2', { + method: 'POST', + headers: { 'stripe-signature': signPayload(payload, 'whsec_wrong_secret') }, + body: payload, + }) + + expect(res.status).toBe(401) + expect(signal).not.toHaveBeenCalled() + }) + + it('returns a non-2xx response when the workflow signal fails, so Stripe retries', async () => { + const pipelineStore = memoryPipelineStore() + await seedStripePipeline(pipelineStore, 'pipe_webhook_3') + const { client: temporalClient, signal } = mockTemporalClientWithHandle() + signal.mockRejectedValueOnce(new Error('workflow not found')) + const webhookApp = createApp({ + temporal: { client: temporalClient, taskQueue: 'test-webhooks' }, + resolver, + pipelineStore, + }) + + const payload = stripeEventPayload() + const res = await webhookApp.request('/webhooks/pipe_webhook_3', { + method: 'POST', + headers: { 'stripe-signature': signPayload(payload) }, + body: payload, + }) + + expect(res.status).toBeGreaterThanOrEqual(500) + expect(signal).toHaveBeenCalledOnce() + }) +}) diff --git a/apps/service/src/api/app.ts b/apps/service/src/api/app.ts index 60acc17f8..1603a3655 100644 --- a/apps/service/src/api/app.ts +++ b/apps/service/src/api/app.ts @@ -10,6 +10,7 @@ import { emptySyncState, EofPayload as EofPayloadSchema, SyncState, + type SourceInputMessage, } from '@stripe/sync-protocol' const engineMsg = createEngineMessageFactory() @@ -972,8 +973,9 @@ export function createApp(options: AppOptions) { // Verify webhook signature const body = await c.req.text() const signature = c.req.header('stripe-signature') ?? '' + let event: ReturnType try { - const event = verifyWebhookSignature(body, signature, webhookSecret) + event = verifyWebhookSignature(body, signature, webhookSecret) log.info( { eventId: event.id, eventType: event.type, pipeline_id }, 'webhook event ingested' @@ -990,10 +992,15 @@ export function createApp(options: AppOptions) { return c.text('temporal is not configured', 503) } - temporal - .getHandle(pipeline_id) - .signal('stripe_event', { body, headers: Object.fromEntries(c.req.raw.headers.entries()) }) - .catch(() => {}) + // Must match the signal name/payload the workflow actually handles + // (sourceInputSignal in temporal/workflows/_shared.ts), or events are silently dropped. + const message: SourceInputMessage = { type: 'source_input', source_input: event } + try { + await temporal.getHandle(pipeline_id).signal('source_input', message) + } catch (err) { + log.error({ err, eventId: event.id, pipeline_id }, 'failed to signal pipeline workflow') + return c.text('failed to enqueue webhook event for processing', 502) + } return c.text('ok', 200) } )