feat(cdp): accept an idempotency key on the transactional email route

nodejs/src/cdp/cdp-api.serial.test.ts

@@ -16,6 +16,7 @@ import { createRedisV2PoolFromConfig } from '~/common/redis/redis-v2'
 import { closeHub, createHub } from '~/common/utils/db/hub'
 import { parseJSON } from '~/common/utils/json-parse'
 import { UUIDT } from '~/common/utils/utils'
+import { waitForExpect } from '~/tests/helpers/expectations'
 
 import { createCdpConsumerDeps } from '../../tests/helpers/cdp'
 import { forSnapshot } from '../../tests/helpers/snapshots'
@@ -33,6 +34,7 @@ import { insertHogFlow as _insertHogFlow } from './_tests/fixtures-hogflows'
 import { CdpApi } from './cdp-api'
 import { CdpConsumerBaseDeps } from './consumers/cdp-base.consumer'
 import { posthogFilterOutPlugin } from './legacy-plugins/_transformations/posthog-filter-out-plugin/template'
+import { IdempotencyKeys, createIdempotencyValkeyPool } from './services/hogflows/idempotency-keys'
 import { transactionalEmailRequestsTotal } from './services/hogflows/transactional-email'
 import { BASE_REDIS_KEY, HogWatcherState } from './services/monitoring/hog-watcher.service'
 import { compileHog } from './templates/compiler'
