diff --git a/nodejs/src/cdp/cdp-api.serial.test.ts b/nodejs/src/cdp/cdp-api.serial.test.ts index a88a53809fba..6520e093dd33 100644 --- a/nodejs/src/cdp/cdp-api.serial.test.ts +++ b/nodejs/src/cdp/cdp-api.serial.test.ts @@ -70,6 +70,7 @@ const fakeJobProducer = (overrides: Partial = {}): 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 = {}, + headers: Record = { Authorization: `Bearer ${mintManageToken(team.id)}` } + ) => supertest(app).get(`/api/projects/${team.id}/transactional_emails/scheduled`).query(query).set(headers) const scheduleSend = async (): Promise => { 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 => ({ + 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): 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 } diff --git a/nodejs/src/cdp/cdp-api.ts b/nodejs/src/cdp/cdp-api.ts index 6f966d40712f..e47249b9d4cf 100644 --- a/nodejs/src/cdp/cdp-api.ts +++ b/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 => { + 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 => { return await this.manageScheduledSend(req, res, 'reschedule', async (sends, teamId, messageId) => { const parsed = RescheduleRequestSchema.safeParse(req.body) diff --git a/nodejs/src/cdp/services/cyclotron-v2/cyclotron-v2.test.ts b/nodejs/src/cdp/services/cyclotron-v2/cyclotron-v2.test.ts index 0839b830898c..f0c5baaa5f99 100644 --- a/nodejs/src/cdp/services/cyclotron-v2/cyclotron-v2.test.ts +++ b/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 { + async function seedScheduledEmail( + teamId = 1, + scheduled = IN_A_WEEK(), + state: unknown = storedState + ): Promise { 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 { + 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 diff --git a/nodejs/src/cdp/services/cyclotron-v2/janitor.ts b/nodejs/src/cdp/services/cyclotron-v2/janitor.ts index 5330a5164bea..7988097b1c36 100644 --- a/nodejs/src/cdp/services/cyclotron-v2/janitor.ts +++ b/nodejs/src/cdp/services/cyclotron-v2/janitor.ts @@ -8,6 +8,7 @@ import { captureException } from '~/common/utils/posthog' import { CYCLOTRON_INVOCATION_JOB_QUEUES, CyclotronJobInvocationHogFlow } from '../../types' import { HogInvocationResultsService } from '../monitoring/hog-invocation-results.service' import { v2JobToInvocation } from './job-codec' +import { countUnclaimedEmailsWaiting } from './manager' import { CYCLOTRON_COUNTER_MAX, CyclotronV2CleanupResult, CyclotronV2JanitorConfig } from './types' // Stable, low-cardinality `error_kind` stamped on the failed invocation result @@ -82,6 +83,14 @@ const queueOldestDueAgeGauge = new Gauge({ labelNames: ['queue'], }) +const scheduledEmailsWaitingGauge = new Gauge({ + name: 'cdp_transactional_email_scheduled_waiting', + help: 'Scheduled transactional emails that wait for their send time, across all teams.', +}) + +// Slower than the cleanup tick, because the count reads every waiting row of every team. +const SCHEDULED_EMAILS_COUNT_INTERVAL_MS = 60_000 + interface PoisonRow { id: string team_id: number @@ -100,6 +109,8 @@ interface PoisonRow { export class CyclotronV2Janitor { private pool: Pool private intervalHandle: ReturnType | null = null + private scheduledEmailsCountHandle: ReturnType | null = null + private scheduledEmailsCountRunning = false // Queue names this process has seen hold work, so a drained queue can report 0. private readonly seenQueues = new Set() @@ -155,11 +166,32 @@ export class CyclotronV2Janitor { logger.error('CyclotronV2Janitor run error', { error: String(err) }) }) }, this.cleanupIntervalMs) + this.scheduledEmailsCountHandle = setInterval( + () => void this.measureScheduledEmailsWaiting(), + SCHEDULED_EMAILS_COUNT_INTERVAL_MS + ) // Run immediately on start + await this.measureScheduledEmailsWaiting() await this.runOnce() } + // A failed count keeps the last value, so the gauge does not drop to 0 and hide a backlog. + // A tick that finds the last count still running skips, so slow counts cannot pile up on the pool. + async measureScheduledEmailsWaiting(): Promise { + if (this.scheduledEmailsCountRunning) { + return + } + this.scheduledEmailsCountRunning = true + try { + scheduledEmailsWaitingGauge.set(await countUnclaimedEmailsWaiting(this.pool)) + } catch (err) { + logger.error('CyclotronV2Janitor scheduled email count failed', { error: String(err) }) + } finally { + this.scheduledEmailsCountRunning = false + } + } + async runOnce(): Promise { // Sweep first, before the cyclotron-cleanup stages below. Each of those issues an unguarded // pool.query, so a failure specific to cyclotron_jobs (a statement timeout or lock pressure on @@ -551,6 +583,10 @@ export class CyclotronV2Janitor { clearInterval(this.intervalHandle) this.intervalHandle = null } + if (this.scheduledEmailsCountHandle) { + clearInterval(this.scheduledEmailsCountHandle) + this.scheduledEmailsCountHandle = null + } await this.pool.end() } } diff --git a/nodejs/src/cdp/services/cyclotron-v2/manager.ts b/nodejs/src/cdp/services/cyclotron-v2/manager.ts index 2d4d3915986a..b0850659f1ae 100644 --- a/nodejs/src/cdp/services/cyclotron-v2/manager.ts +++ b/nodejs/src/cdp/services/cyclotron-v2/manager.ts @@ -9,7 +9,7 @@ import { sleep } from '~/common/utils/utils' import { BatchResolverResumeOutcome, resumeParkedBatchResolver } from '../hogflows/batch-resolver-resume' import { StepResume, StepResumeOutcome, processStepResumes } from '../hogflows/step-resume.service' -import { withScheduledAt } from './job-codec' +import { decodeJobState, withScheduledAt } from './job-codec' import { CYCLOTRON_COUNTER_MAX, CyclotronV2CancelJobsOptions, @@ -22,6 +22,8 @@ import { CyclotronV2RescheduleParkedOptions, CyclotronV2RescheduleParkedResult, CyclotronV2RescheduleUnclaimedResult, + CyclotronV2UnclaimedEmailCursor, + CyclotronV2UnclaimedEmailJob, } from './types' /** @@ -110,7 +112,30 @@ const CANCELED_BEFORE_CLAIM = `(cyclotron_jobs.status = 'canceled' AND cyclotron // The guard of a job no worker has claimed yet. A claim takes the row lock and bumps transition_count, so an // update that waits on a claim re-checks this guard after the claim commits and then matches nothing. -const UNCLAIMED_EMAIL_JOB = `id = $1 AND team_id = $2 AND queue_name = 'email' AND status = 'available' AND transition_count = 0` +// idx_cyclotron_jobs_email_unclaimed has this predicate, so a change here needs a new index. +const UNCLAIMED_EMAIL = `queue_name = 'email' AND status = 'available' AND transition_count = 0` +const UNCLAIMED_EMAIL_JOB = `id = $1 AND team_id = $2 AND ${UNCLAIMED_EMAIL}` + +export const LIST_UNCLAIMED_EMAIL_SQL = `SELECT id, scheduled, created, state FROM cyclotron_jobs + WHERE team_id = $1 AND ${UNCLAIMED_EMAIL} AND scheduled > NOW() + AND (scheduled, id) > ($2::timestamptz, $3::uuid) + ORDER BY scheduled, id + LIMIT $4` + +// Every waiting row is later than this, so the first page uses the same keyset query as the next ones. +const FIRST_PAGE: CyclotronV2UnclaimedEmailCursor = { + scheduled: new Date(0), + id: '00000000-0000-0000-0000-000000000000', +} + +export async function countUnclaimedEmailsWaiting(pool: Pool): Promise { + const result = await pool.query<{ count: number }>( + `SELECT COUNT(*)::int AS count FROM cyclotron_jobs WHERE ${UNCLAIMED_EMAIL} AND scheduled > NOW()` + ) + return result.rows[0].count +} + +type RecipientState = { event?: { properties?: { to?: unknown } } } | null type UnclaimedJobRow = { id: string @@ -820,6 +845,30 @@ export class CyclotronV2Manager { return { outcome: existing.status === 'canceled' || existing.cancelRequested ? 'canceled' : 'claimed' } } + // A worker can claim a listed job right after the read, so the list is a snapshot. + async listUnclaimed( + teamId: number, + after: CyclotronV2UnclaimedEmailCursor | null, + limit: number + ): Promise { + const from = after ?? FIRST_PAGE + const result = await this.pool.query<{ + id: string + scheduled: string | Date + created: string | Date + state: Buffer | null + }>(LIST_UNCLAIMED_EMAIL_SQL, [teamId, from.scheduled, from.id, limit]) + return result.rows.map((row) => { + const to = (decodeJobState(row).state as RecipientState)?.event?.properties?.to + return { + id: row.id, + to: typeof to === 'string' ? to : null, + scheduled: toDateTime(row.scheduled).toJSDate(), + created: toDateTime(row.created).toJSDate(), + } + }) + } + private async findJob(teamId: number, id: string): Promise<{ status: string; cancelRequested: boolean } | null> { const result = await this.pool.query<{ status: string; cancel_requested: boolean }>( `SELECT status, cancel_requested_at IS NOT NULL AS cancel_requested FROM cyclotron_jobs diff --git a/nodejs/src/cdp/services/cyclotron-v2/types.ts b/nodejs/src/cdp/services/cyclotron-v2/types.ts index e5521765210e..30f9ff3ef143 100644 --- a/nodejs/src/cdp/services/cyclotron-v2/types.ts +++ b/nodejs/src/cdp/services/cyclotron-v2/types.ts @@ -183,6 +183,15 @@ export type CyclotronV2CancelUnclaimedResult = export type CyclotronV2RescheduleUnclaimedResult = { outcome: 'rescheduled' | 'canceled' | 'claimed' | 'not_found' } +export type CyclotronV2UnclaimedEmailJob = { + id: string + to: string | null + scheduled: Date + created: Date +} + +export type CyclotronV2UnclaimedEmailCursor = { scheduled: Date; id: string } + /** * Producer-side surface of `CyclotronV2Manager`. Lets API entrypoints depend * on the interface (testable, mockable) without pulling the full manager @@ -205,6 +214,11 @@ export interface CyclotronV2JobProducer { cancelJobs(options: CyclotronV2CancelJobsOptions): Promise cancelUnclaimed(teamId: number, id: string): Promise rescheduleUnclaimed(teamId: number, id: string, scheduled: Date): Promise + listUnclaimed( + teamId: number, + after: CyclotronV2UnclaimedEmailCursor | null, + limit: number + ): Promise resumeParkedSteps(teamId: number, resumes: StepResume[]): Promise> resumeParkedBatchResolver(options: { teamId: number diff --git a/nodejs/src/cdp/services/hogflows/transactional-email.ts b/nodejs/src/cdp/services/hogflows/transactional-email.ts index ce334cebea74..e08418c64964 100644 --- a/nodejs/src/cdp/services/hogflows/transactional-email.ts +++ b/nodejs/src/cdp/services/hogflows/transactional-email.ts @@ -2,6 +2,9 @@ import { Counter, Histogram } from 'prom-client' import { z } from 'zod' import { CyclotronInvocationQueueParametersEmailType } from '~/cdp/schema/cyclotron' +import type { CyclotronV2UnclaimedEmailCursor } from '~/cdp/services/cyclotron-v2/types' +import { parseJSON } from '~/common/utils/json-parse' +import { UUID } from '~/common/utils/utils' export const AcceptedSendSchema = z .object({ @@ -52,6 +55,34 @@ const ScheduledAtSchema = z.string().datetime({ offset: true }) export const RescheduleRequestSchema = z.object({ scheduled_at: ScheduledAtSchema }) +// A list cursor names the last send of a page, as base64url JSON of [scheduled time, message id]. +export const encodeScheduledSendsCursor = ({ scheduled, id }: CyclotronV2UnclaimedEmailCursor): string => + Buffer.from(JSON.stringify([scheduled.toISOString(), id])).toString('base64url') + +const CursorFieldsSchema = z.tuple([ScheduledAtSchema, z.string().refine((id) => UUID.validateString(id, false))]) + +const parseCursor = (cursor: string): unknown => { + try { + return parseJSON(Buffer.from(cursor, 'base64url').toString('utf-8')) + } catch { + return null + } +} + +const ScheduledSendsCursorSchema = z.string().transform((cursor, ctx): CyclotronV2UnclaimedEmailCursor => { + const fields = CursorFieldsSchema.safeParse(parseCursor(cursor)) + if (!fields.success) { + ctx.addIssue({ code: 'custom', message: 'invalid cursor' }) + return z.NEVER + } + return { scheduled: new Date(fields.data[0]), id: fields.data[1] } +}) + +export const ScheduledSendsListQuerySchema = z.object({ + cursor: ScheduledSendsCursorSchema.optional(), + limit: z.coerce.number().int().min(1).max(100).default(100), +}) + // What Django sends to run a transactional email's workflow once. export const TransactionalEmailRequestSchema = z.object({ to: z.string().email({ pattern: RECIPIENT_ADDRESS_PATTERN }), diff --git a/nodejs/src/cdp/services/hogflows/transactional-sends.ts b/nodejs/src/cdp/services/hogflows/transactional-sends.ts index 2aad1fe7e60e..1827ad461fd7 100644 --- a/nodejs/src/cdp/services/hogflows/transactional-sends.ts +++ b/nodejs/src/cdp/services/hogflows/transactional-sends.ts @@ -11,6 +11,7 @@ import { convertToHogFunctionFilterGlobal } from '../../utils/hog-function-filte import { createCanceledInvocationResult } from '../../utils/invocation-utils' import type { CyclotronV2JobProducer } from '../cyclotron-v2' import { v2JobToInvocation } from '../cyclotron-v2/job-codec' +import type { CyclotronV2UnclaimedEmailCursor } from '../cyclotron-v2/types' import type { InvocationResultsService } from '../invocation-results.service' import { JobQueue } from '../job-queue/job-queue.interface' import { TeamWorkflowsConfigService } from '../managers/team-workflows-config.service' @@ -28,6 +29,7 @@ import { TransactionalEmailRequest, TransactionalEmailStageTimer, TransactionalSendRefusal, + encodeScheduledSendsCursor, } from './transactional-email' const ACCEPTED_SEND_FINGERPRINT_MARKER = 'accepted_send' @@ -264,7 +266,7 @@ export class TransactionalSends { } export type ScheduledTransactionalSendsDeps = { - jobs: Pick + jobs: Pick hogFlows: Pick invocationResults: Pick emailStatus: EmailStatusService @@ -276,9 +278,32 @@ export type ScheduledSendRescheduleResult = | { outcome: 'rescheduled'; scheduledAt: string } | { outcome: 'invalid_scheduled_at' | 'canceled' | 'claimed' | 'not_found' } +export type ScheduledSend = { id: string; to: string | null; scheduled_at: string; created_at: string } + +export type ScheduledSendList = { results: ScheduledSend[]; next: string | null } + export class ScheduledTransactionalSends { constructor(private readonly deps: ScheduledTransactionalSendsDeps) {} + async list( + teamId: number, + after: CyclotronV2UnclaimedEmailCursor | null, + limit: number + ): Promise { + // One row past the page tells whether a next page exists. + const jobs = await this.deps.jobs.listUnclaimed(teamId, after, limit + 1) + const page = jobs.slice(0, limit) + return { + results: page.map((job) => ({ + id: job.id, + to: job.to, + scheduled_at: job.scheduled.toISOString(), + created_at: job.created.toISOString(), + })), + next: jobs.length > limit ? encodeScheduledSendsCursor(page[page.length - 1]) : null, + } + } + async cancel(teamId: number, messageId: string): Promise { if (!UUID.validateString(messageId, false)) { return { outcome: 'not_found' } diff --git a/nodejs/src/common/api/middleware/internal-api-auth.test.ts b/nodejs/src/common/api/middleware/internal-api-auth.test.ts index cfad50463e1b..98ab68901b34 100644 --- a/nodejs/src/common/api/middleware/internal-api-auth.test.ts +++ b/nodejs/src/common/api/middleware/internal-api-auth.test.ts @@ -196,6 +196,7 @@ describe('createInternalApiAuthMiddleware', () => { ['/api/projects/1/hog_flows/some-uuid/transactional_email/send'], ['/api/projects/1/transactional_emails/some-uuid/cancel'], ['/api/projects/1/transactional_emails/some-uuid/reschedule'], + ['/api/projects/1/transactional_emails/scheduled'], ])('should skip auth for scoped-JWT route %s, which authenticates in its handler', (path) => { const middleware = createInternalApiAuthMiddleware({ secret: 'test-secret' }) const req = mockRequest(path, {}) @@ -216,6 +217,7 @@ describe('createInternalApiAuthMiddleware', () => { ['/api/projects/1/hog_flows/some-uuid/reschedule_parked/extra'], ['/api/projects/1/hog_flows/some-uuid/transactional_email/send/extra'], ['/api/projects/1/transactional_emails/some-uuid/cancel/extra'], + ['/api/projects/1/transactional_emails/scheduled/extra'], ])('should still require the secret for %s despite resembling a scoped route', (path) => { const middleware = createInternalApiAuthMiddleware({ secret: 'test-secret' }) const req = mockRequest(path, {}) diff --git a/nodejs/src/common/api/middleware/internal-api-auth.ts b/nodejs/src/common/api/middleware/internal-api-auth.ts index b0dea673c16d..4bd171e8ab2c 100644 --- a/nodejs/src/common/api/middleware/internal-api-auth.ts +++ b/nodejs/src/common/api/middleware/internal-api-auth.ts @@ -45,6 +45,8 @@ const SCOPED_AUTH_PATH_PATTERNS = [ /^\/api\/projects\/[^/]+\/transactional_emails\/[^/]+\/cancel$/, // CdpApi.postTransactionalEmailReschedule, verified against WORKFLOWS_TRANSACTIONAL_EMAIL_JWT_SECRET, audience-pinned. /^\/api\/projects\/[^/]+\/transactional_emails\/[^/]+\/reschedule$/, + // CdpApi.getScheduledTransactionalEmails, verified against WORKFLOWS_TRANSACTIONAL_EMAIL_JWT_SECRET, audience-pinned. + /^\/api\/projects\/[^/]+\/transactional_emails\/scheduled$/, ] export interface InternalApiAuthOptions {