From 15324be33877e06590c3d47cae5f3876f663487f Mon Sep 17 00:00:00 2001 From: Rajeh Taher Date: Sat, 26 Sep 2026 12:45:02 +0200 Subject: [PATCH 1/6] Add scheduled campaign reschedule command --- README.md | 6 +- src/commands/campaign/reschedule.ts | 70 +++++++++++++++++ src/services/campaigns.ts | 16 ++++ test/commands/campaigns.test.ts | 117 ++++++++++++++++++++++++++++ 4 files changed, 208 insertions(+), 1 deletion(-) create mode 100644 src/commands/campaign/reschedule.ts diff --git a/README.md b/README.md index 3845eaa7..5b553c68 100644 --- a/README.md +++ b/README.md @@ -1342,13 +1342,17 @@ polymorfa campaign launch campaign_123 \ --scheduled-at 2026-09-01T12:00:00Z polymorfa campaign pause campaign_123 --project-slug launch-project polymorfa campaign resume campaign_123 --project-slug launch-project +polymorfa campaign reschedule campaign_123 \ + --project-slug launch-project \ + --scheduled-at 2026-09-02T12:00:00Z +polymorfa campaign reschedule campaign_123 --project-slug launch-project --now polymorfa campaign stop campaign_123 --project-slug launch-project --yes polymorfa campaign requeue campaign_123 \ --project-slug launch-project \ --include-skipped-error ``` -Launch, pause, and resume return accepted durable work; they do not wait for +Launch, pause, resume, and reschedule return accepted durable work; they do not wait for the state transition. Stop returns accepted work for an active delivery run, or reports an immediate cancellation with no operation ID when none is active. Check the campaign with `polymorfa campaign view`. When an operation ID is diff --git a/src/commands/campaign/reschedule.ts b/src/commands/campaign/reschedule.ts new file mode 100644 index 00000000..b786fa3b --- /dev/null +++ b/src/commands/campaign/reschedule.ts @@ -0,0 +1,70 @@ +import {Args, Flags} from '@oclif/core' +import {PolymorfaValidationError} from '@polymorfa/sdk' + +import {BaseCommand} from '../../base-command.js' +import {idempotencyKey} from '../../input/idempotency.js' +import { + parseCampaignSchedule, + renderCampaignOperation, + rescheduleCampaign, +} from '../../services/campaigns.js' + +export default class CampaignReschedule extends BaseCommand { + static override args = { + campaign: Args.string({description: 'Campaign ID.', required: true}), + } + + static override description = + 'Change a launched campaign start time before it begins sending, or start it now.' + + static override flags = { + 'idempotency-key': Flags.string({ + description: + 'Idempotency-Key header. Generated per invocation when omitted; reuse an explicit key across process retries (24-hour retention).', + }), + now: Flags.boolean({ + description: 'Start the scheduled campaign now.', + }), + 'project-slug': Flags.string({ + description: 'Project slug in the Messaging API path.', + required: true, + }), + 'scheduled-at': Flags.string({ + description: + 'New start time as epoch milliseconds or an ISO 8601 instant.', + }), + } + + async run(): Promise { + const {args, flags} = await this.parse(CampaignReschedule) + if (flags.now === (flags['scheduled-at'] !== undefined)) { + throw new PolymorfaValidationError( + 'Specify exactly one of --scheduled-at or --now.', + {code: 'invalid_campaign_schedule'}, + ) + } + + const scheduledAt = flags.now + ? null + : parseCampaignSchedule(flags['scheduled-at']) + if (scheduledAt === undefined) { + throw new PolymorfaValidationError( + 'Specify exactly one of --scheduled-at or --now.', + {code: 'invalid_campaign_schedule'}, + ) + } + + const result = await rescheduleCampaign( + await this.messagingClient(flags, {serverApiKeyOnly: 'Campaign'}), + flags['project-slug'], + args.campaign, + scheduledAt, + {idempotencyKey: idempotencyKey(flags['idempotency-key'])}, + ) + this.writeValue( + result, + renderCampaignOperation('reschedule', result.data), + flags, + ) + } +} diff --git a/src/services/campaigns.ts b/src/services/campaigns.ts index 89a2b155..774bd023 100644 --- a/src/services/campaigns.ts +++ b/src/services/campaigns.ts @@ -84,6 +84,22 @@ export async function launchCampaign( return result(response.data.data, response.metadata) } +export async function rescheduleCampaign( + client: MessagingClient, + projectSlug: string, + campaignId: string, + scheduledAt: number | null, + options: RequestOptions = {}, +): Promise> { + const response = await client.campaigns.reschedule( + projectSlug, + campaignId, + {scheduledAt}, + options, + ) + return result(response.data.data, response.metadata) +} + export async function pauseCampaign( client: MessagingClient, projectSlug: string, diff --git a/test/commands/campaigns.test.ts b/test/commands/campaigns.test.ts index 41c71447..c6e824f3 100644 --- a/test/commands/campaigns.test.ts +++ b/test/commands/campaigns.test.ts @@ -175,6 +175,123 @@ describe('Messaging campaign commands', () => { expect(server.requests).toHaveLength(0) }) + test.each([ + ['2026-09-02T12:00:00Z', 1_788_350_400_000], + ['1788350400000', 1_788_350_400_000], + ])('reschedules with a precise instant (%s)', async (at, milliseconds) => { + server = await startHttpServer({ + body: { + data: { + ...campaign(), + operationId: 'op_reschedule', + status: 'scheduled', + }, + success: true, + }, + }) + + const result = await apiKeyEnvironment(() => + runCommand( + `campaign reschedule campaign/1 --project-slug project/main --scheduled-at ${at} --idempotency-key move_1 --json`, + ), + ) + + expect(result.error).toBeUndefined() + expect(JSON.parse(result.stdout)).toMatchObject({ + data: {operationId: 'op_reschedule', status: 'scheduled'}, + metadata: {status: 200}, + }) + expect(server.requests[0]).toMatchObject({ + body: JSON.stringify({scheduledAt: milliseconds}), + method: 'POST', + url: '/messaging/projects/project%2Fmain/campaigns/campaign%2F1/reschedule', + }) + expect(server.requests[0]?.headers['idempotency-key']).toBe('move_1') + }) + + test('starts a scheduled campaign now and generates an idempotency key', async () => { + server = await startHttpServer({ + body: {data: {...campaign(), operationId: 'op_now'}, success: true}, + }) + + const result = await apiKeyEnvironment(() => + runCommand( + 'campaign reschedule campaign_1 --project-slug project_1 --now', + ), + ) + + expect(result.error).toBeUndefined() + expect(result.stdout).toContain( + 'Accepted campaign reschedule as operation op_now.', + ) + expect(server.requests[0]).toMatchObject({ + body: JSON.stringify({scheduledAt: null}), + method: 'POST', + url: '/messaging/projects/project_1/campaigns/campaign_1/reschedule', + }) + expect(server.requests[0]?.headers['idempotency-key']).toMatch( + /^pmfa_cli_[0-9a-f-]{36}$/, + ) + }) + + test.each([ + '', + '--now --scheduled-at 2026-09-02T12:00:00Z', + '--scheduled-at someday', + '--scheduled-at 2026-09-02T12:00:00', + ])( + 'rejects invalid reschedule flags before transport (%s)', + async (flags) => { + server = await startHttpServer({body: {data: campaign(), success: true}}) + + const result = await apiKeyEnvironment(() => + runCommand( + `campaign reschedule campaign_1 --project-slug project_1 ${flags} --json`, + ), + ) + + expect(exitCode(result.error)).toBe(2) + expect(server.requests).toHaveLength(0) + }, + ) + + test('rejects a browser client token before reschedule transport', async () => { + server = await startHttpServer({body: {data: campaign(), success: true}}) + + const result = await clientTokenEnvironment(() => + runCommand( + 'campaign reschedule campaign_1 --project-slug project_1 --now', + ), + ) + + expect(exitCode(result.error)).toBe(2) + expect(server.requests).toHaveLength(0) + }) + + test('reports a changed campaign state as a failed reschedule', async () => { + server = await startHttpServer({ + status: 409, + body: { + success: false, + error: { + code: 'campaign_state_conflict', + message: + 'This campaign has already started, so its start time can no longer be changed.', + }, + }, + }) + + const result = await apiKeyEnvironment(() => + runCommand( + 'campaign reschedule campaign_1 --project-slug project_1 --now --json', + ), + ) + + expect(result.error).toBeDefined() + expect(result.stderr).toContain('already started') + expect(server.requests).toHaveLength(1) + }) + test('pauses, resumes, and explicitly stops through distinct lifecycle routes', async () => { server = await startHttpSequenceServer([ {body: {data: {...campaign(), operationId: 'op_pause'}, success: true}}, From 36bb1603daa141f9d82b6c29452038e52d0ec80b Mon Sep 17 00:00:00 2001 From: Rajeh Taher Date: Sat, 26 Sep 2026 12:52:30 +0200 Subject: [PATCH 2/6] feat(cli): add typed Messaging campaign update command --- README.md | 12 +++++- src/commands/campaign/update.ts | 37 ++++++++++++++++ src/services/campaigns.ts | 40 +++++++++++++++++ test/commands/campaigns.test.ts | 76 +++++++++++++++++++++++++++++++++ 4 files changed, 164 insertions(+), 1 deletion(-) create mode 100644 src/commands/campaign/update.ts diff --git a/README.md b/README.md index 3845eaa7..d12582a6 100644 --- a/README.md +++ b/README.md @@ -1334,6 +1334,8 @@ project slug because the source routes are slug-scoped: ```shell polymorfa campaign list --project-slug launch-project polymorfa campaign create --project-slug launch-project --input campaign.json +polymorfa campaign update 0198a2cb-0000-7000-8000-0000000000e3 \ + --project-slug launch-project --input campaign-patch.json polymorfa campaign view campaign_123 --project-slug launch-project polymorfa campaign analytics campaign_123 --project-slug launch-project @@ -1364,6 +1366,14 @@ Launch times accept epoch milliseconds or timezone-qualified ISO 8601 instants. List returns one complete newest-first array with no cursor. Requeue is synchronous and reports only the number moved. +Update reads one JSON object with at least one of `name`, `recipientListId`, +`senderConfig`, or `scheduledAt`. Send only fields to change. Set +`recipientListId` or `scheduledAt` to `null` to detach the audience or start at +launch. The API refuses audience and start-time changes after launch. A sender +change after launch requires the sending numbers to pass live eligibility +checks. If a PATCH result is lost, view the campaign before deciding whether +to send another update. + Create, launch, pause, resume and stop accept `--idempotency-key` to deduplicate the same request for 24 hours. A successful request repeated with the same key returns `409 idempotency_completed`, not the original response. Changed input @@ -1427,7 +1437,7 @@ List returns `{data, page}`. Pass `page.nextCursor` back as `--cursor` while with `--status queued|sending|sent|delivered|read|failed|skipped`. A recipient suppressed by an opt-out appears as `skipped` with `lastError` `opted_out`. -The source exposes no campaign update, delete, validate, archive, duplicate, +The source exposes no campaign delete, validate, archive, duplicate, event, history, or replay route on the Messaging surface. The CLI does not substitute similarly named Platform campaign methods for these gaps. To follow or stop a campaign's operation, use the `operation` commands in diff --git a/src/commands/campaign/update.ts b/src/commands/campaign/update.ts new file mode 100644 index 00000000..972a856a --- /dev/null +++ b/src/commands/campaign/update.ts @@ -0,0 +1,37 @@ +import {Args, Flags} from '@oclif/core' + +import {readJsonInput} from '../../api/request.js' +import {BaseCommand} from '../../base-command.js' +import {renderCampaign, updateCampaign} from '../../services/campaigns.js' + +export default class CampaignUpdate extends BaseCommand { + static override args = { + campaign: Args.string({description: 'Campaign UUID.', required: true}), + } + + static override description = + 'Update campaign fields. A launched campaign cannot change its audience or start time.' + + static override flags = { + input: Flags.string({ + description: + 'Read the campaign patch JSON object from a file or stdin (-).', + required: true, + }), + 'project-slug': Flags.string({ + description: 'Project slug in the Messaging API path.', + required: true, + }), + } + + async run(): Promise { + const {args, flags} = await this.parse(CampaignUpdate) + const result = await updateCampaign( + await this.messagingClient(flags, {serverApiKeyOnly: 'Campaign'}), + flags['project-slug'], + args.campaign, + await readJsonInput(flags.input), + ) + this.writeValue(result, renderCampaign(result.data), flags) + } +} diff --git a/src/services/campaigns.ts b/src/services/campaigns.ts index 89a2b155..0937807a 100644 --- a/src/services/campaigns.ts +++ b/src/services/campaigns.ts @@ -16,6 +16,7 @@ import { PolymorfaValidationError, type RequestOptions, type ResponseMetadata, + type UpdateCampaignRequest, } from '@polymorfa/sdk' export interface CampaignResult { @@ -59,6 +60,20 @@ export async function viewCampaign( return result(response.data.data, response.metadata) } +export async function updateCampaign( + client: MessagingClient, + projectSlug: string, + campaignId: string, + input: unknown, +): Promise> { + const response = await client.campaigns.update( + projectSlug, + campaignId, + campaignUpdateInput(input), + ) + return result(response.data.data, response.metadata) +} + export async function viewCampaignAnalytics( client: MessagingClient, projectSlug: string, @@ -351,6 +366,31 @@ function campaignInput(input: unknown): CreateCampaignRequest { return input as unknown as CreateCampaignRequest } +function campaignUpdateInput(input: unknown): UpdateCampaignRequest { + if (!isRecord(input)) { + throw new PolymorfaValidationError( + 'Campaign update input must contain one JSON object.', + {code: 'invalid_campaign_update_input'}, + ) + } + + const allowed = new Set([ + 'name', + 'recipientListId', + 'senderConfig', + 'scheduledAt', + ]) + const names = Object.keys(input) + if (names.length === 0 || names.some((name) => !allowed.has(name))) { + throw new PolymorfaValidationError( + 'Campaign update input must contain only name, recipientListId, senderConfig, or scheduledAt.', + {code: 'invalid_campaign_update_fields'}, + ) + } + + return input +} + function result(data: T, metadata: ResponseMetadata): CampaignResult { return { data, diff --git a/test/commands/campaigns.test.ts b/test/commands/campaigns.test.ts index 41c71447..c672e456 100644 --- a/test/commands/campaigns.test.ts +++ b/test/commands/campaigns.test.ts @@ -82,6 +82,82 @@ describe('Messaging campaign commands', () => { }, ) + test('updates a campaign with a partial JSON body through Messaging PATCH', async () => { + const input = await jsonInput('campaign-update.json', { + name: 'Autumn launch', + recipientListId: null, + scheduledAt: null, + }) + server = await startHttpServer({ + body: {data: {...campaign(), name: 'Autumn launch'}, success: true}, + headers: {'request-id': 'req_campaign_update'}, + }) + + const result = await apiKeyEnvironment(() => + runCommand( + `campaign update 0198a2cb-0000-7000-8000-0000000000e3 --project-slug project/main --input ${input} --json`, + ), + ) + + expect(result.error).toBeUndefined() + expect(JSON.parse(result.stdout)).toMatchObject({ + data: {name: 'Autumn launch'}, + metadata: {attempts: 1, requestId: 'req_campaign_update', status: 200}, + }) + expect(server.requests[0]).toMatchObject({ + method: 'PATCH', + url: '/messaging/projects/project%2Fmain/campaigns/0198a2cb-0000-7000-8000-0000000000e3', + body: JSON.stringify({ + name: 'Autumn launch', + recipientListId: null, + scheduledAt: null, + }), + }) + }) + + test.each([{}, {status: 'sent'}, []])( + 'rejects unsupported or empty campaign patches before transport', + async (value) => { + const input = await jsonInput('campaign-update-invalid.json', value) + server = await startHttpServer({body: {data: campaign(), success: true}}) + const result = await apiKeyEnvironment(() => + runCommand( + `campaign update 0198a2cb-0000-7000-8000-0000000000e3 --project-slug project_1 --input ${input}`, + ), + ) + expect(exitCode(result.error)).toBe(2) + expect(server.requests).toHaveLength(0) + }, + ) + + test('reports a launched-audience conflict without replaying the PATCH', async () => { + const input = await jsonInput('campaign-update-conflict.json', { + recipientListId: null, + }) + server = await startHttpServer({ + body: { + error: { + code: 'state_conflict', + message: 'The audience is fixed after launch.', + }, + success: false, + }, + status: 409, + }) + + const result = await apiKeyEnvironment(() => + runCommand( + `campaign update 0198a2cb-0000-7000-8000-0000000000e3 --project-slug project_1 --input ${input} --json`, + ), + ) + + expect(exitCode(result.error)).toBe(4) + expect(JSON.parse(result.stderr)).toMatchObject({ + error: {category: 'conflict', code: 'state_conflict'}, + }) + expect(server.requests).toHaveLength(1) + }) + test('views one campaign and its exact live analytics counters', async () => { server = await startHttpSequenceServer([ {body: {data: campaign(), success: true}}, From 1620b6904664f883914fcf0750a88f771487221f Mon Sep 17 00:00:00 2001 From: Rajeh Taher Date: Sat, 26 Sep 2026 12:55:11 +0200 Subject: [PATCH 3/6] docs(cli): name campaign schedule timestamp range --- README.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index d12582a6..20b95c89 100644 --- a/README.md +++ b/README.md @@ -1369,7 +1369,8 @@ is synchronous and reports only the number moved. Update reads one JSON object with at least one of `name`, `recipientListId`, `senderConfig`, or `scheduledAt`. Send only fields to change. Set `recipientListId` or `scheduledAt` to `null` to detach the audience or start at -launch. The API refuses audience and start-time changes after launch. A sender +launch. A non-null `scheduledAt` must be integer Unix milliseconds within the +JavaScript Date range. The API refuses audience and start-time changes after launch. A sender change after launch requires the sending numbers to pass live eligibility checks. If a PATCH result is lost, view the campaign before deciding whether to send another update. From f81a7c8b0d95c93cb8d3217fd14b33aa1779dd8f Mon Sep 17 00:00:00 2001 From: Rajeh Taher Date: Sat, 26 Sep 2026 12:57:46 +0200 Subject: [PATCH 4/6] Reject unrepresentable campaign reschedule times --- src/commands/campaign/reschedule.ts | 9 +++++++++ test/commands/campaigns.test.ts | 2 ++ 2 files changed, 11 insertions(+) diff --git a/src/commands/campaign/reschedule.ts b/src/commands/campaign/reschedule.ts index b786fa3b..0cd2586d 100644 --- a/src/commands/campaign/reschedule.ts +++ b/src/commands/campaign/reschedule.ts @@ -53,6 +53,15 @@ export default class CampaignReschedule extends BaseCommand { {code: 'invalid_campaign_schedule'}, ) } + if ( + scheduledAt !== null && + (scheduledAt <= 0 || scheduledAt > 8_640_000_000_000_000) + ) { + throw new PolymorfaValidationError( + 'Scheduled time must be a positive, representable Unix timestamp in milliseconds.', + {code: 'invalid_campaign_schedule'}, + ) + } const result = await rescheduleCampaign( await this.messagingClient(flags, {serverApiKeyOnly: 'Campaign'}), diff --git a/test/commands/campaigns.test.ts b/test/commands/campaigns.test.ts index c6e824f3..e9dbbee8 100644 --- a/test/commands/campaigns.test.ts +++ b/test/commands/campaigns.test.ts @@ -238,6 +238,8 @@ describe('Messaging campaign commands', () => { '', '--now --scheduled-at 2026-09-02T12:00:00Z', '--scheduled-at someday', + '--scheduled-at 0', + '--scheduled-at 8640000000000001', '--scheduled-at 2026-09-02T12:00:00', ])( 'rejects invalid reschedule flags before transport (%s)', From 9540bc8ff3dc5fab3d0070c5cf12b7a53705df42 Mon Sep 17 00:00:00 2001 From: Rajeh Taher Date: Sat, 26 Sep 2026 23:09:09 +0200 Subject: [PATCH 5/6] feat(campaigns): export recipient outcomes from CLI --- README.md | 17 +++- src/commands/campaign/recipients/export.ts | 110 +++++++++++++++++++++ src/commands/campaign/recipients/list.ts | 9 ++ src/services/campaigns.ts | 72 ++++++++++++++ test/commands/campaign-recipients.test.ts | 64 ++++++++++++ 5 files changed, 270 insertions(+), 2 deletions(-) create mode 100644 src/commands/campaign/recipients/export.ts diff --git a/README.md b/README.md index 349fe3db..46c7eeca 100644 --- a/README.md +++ b/README.md @@ -1399,6 +1399,12 @@ polymorfa campaign recipients list campaign_123 \ --project-slug launch-project \ --status skipped \ --limit 100 + +polymorfa campaign recipients export campaign_123 \ + --project-slug launch-project \ + --status failed \ + --reason send_failed \ + --output recipient-outcomes.csv ``` One append accepts at most 1,000 entries. `--csv` reads the file locally and @@ -1424,8 +1430,15 @@ result arrives, list the recipients before appending again. List returns `{data, page}`. Pass `page.nextCursor` back as `--cursor` while `page.hasMore` is true; the CLI prints the next cursor in human output. Filter -with `--status queued|sending|sent|delivered|read|failed|skipped`. A recipient -suppressed by an opt-out appears as `skipped` with `lastError` `opted_out`. +with `--status queued|sending|sent|delivered|read|failed|skipped` and `--reason` +for a normalized failure reason. A recipient suppressed by an opt-out appears +as `skipped` with `failureReason` and `lastError` `opted_out`. + +`recipients export` fetches up to 1,000 rows per API page and joins them into +one CSV header. It writes to standard output unless `--output` names a file; +an existing file requires `--force`. Use `--max-pages` to stop early, then pass +the cursor printed on standard error back with `--cursor` and the same filters. +Each page reflects recipient outcomes when that page was read. The source exposes no campaign update, delete, validate, archive, duplicate, event, history, or replay route on the Messaging surface. The CLI does not diff --git a/src/commands/campaign/recipients/export.ts b/src/commands/campaign/recipients/export.ts new file mode 100644 index 00000000..b6474e9d --- /dev/null +++ b/src/commands/campaign/recipients/export.ts @@ -0,0 +1,110 @@ +import {Args, Flags} from '@oclif/core' +import {PolymorfaConfigurationError} from '@polymorfa/sdk' +import {stat, writeFile} from 'node:fs/promises' +import {resolve} from 'node:path' + +import {BaseCommand} from '../../../base-command.js' +import { + exportCampaignRecipients, + recipientFailureReason, + recipientStatus, + RECIPIENT_FAILURE_REASONS, + RECIPIENT_STATUSES, +} from '../../../services/campaigns.js' + +export default class CampaignRecipientsExport extends BaseCommand { + static override args = { + campaign: Args.string({description: 'Campaign ID.', required: true}), + } + + static override description = `Export campaign recipient outcomes as CSV. + +The command reads cursor pages with the same filters and writes one header. Use --max-pages to stop early; the remaining cursor is printed on standard error. Resume with --cursor and the same filters. This read reflects each recipient when its page was fetched.` + + static override examples = [ + '<%= config.bin %> <%= command.id %> campaign_1 --project-slug launch --output recipients.csv', + '<%= config.bin %> <%= command.id %> campaign_1 --project-slug launch --status failed --reason send_failed --max-pages 1', + ] + + static override flags = { + cursor: Flags.string({ + description: 'Continue from an earlier export cursor.', + }), + force: Flags.boolean({description: 'Overwrite an existing --output file.'}), + limit: Flags.integer({ + description: 'Rows per API page.', + min: 1, + max: 1000, + default: 1000, + }), + 'max-pages': Flags.integer({ + description: 'Stop after this many pages.', + min: 1, + max: 500, + }), + output: Flags.string({ + char: 'o', + description: 'Write CSV to this file instead of standard output.', + }), + 'project-slug': Flags.string({ + description: 'Project slug in the Messaging API path.', + required: true, + }), + reason: Flags.string({ + description: 'Filter by failure reason.', + options: [...RECIPIENT_FAILURE_REASONS], + }), + status: Flags.string({ + description: 'Filter by delivery status.', + options: [...RECIPIENT_STATUSES], + }), + } + + async run(): Promise { + const {args, flags} = await this.parse(CampaignRecipientsExport) + const cursor = flags.cursor?.trim() + if (cursor !== undefined && (cursor.length === 0 || cursor.length > 1024)) + throw new PolymorfaConfigurationError( + '--cursor must be 1 to 1024 characters.', + 'cursor', + ) + const outputPath = + flags.output && flags.output !== '-' ? resolve(flags.output) : undefined + if (outputPath && !flags.force) { + try { + await stat(outputPath) + throw new PolymorfaConfigurationError( + 'The output file already exists; pass --force to replace it.', + 'output', + ) + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error + } + } + const result = await exportCampaignRecipients( + await this.messagingClient(flags, {serverApiKeyOnly: 'Campaign'}), + flags['project-slug'], + args.campaign, + { + ...(cursor === undefined ? {} : {cursor}), + limit: flags.limit, + ...(flags.status === undefined + ? {} + : {status: recipientStatus(flags.status)!}), + ...(flags.reason === undefined + ? {} + : {reason: recipientFailureReason(flags.reason)!}), + }, + flags['max-pages'] ?? 500, + ) + if (outputPath) { + await writeFile(outputPath, result.csv, { + flag: flags.force ? 'w' : 'wx', + }) + } else { + process.stdout.write(result.csv) + } + if (result.nextCursor !== null) + process.stderr.write(`Continue with --cursor ${result.nextCursor}\n`) + } +} diff --git a/src/commands/campaign/recipients/list.ts b/src/commands/campaign/recipients/list.ts index 9a7354d2..a1888560 100644 --- a/src/commands/campaign/recipients/list.ts +++ b/src/commands/campaign/recipients/list.ts @@ -3,7 +3,9 @@ import {Args, Flags} from '@oclif/core' import {BaseCommand} from '../../../base-command.js' import { listCampaignRecipients, + recipientFailureReason, recipientStatus, + RECIPIENT_FAILURE_REASONS, renderCampaignRecipients, RECIPIENT_STATUSES, } from '../../../services/campaigns.js' @@ -31,6 +33,10 @@ export default class CampaignRecipientsList extends BaseCommand { description: 'Project slug in the Messaging API path.', required: true, }), + reason: Flags.string({ + description: 'Filter by the normalized failure reason.', + options: [...RECIPIENT_FAILURE_REASONS], + }), status: Flags.string({ description: 'Filter by delivery status. Use skipped to find recipients suppressed by an opt-out.', @@ -50,6 +56,9 @@ export default class CampaignRecipientsList extends BaseCommand { ...(flags.status === undefined ? {} : {status: recipientStatus(flags.status)!}), + ...(flags.reason === undefined + ? {} + : {reason: recipientFailureReason(flags.reason)!}), }, ) this.writeValue(result, renderCampaignRecipients(result.data), flags) diff --git a/src/services/campaigns.ts b/src/services/campaigns.ts index 89a2b155..7765942e 100644 --- a/src/services/campaigns.ts +++ b/src/services/campaigns.ts @@ -4,9 +4,11 @@ import { type CampaignAnalytics, type CampaignOperation, type CampaignRecipient, + type CampaignRecipientFailureReason, type CampaignRecipientInput, type CampaignRecipientPage, type CampaignRecipientStatus, + type ExportCampaignRecipientsParams, type CampaignRequeueResult, type CampaignStopOperation, type CreateCampaignRequest, @@ -144,6 +146,58 @@ export async function listCampaignRecipients( ) } +export interface CampaignRecipientCsvResult { + readonly csv: string + readonly nextCursor: string | null + readonly pages: number +} + +/** Download bounded CSV pages, keeping the header only once. */ +export async function exportCampaignRecipients( + client: MessagingClient, + projectSlug: string, + campaignId: string, + params: ExportCampaignRecipientsParams, + maxPages = 500, +): Promise { + const chunks: string[] = [] + let header: string | null = null + let cursor = params.cursor + const seen = new Set() + for (let page = 0; page < maxPages; page += 1) { + const response = await client.campaigns.exportRecipients( + projectSlug, + campaignId, + { + ...params, + ...(cursor === undefined ? {} : {cursor}), + }, + ) + const {csv, nextCursor} = response.data + const headerEnd = csv.indexOf('\r\n') + if (headerEnd < 0) + throw new PolymorfaValidationError( + 'The recipient export did not return a CSV header.', + ) + const pageHeader = csv.slice(0, headerEnd) + if (header === null) header = pageHeader + else if (pageHeader !== header) + throw new PolymorfaValidationError( + 'The recipient export columns changed between pages.', + ) + chunks.push(page === 0 ? csv : csv.slice(headerEnd + 2)) + if (nextCursor === null) + return {csv: chunks.join(''), nextCursor: null, pages: page + 1} + if (seen.has(nextCursor) || nextCursor === cursor) + throw new PolymorfaValidationError( + 'The recipient export returned a repeated cursor.', + ) + seen.add(nextCursor) + cursor = nextCursor + } + return {csv: chunks.join(''), nextCursor: cursor ?? null, pages: maxPages} +} + export async function addCampaignRecipients( client: MessagingClient, projectSlug: string, @@ -325,6 +379,24 @@ export function recipientStatus( return value as CampaignRecipientStatus } +export function recipientFailureReason( + value?: string, +): CampaignRecipientFailureReason | undefined { + return value as CampaignRecipientFailureReason | undefined +} + +export const RECIPIENT_FAILURE_REASONS = [ + 'opted_out', + 'cold_held', + 'campaign_cancelled', + 'invalid_recipient', + 'session_not_connected', + 'ack_timeout', + 'blocked_by_safety', + 'send_failed', + 'other', +] as const + export const RECIPIENT_STATUSES = [ 'queued', 'sending', diff --git a/test/commands/campaign-recipients.test.ts b/test/commands/campaign-recipients.test.ts index f8bcfe7a..99623ecd 100644 --- a/test/commands/campaign-recipients.test.ts +++ b/test/commands/campaign-recipients.test.ts @@ -7,6 +7,7 @@ import {afterEach, describe, expect, test} from 'vitest' import { startHttpServer, + startHttpSequenceServer, withEnvironment, type TestHttpServer, } from '../helpers/http-server.js' @@ -52,6 +53,69 @@ describe('Campaign recipient commands', () => { }) }) + test('filters recipient pages by failure reason', async () => { + server = await startHttpServer({ + body: { + data: [recipient()], + page: {hasMore: false, nextCursor: null}, + success: true, + }, + }) + const result = await apiKeyEnvironment(() => + runCommand( + 'campaign recipients list campaign_1 --project-slug project_1 --reason opted_out', + ), + ) + expect(result.error).toBeUndefined() + expect(server.requests[0]?.url).toContain('reason=opted_out') + }) + + test('exports paged CSV with one header and a stable filter', async () => { + server = await startHttpSequenceServer([ + { + rawBody: 'recipientId,phone\r\nr1,+14155550100\r\n', + headers: { + 'content-type': 'text/csv', + 'Polymorfa-Next-Cursor': 'next_2', + }, + }, + { + rawBody: 'recipientId,phone\r\nr2,+14155550101\r\n', + headers: {'content-type': 'text/csv'}, + }, + ]) + const result = await apiKeyEnvironment(() => + runCommand( + 'campaign recipients export campaign_1 --project-slug project_1 --reason send_failed --limit 1', + ), + ) + expect(result.error).toBeUndefined() + expect(result.stdout).toBe( + 'recipientId,phone\r\nr1,+14155550100\r\nr2,+14155550101\r\n', + ) + expect(server.requests.map((request) => request.url)).toEqual([ + '/messaging/projects/project_1/campaigns/campaign_1/recipients/export?reason=send_failed&limit=1', + '/messaging/projects/project_1/campaigns/campaign_1/recipients/export?reason=send_failed&cursor=next_2&limit=1', + ]) + }) + + test('refuses to replace an export file before making an API request', async () => { + directory = await mkdtemp(join(tmpdir(), 'polymorfa-export-')) + const output = join(directory, 'recipients.csv') + await writeFile(output, 'existing') + server = await startHttpServer({ + rawBody: 'recipientId,phone\r\n', + headers: {'content-type': 'text/csv'}, + }) + const result = await apiKeyEnvironment(() => + runCommand( + `campaign recipients export campaign_1 --project-slug project_1 --output ${output}`, + ), + ) + expect(result.error).toBeDefined() + expect(server.requests).toHaveLength(0) + }) + test('offers the next cursor in human output when a page remains', async () => { server = await startHttpServer({ body: { From 7fd4bad81350d3c142680e49bca63f6f892af6cb Mon Sep 17 00:00:00 2001 From: Rajeh Taher Date: Tue, 6 Oct 2026 02:45:48 +0300 Subject: [PATCH 6/6] Stream recipient exports, honour output modes, and validate campaign updates - recipients export writes each page as it arrives (stdout waits for drain), so memory stays at one page; a later failure keeps the written pages and prints the cursor to resume from. - --json is refused and --quiet needs --output, before any recipient is read. - campaign update validates every field (name, recipientListId, senderConfig, scheduledAt range, sendWindow, templateId, variableMapping, trackedLinks) and sends only the checked fields. --- README.md | 13 +- src/commands/campaign/recipients/export.ts | 82 ++++++++---- src/commands/campaign/update.ts | 2 +- src/services/campaigns.ts | 138 +++++++++++++++------ test/commands/campaign-recipients.test.ts | 48 +++++++ test/commands/campaigns.test.ts | 30 ++++- 6 files changed, 242 insertions(+), 71 deletions(-) diff --git a/README.md b/README.md index 35b0806c..90d448e9 100644 --- a/README.md +++ b/README.md @@ -1784,11 +1784,14 @@ with `--status queued|sending|sent|delivered|read|failed|skipped` and `--reason` for a normalized failure reason. A recipient suppressed by an opt-out appears as `skipped` with `failureReason` and `lastError` `opted_out`. -`recipients export` fetches up to 1,000 rows per API page and joins them into -one CSV header. It writes to standard output unless `--output` names a file; -an existing file requires `--force`. Use `--max-pages` to stop early, then pass -the cursor printed on standard error back with `--cursor` and the same filters. -Each page reflects recipient outcomes when that page was read. +`recipients export` fetches up to 1,000 rows per API page and writes each page +as it arrives, under one CSV header. It writes to standard output unless +`--output` names a file; an existing file requires `--force`. Use `--max-pages` +to stop early, then pass the cursor printed on standard error back with +`--cursor` and the same filters. If a later page fails, the pages already +written stay and the cursor to resume from is printed. The export is CSV only: +`--json` is refused, and `--quiet` needs `--output`. Each page reflects +recipient outcomes when that page was read. #### Campaign conversions (beta) diff --git a/src/commands/campaign/recipients/export.ts b/src/commands/campaign/recipients/export.ts index b6474e9d..1d25ec12 100644 --- a/src/commands/campaign/recipients/export.ts +++ b/src/commands/campaign/recipients/export.ts @@ -1,10 +1,11 @@ import {Args, Flags} from '@oclif/core' import {PolymorfaConfigurationError} from '@polymorfa/sdk' -import {stat, writeFile} from 'node:fs/promises' +import {open, stat} from 'node:fs/promises' import {resolve} from 'node:path' import {BaseCommand} from '../../../base-command.js' import { + CampaignRecipientExportInterrupted, exportCampaignRecipients, recipientFailureReason, recipientStatus, @@ -70,6 +71,19 @@ The command reads cursor pages with the same filters and writes one header. Use ) const outputPath = flags.output && flags.output !== '-' ? resolve(flags.output) : undefined + // The export is CSV only. Refuse modes it cannot honour before reading + // any recipient, so --quiet never prints recipient data. + const mode = this.outputMode(flags) + if (mode === 'json') + throw new PolymorfaConfigurationError( + 'Recipient export writes CSV; --json is not supported. Use campaign recipients list --json for JSON.', + 'json', + ) + if (mode === 'quiet' && outputPath === undefined) + throw new PolymorfaConfigurationError( + '--quiet would discard the CSV; pass --output to write it to a file.', + 'quiet', + ) if (outputPath && !flags.force) { try { await stat(outputPath) @@ -81,28 +95,50 @@ The command reads cursor pages with the same filters and writes one header. Use if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error } } - const result = await exportCampaignRecipients( - await this.messagingClient(flags, {serverApiKeyOnly: 'Campaign'}), - flags['project-slug'], - args.campaign, - { - ...(cursor === undefined ? {} : {cursor}), - limit: flags.limit, - ...(flags.status === undefined - ? {} - : {status: recipientStatus(flags.status)!}), - ...(flags.reason === undefined - ? {} - : {reason: recipientFailureReason(flags.reason)!}), - }, - flags['max-pages'] ?? 500, - ) - if (outputPath) { - await writeFile(outputPath, result.csv, { - flag: flags.force ? 'w' : 'wx', - }) - } else { - process.stdout.write(result.csv) + const file = outputPath + ? await open(outputPath, flags.force ? 'w' : 'wx') + : undefined + const write = file + ? async (chunk: string) => { + await file.write(chunk) + } + : async (chunk: string) => { + // Wait for the pipe to drain so a slow reader bounds memory. + if (!process.stdout.write(chunk)) + await new Promise((done) => + process.stdout.once('drain', () => done()), + ) + } + let result + try { + result = await exportCampaignRecipients( + await this.messagingClient(flags, {serverApiKeyOnly: 'Campaign'}), + flags['project-slug'], + args.campaign, + { + ...(cursor === undefined ? {} : {cursor}), + limit: flags.limit, + ...(flags.status === undefined + ? {} + : {status: recipientStatus(flags.status)!}), + ...(flags.reason === undefined + ? {} + : {reason: recipientFailureReason(flags.reason)!}), + }, + write, + flags['max-pages'] ?? 500, + ) + } catch (error) { + if (error instanceof CampaignRecipientExportInterrupted) { + // The pages already written stay; resume after them. + process.stderr.write( + `Wrote ${error.pages} page(s) before the export stopped. Continue with --cursor ${error.nextCursor}\n`, + ) + throw error.cause + } + throw error + } finally { + await file?.close() } if (result.nextCursor !== null) process.stderr.write(`Continue with --cursor ${result.nextCursor}\n`) diff --git a/src/commands/campaign/update.ts b/src/commands/campaign/update.ts index 972a856a..3f1a9b0b 100644 --- a/src/commands/campaign/update.ts +++ b/src/commands/campaign/update.ts @@ -10,7 +10,7 @@ export default class CampaignUpdate extends BaseCommand { } static override description = - 'Update campaign fields. A launched campaign cannot change its audience or start time.' + 'Update campaign fields: name, recipientListId, senderConfig, scheduledAt, sendWindow, templateId, variableMapping or trackedLinks. A launched campaign cannot change its audience, start time, template, variable mapping or tracked links.' static override flags = { input: Flags.string({ diff --git a/src/services/campaigns.ts b/src/services/campaigns.ts index 61018d4b..3dcb4140 100644 --- a/src/services/campaigns.ts +++ b/src/services/campaigns.ts @@ -1,3 +1,4 @@ +import {z} from 'zod' import { type AddCampaignRecipientsResult, type Campaign, @@ -178,55 +179,80 @@ export async function listCampaignRecipients( } export interface CampaignRecipientCsvResult { - readonly csv: string + /** Cursor after the last written page, or null when the export finished. */ readonly nextCursor: string | null readonly pages: number } -/** Download bounded CSV pages, keeping the header only once. */ +/** Raised after at least one page was written; `nextCursor` resumes after it. */ +export class CampaignRecipientExportInterrupted extends Error { + constructor( + override readonly cause: unknown, + readonly nextCursor: string, + readonly pages: number, + ) { + super('The recipient export stopped before the last page.') + } +} + +/** + * Download bounded CSV pages, keeping the header only once. Each page is + * handed to `write` as it arrives, so memory stays at one page and a later + * failure leaves the earlier pages written with a cursor to resume from. + */ export async function exportCampaignRecipients( client: MessagingClient, projectSlug: string, campaignId: string, params: ExportCampaignRecipientsParams, + write: (chunk: string) => Promise, maxPages = 500, ): Promise { - const chunks: string[] = [] let header: string | null = null let cursor = params.cursor const seen = new Set() for (let page = 0; page < maxPages; page += 1) { - const response = await client.campaigns.exportRecipients( - projectSlug, - campaignId, - { - ...params, - ...(cursor === undefined ? {} : {cursor}), - }, - ) - const {csv, nextCursor} = response.data - const headerEnd = csv.indexOf('\r\n') - if (headerEnd < 0) - throw new PolymorfaValidationError( - 'The recipient export did not return a CSV header.', + let csv: string + let nextCursor: string | null + try { + const response = await client.campaigns.exportRecipients( + projectSlug, + campaignId, + { + ...params, + ...(cursor === undefined ? {} : {cursor}), + }, ) - const pageHeader = csv.slice(0, headerEnd) - if (header === null) header = pageHeader - else if (pageHeader !== header) - throw new PolymorfaValidationError( - 'The recipient export columns changed between pages.', - ) - chunks.push(page === 0 ? csv : csv.slice(headerEnd + 2)) - if (nextCursor === null) - return {csv: chunks.join(''), nextCursor: null, pages: page + 1} - if (seen.has(nextCursor) || nextCursor === cursor) - throw new PolymorfaValidationError( - 'The recipient export returned a repeated cursor.', + ;({csv, nextCursor} = response.data) + const headerEnd = csv.indexOf('\r\n') + if (headerEnd < 0) + throw new PolymorfaValidationError( + 'The recipient export did not return a CSV header.', + ) + const pageHeader = csv.slice(0, headerEnd) + if (header === null) header = pageHeader + else if (pageHeader !== header) + throw new PolymorfaValidationError( + 'The recipient export columns changed between pages.', + ) + if ( + nextCursor !== null && + (seen.has(nextCursor) || nextCursor === cursor) ) + throw new PolymorfaValidationError( + 'The recipient export returned a repeated cursor.', + ) + await write(page === 0 ? csv : csv.slice(headerEnd + 2)) + } catch (error) { + if (page > 0 && cursor !== undefined) + throw new CampaignRecipientExportInterrupted(error, cursor, page) + throw error + } + if (nextCursor === null) return {nextCursor: null, pages: page + 1} seen.add(nextCursor) cursor = nextCursor } - return {csv: chunks.join(''), nextCursor: cursor ?? null, pages: maxPages} + return {nextCursor: cursor ?? null, pages: maxPages} } export async function addCampaignRecipients( @@ -454,6 +480,40 @@ function campaignInput(input: unknown): CreateCampaignRequest { return input as unknown as CreateCampaignRequest } +const MAX_DATE_MS = 8_640_000_000_000_000 +const UUID_PATTERN = + /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i +const CampaignUpdate = z + .object({ + name: z + .string() + .min(1) + .max(200) + .refine((value) => value.trim().length > 0), + recipientListId: z.string().min(1).nullable(), + senderConfig: z.record(z.string(), z.unknown()), + scheduledAt: z.number().int().min(0).max(MAX_DATE_MS).nullable(), + sendWindow: z.record(z.string(), z.unknown()).nullable(), + templateId: z.string().regex(UUID_PATTERN).nullable(), + variableMapping: z + .record( + z.string(), + z + .object({ + source: z.string().min(1).optional(), + fallback: z.string().min(1).max(1024).optional(), + }) + .strict(), + ) + .nullable(), + trackedLinks: z + .array(z.object({key: z.string(), url: z.string()}).strict()) + .max(5) + .nullable(), + }) + .partial() + .strict() + function campaignUpdateInput(input: unknown): UpdateCampaignRequest { if (!isRecord(input)) { throw new PolymorfaValidationError( @@ -461,22 +521,18 @@ function campaignUpdateInput(input: unknown): UpdateCampaignRequest { {code: 'invalid_campaign_update_input'}, ) } - - const allowed = new Set([ - 'name', - 'recipientListId', - 'senderConfig', - 'scheduledAt', - ]) - const names = Object.keys(input) - if (names.length === 0 || names.some((name) => !allowed.has(name))) { + const parsed = CampaignUpdate.safeParse(input) + if (!parsed.success || Object.keys(parsed.data).length === 0) { throw new PolymorfaValidationError( - 'Campaign update input must contain only name, recipientListId, senderConfig, or scheduledAt.', + 'Campaign update input must contain name, recipientListId, senderConfig, scheduledAt, sendWindow, templateId, variableMapping, or trackedLinks, with valid values. scheduledAt is integer Unix milliseconds or null.', {code: 'invalid_campaign_update_fields'}, ) } - - return input + // Only the checked, present fields reach the SDK request. + const request: Record = {} + for (const [key, value] of Object.entries(parsed.data)) + if (value !== undefined) request[key] = value + return request } function result(data: T, metadata: ResponseMetadata): CampaignResult { diff --git a/test/commands/campaign-recipients.test.ts b/test/commands/campaign-recipients.test.ts index 99623ecd..f981a486 100644 --- a/test/commands/campaign-recipients.test.ts +++ b/test/commands/campaign-recipients.test.ts @@ -99,6 +99,54 @@ describe('Campaign recipient commands', () => { ]) }) + test.each([ + ['--json', 'json'], + ['--quiet', 'quiet'], + ])( + 'refuses %s without an output file before reading recipients', + async (flag) => { + server = await startHttpServer({ + rawBody: 'recipientId,phone\r\n', + headers: {'content-type': 'text/csv'}, + }) + const result = await apiKeyEnvironment(() => + runCommand( + `campaign recipients export campaign_1 --project-slug project_1 ${flag}`, + ), + ) + expect(result.error).toBeDefined() + expect(result.stdout).toBe('') + expect(server.requests).toHaveLength(0) + }, + ) + + test('keeps written pages and prints the resume cursor when a later page fails', async () => { + server = await startHttpSequenceServer([ + { + rawBody: 'recipientId,phone\r\nr1,+14155550100\r\n', + headers: { + 'content-type': 'text/csv', + 'Polymorfa-Next-Cursor': 'next_2', + }, + }, + { + status: 400, + body: { + success: false, + error: {code: 'invalid_parameter', message: 'bad cursor'}, + }, + }, + ]) + const result = await apiKeyEnvironment(() => + runCommand( + 'campaign recipients export campaign_1 --project-slug project_1 --limit 1', + ), + ) + expect(result.error).toBeDefined() + expect(result.stdout).toBe('recipientId,phone\r\nr1,+14155550100\r\n') + expect(result.stderr).toContain('--cursor next_2') + }) + test('refuses to replace an export file before making an API request', async () => { directory = await mkdtemp(join(tmpdir(), 'polymorfa-export-')) const output = join(directory, 'recipients.csv') diff --git a/test/commands/campaigns.test.ts b/test/commands/campaigns.test.ts index 0edc7070..63951d53 100644 --- a/test/commands/campaigns.test.ts +++ b/test/commands/campaigns.test.ts @@ -115,7 +115,17 @@ describe('Messaging campaign commands', () => { }) }) - test.each([{}, {status: 'sent'}, []])( + test.each([ + {}, + {status: 'sent'}, + [], + {name: null}, + {name: ' '}, + {scheduledAt: 8_640_000_000_000_001}, + {scheduledAt: 1.5}, + {templateId: 'not-a-uuid'}, + {variableMapping: {first_name: {fallback: ''}}}, + ])( 'rejects unsupported or empty campaign patches before transport', async (value) => { const input = await jsonInput('campaign-update-invalid.json', value) @@ -130,6 +140,24 @@ describe('Messaging campaign commands', () => { }, ) + test('sends template and mapping changes the API accepts', async () => { + const input = await jsonInput('campaign-update-content.json', { + templateId: null, + variableMapping: {first_name: {source: 'FirstName', fallback: 'there'}}, + }) + server = await startHttpServer({body: {data: campaign(), success: true}}) + const result = await apiKeyEnvironment(() => + runCommand( + `campaign update 0198a2cb-0000-7000-8000-0000000000e3 --project-slug project_1 --input ${input}`, + ), + ) + expect(result.error).toBeUndefined() + expect(JSON.parse(server.requests[0]!.body)).toEqual({ + templateId: null, + variableMapping: {first_name: {source: 'FirstName', fallback: 'there'}}, + }) + }) + test('reports a launched-audience conflict without replaying the PATCH', async () => { const input = await jsonInput('campaign-update-conflict.json', { recipientListId: null,