@@ -111,10 +113,14 @@ describe('CDP API', () => {
         team = await getFirstTeam(hub.postgres)
 
         cdpDeps = createCdpConsumerDeps(hub)
-        api = new CdpApi(hub, cdpDeps, {
-            hogQueue: createMockJobQueue(),
-            hogflowQueue: createMockJobQueue(),
-        })
+        api = new CdpApi(
+            hub,
+            { ...cdpDeps, idempotencyValkey: createIdempotencyValkeyPool(hub) },
+            {
+                hogQueue: createMockJobQueue(),
+                hogflowQueue: createMockJobQueue(),
+            }
+        )
         app = setupExpressApp()
         app.use('/', api.router())
         server = app.listen(0, () => {})
@@ -2128,11 +2134,11 @@ describe('CDP API', () => {
         const outcomeCount = async (outcome: string): Promise<number> =>
             (await transactionalEmailRequestsTotal.get()).values.find((v) => v.labels.outcome === outcome)?.value ?? 0
 
-        const send = (flowId: string, headers: Record<string, string>) =>
+        const send = (flowId: string, headers: Record<string, string>, payload: Record<string, unknown> = body) =>
             supertest(app)
                 .post(`/api/projects/${team.id}/hog_flows/${flowId}/transactional_email`)
                 .set(headers)
-                .send(body)
+                .send(payload)
 
         beforeEach(async () => {
             mockQueueInvocations = jest.fn().mockResolvedValue(undefined)
@@ -2168,6 +2174,84 @@ describe('CDP API', () => {
             )
         })
 
+        describe('with an idempotency key', () => {
+            let keyed: Record<string, unknown>
+            const sendKeyed = (payload: Record<string, unknown> = keyed) =>
+                send(hogFlow.id, { Authorization: `Bearer ${mintToken(team.id, hogFlow.id)}` }, payload)
+
+            beforeEach(() => {
+                // A fresh key per test, so no test reads a key another one left in Valkey.
+                keyed = { ...body, idempotency_key: `key-${new UUIDT().toString()}` }
+            })
+
+            it('returns the first send for a retry and queues nothing new', async () => {
+                const first = await sendKeyed()
+                const retry = await sendKeyed()
+
+                expect(first.status).toEqual(202)
+                expect(first.body).toMatchObject({ replayed: false })
+                expect(retry.status).toEqual(202)
+                expect(retry.body).toEqual({
+                    status: 'queued',
+                    invocation_id: first.body.invocation_id,
+                    replayed: true,
+                })
+                expect(mockQueueInvocations).toHaveBeenCalledTimes(1)
+            })
+
+            it('rejects the key for a different request', async () => {
+                await sendKeyed()
+
+                const res = await sendKeyed({ ...keyed, to: 'someone-else@example.com' })
+
+                expect(res.status).toEqual(422)
+                expect(res.body).toEqual({ error: 'idempotency_key_reused' })
+                expect(mockQueueInvocations).toHaveBeenCalledTimes(1)
+            })
+
+            it('tells a retry to wait while the first request is still queueing', async () => {
+                let finishQueueing: () => void = () => {}
+                mockQueueInvocations.mockImplementationOnce(
+                    () => new Promise<void>((resolve) => (finishQueueing = resolve))
+                )
+                const first = sendKeyed().then((res) => res)
+                await waitForExpect(() => expect(mockQueueInvocations).toHaveBeenCalledTimes(1))
+
+                const retry = await sendKeyed()
+                finishQueueing()
+
+                expect(retry.status).toEqual(409)
+                expect((await first).status).toEqual(202)
+                expect(mockQueueInvocations).toHaveBeenCalledTimes(1)
+            })
+
+            it('lets a retry send when the first request failed to queue', async () => {
+                mockQueueInvocations.mockRejectedValueOnce(new Error('queue down'))
+
+                const first = await sendKeyed()
+                const retry = await sendKeyed()
+
+                expect(first.status).toEqual(500)
+                expect(retry.status).toEqual(202)
+                expect(retry.body).toMatchObject({ replayed: false })
+                expect(mockQueueInvocations).toHaveBeenCalledTimes(2)
+            })
+
+            it('refuses a keyed send when no idempotency store is configured', async () => {
+                const keys = api['transactionalEmailIdempotencyKeys']
+                api['transactionalEmailIdempotencyKeys'] = new IdempotencyKeys(null, 'transactional-email')
+                try {
+                    const res = await sendKeyed()
+
+                    expect(res.status).toEqual(503)
+                    expect(res.body).toEqual({ error: 'idempotency_unavailable' })
+                    expect(mockQueueInvocations).not.toHaveBeenCalled()
+                } finally {
+                    api['transactionalEmailIdempotencyKeys'] = keys
+                }
+            })
+        })
+
         it('refuses a send when the team is over its email quota', async () => {
             jest.spyOn(api['deps'].quotaLimiting, 'isTeamQuotaLimited').mockResolvedValue(true)
             const limitedBefore = await outcomeCount('quota_limited')

nodejs/src/cdp/cdp-api.ts

@@ -50,6 +50,7 @@ import { HogFlowExecutorService, createHogFlowInvocation } from './services/hogf
 import { HogFlowManagerService } from './services/hogflows/hogflow-manager.service'
 import { checkHogFlowQuotaLimits } from './services/hogflows/hogflow-quota-limiting'
 import { matchesWaitUntilCondition } from './services/hogflows/hogflow-utils'
+import { IdempotencyKeys } from './services/hogflows/idempotency-keys'
 import { WorkflowStepResumeSchema } from './services/hogflows/step-resume.service'
 import {
     TransactionalEmailOutcome,
@@ -192,6 +193,7 @@ export class CdpApi {
     private cancelBatchJwt: ScopedServiceJwt
     private stepResumeJwt: ScopedServiceJwt
     private transactionalEmailJwt: ScopedServiceJwt
+    private transactionalEmailIdempotencyKeys: IdempotencyKeys
     private hogFunctionMonitoringService: HogFunctionMonitoringService
     private resumeBatchJwt: ScopedServiceJwt
 
@@ -266,6 +268,10 @@ export class CdpApi {
             PosthogJwtAudience.WORKFLOWS_SEND_TRANSACTIONAL_EMAIL,
             config.WORKFLOWS_TRANSACTIONAL_EMAIL_JWT_SECRET || ''
         )
+        this.transactionalEmailIdempotencyKeys = new IdempotencyKeys(
+            deps.idempotencyValkey ?? null,
+            'transactional-email'
+        )
         // The same workers that wake parked steps wake parked batch resolvers, so they share the
         // key. The audience keeps a step-resume token from waking a batch, and the reverse.
         this.resumeBatchJwt = new ScopedServiceJwt(
@@ -1151,7 +1157,7 @@ export class CdpApi {
             if (!parsed.success) {
                 return reply('invalid_request', 400, { error: parsed.error.message })
             }
-            const { to, distinct_id: distinctId, variables } = parsed.data
+            const { to, distinct_id: distinctId, variables, idempotency_key: idempotencyKey } = parsed.data
 
             const team = await this.deps.teamManager.getTeam(parseInt(req.params.team_id)).catch(() => null)
             if (!team) {
@@ -1164,10 +1170,6 @@ export class CdpApi {
             if (hogFlow.trigger?.type !== 'api' || hogFlow.status !== 'active') {
                 return reply('not_sendable', 400, { error: 'Workflow is not an active transactional email' })
             }
-            const { isLimited } = await checkHogFlowQuotaLimits(hogFlow, team.id, this.deps.quotaLimiting)
-            if (isLimited) {
-                return reply('quota_limited', 429, { error: 'quota_limited' })
-            }
 
             const event: HogFunctionInvocationGlobals['event'] = {
                 uuid: new UUIDT().toString(),
@@ -1196,13 +1198,52 @@ export class CdpApi {
             invocation.queue = 'email'
             invocation.queuePriority = EMAIL_QUEUE_PRIORITY.fast
 
-            this.invocationResultsService.invocationResultsRowsService.queueLifecycleRow(invocation, 'running')
+            const idempotencyKeys = this.transactionalEmailIdempotencyKeys
+            const fingerprint = IdempotencyKeys.fingerprint([hogFlow.id, to, distinctId, variables])
+            if (idempotencyKey) {
+                // Checked before the quota, so a retry of a send that went out gets its answer even
+                // when the team has hit its quota since.
+                const reservation = await idempotencyKeys.reserve(team.id, idempotencyKey, fingerprint, invocation.id)
+                switch (reservation.status) {
+                    case 'replayed':
+                        return reply('replayed', 202, {
+                            status: 'queued',
+                            invocation_id: reservation.invocationId,
+                            replayed: true,
+                        })
+                    case 'in_flight':
+                        return reply('idempotency_key_in_flight', 409, { error: 'idempotency_key_in_flight' })
+                    case 'conflict':
+                        return reply('idempotency_key_reused', 422, { error: 'idempotency_key_reused' })
+                    case 'unavailable':
+                        return reply('idempotency_unavailable', 503, { error: 'idempotency_unavailable' })
+                }
+            }
             try {
-                await this.hogflowQueue.queueInvocations([invocation])
+                const { isLimited } = await checkHogFlowQuotaLimits(hogFlow, team.id, this.deps.quotaLimiting)
+                if (isLimited) {
+                    if (idempotencyKey) {
+                        await idempotencyKeys.release(team.id, idempotencyKey, invocation.id)
+                    }
+                    return reply('quota_limited', 429, { error: 'quota_limited' })
+                }
+                this.invocationResultsService.invocationResultsRowsService.queueLifecycleRow(invocation, 'running')
+                try {
+                    await this.hogflowQueue.queueInvocations([invocation])
+                } catch (error) {
+                    this.invocationResultsService.invocationResultsRowsService.dropQueuedRowsFor([invocation.id])
+                    throw error
+                }
             } catch (error) {
-                this.invocationResultsService.invocationResultsRowsService.dropQueuedRowsFor([invocation.id])
+                // Nothing was queued, so the caller's retry with the same key must be able to send.
+                if (idempotencyKey) {
+                    await idempotencyKeys.release(team.id, idempotencyKey, invocation.id)
+                }
                 throw error
             }
+            if (idempotencyKey) {
+                await idempotencyKeys.markQueued(team.id, idempotencyKey, fingerprint, invocation.id)
+            }
             await this.invocationResultsService.invocationResultsRowsService.flush()
             // The run skips the trigger pipeline that counts runs for event-triggered workflows.
             this.hogFunctionMonitoringService.queueAppMetric(
@@ -1218,7 +1259,7 @@ export class CdpApi {
             )
             await this.hogFunctionMonitoringService.flush()
 
-            return reply('queued', 202, { status: 'queued', invocation_id: invocation.id })
+            return reply('queued', 202, { status: 'queued', invocation_id: invocation.id, replayed: false })
         } catch (e) {
             logger.error('Error queueing transactional email', {
                 error: e instanceof Error ? e.message : String(e),

nodejs/src/cdp/services/hogflows/transactional-email.ts

@@ -8,10 +8,16 @@ export const TransactionalEmailRequestSchema = z.object({
     distinct_id: z.string().min(1),
     // The caller's `message_data`, available to the template as `variables`.
     variables: z.record(z.string(), z.any()).default({}),
+    // The caller's `Idempotency-Key`. A repeat with the same key and request returns the first send.
+    idempotency_key: z.string().min(1).max(255).optional(),
 })
 
 export type TransactionalEmailOutcome =
     | 'queued'
+    | 'replayed'
+    | 'idempotency_key_in_flight'
+    | 'idempotency_key_reused'
+    | 'idempotency_unavailable'
     | 'not_configured'
     | 'unauthorized'
     | 'invalid_request'
@@ -22,6 +28,6 @@ export type TransactionalEmailOutcome =
 
 export const transactionalEmailRequestsTotal = new Counter({
     name: 'cdp_transactional_email_requests_total',
-    help: 'Requests from Django to run a transactional email, by outcome. Every outcome except `queued` means no email was queued.',
+    help: 'Requests from Django to run a transactional email, by outcome. Every outcome except `queued` means this request queued no email.',
     labelNames: ['outcome'],
 })