From 43e29d41bf3c0d34ae58e17207c2eddb07c0a73a Mon Sep 17 00:00:00 2001 From: eleith Date: Tue, 20 Oct 2026 08:42:02 -0700 Subject: [PATCH 1/2] fix(worker): checkpoint email preparation and isolate native delivery retries --- .changeset/appointment-retries.md | 7 + apps/worker/README.md | 28 +- .../appointment-cancelled-by-guest.ts | 9 +- .../builders/appointment-cancelled-by-host.ts | 9 +- .../email/builders/appointment-confirmed.ts | 9 +- .../email/builders/appointment-declined.ts | 9 +- .../builders/appointment-edited-by-host.ts | 9 +- .../src/email/builders/appointment-pending.ts | 9 +- .../appointment-rescheduled-by-guest.ts | 9 +- .../appointment-rescheduled-by-host.ts | 9 +- .../src/email/builders/test-email.test.ts | 1 + apps/worker/src/email/dispatch.test.ts | 123 +++- apps/worker/src/email/dispatch.ts | 19 +- apps/worker/src/email/recipients.test.ts | 44 +- apps/worker/src/email/recipients.ts | 23 +- apps/worker/src/email/status.test.ts | 8 +- apps/worker/src/email/status.ts | 9 +- apps/worker/src/services/job-log.test.ts | 60 ++ apps/worker/src/services/job-log.ts | 16 + .../deliver-appointment-email.test.ts | 158 +++++ .../workflows/deliver-appointment-email.ts | 82 +++ apps/worker/src/workflows/index.ts | 2 + .../reconcile-appointment.integration.test.ts | 631 ++++++++++++++++++ .../workflows/reconcile-appointment.test.ts | 40 +- .../src/workflows/reconcile-appointment.ts | 175 +++-- packages/jobs/src/specs.ts | 2 +- 26 files changed, 1350 insertions(+), 150 deletions(-) create mode 100644 .changeset/appointment-retries.md create mode 100644 apps/worker/src/services/job-log.test.ts create mode 100644 apps/worker/src/workflows/deliver-appointment-email.test.ts create mode 100644 apps/worker/src/workflows/deliver-appointment-email.ts create mode 100644 apps/worker/src/workflows/reconcile-appointment.integration.test.ts diff --git a/.changeset/appointment-retries.md b/.changeset/appointment-retries.md new file mode 100644 index 00000000..9716ac3c --- /dev/null +++ b/.changeset/appointment-retries.md @@ -0,0 +1,7 @@ +--- +'@when/web': patch +'@when/worker': patch +'@when/cli': patch +--- + +Use native retries for appointment email delivery. Guest and host emails run independently with checkpointed content, so retrying one delivery or recording SMTP status does not repeat another saved send. diff --git a/apps/worker/README.md b/apps/worker/README.md index ddffa5af..e2678da2 100644 --- a/apps/worker/README.md +++ b/apps/worker/README.md @@ -55,6 +55,27 @@ Relative paths resolve against the config file's directory, so web and worker - `pnpm start` — `node dist/src/index.js`. - `pnpm check` / `pnpm lint` / `pnpm test` / `pnpm test:coverage`. +## Appointment jobs + +Reconciliation saves prepared email content and recipient roles, then starts one +`deliver-appointment-email` child per recipient. Each child has five native SMTP +attempts (30-second initial delay, doubling up to ten minutes). A successful send +is checkpointed separately from its status write. Host and guest keep independent +retry budgets, even when they share an email address. + +An email `failed` action means a provider attempt failed and retries may remain. +The parent records `done` only when every intended email was sent and the appointment +still exists and is not purged. Ineligible deliveries are skipped. Exhaustion fails +the engine run rather than returning a successful summary of failed delivery. + +Prepared envelopes include personal data and guest capability links: treat queue +backups like application backups. Checkpoints do not guarantee exactly-once SMTP; +a crash after remote acceptance but before saving success can still cause a duplicate. + +Before deploying these changed workflows, stop producers, let the old worker drain +unfinished jobs, then stop it and back up both databases before a grouped restart. +Do not replay unfinished old workflow histories through the changed step graph. + ## Engine integration Engine polling, schedule, heartbeat, and execution diagnostics go through the @@ -66,7 +87,6 @@ escaped handler execution may mean durable suspension rather than a terminal job failure. Email metrics count physical SMTP attempt outcomes, not an exactly-once delivery ledger. -OpenWorkflow `0.11.0` supports persisted cron schedules. This engine-upgrade step -still uses the existing process timers for refresh, expiry, and repair; migrating -those tasks to awaited cron workflows is a separate application change. Likewise, -upgrading the engine alone does not change the current workflow error handling. +OpenWorkflow `0.11.0` supports persisted cron schedules. Refresh, expiry, and repair +still use the existing process timers; migrating those tasks to awaited cron +workflows is a separate application change. diff --git a/apps/worker/src/email/builders/appointment-cancelled-by-guest.ts b/apps/worker/src/email/builders/appointment-cancelled-by-guest.ts index eeff70f0..e60639b3 100644 --- a/apps/worker/src/email/builders/appointment-cancelled-by-guest.ts +++ b/apps/worker/src/email/builders/appointment-cancelled-by-guest.ts @@ -7,7 +7,12 @@ import { whenForHost } from '../format.js'; import { cancelIcs } from '../ics.js'; -import { guestMessage, messages, hostMessage, type EmailMessage } from '../recipients.js'; +import { + guestMessage, + messages, + hostMessage, + type AppointmentEmailMessage +} from '../recipients.js'; import type { EmailContent } from '../content.js'; import type { AppointmentEmailInput } from '../types.js'; @@ -19,7 +24,7 @@ function reasonParagraph(a: Appointment): string[] { return cancel?.payload?.note ? [`Reason: ${cancel.payload.note}`] : []; } -export function appointmentCancelledByGuest(i: AppointmentEmailInput): EmailMessage[] { +export function appointmentCancelledByGuest(i: AppointmentEmailInput): AppointmentEmailMessage[] { const a = i.appointment; const brand = deriveBrand(i.cfg, i.logo?.cid); const eventName = eventTypeName(i.eventType, a); diff --git a/apps/worker/src/email/builders/appointment-cancelled-by-host.ts b/apps/worker/src/email/builders/appointment-cancelled-by-host.ts index a56f2da5..62018a30 100644 --- a/apps/worker/src/email/builders/appointment-cancelled-by-host.ts +++ b/apps/worker/src/email/builders/appointment-cancelled-by-host.ts @@ -7,7 +7,12 @@ import { whenForHost } from '../format.js'; import { cancelIcs } from '../ics.js'; -import { guestMessage, messages, hostMessage, type EmailMessage } from '../recipients.js'; +import { + guestMessage, + messages, + hostMessage, + type AppointmentEmailMessage +} from '../recipients.js'; import type { EmailContent } from '../content.js'; import type { AppointmentEmailInput } from '../types.js'; @@ -19,7 +24,7 @@ function reasonParagraph(a: Appointment): string[] { return cancel?.payload?.note ? [`Reason: ${cancel.payload.note}`] : []; } -export function appointmentCancelledByHost(i: AppointmentEmailInput): EmailMessage[] { +export function appointmentCancelledByHost(i: AppointmentEmailInput): AppointmentEmailMessage[] { const a = i.appointment; const brand = deriveBrand(i.cfg, i.logo?.cid); const eventName = eventTypeName(i.eventType, a); diff --git a/apps/worker/src/email/builders/appointment-confirmed.ts b/apps/worker/src/email/builders/appointment-confirmed.ts index 3aae7b6e..0f7f3c66 100644 --- a/apps/worker/src/email/builders/appointment-confirmed.ts +++ b/apps/worker/src/email/builders/appointment-confirmed.ts @@ -7,11 +7,16 @@ import { whenForHost } from '../format.js'; import { requestIcs } from '../ics.js'; -import { guestMessage, messages, hostMessage, type EmailMessage } from '../recipients.js'; +import { + guestMessage, + messages, + hostMessage, + type AppointmentEmailMessage +} from '../recipients.js'; import type { EmailContent } from '../content.js'; import type { AppointmentEmailInput } from '../types.js'; -export function appointmentConfirmed(i: AppointmentEmailInput): EmailMessage[] { +export function appointmentConfirmed(i: AppointmentEmailInput): AppointmentEmailMessage[] { const a = i.appointment; const brand = deriveBrand(i.cfg, i.logo?.cid); const eventName = eventTypeName(i.eventType, a); diff --git a/apps/worker/src/email/builders/appointment-declined.ts b/apps/worker/src/email/builders/appointment-declined.ts index 683af5a3..0cc79f54 100644 --- a/apps/worker/src/email/builders/appointment-declined.ts +++ b/apps/worker/src/email/builders/appointment-declined.ts @@ -6,11 +6,16 @@ import { whenForGuest, whenForHost } from '../format.js'; -import { guestMessage, messages, hostMessage, type EmailMessage } from '../recipients.js'; +import { + guestMessage, + messages, + hostMessage, + type AppointmentEmailMessage +} from '../recipients.js'; import type { EmailContent } from '../content.js'; import type { AppointmentEmailInput } from '../types.js'; -export function appointmentDeclined(i: AppointmentEmailInput): EmailMessage[] { +export function appointmentDeclined(i: AppointmentEmailInput): AppointmentEmailMessage[] { const a = i.appointment; const brand = deriveBrand(i.cfg, i.logo?.cid); const eventName = eventTypeName(i.eventType, a); diff --git a/apps/worker/src/email/builders/appointment-edited-by-host.ts b/apps/worker/src/email/builders/appointment-edited-by-host.ts index a5026cb5..8b251995 100644 --- a/apps/worker/src/email/builders/appointment-edited-by-host.ts +++ b/apps/worker/src/email/builders/appointment-edited-by-host.ts @@ -7,12 +7,17 @@ import { whenForHost } from '../format.js'; import { requestIcs } from '../ics.js'; -import { guestMessage, messages, hostMessage, type EmailMessage } from '../recipients.js'; +import { + guestMessage, + messages, + hostMessage, + type AppointmentEmailMessage +} from '../recipients.js'; import type { EmailContent } from '../content.js'; import type { AppointmentEmailInput } from '../types.js'; import { parseActionLog } from '@when/db'; -export function appointmentEditedByHost(i: AppointmentEmailInput): EmailMessage[] { +export function appointmentEditedByHost(i: AppointmentEmailInput): AppointmentEmailMessage[] { const a = i.appointment; const brand = deriveBrand(i.cfg, i.logo?.cid); const eventName = eventTypeName(i.eventType, a); diff --git a/apps/worker/src/email/builders/appointment-pending.ts b/apps/worker/src/email/builders/appointment-pending.ts index 7ef66948..5383b7cd 100644 --- a/apps/worker/src/email/builders/appointment-pending.ts +++ b/apps/worker/src/email/builders/appointment-pending.ts @@ -6,11 +6,16 @@ import { whenForGuest, whenForHost } from '../format.js'; -import { guestMessage, messages, hostMessage, type EmailMessage } from '../recipients.js'; +import { + guestMessage, + messages, + hostMessage, + type AppointmentEmailMessage +} from '../recipients.js'; import type { EmailContent } from '../content.js'; import type { AppointmentEmailInput } from '../types.js'; -export function appointmentPending(i: AppointmentEmailInput): EmailMessage[] { +export function appointmentPending(i: AppointmentEmailInput): AppointmentEmailMessage[] { const a = i.appointment; const brand = deriveBrand(i.cfg, i.logo?.cid); const eventName = eventTypeName(i.eventType, a); diff --git a/apps/worker/src/email/builders/appointment-rescheduled-by-guest.ts b/apps/worker/src/email/builders/appointment-rescheduled-by-guest.ts index 2cdb3052..66bafc3f 100644 --- a/apps/worker/src/email/builders/appointment-rescheduled-by-guest.ts +++ b/apps/worker/src/email/builders/appointment-rescheduled-by-guest.ts @@ -7,11 +7,16 @@ import { whenForHost } from '../format.js'; import { requestIcs } from '../ics.js'; -import { guestMessage, messages, hostMessage, type EmailMessage } from '../recipients.js'; +import { + guestMessage, + messages, + hostMessage, + type AppointmentEmailMessage +} from '../recipients.js'; import type { EmailContent } from '../content.js'; import type { AppointmentEmailInput } from '../types.js'; -export function appointmentRescheduledByGuest(i: AppointmentEmailInput): EmailMessage[] { +export function appointmentRescheduledByGuest(i: AppointmentEmailInput): AppointmentEmailMessage[] { const a = i.appointment; const brand = deriveBrand(i.cfg, i.logo?.cid); const eventName = eventTypeName(i.eventType, a); diff --git a/apps/worker/src/email/builders/appointment-rescheduled-by-host.ts b/apps/worker/src/email/builders/appointment-rescheduled-by-host.ts index fddf9fe5..650fb49f 100644 --- a/apps/worker/src/email/builders/appointment-rescheduled-by-host.ts +++ b/apps/worker/src/email/builders/appointment-rescheduled-by-host.ts @@ -7,11 +7,16 @@ import { whenForHost } from '../format.js'; import { requestIcs } from '../ics.js'; -import { guestMessage, messages, hostMessage, type EmailMessage } from '../recipients.js'; +import { + guestMessage, + messages, + hostMessage, + type AppointmentEmailMessage +} from '../recipients.js'; import type { EmailContent } from '../content.js'; import type { AppointmentEmailInput } from '../types.js'; -export function appointmentRescheduledByHost(i: AppointmentEmailInput): EmailMessage[] { +export function appointmentRescheduledByHost(i: AppointmentEmailInput): AppointmentEmailMessage[] { const a = i.appointment; const brand = deriveBrand(i.cfg, i.logo?.cid); const eventName = eventTypeName(i.eventType, a); diff --git a/apps/worker/src/email/builders/test-email.test.ts b/apps/worker/src/email/builders/test-email.test.ts index af0a26cc..6fbee51a 100644 --- a/apps/worker/src/email/builders/test-email.test.ts +++ b/apps/worker/src/email/builders/test-email.test.ts @@ -8,6 +8,7 @@ describe('testEmailMessage', () => { test('addresses the recipient and builds test content', () => { const msg = testEmailMessage(sampleConfig, 'me@example.com', null); expect(msg.to).toBe('me@example.com'); + expect(msg).not.toHaveProperty('role'); expect(msg.content.subject).toContain('Test email'); expect(msg.content.heading).toContain('test email from When'); expect(msg.content.actions).toEqual([ diff --git a/apps/worker/src/email/dispatch.test.ts b/apps/worker/src/email/dispatch.test.ts index 5378301c..369a4b0c 100644 --- a/apps/worker/src/email/dispatch.test.ts +++ b/apps/worker/src/email/dispatch.test.ts @@ -1,7 +1,14 @@ -import { describe, expect, test } from 'vitest'; +import { afterEach, describe, expect, test, vi } from 'vitest'; import type { AppointmentEmailKind } from '@when/jobs'; -import { dispatch } from './dispatch.js'; +import { dispatch, prepareAppointmentEmails } from './dispatch.js'; import { sampleInput } from './__fixtures__/appointment.js'; +import * as logo from './logo.js'; +import type { PreparedAppointmentEmail } from './recipients.js'; + +afterEach(() => { + vi.restoreAllMocks(); + logo.clearLogoCache(); +}); const base = { appointment: sampleInput.appointment, @@ -51,7 +58,8 @@ describe('dispatch', () => { 'cancelled-by-host', 'rescheduled-by-guest', 'rescheduled-by-host', - 'declined' + 'declined', + 'edited-by-host' ]; for (const kind of kinds) { const e = await run(kind); @@ -60,3 +68,112 @@ describe('dispatch', () => { } }); }); + +describe('prepareAppointmentEmails', () => { + test.each([ + 'confirmed', + 'pending', + 'cancelled-by-guest', + 'cancelled-by-host', + 'rescheduled-by-guest', + 'rescheduled-by-host', + 'declined', + 'edited-by-host' + ])('%s preserves builder roles and renders addressed envelopes', async (kind) => { + const prepared = await prepareAppointmentEmails({ kind, ...base }, sampleInput.cfg); + expect(prepared.map(({ role, envelope }) => ({ role, to: envelope.to }))).toEqual([ + { role: 'guest', to: 'jane@example.com' }, + { role: 'host', to: 'owner@acme.test' } + ]); + for (const { envelope } of prepared) { + expect(envelope.subject).toBeTruthy(); + expect(envelope.text).toBeTruthy(); + expect(envelope.html).toBeTruthy(); + expect(envelope).not.toHaveProperty('role'); + } + }); + + test('no guest email prepares only the host', async () => { + const prepared = await prepareAppointmentEmails( + { + kind: 'confirmed', + appointment: { ...sampleInput.appointment, guest_email: null } + }, + sampleInput.cfg + ); + expect(prepared.map(({ role }) => role)).toEqual(['host']); + expect(prepared[0].envelope.to).toBe('owner@acme.test'); + }); + + test('pending edits prepare only the host even with a guest address', async () => { + const prepared = await prepareAppointmentEmails( + { + kind: 'edited-by-host', + appointment: { ...sampleInput.appointment, status: 'pending' } + }, + sampleInput.cfg + ); + expect(prepared.map(({ role }) => role)).toEqual(['host']); + }); + + test('equal addresses prepare two distinct recipient roles', async () => { + const prepared = await prepareAppointmentEmails( + { + kind: 'confirmed', + appointment: { ...sampleInput.appointment, guest_email: sampleInput.cfg.user.email } + }, + sampleInput.cfg + ); + expect(prepared.map(({ role, envelope }) => ({ role, to: envelope.to }))).toEqual([ + { role: 'guest', to: 'owner@acme.test' }, + { role: 'host', to: 'owner@acme.test' } + ]); + expect(prepared[0].envelope.subject).not.toBe(prepared[1].envelope.subject); + }); + + test('prepared envelopes round-trip JSON with string ICS and base64 logo attachments', async () => { + const bytes = Buffer.from([0, 1, 2, 127, 128, 254, 255]); + const fetch = vi + .spyOn(globalThis, 'fetch') + .mockResolvedValue(new Response(bytes, { headers: { 'content-type': 'image/png' } })); + const cfg = { + ...sampleInput.cfg, + user: { + ...sampleInput.cfg.user, + appearance: { ...sampleInput.cfg.user.appearance, app_icon_path: '/public/logo.png' } + } + }; + const prepared = await prepareAppointmentEmails({ kind: 'confirmed', ...base }, cfg); + const restored: PreparedAppointmentEmail[] = JSON.parse(JSON.stringify(prepared)); + expect(restored).toStrictEqual(prepared); + expect(fetch).toHaveBeenCalledTimes(1); + + const guest = restored.find(({ role }) => role === 'guest'); + const ics = guest?.envelope.attachments?.find(({ filename }) => filename === 'invite.ics'); + expect(ics?.content).toContain('BEGIN:VCALENDAR'); + expect(ics?.content).toContain('METHOD:REQUEST'); + expect(ics?.contentType).toBe('text/calendar; charset=utf-8'); + expect(ics?.encoding).toBeUndefined(); + for (const { envelope } of restored) { + const image = envelope.attachments?.find(({ cid }) => cid === logo.BRAND_LOGO_CID); + expect(image).toEqual({ + filename: 'logo.png', + content: bytes.toString('base64'), + contentType: 'image/png', + cid: logo.BRAND_LOGO_CID, + encoding: 'base64' + }); + expect(Buffer.from(image?.content ?? '', 'base64')).toEqual(bytes); + expect(envelope.html).toContain(`cid:${logo.BRAND_LOGO_CID}`); + } + }); + + test('dispatch returns the same envelopes and prepares only once per call', async () => { + const input = { kind: 'pending' as const, ...base }; + const prepared = await prepareAppointmentEmails(input, sampleInput.cfg); + const fetchLogo = vi.spyOn(logo, 'fetchBrandLogo'); + const envelopes = await dispatch(input, sampleInput.cfg); + expect(envelopes).toEqual(prepared.map(({ envelope }) => envelope)); + expect(fetchLogo).toHaveBeenCalledTimes(1); + }); +}); diff --git a/apps/worker/src/email/dispatch.ts b/apps/worker/src/email/dispatch.ts index 09cbb121..9d80c0ef 100644 --- a/apps/worker/src/email/dispatch.ts +++ b/apps/worker/src/email/dispatch.ts @@ -9,7 +9,7 @@ import { appointmentPending } from './builders/appointment-pending.js'; import { appointmentRescheduledByGuest } from './builders/appointment-rescheduled-by-guest.js'; import { appointmentRescheduledByHost } from './builders/appointment-rescheduled-by-host.js'; import { appointmentEditedByHost } from './builders/appointment-edited-by-host.js'; -import type { EmailMessage, Envelope } from './recipients.js'; +import type { AppointmentEmailMessage, Envelope, PreparedAppointmentEmail } from './recipients.js'; import { renderMessage } from './render.js'; import { emailTheme } from './theme.js'; import { appointmentLinks } from '../links.js'; @@ -22,7 +22,7 @@ export interface DispatchInput { appointment: Appointment; } -function build(i: AppointmentEmailInput, kind: AppointmentEmailKind): EmailMessage[] { +function build(i: AppointmentEmailInput, kind: AppointmentEmailKind): AppointmentEmailMessage[] { switch (kind) { case 'confirmed': return appointmentConfirmed(i); @@ -47,7 +47,10 @@ function build(i: AppointmentEmailInput, kind: AppointmentEmailKind): EmailMessa } } -export async function dispatch(input: DispatchInput, cfg: WhenConfiguration): Promise { +export async function prepareAppointmentEmails( + input: DispatchInput, + cfg: WhenConfiguration +): Promise { const eventType = cfg.meetings[input.appointment.event_type_id]; const logo = await fetchBrandLogo(cfg); @@ -65,5 +68,13 @@ export async function dispatch(input: DispatchInput, cfg: WhenConfiguration): Pr }; const theme = emailTheme(cfg.user.appearance); - return build(i, input.kind).map((m) => renderMessage(m, logo, theme)); + return build(i, input.kind).map((m) => ({ + role: m.role, + envelope: renderMessage(m, logo, theme) + })); +} + +export async function dispatch(input: DispatchInput, cfg: WhenConfiguration): Promise { + const emails = await prepareAppointmentEmails(input, cfg); + return emails.map((email) => email.envelope); } diff --git a/apps/worker/src/email/recipients.test.ts b/apps/worker/src/email/recipients.test.ts index acc68d0d..aa8431e5 100644 --- a/apps/worker/src/email/recipients.test.ts +++ b/apps/worker/src/email/recipients.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from 'vitest'; -import { guestMessage, hostMessage } from './recipients.js'; +import { guestMessage, hostMessage, messages } from './recipients.js'; import { sampleInput } from './__fixtures__/appointment.js'; import type { EmailContent } from './content.js'; @@ -18,12 +18,46 @@ const content: EmailContent = { }; describe('recipients', () => { - test('guestMessage addresses the guest', () => { - expect(guestMessage(sampleInput, content)!.to).toBe('jane@example.com'); + test('guestMessage assigns the guest role and address', () => { + expect(guestMessage(sampleInput, content)).toEqual({ + role: 'guest', + to: 'jane@example.com', + content, + ics: undefined + }); }); - test('hostMessage addresses the configured user', () => { - expect(hostMessage(sampleInput, content).to).toBe('owner@acme.test'); + test('hostMessage assigns the host role and configured address', () => { + expect(hostMessage(sampleInput, content)).toEqual({ + role: 'host', + to: 'owner@acme.test', + content + }); + }); + + test('missing guest email produces null and is dropped without changing the host role', () => { + const input = { + ...sampleInput, + appointment: { ...sampleInput.appointment, guest_email: null } + }; + const guest = guestMessage(input, content); + const host = hostMessage(input, content); + expect(guest).toBeNull(); + expect(messages(guest, host)).toEqual([host]); + expect(messages(null)).toEqual([]); + }); + + test('equal addresses keep distinct roles regardless of collection order', () => { + const input = { + ...sampleInput, + appointment: { ...sampleInput.appointment, guest_email: sampleInput.cfg.user.email } + }; + const host = hostMessage(input, content); + const guest = guestMessage(input, content); + expect(messages(host, guest).map(({ role, to }) => ({ role, to }))).toEqual([ + { role: 'host', to: 'owner@acme.test' }, + { role: 'guest', to: 'owner@acme.test' } + ]); }); test('guestMessage carries an ics when given; otherwise none', () => { diff --git a/apps/worker/src/email/recipients.ts b/apps/worker/src/email/recipients.ts index 7d56da18..bf32f278 100644 --- a/apps/worker/src/email/recipients.ts +++ b/apps/worker/src/email/recipients.ts @@ -25,25 +25,36 @@ export interface EmailMessage { ics?: Attachment; } +export type RecipientRole = 'host' | 'guest'; + +export interface AppointmentEmailMessage extends EmailMessage { + role: RecipientRole; +} + +export interface PreparedAppointmentEmail { + role: RecipientRole; + envelope: Envelope; +} + /** Message addressed to the appointment's guest, or null when no email was collected. */ export function guestMessage( i: Pick, content: EmailContent, ics?: Attachment -): EmailMessage | null { +): AppointmentEmailMessage | null { if (!i.appointment.guest_email) return null; - return { to: i.appointment.guest_email, content, ics }; + return { role: 'guest', to: i.appointment.guest_email, content, ics }; } /** Message addressed to the host (the single configured user). */ export function hostMessage( i: Pick, content: EmailContent -): EmailMessage { - return { to: i.cfg.user.email, content }; +): AppointmentEmailMessage { + return { role: 'host', to: i.cfg.user.email, content }; } /** Collect a builder's messages, dropping any that had no recipient (e.g. no guest email). */ -export function messages(...list: (EmailMessage | null)[]): EmailMessage[] { - return list.filter((m): m is EmailMessage => m !== null); +export function messages(...list: (AppointmentEmailMessage | null)[]): AppointmentEmailMessage[] { + return list.filter((m): m is AppointmentEmailMessage => m !== null); } diff --git a/apps/worker/src/email/status.test.ts b/apps/worker/src/email/status.test.ts index 3c8edce2..49b3b64e 100644 --- a/apps/worker/src/email/status.test.ts +++ b/apps/worker/src/email/status.test.ts @@ -23,14 +23,16 @@ test('a sent email records smtp as working', async () => { } }); -test('a failed send records the reason', async () => { +test('a failed send records the reason using the supplied clock', async () => { const ctx = await freshCtx(); try { - await recordSendOutcome(ctx, { ok: false, reason: 'EAUTH' }); + const at = '2026-02-01T12:00:00.000Z'; + await recordSendOutcome(ctx, { ok: false, reason: 'EAUTH' }, { now: () => new Date(at) }); const [smtp] = await listServiceStatus(ctx.db, 'smtp'); expect(smtp.error).toBe('EAUTH'); - expect(smtp.failing_since).toBeTruthy(); + expect(smtp.last_attempt_at).toBe(at); + expect(smtp.failing_since).toBe(at); expect(smtp.last_ok_at).toBeNull(); } finally { await ctx.db.destroy(); diff --git a/apps/worker/src/email/status.ts b/apps/worker/src/email/status.ts index 10325fe6..d7d75079 100644 --- a/apps/worker/src/email/status.ts +++ b/apps/worker/src/email/status.ts @@ -1,13 +1,18 @@ import { recordServiceOutcome } from '@when/db'; import type { WorkerContext } from '../services/context.js'; import type { SendResult } from './smtp.js'; +import { systemClock, type Clock } from './clock.js'; -export async function recordSendOutcome(ctx: WorkerContext, result: SendResult): Promise { +export async function recordSendOutcome( + ctx: WorkerContext, + result: SendResult, + clock: Pick = systemClock +): Promise { await recordServiceOutcome( ctx.db, { kind: 'smtp' }, { - at: Temporal.Now.instant().toString(), + at: clock.now().toISOString(), via: 'send', error: result.ok ? null : result.reason } diff --git a/apps/worker/src/services/job-log.test.ts b/apps/worker/src/services/job-log.test.ts new file mode 100644 index 00000000..864b89cd --- /dev/null +++ b/apps/worker/src/services/job-log.test.ts @@ -0,0 +1,60 @@ +import { afterEach, beforeEach, describe, expect, test } from 'vitest'; +import { openDb, parseActionLog, runMigrations } from '@when/db'; +import { sampleAppointment } from '../email/__fixtures__/appointment.js'; +import { appendEmailJobLog, appendJobLog } from './job-log.js'; + +const at = '2026-02-01T12:00:00.000Z'; +let db: ReturnType; +beforeEach(async () => { + db = openDb(':memory:'); + await runMigrations(db); + await db.insertInto('appointments').values(sampleAppointment).execute(); +}); +afterEach(async () => { + await db.destroy(); +}); + +async function entries() { + const row = await db.selectFrom('appointments').select('action_log').executeTakeFirstOrThrow(); + return parseActionLog(row.action_log); +} + +describe('conditional email action log', () => { + test('appends failures and recovery with the supplied timestamp and appointment id', async () => { + await appendEmailJobLog(db, sampleAppointment.id, 'failed', at); + await appendEmailJobLog(db, sampleAppointment.id, 'done', at); + expect(await entries()).toEqual( + ['failed', 'done'].map((state) => + expect.objectContaining({ + action: 'email', + at, + payload: expect.objectContaining({ + metadata: { state, appointment_id: sampleAppointment.id } + }) + }) + ) + ); + }); + + test.each(['failed', 'done'] as const)( + 'does not append %s after the row is purged', + async (state) => { + await appendJobLog(db, sampleAppointment.id, 'calendar', 'done', at); + const before = await entries(); + await db.updateTable('appointments').set({ status: 'purged' }).execute(); + await appendEmailJobLog(db, sampleAppointment.id, state, at); + expect(await entries()).toEqual(before); + } + ); + + test('a missing appointment is a no-op', async () => { + await expect(appendEmailJobLog(db, 'missing', 'done', at)).resolves.toBeUndefined(); + expect(await entries()).toEqual([]); + }); + + test('does not change the general job logger on purged rows', async () => { + await db.updateTable('appointments').set({ status: 'purged' }).execute(); + await appendJobLog(db, sampleAppointment.id, 'calendar', 'failed', at); + expect(await entries()).toEqual([expect.objectContaining({ action: 'calendar' })]); + }); +}); diff --git a/apps/worker/src/services/job-log.ts b/apps/worker/src/services/job-log.ts index 6726c932..cabe18ef 100644 --- a/apps/worker/src/services/job-log.ts +++ b/apps/worker/src/services/job-log.ts @@ -22,6 +22,22 @@ export async function appendJobLog( .execute(); } +export async function appendEmailJobLog( + db: Kysely, + appointmentId: string, + state: JobState, + at: string +): Promise { + await db + .updateTable('appointments') + .set({ + action_log: appendJobLogSql({ kind: 'email', at, state, appointment_id: appointmentId }) + }) + .where('id', '=', appointmentId) + .where('status', '!=', 'purged') + .execute(); +} + // The `at` of an unclosed calendar `failed` (no later `done`) for this row, or // null. Scoped to appointment_id: the log is inherited across a reschedule chain. // This timestamp is the log's "failing since". diff --git a/apps/worker/src/workflows/deliver-appointment-email.test.ts b/apps/worker/src/workflows/deliver-appointment-email.test.ts new file mode 100644 index 00000000..68f6c040 --- /dev/null +++ b/apps/worker/src/workflows/deliver-appointment-email.test.ts @@ -0,0 +1,158 @@ +import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'; +import { openDb, parseActionLog, runMigrations } from '@when/db'; +import type { RetryPolicy } from 'openworkflow'; +import { sampleAppointment, sampleConfig } from '../email/__fixtures__/appointment.js'; +import type { Envelope } from '../email/recipients.js'; +import type { SendResult } from '../email/smtp.js'; +import * as status from '../email/status.js'; +import { setWorkerContext } from '../services/context.js'; +import { createLogger } from '../services/logger.js'; +import { emailsTotal } from '../services/metrics.js'; +import { + deliverAppointmentEmail, + runDeliverAppointmentEmail +} from './deliver-appointment-email.js'; + +const at = '2026-02-01T12:00:00.000Z'; +const clock = { now: () => new Date(at) }; +const envelope: Envelope = { to: 'jane@example.com', subject: 'Prepared', text: 'Saved content' }; +const input = { + appointmentId: sampleAppointment.id, + kind: 'confirmed', + role: 'guest', + envelope +} as const; + +type StepConfig = { name: string; retryPolicy?: Partial }; +function directStep(before?: (config: StepConfig) => Promise) { + const configs: StepConfig[] = []; + return { + configs, + async run(config: StepConfig, fn: () => Promise | T): Promise { + configs.push(config); + await before?.(config); + return fn(); + } + }; +} + +let db: ReturnType; +let send: ReturnType Promise>>; +beforeEach(async () => { + db = openDb(':memory:'); + await runMigrations(db); + await db.insertInto('appointments').values(sampleAppointment).execute(); + send = vi.fn(async () => ({ ok: true })); + setWorkerContext({ db, config: sampleConfig, logger: createLogger(), mailer: { send } }); +}); +afterEach(async () => { + vi.restoreAllMocks(); + await db.destroy(); +}); + +async function emailEntries() { + const row = await db.selectFrom('appointments').select('action_log').executeTakeFirstOrThrow(); + return parseActionLog(row.action_log).filter((entry) => entry.action === 'email'); +} + +describe('delivery attempt boundaries', () => { + test('uses the internal spec and five native SMTP attempts, then a separate success checkpoint', async () => { + const step = directStep(); + expect(deliverAppointmentEmail.name).toBe('deliver-appointment-email'); + await expect(runDeliverAppointmentEmail(input, step, clock)).resolves.toBe('sent'); + expect(step.configs).toEqual([ + { + name: 'smtp', + retryPolicy: { + maximumAttempts: 5, + initialInterval: '30s', + backoffCoefficient: 2, + maximumInterval: '10m' + } + }, + { name: 'smtp:status' } + ]); + expect(send).toHaveBeenCalledExactlyOnceWith(envelope); + expect(await emailEntries()).toEqual([]); + expect( + await db.selectFrom('service_status').selectAll().executeTakeFirstOrThrow() + ).toMatchObject({ + kind: 'smtp', + error: null, + via: 'send', + last_attempt_at: at, + last_ok_at: at + }); + }); + + test('records an observed failure inside the callback before throwing to the engine', async () => { + send.mockResolvedValue({ ok: false, reason: 'SMTP unavailable' }); + const increment = vi.spyOn(emailsTotal, 'inc'); + const step = directStep(); + await expect(runDeliverAppointmentEmail(input, step, clock)).rejects.toThrow( + 'SMTP unavailable' + ); + expect(step.configs.map((config) => config.name)).toEqual(['smtp']); + expect(increment).toHaveBeenCalledWith({ + recipient_type: 'guest', + email_kind: 'confirmed', + status: 'failure' + }); + expect(await emailEntries()).toEqual([ + expect.objectContaining({ + at, + payload: expect.objectContaining({ + metadata: expect.objectContaining({ + state: 'failed', + appointment_id: input.appointmentId + }) + }) + }) + ]); + expect( + await db + .selectFrom('service_status') + .select(['error', 'last_attempt_at', 'failing_since']) + .executeTakeFirstOrThrow() + ).toEqual({ error: 'SMTP unavailable', last_attempt_at: at, failing_since: at }); + }); + + test.each(['missing', 'purged'] as const)( + 'checks live %s eligibility inside the SMTP callback', + async (state) => { + const step = directStep(async ({ name }) => { + if (name !== 'smtp') return; + if (state === 'missing') await db.deleteFrom('appointments').execute(); + else await db.updateTable('appointments').set({ status: 'purged' }).execute(); + }); + await expect(runDeliverAppointmentEmail(input, step, clock)).resolves.toBe('skipped'); + expect(send).not.toHaveBeenCalled(); + expect(await db.selectFrom('service_status').selectAll().execute()).toEqual([]); + if (state === 'purged') expect(await emailEntries()).toEqual([]); + } + ); + + test('a success-status error escapes without a provider-failure log or metric', async () => { + const fault = new Error('local status write failed'); + vi.spyOn(status, 'recordSendOutcome').mockRejectedValueOnce(fault); + const increment = vi.spyOn(emailsTotal, 'inc'); + const step = directStep(); + await expect(runDeliverAppointmentEmail(input, step, clock)).rejects.toBe(fault); + expect(step.configs.map((config) => config.name)).toEqual(['smtp', 'smtp:status']); + expect(send).toHaveBeenCalledTimes(1); + expect(await emailEntries()).toEqual([]); + expect(increment).not.toHaveBeenCalledWith(expect.objectContaining({ status: 'failure' })); + }); + + test('does not catch opaque errors from a durable call', async () => { + const opaque = { marker: 'engine control flow' }; + const step = { + run: vi.fn(async () => { + throw opaque; + }) + }; + await expect(runDeliverAppointmentEmail(input, step, clock)).rejects.toBe(opaque); + expect(send).not.toHaveBeenCalled(); + expect(await emailEntries()).toEqual([]); + }); +}); diff --git a/apps/worker/src/workflows/deliver-appointment-email.ts b/apps/worker/src/workflows/deliver-appointment-email.ts new file mode 100644 index 00000000..753cc1ae --- /dev/null +++ b/apps/worker/src/workflows/deliver-appointment-email.ts @@ -0,0 +1,82 @@ +import { defineWorkflowSpec, type RetryPolicy } from 'openworkflow'; +import type { AppointmentEmailKind, WorkflowSpec } from '@when/jobs'; +import type { Envelope, RecipientRole } from '../email/recipients.js'; +import { systemClock, type Clock } from '../email/clock.js'; +import { recordSendOutcome } from '../email/status.js'; +import { getWorkerContext } from '../services/context.js'; +import { appendEmailJobLog } from '../services/job-log.js'; +import { implementObservedWorkflow, emailsTotal } from '../services/metrics.js'; + +export interface DeliverAppointmentEmailInput { + appointmentId: string; + kind: AppointmentEmailKind; + role: RecipientRole; + envelope: Envelope; +} + +export type DeliverAppointmentEmailResult = 'sent' | 'skipped'; + +export const deliverAppointmentEmail: WorkflowSpec< + DeliverAppointmentEmailInput, + DeliverAppointmentEmailResult +> = defineWorkflowSpec({ name: 'deliver-appointment-email' }); + +interface DeliveryStep { + run( + config: { name: string; retryPolicy?: Partial }, + fn: () => Promise | T + ): Promise; +} + +export async function runDeliverAppointmentEmail( + input: DeliverAppointmentEmailInput, + step: DeliveryStep, + clock: Pick = systemClock +): Promise { + const ctx = getWorkerContext(); + const outcome = await step.run( + { + name: 'smtp', + retryPolicy: { + maximumAttempts: 5, + initialInterval: '30s', + backoffCoefficient: 2, + maximumInterval: '10m' + } + }, + async () => { + const row = await ctx.db + .selectFrom('appointments') + .select('id') + .where('id', '=', input.appointmentId) + .where('status', '!=', 'purged') + .executeTakeFirst(); + if (!row) return 'skipped' as const; + + const result = await ctx.mailer.send(input.envelope); + const deliveryStatus = result.ok ? 'success' : 'failure'; + emailsTotal.inc({ + recipient_type: input.role, + email_kind: input.kind, + status: deliveryStatus + }); + if (!result.ok) { + await recordSendOutcome(ctx, result, clock); + await appendEmailJobLog(ctx.db, input.appointmentId, 'failed', clock.now().toISOString()); + throw new Error(result.reason); + } + return 'sent' as const; + } + ); + + if (outcome === 'sent') { + await step.run({ name: 'smtp:status' }, () => recordSendOutcome(ctx, { ok: true }, clock)); + } + return outcome; +} + +export function registerDeliverAppointmentEmailWorkflow(): void { + implementObservedWorkflow(deliverAppointmentEmail, ({ input, step }) => + runDeliverAppointmentEmail(input, step) + ); +} diff --git a/apps/worker/src/workflows/index.ts b/apps/worker/src/workflows/index.ts index 616d1d35..ae096b4c 100644 --- a/apps/worker/src/workflows/index.ts +++ b/apps/worker/src/workflows/index.ts @@ -1,4 +1,5 @@ import { registerReconcileAppointmentWorkflow } from './reconcile-appointment.js'; +import { registerDeliverAppointmentEmailWorkflow } from './deliver-appointment-email.js'; import { registerPurgeAppointmentWorkflow } from './purge-appointment.js'; import { registerTestEmailWorkflow } from './test-email.js'; import { registerProbeProviderWorkflows } from './probe-provider.js'; @@ -10,6 +11,7 @@ import { registerGenerateVideoChatWorkflow } from './generate-video-chat.js'; * once at boot, after `initOpenWorkflow`. */ export function registerWorkflows(): void { + registerDeliverAppointmentEmailWorkflow(); registerReconcileAppointmentWorkflow(); registerPurgeAppointmentWorkflow(); registerTestEmailWorkflow(); diff --git a/apps/worker/src/workflows/reconcile-appointment.integration.test.ts b/apps/worker/src/workflows/reconcile-appointment.integration.test.ts new file mode 100644 index 00000000..068c0298 --- /dev/null +++ b/apps/worker/src/workflows/reconcile-appointment.integration.test.ts @@ -0,0 +1,631 @@ +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'; +import { OpenWorkflow, type Worker } from 'openworkflow'; +import { BackendSqlite } from 'openworkflow/sqlite'; +import { reconcileAppointment, type ReconcileAppointmentInput } from '@when/jobs'; +import { openDb, parseActionLog, runMigrations } from '@when/db'; +import type { WhenConfiguration } from '@when/config'; +import { sampleAppointment, sampleConfig } from '../email/__fixtures__/appointment.js'; +import type { Envelope, RecipientRole } from '../email/recipients.js'; +import type { SendResult } from '../email/smtp.js'; +import { clearLogoCache } from '../email/logo.js'; +import * as status from '../email/status.js'; +import { setWorkerContext, type WorkerContext } from '../services/context.js'; +import { createLogger } from '../services/logger.js'; +import { runReconcileAppointment } from './reconcile-appointment.js'; +import { + deliverAppointmentEmail, + runDeliverAppointmentEmail +} from './deliver-appointment-email.js'; + +type ParentStep = Parameters[1]; +type DeliveryStep = Parameters[1]; +const at = '2026-02-01T12:00:00.000Z'; +const clock = { now: () => new Date(at) }; +const input: ReconcileAppointmentInput = { + appointmentId: sampleAppointment.id, + emailKind: 'pending' +}; +const roles: RecipientRole[] = ['guest', 'host']; + +function openQueue(path = ':memory:', beforeParentStep?: (name: string) => Promise) { + const backend = BackendSqlite.connect(path); + const client = new OpenWorkflow({ backend }); + client.implementWorkflow(deliverAppointmentEmail, ({ input, step }) => { + const adapted: DeliveryStep = { + run: (config, fn) => + step.run( + { + ...config, + retryPolicy: { ...config.retryPolicy, initialInterval: '1ms', maximumInterval: '1ms' } + }, + fn + ) + }; + return runDeliverAppointmentEmail(input, adapted, clock); + }); + client.implementWorkflow(reconcileAppointment, ({ input, step }) => { + const adapted: ParentStep = { + ...step, + run: (config, fn) => + step.run( + { + ...config, + retryPolicy: { initialInterval: '1ms', maximumInterval: '1ms' } + }, + async () => { + await beforeParentStep?.(config.name); + return fn(); + } + ), + runWorkflow: (...args) => step.runWorkflow(...args) + }; + return runReconcileAppointment(input, adapted, clock); + }); + const worker = client.newWorker({ concurrency: 1 }); + return { + backend, + client, + worker, + async close() { + try { + await worker.stop(); + } finally { + await backend.stop(); + } + } + }; +} + +type Queue = ReturnType; +async function runs(queue: Queue) { + const page = await queue.backend.listWorkflowRuns({ limit: 100 }); + expect(page.pagination.next).toBeNull(); + return page.data; +} +async function attempts(queue: Queue, workflowRunId: string, stepName?: string) { + const page = await queue.backend.listStepAttempts({ workflowRunId, limit: 100 }); + expect(page.pagination.next).toBeNull(); + return page.data.filter((attempt) => !stepName || attempt.stepName === stepName); +} +async function driveUntil(worker: Worker, assertion: () => Promise) { + await vi.waitFor( + async () => { + await worker.tick(); + await worker.stop(); + await assertion(); + }, + { timeout: 5000, interval: 5 } + ); +} +async function children(queue: Queue) { + const workflowRuns = await runs(queue); + return workflowRuns.filter((run) => run.workflowName === deliverAppointmentEmail.name); +} +async function startParentAndExpectChildren(queue: Queue, count = 2) { + expect(await queue.worker.tick()).toBe(1); + await queue.worker.stop(); + expect(await children(queue)).toHaveLength(count); +} +async function finish(queue: Queue, parentId: string, state = 'completed') { + await driveUntil(queue.worker, async () => { + expect(await queue.backend.getWorkflowRun({ workflowRunId: parentId })).toMatchObject({ + status: state + }); + }); +} + +let db: ReturnType; +let config: WhenConfiguration; +let ctx: WorkerContext; +let send: ReturnType Promise>>; +beforeEach(async () => { + db = openDb(':memory:'); + await runMigrations(db); + await db + .insertInto('appointments') + .values({ ...sampleAppointment, status: 'pending' }) + .execute(); + config = structuredClone(sampleConfig); + config.providers = {}; + config.meetings = { + '30-min': { title: 'Original meeting', booking_calendar: 'unconfigured', schedule: 'standard' } + } as unknown as WhenConfiguration['meetings']; + send = vi.fn(async () => ({ ok: true })); + ctx = { db, config, logger: createLogger(), mailer: { send } }; + setWorkerContext(ctx); +}); +afterEach(async () => { + vi.restoreAllMocks(); + clearLogoCache(); + await db.destroy(); +}); +async function emailStates() { + const row = await db.selectFrom('appointments').select('action_log').executeTakeFirst(); + return parseActionLog(row?.action_log ?? null) + .filter((entry) => entry.action === 'email') + .map((entry) => entry.payload?.metadata?.state); +} +function failingMailer(failures: Record) { + const counts = { guest: 0, host: 0 }; + send.mockImplementation(async (envelope) => { + const role = envelope.to === config.user.email ? 'host' : 'guest'; + counts[role] += 1; + return counts[role] <= failures[role] + ? { ok: false, reason: `SMTP ${role} unavailable` } + : { ok: true }; + }); + return counts; +} + +describe('appointment email reconciliation with native children', () => { + test('retries transient SMTP failures and logs done only after both roles recover', async () => { + const counts = failingMailer({ guest: 2, host: 2 }); + const queue = openQueue(); + try { + const parent = await queue.client.runWorkflow(reconcileAppointment, input); + await startParentAndExpectChildren(queue); + await finish(queue, parent.workflowRun.id); + expect(await parent.result({ timeoutMs: 500 })).toBe('reconciled'); + expect(counts).toEqual({ guest: 3, host: 3 }); + expect(await emailStates()).toEqual(['failed', 'failed', 'failed', 'failed', 'done']); + expect( + await db + .selectFrom('service_status') + .select(['error', 'via']) + .where('kind', '=', 'smtp') + .executeTakeFirstOrThrow() + ).toEqual({ error: null, via: 'send' }); + for (const child of await children(queue)) { + expect((await attempts(queue, child.id, 'smtp')).map((attempt) => attempt.status)).toEqual([ + 'failed', + 'failed', + 'completed' + ]); + } + expect(await runs(queue)).toHaveLength(3); + } finally { + await queue.close(); + } + }); + + test.each(roles)( + '%s exhausts five attempts without taking its sibling budget', + async (failedRole) => { + const sibling = failedRole === 'guest' ? 'host' : 'guest'; + const counts = failingMailer({ [failedRole]: 5, [sibling]: 4 } as Record< + RecipientRole, + number + >); + const queue = openQueue(); + try { + const parent = await queue.client.runWorkflow(reconcileAppointment, input); + await startParentAndExpectChildren(queue); + const links = await attempts(queue, parent.workflowRun.id); + expect( + links + .filter((attempt) => attempt.childWorkflowRunId) + .map((attempt) => attempt.stepName) + .sort() + ).toEqual(['email:guest', 'email:host']); + await finish(queue, parent.workflowRun.id, 'failed'); + await expect(parent.result({ timeoutMs: 500 })).rejects.toThrow( + `SMTP ${failedRole} unavailable` + ); + expect(counts).toEqual({ guest: 5, host: 5 }); + expect(await emailStates()).toEqual(Array(9).fill('failed')); + for (const child of await children(queue)) { + const role = (child.input as { role: RecipientRole }).role; + expect(child.status).toBe(role === failedRole ? 'failed' : 'completed'); + expect(await attempts(queue, child.id, 'smtp')).toHaveLength(5); + } + } finally { + await queue.close(); + } + } + ); + + test('both recipients exhaust independent five-attempt policies with no terminal summary', async () => { + const counts = failingMailer({ guest: 5, host: 5 }); + const queue = openQueue(); + try { + const parent = await queue.client.runWorkflow(reconcileAppointment, input); + await finish(queue, parent.workflowRun.id, 'failed'); + expect(counts).toEqual({ guest: 5, host: 5 }); + expect((await children(queue)).map((child) => child.status)).toEqual(['failed', 'failed']); + expect(await emailStates()).toEqual(Array(10).fill('failed')); + expect( + await db + .selectFrom('service_status') + .select('error') + .where('kind', '=', 'smtp') + .executeTakeFirstOrThrow() + ).toEqual({ error: expect.stringMatching(/^SMTP (guest|host) unavailable$/) }); + } finally { + await queue.close(); + } + }); + + test('equal addresses still produce two role-named children', async () => { + await db.updateTable('appointments').set({ guest_email: config.user.email }).execute(); + const queue = openQueue(); + try { + const parent = await queue.client.runWorkflow(reconcileAppointment, input); + await finish(queue, parent.workflowRun.id); + expect(send).toHaveBeenCalledTimes(2); + expect(send.mock.calls.map(([envelope]) => envelope.to)).toEqual([ + config.user.email, + config.user.email + ]); + expect( + (await children(queue)).map((child) => (child.input as { role: RecipientRole }).role).sort() + ).toEqual(['guest', 'host']); + expect(await emailStates()).toEqual(['done']); + } finally { + await queue.close(); + } + }); + + test('a booking without guest email starts only the host child', async () => { + await db.updateTable('appointments').set({ guest_email: null }).execute(); + const queue = openQueue(); + try { + const parent = await queue.client.runWorkflow(reconcileAppointment, input); + await finish(queue, parent.workflowRun.id); + expect(send).toHaveBeenCalledTimes(1); + expect((await children(queue)).map((child) => child.input)).toEqual([ + expect.objectContaining({ role: 'host' }) + ]); + expect(await emailStates()).toEqual(['done']); + } finally { + await queue.close(); + } + }); + + test.each(['missing', 'purged', 'no-email'] as const)( + '%s reconciliation sends nothing and has no email summary', + async (state) => { + if (state === 'missing') await db.deleteFrom('appointments').execute(); + if (state === 'purged') + await db.updateTable('appointments').set({ status: 'purged' }).execute(); + const queue = openQueue(); + try { + const parent = await queue.client.runWorkflow( + reconcileAppointment, + state === 'no-email' ? { appointmentId: input.appointmentId } : input + ); + await finish(queue, parent.workflowRun.id); + expect(await parent.result({ timeoutMs: 500 })).toBe('reconciled'); + expect(send).not.toHaveBeenCalled(); + expect(await children(queue)).toEqual([]); + expect(await emailStates()).toEqual([]); + } finally { + await queue.close(); + } + } + ); + + test.each(['missing', 'purged'] as const)( + 'prepared children skip a now-%s row without SMTP or done', + async (state) => { + const queue = openQueue(); + try { + const parent = await queue.client.runWorkflow(reconcileAppointment, input); + await startParentAndExpectChildren(queue); + if (state === 'missing') await db.deleteFrom('appointments').execute(); + else await db.updateTable('appointments').set({ status: 'purged' }).execute(); + await finish(queue, parent.workflowRun.id); + expect(send).not.toHaveBeenCalled(); + expect((await children(queue)).map((child) => child.output)).toEqual([ + 'skipped', + 'skipped' + ]); + expect(await emailStates()).toEqual([]); + expect(await db.selectFrom('service_status').selectAll().execute()).toEqual([]); + } finally { + await queue.close(); + } + } + ); + + test('a mixed sent/skipped set has no done summary', async () => { + send.mockImplementation(async () => { + await db.updateTable('appointments').set({ status: 'purged' }).execute(); + return { ok: true }; + }); + const queue = openQueue(); + try { + const parent = await queue.client.runWorkflow(reconcileAppointment, input); + await finish(queue, parent.workflowRun.id); + expect(send).toHaveBeenCalledTimes(1); + expect((await children(queue)).map((child) => child.output).sort()).toEqual([ + 'sent', + 'skipped' + ]); + expect(await emailStates()).toEqual([]); + } finally { + await queue.close(); + } + }); + + test.each(['cleanup-video-chat', 'resolve-video-chat', 'sync-calendar'])( + 'rechecks live eligibility inside %s after initial preparation', + async (blockedStep) => { + config.providers = { + cloud: { + type: 'nextcloud', + url: 'https://cloud.example.com', + username: 'user', + password: 'secret', + calendars: {} + } + }; + config.meetings['30-min'].video_chat = { provider: 'cloud', attach: { auto: true } }; + const external = vi.spyOn(globalThis, 'fetch').mockResolvedValue( + new Response(JSON.stringify({ ocs: { data: { token: 'created-room' } } }), { + status: 201, + headers: { 'content-type': 'application/json' } + }) + ); + const queue = openQueue(':memory:', async (name) => { + if (name === blockedStep) + await db.updateTable('appointments').set({ status: 'purged' }).execute(); + }); + try { + const cleanupVideoChatUrl = + blockedStep === 'cleanup-video-chat' + ? 'https://cloud.example.com/call/old-room' + : undefined; + const parent = await queue.client.runWorkflow(reconcileAppointment, { + ...input, + cleanupVideoChatUrl + }); + await finish(queue, parent.workflowRun.id); + expect(external).toHaveBeenCalledTimes(blockedStep === 'sync-calendar' ? 1 : 0); + expect(send).not.toHaveBeenCalled(); + expect( + await db + .selectFrom('appointments') + .select('calendar_synced_revision') + .executeTakeFirstOrThrow() + ).toEqual({ calendar_synced_revision: null }); + expect(await emailStates()).toEqual([]); + } finally { + await queue.close(); + } + } + ); + + test('prepared envelopes and role names survive row/config/logo changes and a queue restart', async () => { + const directory = await mkdtemp(join(tmpdir(), 'when-reconcile-')); + config.user.appearance.app_icon_path = '/public/original-logo.png'; + const external = vi + .spyOn(globalThis, 'fetch') + .mockResolvedValue( + new Response('original image', { headers: { 'content-type': 'image/png' } }) + ); + let queue = openQueue(join(directory, 'queue.sqlite')); + try { + const parent = await queue.client.runWorkflow(reconcileAppointment, { + ...input, + emailKind: 'confirmed' + }); + await startParentAndExpectChildren(queue); + const savedChildren = await children(queue); + const prepared = savedChildren.map( + (child) => child.input as unknown as { role: RecipientRole; envelope: Envelope } + ); + expect(prepared.map((item) => item.role).sort()).toEqual(['guest', 'host']); + expect(prepared.find((item) => item.role === 'guest')?.envelope.attachments).toEqual( + expect.arrayContaining([ + expect.objectContaining({ content: expect.stringContaining('BEGIN:VCALENDAR') }) + ]) + ); + expect( + prepared.every((item) => + item.envelope.attachments?.some( + (attachment) => attachment.content === Buffer.from('original image').toString('base64') + ) + ) + ).toBe(true); + await queue.close(); + await db + .updateTable('appointments') + .set({ + guest_email: 'changed@example.com', + guest_name: 'Changed guest', + start_time: '2026-03-02T15:00:00Z', + cancel_token: 'changed-token', + video_chat: 'https://changed.example.com/call' + }) + .execute(); + config.user.email = 'new-owner@example.com'; + config.user.name = 'Changed host'; + config.user.appearance.app_icon_path = '/public/changed-logo.png'; + config.meetings['30-min'].title = 'Changed meeting'; + external.mockResolvedValue( + new Response('changed image', { headers: { 'content-type': 'image/png' } }) + ); + queue = openQueue(join(directory, 'queue.sqlite')); + await finish(queue, parent.workflowRun.id); + expect(send.mock.calls.map(([envelope]) => envelope)).toEqual( + expect.arrayContaining(prepared.map((item) => item.envelope)) + ); + expect(send).toHaveBeenCalledTimes(2); + expect((await children(queue)).map((child) => child.id).sort()).toEqual( + savedChildren.map((child) => child.id).sort() + ); + expect(external).toHaveBeenCalledTimes(1); + expect(await attempts(queue, parent.workflowRun.id, 'prepare-emails')).toHaveLength(1); + expect(await emailStates()).toEqual(['done']); + } finally { + try { + await queue.close(); + } finally { + await rm(directory, { recursive: true, force: true }); + } + } + }); + + test.each([true, false])( + 'the initial optional-video decision (%s) survives a retry and restart', + async (resolveVideo) => { + const directory = await mkdtemp(join(tmpdir(), 'when-video-branch-')); + config.providers = { + cloud: { + type: 'nextcloud', + url: 'https://cloud.example.com', + username: 'user', + password: 'secret', + calendars: {} + } + }; + config.meetings['30-min'].video_chat = { provider: 'cloud', attach: { auto: true } }; + if (!resolveVideo) + await db + .updateTable('appointments') + .set({ video_chat: 'https://cloud.example.com/call/original' }) + .execute(); + const external = vi.spyOn(globalThis, 'fetch').mockResolvedValue( + new Response(JSON.stringify({ ocs: { data: { token: 'created-room' } } }), { + status: 201, + headers: { 'content-type': 'application/json' } + }) + ); + let failSync = true; + let queue = openQueue(join(directory, 'queue.sqlite'), async (name) => { + if (name === 'sync-calendar' && failSync) { + failSync = false; + throw new Error('local calendar callback interrupted'); + } + }); + try { + const parent = await queue.client.runWorkflow(reconcileAppointment, input); + expect(await queue.worker.tick()).toBe(1); + await queue.worker.stop(); + expect( + await queue.backend.getWorkflowRun({ workflowRunId: parent.workflowRun.id }) + ).toMatchObject({ status: 'pending', workerId: null }); + const savedVideo = await attempts(queue, parent.workflowRun.id, 'resolve-video-chat'); + expect(savedVideo).toHaveLength(resolveVideo ? 1 : 0); + await queue.close(); + const changedVideoChatUrl = resolveVideo ? 'https://changed.example.com/call' : null; + await db.updateTable('appointments').set({ video_chat: changedVideoChatUrl }).execute(); + queue = openQueue(join(directory, 'queue.sqlite')); + await finish(queue, parent.workflowRun.id); + expect(await attempts(queue, parent.workflowRun.id, 'resolve-video-chat')).toEqual( + savedVideo + ); + expect(await attempts(queue, parent.workflowRun.id, 'prepare-reconciliation')).toHaveLength( + 1 + ); + expect(external).toHaveBeenCalledTimes(resolveVideo ? 1 : 0); + expect( + (await attempts(queue, parent.workflowRun.id, 'sync-calendar')).map( + (attempt) => attempt.status + ) + ).toEqual(['failed', 'completed']); + expect(await emailStates()).toEqual(['done']); + } finally { + try { + await queue.close(); + } finally { + await rm(directory, { recursive: true, force: true }); + } + } + } + ); + + test('a native SMTP retry rechecks eligibility after a queue restart', async () => { + const directory = await mkdtemp(join(tmpdir(), 'when-email-eligibility-')); + send.mockResolvedValue({ ok: false, reason: 'temporary SMTP failure' }); + let queue = openQueue(join(directory, 'queue.sqlite')); + try { + const parent = await queue.client.runWorkflow(reconcileAppointment, input); + await startParentAndExpectChildren(queue); + expect(await queue.worker.tick()).toBe(1); + await queue.worker.stop(); + expect(send).toHaveBeenCalledTimes(1); + expect(await emailStates()).toEqual(['failed']); + const failedChild = (await children(queue)).find( + (child) => child.status === 'pending' && child.error !== null + ); + expect(failedChild).toBeDefined(); + await queue.close(); + await db.updateTable('appointments').set({ status: 'purged' }).execute(); + queue = openQueue(join(directory, 'queue.sqlite')); + await finish(queue, parent.workflowRun.id); + expect(send).toHaveBeenCalledTimes(1); + expect((await children(queue)).map((child) => child.output)).toEqual(['skipped', 'skipped']); + expect(await emailStates()).toEqual(['failed']); + expect( + (await attempts(queue, failedChild!.id, 'smtp')).map((attempt) => attempt.status) + ).toEqual(['failed', 'completed']); + } finally { + try { + await queue.close(); + } finally { + await rm(directory, { recursive: true, force: true }); + } + } + }); + + test('the final done callback checks eligibility even when both deliveries sent', async () => { + const queue = openQueue(':memory:', async (name) => { + if (name === 'log:result') + await db.updateTable('appointments').set({ status: 'purged' }).execute(); + }); + try { + const parent = await queue.client.runWorkflow(reconcileAppointment, input); + await finish(queue, parent.workflowRun.id); + expect(send).toHaveBeenCalledTimes(2); + expect((await children(queue)).map((child) => child.output)).toEqual(['sent', 'sent']); + expect(await emailStates()).toEqual([]); + } finally { + await queue.close(); + } + }); + + test('saved SMTP success survives a status failure and restart without resending', async () => { + const directory = await mkdtemp(join(tmpdir(), 'when-email-status-')); + await db.updateTable('appointments').set({ guest_email: null }).execute(); + const bookkeeping = vi + .spyOn(status, 'recordSendOutcome') + .mockRejectedValueOnce(new Error('local status failure')); + let queue = openQueue(join(directory, 'queue.sqlite')); + try { + const parent = await queue.client.runWorkflow(reconcileAppointment, input); + await startParentAndExpectChildren(queue, 1); + await driveUntil(queue.worker, async () => { + const [child] = await children(queue); + expect(child).toMatchObject({ status: 'pending', workerId: null }); + expect( + (await attempts(queue, child.id, 'smtp:status')).map((attempt) => attempt.status) + ).toEqual(['failed']); + }); + const [child] = await children(queue); + const savedSmtp = await attempts(queue, child.id, 'smtp'); + expect(savedSmtp).toHaveLength(1); + expect(savedSmtp[0].status).toBe('completed'); + expect(await emailStates()).toEqual([]); + await queue.close(); + bookkeeping.mockRestore(); + queue = openQueue(join(directory, 'queue.sqlite')); + await finish(queue, parent.workflowRun.id); + expect(send).toHaveBeenCalledTimes(1); + expect(await attempts(queue, child.id, 'smtp')).toEqual(savedSmtp); + expect( + (await attempts(queue, child.id, 'smtp:status')).map((attempt) => attempt.status) + ).toEqual(['failed', 'completed']); + expect(await emailStates()).toEqual(['done']); + } finally { + try { + await queue.close(); + } finally { + await rm(directory, { recursive: true, force: true }); + } + } + }); +}); diff --git a/apps/worker/src/workflows/reconcile-appointment.test.ts b/apps/worker/src/workflows/reconcile-appointment.test.ts index 80a9a197..b4777075 100644 --- a/apps/worker/src/workflows/reconcile-appointment.test.ts +++ b/apps/worker/src/workflows/reconcile-appointment.test.ts @@ -8,6 +8,12 @@ import { createLogger } from '../services/logger.js'; import type { Mailer } from '../email/smtp.js'; import { sampleInput } from '../email/__fixtures__/appointment.js'; import { runReconcileAppointment } from './reconcile-appointment.js'; +import { + deliverAppointmentEmail, + runDeliverAppointmentEmail, + type DeliverAppointmentEmailInput, + type DeliverAppointmentEmailResult +} from './deliver-appointment-email.js'; import type { WhenConfiguration } from '@when/config'; @@ -41,13 +47,27 @@ const testConfig = { function makeStep() { const names: string[] = []; + const childRuns: Promise[] = []; const step = { run: async (config: { name: string }, fn: () => Promise | T): Promise => { names.push(config.name); return fn(); + }, + runWorkflow( + spec: typeof deliverAppointmentEmail, + childInput: DeliverAppointmentEmailInput, + config: { name: string } + ) { + expect(spec).toBe(deliverAppointmentEmail); + names.push(config.name); + const child = runDeliverAppointmentEmail(childInput, { + run: async (_config: { name: string }, fn: () => Promise | T): Promise => fn() + }); + childRuns.push(child); + return child; } }; - return { step, names }; + return { step, names, childRuns }; } function makeMailer() { @@ -133,17 +153,19 @@ describe('runReconcileAppointment', () => { expect(result).toBe('reconciled'); expect(send).toHaveBeenCalledTimes(2); // guest + host expect(names).toEqual([ + 'prepare-reconciliation', 'resolve-video-chat', 'sync-calendar', - 'smtp:guest', - 'smtp:host', + 'prepare-emails', + 'email:guest', + 'email:host', 'log:result' ]); expect(await readEmailJobStates(db)).toEqual(['done']); await db.destroy(); }); - test('records email:failed when a send keeps failing', async () => { + test('records failed attempts and lets delivery errors escape', async () => { const db = await seedDb(); const { mailer, send } = makeMailer(); send.mockResolvedValue({ ok: false, reason: 'smtp down' }); @@ -151,11 +173,13 @@ describe('runReconcileAppointment', () => { setWorkerContext({ config: testConfig, db, logger: createLogger(), mailer }); - const { step } = makeStep(); - const result = await runReconcileAppointment(input, step); + const { step, names, childRuns } = makeStep(); + await expect(runReconcileAppointment(input, step)).rejects.toThrow('smtp down'); + await Promise.allSettled(childRuns); - expect(result).toBe('reconciled'); - expect(await readEmailJobStates(db)).toEqual(['failed']); + expect(send).toHaveBeenCalledTimes(2); + expect(names).not.toContain('log:result'); + expect(await readEmailJobStates(db)).toEqual(['failed', 'failed']); await db.destroy(); }); diff --git a/apps/worker/src/workflows/reconcile-appointment.ts b/apps/worker/src/workflows/reconcile-appointment.ts index ec9b0347..2ec46af7 100644 --- a/apps/worker/src/workflows/reconcile-appointment.ts +++ b/apps/worker/src/workflows/reconcile-appointment.ts @@ -1,127 +1,106 @@ -import type { RetryPolicy } from 'openworkflow'; import { reconcileAppointment } from '@when/jobs'; import type { ReconcileAppointmentInput, ReconcileAppointmentResult } from '@when/jobs'; -import { dispatch } from '../email/dispatch.js'; -import { getWorkerContext } from '../services/context.js'; -import { appendJobLog } from '../services/job-log.js'; import { parseGuestAnswers } from '@when/config'; import { shouldAttachVideoChat } from '@when/video-chat'; -import { recordSendOutcome } from '../email/status.js'; +import { prepareAppointmentEmails } from '../email/dispatch.js'; +import { systemClock, type Clock } from '../email/clock.js'; +import { getWorkerContext } from '../services/context.js'; +import { appendEmailJobLog } from '../services/job-log.js'; import { createStandaloneVideoChat, deleteStandaloneVideoChatByUrl } from '../services/video-chat.js'; import { reconcileAppointment as syncCalendarForAppointment } from '../calendar/sync.js'; -import { implementObservedWorkflow, emailsTotal } from '../services/metrics.js'; - -const SEND_RETRY: Partial = { - maximumAttempts: 5, - initialInterval: '30s', - backoffCoefficient: 2, - maximumInterval: '10m' -}; +import { implementObservedWorkflow } from '../services/metrics.js'; +import { + deliverAppointmentEmail, + type DeliverAppointmentEmailInput, + type DeliverAppointmentEmailResult +} from './deliver-appointment-email.js'; interface ReconcileStep { - run( - config: { name: string; retryPolicy?: Partial }, - fn: () => Promise | T - ): Promise; + run(config: { name: string }, fn: () => Promise | T): Promise; + runWorkflow( + spec: typeof deliverAppointmentEmail, + input: DeliverAppointmentEmailInput, + options: { name: string } + ): Promise; } export async function runReconcileAppointment( input: ReconcileAppointmentInput, - step: ReconcileStep + step: ReconcileStep, + clock: Pick = systemClock ): Promise { const ctx = getWorkerContext(); - - let resolvedRow = await ctx.db - .selectFrom('appointments') - .selectAll() - .where('id', '=', input.appointmentId) - .executeTakeFirstOrThrow(); - - // 1. Cleanup removed standalone video chat if specified - if (input.cleanupVideoChatUrl) { - await step.run({ name: 'cleanup-video-chat' }, () => - deleteStandaloneVideoChatByUrl( - ctx.config, - resolvedRow.event_type_id, - input.cleanupVideoChatUrl! - ) - ); - } - - // 2. Resolve Standalone Video Chat (e.g. Nextcloud Talk) ONLY on initial booking/confirmation - const isInitialCreation = input.emailKind === 'confirmed' || input.emailKind === 'pending'; - const meeting = ctx.config.meetings[resolvedRow.event_type_id]; - if (isInitialCreation && meeting?.video_chat && !resolvedRow.video_chat) { - const answers = parseGuestAnswers(resolvedRow.guest_answers); - if (shouldAttachVideoChat(meeting, ctx.config, answers)) { - resolvedRow = await step.run({ name: 'resolve-video-chat' }, () => - createStandaloneVideoChat(ctx.db, input.appointmentId, ctx.config) - ); - } - } - - // 3. Sync Calendar (which also handles Google Meet generation if initially requested) - await step.run({ name: 'sync-calendar' }, () => syncCalendarForAppointment(ctx, resolvedRow)); - - // 3. Send email if requested - if (input.emailKind) { - // Fetch the latest state (which has resolved video chat link and external event details) - const finalRow = await ctx.db + const liveAppointment = () => + ctx.db .selectFrom('appointments') .selectAll() .where('id', '=', input.appointmentId) - .executeTakeFirstOrThrow(); + .where('status', '!=', 'purged') + .executeTakeFirst(); - const envelopes = await dispatch({ kind: input.emailKind, appointment: finalRow }, ctx.config); - let allSent = true; + const preparation = await step.run({ name: 'prepare-reconciliation' }, async () => { + const row = await liveAppointment(); + if (!row) return null; + const meeting = ctx.config.meetings[row.event_type_id]; + const initialCreation = input.emailKind === 'confirmed' || input.emailKind === 'pending'; + let resolveVideoChat = false; + if (initialCreation && meeting?.video_chat && !row.video_chat) { + const answers = parseGuestAnswers(row.guest_answers); + resolveVideoChat = shouldAttachVideoChat(meeting, ctx.config, answers); + } + return { resolveVideoChat }; + }); + if (!preparation) return 'reconciled'; - for (const envelope of envelopes) { - const recipientType = envelope.to === ctx.config.user.email ? 'host' : 'guest'; - try { - await step.run({ name: `smtp:${recipientType}`, retryPolicy: SEND_RETRY }, async () => { - const result = await ctx.mailer.send(envelope); - await recordSendOutcome(ctx, result); - if (!result.ok) { - emailsTotal.inc({ - recipient_type: recipientType, - email_kind: input.emailKind!, - status: 'failure' - }); - throw new Error(result.reason); - } - emailsTotal.inc({ - recipient_type: recipientType, - email_kind: input.emailKind!, - status: 'success' - }); - }); - } catch (err) { - allSent = false; - ctx.logger.error( - { - appointmentId: input.appointmentId, - recipientType, - error: err instanceof Error ? err.message : String(err) - }, - 'appointment email send failed after retries' + if (input.cleanupVideoChatUrl) { + await step.run({ name: 'cleanup-video-chat' }, async () => { + const row = await liveAppointment(); + if (row) { + await deleteStandaloneVideoChatByUrl( + ctx.config, + row.event_type_id, + input.cleanupVideoChatUrl! ); } - } - - await step.run({ name: 'log:result' }, () => - appendJobLog( - ctx.db, - input.appointmentId, - 'email', - allSent ? 'done' : 'failed', - Temporal.Now.instant().toString() - ) - ); + }); } + if (preparation.resolveVideoChat) { + await step.run({ name: 'resolve-video-chat' }, async () => { + if (await liveAppointment()) { + await createStandaloneVideoChat(ctx.db, input.appointmentId, ctx.config); + } + }); + } + + await step.run({ name: 'sync-calendar' }, async () => { + const row = await liveAppointment(); + if (row) await syncCalendarForAppointment(ctx, row); + }); + + const kind = input.emailKind; + if (!kind) return 'reconciled'; + const deliveries = await step.run({ name: 'prepare-emails' }, async () => { + const row = await liveAppointment(); + return row ? prepareAppointmentEmails({ kind, appointment: row }, ctx.config) : []; + }); + const outcomes = await Promise.all( + deliveries.map(({ role, envelope }) => + step.runWorkflow( + deliverAppointmentEmail, + { appointmentId: input.appointmentId, kind, role, envelope }, + { name: `email:${role}` } + ) + ) + ); + if (deliveries.length > 0 && outcomes.every((outcome) => outcome === 'sent')) { + await step.run({ name: 'log:result' }, () => + appendEmailJobLog(ctx.db, input.appointmentId, 'done', clock.now().toISOString()) + ); + } return 'reconciled'; } diff --git a/packages/jobs/src/specs.ts b/packages/jobs/src/specs.ts index 429086cf..5ff4ccbb 100644 --- a/packages/jobs/src/specs.ts +++ b/packages/jobs/src/specs.ts @@ -28,7 +28,7 @@ export interface ReconcileAppointmentInput { cleanupVideoChatUrl?: string | null; } -// 'reconciled' = completed successfully; 'failed' = failed after retries. +// Native retry exhaustion rejects result(); 'failed' remains for contract compatibility. export type ReconcileAppointmentResult = 'reconciled' | 'failed'; /** -- 2.54.0 From 1909f88488b6b8b9a585b5762573feb08f55713d Mon Sep 17 00:00:00 2001 From: eleith Date: Tue, 20 Oct 2026 08:42:12 -0700 Subject: [PATCH 2/2] fix(worker): retain purged rows when native remote deletion exhausts --- .changeset/appointment-retries.md | 2 +- apps/worker/README.md | 7 + .../purge-appointment.integration.test.ts | 312 ++++++++++++++++++ .../src/workflows/purge-appointment.test.ts | 88 ++++- .../worker/src/workflows/purge-appointment.ts | 30 +- 5 files changed, 405 insertions(+), 34 deletions(-) create mode 100644 apps/worker/src/workflows/purge-appointment.integration.test.ts diff --git a/.changeset/appointment-retries.md b/.changeset/appointment-retries.md index 9716ac3c..95338b74 100644 --- a/.changeset/appointment-retries.md +++ b/.changeset/appointment-retries.md @@ -4,4 +4,4 @@ '@when/cli': patch --- -Use native retries for appointment email delivery. Guest and host emails run independently with checkpointed content, so retrying one delivery or recording SMTP status does not repeat another saved send. +Use native retries for appointment email delivery and purge cleanup. Guest and host emails run independently with checkpointed content, so retrying one delivery or recording SMTP status does not repeat another saved send. Exhausted remote-deletion retries now fail the purge and retain its remaining rows instead of abandoning remote cleanup and deleting local data. diff --git a/apps/worker/README.md b/apps/worker/README.md index e2678da2..ad634bd0 100644 --- a/apps/worker/README.md +++ b/apps/worker/README.md @@ -68,6 +68,13 @@ The parent records `done` only when every intended email was sent and the appoin still exists and is not purged. Ineligible deliveries are skipped. Exhaustion fails the engine run rather than returning a successful summary of failed delivery. +Purge deletes each remote event with up to thirty native attempts (one-minute +initial delay, doubling up to one hour). Success or an already-missing event permits +video cleanup and local deletion. Exhaustion fails the job, retaining that row and +all later rows in the chain. These rows are already marked `purged` and hidden from +normal scheduling; another purge click does not automatically recover them. Review +failed jobs before deciding how to finish cleanup. + Prepared envelopes include personal data and guest capability links: treat queue backups like application backups. Checkpoints do not guarantee exactly-once SMTP; a crash after remote acceptance but before saving success can still cause a duplicate. diff --git a/apps/worker/src/workflows/purge-appointment.integration.test.ts b/apps/worker/src/workflows/purge-appointment.integration.test.ts new file mode 100644 index 00000000..c5a51825 --- /dev/null +++ b/apps/worker/src/workflows/purge-appointment.integration.test.ts @@ -0,0 +1,312 @@ +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { OpenWorkflow } from 'openworkflow'; +import { BackendSqlite } from 'openworkflow/sqlite'; +import { openDb, runMigrations } from '@when/db'; +import { purgeAppointment, type PurgeAppointmentInput } from '@when/jobs'; +import type { WhenConfiguration } from '@when/config'; +import { afterEach, beforeEach, expect, test, vi } from 'vitest'; +import { setWorkerContext } from '../services/context.js'; +import type { Logger } from '../services/logger.js'; +import { runPurgeAppointment } from './purge-appointment.js'; + +const config = { + providers: { + dav: { + type: 'caldav', + url: 'https://cal.example.com/', + username: 'u', + password: 'p', + calendars: { work: { href: 'https://cal.example.com/work/' } } + }, + talk: { + type: 'nextcloud', + url: 'https://talk.example.com', + username: 'u', + password: 'p', + calendars: {} + } + }, + meetings: { chat: { video_chat: { provider: 'talk' } } } +} as unknown as WhenConfiguration; + +beforeEach(() => { + // Only the wall clock moves; engine timers and the production retry policy stay real. + vi.useFakeTimers({ toFake: ['Date'] }); + vi.setSystemTime(new Date('2026-05-01T00:00:00Z')); +}); + +afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); +}); + +async function seed(ids: string[]) { + const db = openDb(':memory:'); + await runMigrations(db); + for (const id of ids) { + await db + .insertInto('appointments') + .values({ + id, + event_type_id: 'chat', + start_time: '2026-05-01T10:00:00Z', + end_time: '2026-05-01T10:30:00Z', + guest_name: 'Guest', + guest_email: 'guest@example.com', + guest_timezone: 'UTC', + status: 'purged', + cancel_token: `token-${id}`, + external_event_id: id, + external_calendar_id: 'work', + video_chat: `https://talk.example.com/call/${id}`, + created_at: '', + updated_at: '' + }) + .execute(); + } + const logger = { warn: vi.fn(), info: vi.fn(), debug: vi.fn(), error: vi.fn() }; + setWorkerContext({ config, db, logger: logger as unknown as Logger, mailer: { send: vi.fn() } }); + return { db, logger }; +} + +function input(ids: string[]): PurgeAppointmentInput { + return { rows: ids.map((id) => ({ id, externalEventId: id, externalCalendarId: 'work' })) }; +} + +function provider(failures: Record, successStatus = 204, videoStatus = 204) { + const calendar: Record = {}; + const video: string[] = []; + vi.spyOn(globalThis, 'fetch').mockImplementation(async (url, init) => { + expect(init?.method).toBe('DELETE'); + const target = new URL(String(url)); + const id = target.pathname + .split('/') + .at(-1)! + .replace(/\.ics$/, ''); + if (target.hostname === 'cal.example.com') { + calendar[id] = (calendar[id] ?? 0) + 1; + const status = calendar[id] <= (failures[id] ?? 0) ? 500 : successStatus; + return new Response(null, { status }); + } + expect(target.hostname).toBe('talk.example.com'); + video.push(id); + return new Response(null, { status: videoStatus }); + }); + return { calendar, video }; +} + +function openQueue(path = ':memory:') { + const backend = BackendSqlite.connect(path); + const client = new OpenWorkflow({ backend }); + client.implementWorkflow(purgeAppointment, ({ input, step }) => runPurgeAppointment(input, step)); + const worker = client.newWorker({ concurrency: 1 }); + return { + backend, + client, + worker, + async close() { + try { + await worker.stop(); + } finally { + await backend.stop(); + } + } + }; +} + +type Queue = ReturnType; + +async function readRun(queue: Queue, workflowRunId: string) { + const run = await queue.backend.getWorkflowRun({ workflowRunId }); + expect(run).not.toBeNull(); + return run!; +} + +async function attempts(queue: Queue, workflowRunId: string) { + const first = await queue.backend.listStepAttempts({ workflowRunId, limit: 100 }); + const all = first.data; + let after = first.pagination.next; + while (after) { + const page = await queue.backend.listStepAttempts({ workflowRunId, limit: 100, after }); + all.push(...page.data); + after = page.pagination.next; + } + return all; +} + +async function runNextPurgePass(queue: Queue, id: string) { + const before = await readRun(queue, id); + if (before.availableAt && before.availableAt.getTime() > Date.now()) { + vi.setSystemTime(before.availableAt); + } + await vi.waitFor(async () => expect(await queue.worker.tick()).toBe(1), { + timeout: 2000, + interval: 1 + }); + await vi.waitFor( + async () => { + const run = await readRun(queue, id); + expect(['pending', 'completed', 'failed']).toContain(run.status); + if (run.status === 'pending') expect(run.workerId).toBeNull(); + }, + { timeout: 2000, interval: 1 } + ); + await queue.worker.stop(); + return readRun(queue, id); +} + +async function finish(queue: Queue, id: string) { + for (let i = 0; i < 1000; i++) { + const run = await runNextPurgePass(queue, id); + if (run.status !== 'pending') return run; + } + throw new Error('Purge did not settle within 1000 passes'); +} + +test('native transient retries retain the row until remote deletion succeeds', async () => { + const { db } = await seed(['one']); + const remote = provider({ one: 2 }); + const queue = openQueue(); + try { + const job = await queue.client.runWorkflow(purgeAppointment, input(['one'])); + const id = job.workflowRun.id; + expect(await runNextPurgePass(queue, id)).toMatchObject({ status: 'pending', workerId: null }); + expect(remote.video).toEqual([]); + expect(await db.selectFrom('appointments').selectAll().execute()).toHaveLength(1); + expect(await finish(queue, id)).toMatchObject({ status: 'completed', output: 'purged' }); + expect(await job.result({ timeoutMs: 500 })).toBe('purged'); + expect(remote.calendar).toEqual({ one: 3 }); + expect(remote.video).toEqual(['one']); + expect(await db.selectFrom('appointments').selectAll().execute()).toEqual([]); + expect((await attempts(queue, id)).map((attempt) => attempt.status)).toEqual([ + 'failed', + 'failed', + 'completed', + 'completed', + 'completed' + ]); + } finally { + await queue.close(); + await db.destroy(); + } +}); + +test('calendar 404 enables best-effort video cleanup and local deletion', async () => { + const { db } = await seed(['gone']); + const remote = provider({}, 404, 500); + const warning = vi.spyOn(console, 'warn').mockImplementation(() => {}); + const queue = openQueue(); + try { + const job = await queue.client.runWorkflow(purgeAppointment, input(['gone'])); + expect(await finish(queue, job.workflowRun.id)).toMatchObject({ status: 'completed' }); + expect(remote.calendar).toEqual({ gone: 1 }); + expect(remote.video).toEqual(['gone']); + expect(warning).toHaveBeenCalledTimes(1); + expect(await db.selectFrom('appointments').selectAll().execute()).toEqual([]); + expect(await attempts(queue, job.workflowRun.id)).toHaveLength(3); + } finally { + await queue.close(); + await db.destroy(); + } +}); + +test('thirty native failures fail the job, retaining current/later rows without redoing earlier success', async () => { + const ids = ['earlier', 'current', 'later']; + const { db, logger } = await seed(ids); + const remote = provider({ current: 30 }); + const queue = openQueue(); + try { + const job = await queue.client.runWorkflow(purgeAppointment, input(ids)); + const id = job.workflowRun.id; + expect(await runNextPurgePass(queue, id)).toMatchObject({ status: 'pending' }); + const earlier = (await attempts(queue, id)).filter((attempt) => + attempt.stepName.endsWith(':earlier') + ); + expect(earlier).toHaveLength(3); + expect(await finish(queue, id)).toMatchObject({ status: 'failed', output: null }); + await expect(job.result({ timeoutMs: 500 })).rejects.toThrow(/CalDAV DELETE/); + const recorded = await attempts(queue, id); + expect(recorded).toHaveLength(33); + expect(recorded.filter((attempt) => attempt.stepName.endsWith(':earlier'))).toEqual(earlier); + const failed = recorded.filter((attempt) => attempt.stepName === 'calendar:current'); + expect(failed).toHaveLength(30); + expect(failed.every((attempt) => attempt.status === 'failed')).toBe(true); + expect(remote.calendar).toEqual({ earlier: 1, current: 30 }); + expect(remote.video).toEqual(['earlier']); + const retained = await db.selectFrom('appointments').selectAll().orderBy('id').execute(); + expect(retained.map((row) => [row.id, row.status, row.video_chat])).toEqual([ + ['current', 'purged', 'https://talk.example.com/call/current'], + ['later', 'purged', 'https://talk.example.com/call/later'] + ]); + expect(logger.warn).not.toHaveBeenCalled(); + expect(await queue.worker.tick()).toBe(0); + } finally { + await queue.close(); + await db.destroy(); + } +}); + +test('reopens a file queue during retry delay without replaying earlier cleanup', async () => { + const directory = await mkdtemp(join(tmpdir(), 'when-purge-')); + const path = join(directory, 'queue.sqlite'); + const { db } = await seed(['earlier', 'retry']); + const remote = provider({ retry: 2 }); + let queue = openQueue(path); + try { + const job = await queue.client.runWorkflow(purgeAppointment, input(['earlier', 'retry'])); + const id = job.workflowRun.id; + const parked = await runNextPurgePass(queue, id); + expect(parked).toMatchObject({ status: 'pending', workerId: null }); + expect(parked.availableAt!.getTime()).toBeGreaterThan(Date.now()); + const saved = await attempts(queue, id); + expect(saved).toHaveLength(4); + await queue.close(); + queue = openQueue(path); + expect(await readRun(queue, id)).toEqual(parked); + expect(await queue.worker.tick()).toBe(0); + expect(await attempts(queue, id)).toEqual(saved); + expect(await finish(queue, id)).toMatchObject({ status: 'completed', output: 'purged' }); + expect(remote.calendar).toEqual({ earlier: 1, retry: 3 }); + expect(remote.video).toEqual(['earlier', 'retry']); + expect((await attempts(queue, id)).slice(0, 4)).toEqual(saved); + expect(await db.selectFrom('appointments').selectAll().execute()).toEqual([]); + } finally { + await queue.close(); + await db.destroy(); + await rm(directory, { recursive: true, force: true }); + } +}); + +test('twenty near-budget rows complete with 640 actual step attempts, below the 1000-attempt cap', async () => { + const ids = Array.from({ length: 20 }, (_, i) => `row-${i}`); + const { db } = await seed(ids); + const remote = provider(Object.fromEntries(ids.map((id) => [id, 29]))); + const queue = openQueue(); + try { + const job = await queue.client.runWorkflow(purgeAppointment, input(ids)); + expect(await finish(queue, job.workflowRun.id)).toMatchObject({ + status: 'completed', + output: 'purged' + }); + const recorded = await attempts(queue, job.workflowRun.id); + expect(recorded).toHaveLength(640); + expect(recorded.length).toBeLessThan(1000); + for (const id of ids) { + const calendar = recorded.filter((attempt) => attempt.stepName === `calendar:${id}`); + expect(calendar.filter((attempt) => attempt.status === 'failed')).toHaveLength(29); + expect(calendar.filter((attempt) => attempt.status === 'completed')).toHaveLength(1); + expect(remote.calendar[id]).toBe(30); + } + expect( + recorded.filter((attempt) => attempt.stepName.startsWith('video-chat-cleanup:')) + ).toHaveLength(20); + expect(recorded.filter((attempt) => attempt.stepName.startsWith('delete:'))).toHaveLength(20); + expect(remote.video).toEqual(ids); + expect(await db.selectFrom('appointments').selectAll().execute()).toEqual([]); + } finally { + await queue.close(); + await db.destroy(); + } +}, 15000); diff --git a/apps/worker/src/workflows/purge-appointment.test.ts b/apps/worker/src/workflows/purge-appointment.test.ts index 1a91c755..803ab6c4 100644 --- a/apps/worker/src/workflows/purge-appointment.test.ts +++ b/apps/worker/src/workflows/purge-appointment.test.ts @@ -1,7 +1,8 @@ import { expect, test, vi, beforeEach } from 'vitest'; import { openDb, runMigrations } from '@when/db'; import type { Database } from '@when/db'; -import type { Kysely } from 'kysely'; +import { sql, type Kysely } from 'kysely'; +import type { RetryPolicy } from 'openworkflow'; import type { WhenConfiguration } from '@when/config'; import type { PurgeAppointmentInput } from '@when/jobs'; import { setWorkerContext } from '../services/context.js'; @@ -25,13 +26,18 @@ const config = { function makeStep() { const names: string[] = []; + const policies: (Partial | undefined)[] = []; const step = { - run: async (cfg: { name: string }, fn: () => Promise | T): Promise => { + run: async ( + cfg: { name: string; retryPolicy?: Partial }, + fn: () => Promise | T + ): Promise => { names.push(cfg.name); + policies.push(cfg.retryPolicy); return fn(); } }; - return { step, names }; + return { step, names, policies }; } function recordingFetch(status = 204) { @@ -107,9 +113,19 @@ test('reachable calendar: deletes the remote event and the row', async () => { const input: PurgeAppointmentInput = { rows: [{ id: '1', externalEventId: '1', externalCalendarId: 'work' }] }; - const { step, names } = makeStep(); + const { step, names, policies } = makeStep(); expect(await runPurgeAppointment(input, step)).toBe('purged'); + expect(policies).toEqual([ + { + maximumAttempts: 30, + initialInterval: '1m', + backoffCoefficient: 2, + maximumInterval: '1h' + }, + undefined, + undefined + ]); expect(calls.some((c) => c.method === 'DELETE')).toBe(true); expect(names).toEqual(['calendar:1', 'video-chat-cleanup:1', 'delete:1']); expect(await exists(db, '1')).toBe(false); @@ -119,7 +135,7 @@ test('reachable calendar: deletes the remote event and the row', async () => { } }); -test('unreachable calendar: logs the orphan but still deletes the row', async () => { +test('calendar failure escapes without cleanup, deletion or an orphan warning', async () => { const db = await seedDb([ { id: '1', cancel_token: 't1', external_event_id: '1', external_calendar_id: 'work' } ]); @@ -130,12 +146,11 @@ test('unreachable calendar: logs the orphan but still deletes the row', async () rows: [{ id: '1', externalEventId: '1', externalCalendarId: 'work' }] }; - expect(await runPurgeAppointment(input, makeStep().step)).toBe('purged'); - expect(warn).toHaveBeenCalledWith( - expect.objectContaining({ appointmentId: '1', externalEventId: '1' }), - 'purge left an orphaned calendar event' - ); - expect(await exists(db, '1')).toBe(false); + const { step, names } = makeStep(); + await expect(runPurgeAppointment(input, step)).rejects.toThrow(); + expect(names).toEqual(['calendar:1']); + expect(warn).not.toHaveBeenCalled(); + expect(await exists(db, '1')).toBe(true); } finally { await db.destroy(); } @@ -182,3 +197,54 @@ test('purges every row in the chain', async () => { await db.destroy(); } }); + +test('opaque step rejection escapes unchanged', async () => { + const db = await seedDb([{ id: '1' }]); + try { + const { warn } = useContext(db); + const rejection = new Error('opaque step rejection'); + const run = vi.fn().mockRejectedValue(rejection); + await expect( + runPurgeAppointment( + { rows: [{ id: '1', externalEventId: '1', externalCalendarId: 'work' }] }, + { run } + ) + ).rejects.toBe(rejection); + expect(run).toHaveBeenCalledTimes(1); + expect(warn).not.toHaveBeenCalled(); + expect(await exists(db, '1')).toBe(true); + } finally { + await db.destroy(); + } +}); + +test.each(['video cleanup', 'local delete'])( + '%s database failure escapes and retains the row', + async (stage) => { + const videoChatUrl = stage === 'video cleanup' ? 'https://talk.example.com/call/room' : null; + const db = await seedDb([{ id: '1', video_chat: videoChatUrl }]); + try { + useContext(db); + const operation = stage === 'video cleanup' ? 'UPDATE' : 'DELETE'; + await sql + .raw( + `CREATE TRIGGER fail_cleanup BEFORE ${operation} ON appointments + BEGIN SELECT RAISE(FAIL, 'local cleanup failed'); END` + ) + .execute(db); + const { step, names } = makeStep(); + await expect( + runPurgeAppointment( + { rows: [{ id: '1', externalEventId: null, externalCalendarId: null }] }, + step + ) + ).rejects.toThrow('local cleanup failed'); + expect(names).toEqual( + stage === 'video cleanup' ? ['video-chat-cleanup:1'] : ['video-chat-cleanup:1', 'delete:1'] + ); + expect(await exists(db, '1')).toBe(true); + } finally { + await db.destroy(); + } + } +); diff --git a/apps/worker/src/workflows/purge-appointment.ts b/apps/worker/src/workflows/purge-appointment.ts index 723fd896..bfeb379f 100644 --- a/apps/worker/src/workflows/purge-appointment.ts +++ b/apps/worker/src/workflows/purge-appointment.ts @@ -6,7 +6,6 @@ import { getWorkerContext } from '../services/context.js'; import { implementObservedWorkflow } from '../services/metrics.js'; import { deleteStandaloneVideoChat } from '../services/video-chat.js'; -// ~24h of bounded effort before we give up and orphan the remote event. const CALENDAR_RETRY: Partial = { maximumAttempts: 30, initialInterval: '1m', @@ -25,32 +24,19 @@ export async function runPurgeAppointment( input: PurgeAppointmentInput, step: PurgeStep ): Promise { - const { config, db, logger } = getWorkerContext(); + const { config, db } = getWorkerContext(); for (const row of input.rows) { const { externalEventId, externalCalendarId } = row; if (externalEventId && externalCalendarId) { - try { - await step.run({ name: `calendar:${row.id}`, retryPolicy: CALENDAR_RETRY }, async () => { - const res = await deleteAppointmentFromCalendar( - config, - externalCalendarId, - externalEventId - ); - if (!res.ok) throw new Error(res.reason); - }); - } catch (err) { - // Give up but still delete the row; the orphan is logged, not retried forever. - logger.warn( - { - appointmentId: row.id, - calendarId: externalCalendarId, - externalEventId, - error: err instanceof Error ? err.message : String(err) - }, - 'purge left an orphaned calendar event' + await step.run({ name: `calendar:${row.id}`, retryPolicy: CALENDAR_RETRY }, async () => { + const res = await deleteAppointmentFromCalendar( + config, + externalCalendarId, + externalEventId ); - } + if (!res.ok) throw new Error(res.reason); + }); } await step.run({ name: `video-chat-cleanup:${row.id}` }, async () => { await deleteStandaloneVideoChat(db, row.id, config); -- 2.54.0