feat(cdp): list waiting scheduled sends and count them

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

@@ -70,6 +70,7 @@ const fakeJobProducer = (overrides: Partial<CyclotronV2JobProducer> = {}): Cyclo
     resumeParkedBatchResolver: jest.fn(),
     cancelUnclaimed: jest.fn(),
     rescheduleUnclaimed: jest.fn(),
+    listUnclaimed: jest.fn(),
     disconnect: jest.fn().mockResolvedValue(undefined),
     ...overrides,
 })
@@ -2900,6 +2901,7 @@ describe('CDP API', () => {
         describe('cancel and reschedule of a scheduled send', () => {
             let cancelUnclaimed: jest.Mock
             let rescheduleUnclaimed: jest.Mock
+            let listUnclaimed: jest.Mock
 
             const mintManageToken = (
                 teamId: number,
@@ -2918,6 +2920,10 @@ describe('CDP API', () => {
                     .post(`/api/projects/${team.id}/transactional_emails/${messageId}/${action}`)
                     .set(headers)
                     .send(payload)
+            const list = (
+                query: Record<string, string> = {},
+                headers: Record<string, string> = { Authorization: `Bearer ${mintManageToken(team.id)}` }
+            ) => supertest(app).get(`/api/projects/${team.id}/transactional_emails/scheduled`).query(query).set(headers)
 
             const scheduleSend = async (): Promise<CyclotronJobInvocation> => {
                 const res = await send(
@@ -2932,7 +2938,8 @@ describe('CDP API', () => {
             beforeEach(() => {
                 cancelUnclaimed = jest.fn()
                 rescheduleUnclaimed = jest.fn()
-                api['batchResolverProducer'] = fakeJobProducer({ cancelUnclaimed, rescheduleUnclaimed })
+                listUnclaimed = jest.fn()
+                api['batchResolverProducer'] = fakeJobProducer({ cancelUnclaimed, rescheduleUnclaimed, listUnclaimed })
             })
 
             afterEach(() => {
@@ -3103,7 +3110,68 @@ describe('CDP API', () => {
                 expect(emailStatusRows()).toEqual([])
             })
 
-            describe.each(['cancel', 'reschedule'] as const)('on %s', (action) => {
+            it('lists waiting sends a page at a time, and its next cursor continues after the last one', async () => {
+                const jobs = [1, 2, 3].map((days) => ({
+                    id: new UUIDT().toString(),
+                    to: `user${days}@example.com`,
+                    scheduled: new Date(inDays(days)),
+                    created: new Date(Date.now() - 60_000),
+                }))
+                const asResult = (job: (typeof jobs)[number]): Record<string, string> => ({
+                    id: job.id,
+                    to: job.to,
+                    scheduled_at: job.scheduled.toISOString(),
+                    created_at: job.created.toISOString(),
+                })
+                listUnclaimed.mockResolvedValueOnce(jobs).mockResolvedValueOnce(jobs.slice(2)).mockResolvedValueOnce([])
+
+                const first = await list({ limit: '2' })
+                const second = await list({ limit: '2', cursor: first.body.next })
+                const byDefault = await list()
+
+                expect(first.status).toEqual(200)
+                expect(first.body).toEqual({ results: jobs.slice(0, 2).map(asResult), next: expect.any(String) })
+                expect(second.status).toEqual(200)
+                expect(second.body).toEqual({ results: [asResult(jobs[2])], next: null })
+                expect(byDefault.body).toEqual({ results: [], next: null })
+                expect(listUnclaimed.mock.calls).toEqual([
+                    [team.id, null, 3],
+                    [team.id, { scheduled: jobs[1].scheduled, id: jobs[1].id }, 3],
+                    [team.id, null, 101],
+                ])
+            })
+
+            it.each([
+                ['a limit of 0', { limit: '0' }],
+                ['a limit above 100', { limit: '101' }],
+                ['a limit that is not a number', { limit: 'ten' }],
+                ['a cursor that does not decode', { cursor: 'not-a-cursor' }],
+                [
+                    'a cursor without a time',
+                    { cursor: Buffer.from(JSON.stringify(['soon', new UUIDT().toString()])).toString('base64url') },
+                ],
+                [
+                    'a cursor with a time Postgres cannot store',
+                    {
+                        cursor: Buffer.from(
+                            JSON.stringify(['-010000-01-01T00:00:00.000Z', new UUIDT().toString()])
+                        ).toString('base64url'),
+                    },
+                ],
+            ])('refuses a list with %s', async (_desc, query) => {
+                const res = await list(query)
+
+                expect(res.status).toEqual(400)
+                expect(res.body).toEqual({ error: 'invalid_request' })
+                expect(listUnclaimed).not.toHaveBeenCalled()
+            })
+
+            describe.each(['cancel', 'reschedule', 'list'] as const)('on %s', (action) => {
+                const request = (headers?: Record<string, string>): supertest.Test =>
+                    action === 'list'
+                        ? list({}, headers)
+                        : manage(action, new UUIDT().toString(), { scheduled_at: inDays(2) }, headers)
+
                 it.each([
                     ['no token', () => ({})],
                     ["another team's token", () => ({ Authorization: `Bearer ${mintManageToken(team.id + 1)}` })],
@@ -3122,11 +3190,12 @@ describe('CDP API', () => {
                         }),
                     ],
                 ])('rejects a request with %s', async (_desc, headers) => {
-                    const res = await manage(action, new UUIDT().toString(), { scheduled_at: inDays(2) }, headers())
+                    const res = await request(headers())
 
                     expect(res.status).toEqual(401)
                     expect(cancelUnclaimed).not.toHaveBeenCalled()
                     expect(rescheduleUnclaimed).not.toHaveBeenCalled()
+                    expect(listUnclaimed).not.toHaveBeenCalled()
                 })
 
                 it('fails closed when the transactional email JWT key is not provisioned', async () => {
@@ -3136,11 +3205,12 @@ describe('CDP API', () => {
                         ''
                     )
                     try {
-                        const res = await manage(action, new UUIDT().toString(), { scheduled_at: inDays(2) })
+                        const res = await request()
 
                         expect(res.status).toEqual(503)
                         expect(cancelUnclaimed).not.toHaveBeenCalled()
                         expect(rescheduleUnclaimed).not.toHaveBeenCalled()
+                        expect(listUnclaimed).not.toHaveBeenCalled()
                     } finally {
                         api['manageTransactionalEmailJwt'] = savedJwt
                     }

nodejs/src/cdp/cdp-api.ts

@@ -55,6 +55,7 @@ import {
     AcceptedSendRequestSchema,
     CODE_SEND_TOKEN_CLAIMS,
     RescheduleRequestSchema,
+    ScheduledSendsListQuerySchema,
     TEMPLATE_SEND_TOKEN_CLAIMS,
     TransactionalEmailOutcome,
     TransactionalEmailRequest,
@@ -411,6 +412,10 @@ export class CdpApi {
             '/api/projects/:team_id/hog_flows/:id/transactional_email/send',
             asyncHandler(this.postTransactionalEmail(AcceptedSendRequestSchema, CODE_SEND_TOKEN_CLAIMS))
         )
+        router.get(
+            '/api/projects/:team_id/transactional_emails/scheduled',
+            asyncHandler(this.getScheduledTransactionalEmails)
+        )
         // `:message_id`, not `:id`: verifyScopedWorkflowJwt matches `:id` against a workflow claim.
         router.post(
             '/api/projects/:team_id/transactional_emails/:message_id/cancel',
@@ -1306,6 +1311,16 @@ export class CdpApi {
         })
     }
 
+    private getScheduledTransactionalEmails = async (req: ModifiedRequest, res: express.Response): Promise<any> => {
+        return await this.manageScheduledSend(req, res, 'list', async (sends, teamId) => {
+            const query = ScheduledSendsListQuerySchema.safeParse(req.query)
+            if (!query.success) {
+                return res.status(400).json({ error: 'invalid_request' })
+            }
+            return res.status(200).json(await sends.list(teamId, query.data.cursor ?? null, query.data.limit))
+        })
+    }
+
     private postTransactionalEmailReschedule = async (req: ModifiedRequest, res: express.Response): Promise<any> => {
         return await this.manageScheduledSend(req, res, 'reschedule', async (sends, teamId, messageId) => {
             const parsed = RescheduleRequestSchema.safeParse(req.body)

nodejs/src/cdp/services/cyclotron-v2/cyclotron-v2.test.ts

@@ -12,7 +12,7 @@ import {
 } from '../hogflows/batch-resolver.types'
 import { HogInvocationResultsService } from '../monitoring/hog-invocation-results.service'
 import { CyclotronV2Janitor, JANITOR_POISON_PILL_ERROR_KIND } from './janitor'
-import { CyclotronV2Manager } from './manager'
+import { CyclotronV2Manager, LIST_UNCLAIMED_EMAIL_SQL } from './manager'
 import {
     CYCLOTRON_COUNTER_MAX,
     CYCLOTRON_TRANSITION_CHURN_THRESHOLD,
@@ -1186,29 +1186,43 @@ describe('Cyclotron V2', () => {
             })
         })
 
-        describe('cancelUnclaimed and rescheduleUnclaimed', () => {
+        describe('unclaimed email jobs', () => {
             const IN_A_WEEK = (): Date => new Date(Date.now() + 7 * 24 * 3600 * 1000)
             const IN_TWO_DAYS = new Date(Date.now() + 2 * 24 * 3600 * 1000)
             const storedState = { state: { scheduledAt: 'original', html: 'a\u0000b' }, queueParameters: { x: 1 } }
 
-            async function seedScheduledEmail(teamId = 1): Promise<string> {
+            async function seedScheduledEmail(
+                teamId = 1,
+                scheduled = IN_A_WEEK(),
+                state: unknown = storedState
+            ): Promise<string> {
                 const [id] = await manager.bulkCreateJobs([
                     {
                         teamId,
                         queueName: 'email',
                         functionId: uuidv7(),
-                        scheduled: IN_A_WEEK(),
-                        state: Buffer.from(JSON.stringify(storedState)),
+                        scheduled,
+                        state: Buffer.from(JSON.stringify(state)),
                     },
                 ])
                 return id
             }
 
+            const sendTo = (to: string, html = ''): unknown => ({ state: { event: { properties: { to } }, html } })
+
             // What a worker's dequeue writes when it claims the row.
             const claimSql = `UPDATE cyclotron_jobs
                 SET status = 'running', lock_id = gen_random_uuid(), transition_count = transition_count + 1
                 WHERE id = $1`
 
+            // A worker that claimed the job and gave it back, so only transition_count tells it apart.
+            async function claimAndRelease(id: string): Promise<void> {
+                await assertPool.query(claimSql, [id])
+                await assertPool.query(`UPDATE cyclotron_jobs SET status = 'available', lock_id = NULL WHERE id = $1`, [
+                    id,
+                ])
+            }
+
             const operations = [
                 ['cancelUnclaimed', (teamId: number, id: string) => manager.cancelUnclaimed(teamId, id)],
                 [
@@ -1309,6 +1323,78 @@ describe('Cyclotron V2', () => {
                 expect(result).toEqual({ outcome })
                 expect(await queryJob(before.id)).toEqual(before)
             })
+
+            it("lists only the team's waiting sends in (scheduled, id) order, and a cursor walks them page by page", async () => {
+                const later = IN_A_WEEK()
+                const sooner = new Date(later.getTime() - 3600 * 1000)
+                const [a, b] = [await seedScheduledEmail(1, later, sendTo('a@example.com')), uuidv7()]
+                await seedScheduledEmail(2, sooner, sendTo('other-team@example.com'))
+                // Same time as `a`, so the id breaks the tie.
+                await manager.bulkCreateJobs([
+                    {
+                        id: b,
+                        teamId: 1,
+                        queueName: 'email',
+                        functionId: uuidv7(),
+                        scheduled: later,
+                        state: Buffer.from(JSON.stringify(sendTo('b@example.com'))),
+                    },
+                ])
+                // Caller content can hold a \u0000 or a lone surrogate. Postgres json and jsonb reject both anywhere in the
+                // document, so this row fails if the recipient is read from the state in SQL.
+                const c = await seedScheduledEmail(1, sooner, sendTo('c@example.com', 'a\u0000b\ud800'))
+                await claimAndRelease(await seedScheduledEmail(1, sooner))
+                await manager.cancelUnclaimed(1, await seedScheduledEmail(1, sooner))
+                await seedScheduledEmail(1, new Date(Date.now() - 1000))
+
+                const all = await manager.listUnclaimed(1, null, 10)
+                const pages: string[][] = []
+                let after: { scheduled: Date; id: string } | null = null
+                for (let i = 0; i < 4; i++) {
+                    const page = await manager.listUnclaimed(1, after, 1)
+                    pages.push(page.map((job) => job.id))
+                    after = page.length ? { scheduled: page[0].scheduled, id: page[0].id } : after
+                }
+
+                const [first, second] = [a, b].sort()
+                expect(all).toEqual([
+                    { id: c, to: 'c@example.com', scheduled: sooner, created: expect.any(Date) },
+                    {
+                        id: first,
+                        to: first === a ? 'a@example.com' : 'b@example.com',
+                        scheduled: later,
+                        created: expect.any(Date),
+                    },
+                    {
+                        id: second,
+                        to: second === a ? 'a@example.com' : 'b@example.com',
+                        scheduled: later,
+                        created: expect.any(Date),
+                    },
+                ])
+                expect(pages).toEqual([[c], [first], [second], []])
+            })
+
+            it('lists through the unclaimed email index', async () => {
+                const client = await assertPool.connect()
+                try {
+                    // The test table holds a few rows, so the planner prefers a seq scan unless it is off.
+                    await client.query('SET enable_seqscan = off')
+                    const plan = await client.query<{ 'QUERY PLAN': string }>(`EXPLAIN ${LIST_UNCLAIMED_EMAIL_SQL}`, [
+                        1,
+                        new Date(),
+                        uuidv7(),
+                        10,
+                    ])
+
+                    expect(plan.rows.map((row) => row['QUERY PLAN']).join('\n')).toContain(
+                        'idx_cyclotron_jobs_email_unclaimed'
+                    )
+                } finally {
+                    await client.query('RESET enable_seqscan')
+                    client.release()
+                }
+            })
         })
     })
 
@@ -2978,6 +3064,33 @@ describe('Cyclotron V2', () => {
             expect(result.depths.get('queue-b')).toBe(1)
         })
 
+        it('reports the scheduled emails of all teams that still wait, on start', async () => {
+            const inADay = new Date(Date.now() + 24 * 3600 * 1000)
+            await insertRawJob({ id: uuidv7(), queue_name: 'email', scheduled: inADay })
+            await insertRawJob({ id: uuidv7(), queue_name: 'email', team_id: 2, scheduled: inADay })
+            await insertRawJob({ id: uuidv7(), queue_name: 'email', scheduled: inADay, transition_count: 1 })
+            await insertRawJob({ id: uuidv7(), queue_name: 'email', scheduled: inADay, status: 'canceled' })
+            await insertRawJob({ id: uuidv7(), queue_name: 'email', scheduled: new Date(Date.now() - 1000) })
+
+            const janitor = createJanitor({ stallTimeoutMs: 60_000 })
+            await janitor.start()
+            await janitor.stop()
+
+            const metric = await register.getSingleMetricAsString('cdp_transactional_email_scheduled_waiting')
+            expect(Number(metric.trim().split('\n').pop()!.split(' ').pop())).toBe(2)
+        })
+
+        it('runs one scheduled email count at a time', async () => {
+            const janitor = createJanitor({ stallTimeoutMs: 60_000 })
+            const query = jest.spyOn(janitor['pool'], 'query')
+
+            await Promise.all([janitor.measureScheduledEmailsWaiting(), janitor.measureScheduledEmailsWaiting()])
+            await janitor.measureScheduledEmailsWaiting()
+            await janitor.stop()
+
+            expect(query).toHaveBeenCalledTimes(2)
+        })
+
         it('sweeps expired conversion watchers and keeps live ones', async () => {
             // The sweep is the only thing that removes a watcher that never converts. If a refactor
             // drops it from runOnce the table grows without bound and nothing else fails, so assert