diff --git a/.changeset/email-durable-queue-delivery.md b/.changeset/email-durable-queue-delivery.md new file mode 100644 index 0000000000..0fe4154680 --- /dev/null +++ b/.changeset/email-durable-queue-delivery.md @@ -0,0 +1,64 @@ +--- +"@objectstack/plugin-email": minor +"@objectstack/service-settings": minor +"@objectstack/cli": minor +--- + +feat(plugin-email): durable email delivery through `sys_job_queue`, opt-in (#5160) + +`IEmailService.send()` has always delivered **inline**: the SMTP session ran +inside the caller's `await`, and `EmailService`'s retry loop lived in the same +process — so a crash between the attempt and the retry dropped the message with +no trace beyond a `sys_email` row stuck at `queued`. The pieces for a durable +path all existed (`sys_job_queue`, the `DbQueueAdapter`, an `email.send.async` +subscriber) but nothing in the repo ever published to that topic. + +**New: `queueDelivery`.** With it on, `send()` persists the `sys_email` row, +publishes an `email.send.async` job **referencing that row**, and returns +`{ status: 'queued' }` immediately. A worker delivers the row and finalizes it +in place (`sent` + `message_id`, or `failed` + `error`); the queue retries with +exponential backoff (1s → 5min cap) and dead-letters the job when the attempts +run out, so a restart resumes delivery instead of losing it. The `'queued'` +status was already in `EmailDeliveryStatus` — no spec change. + +Three ways to turn it on, all default-off: + +- `new EmailServicePlugin({ queueDelivery: true })` +- `OS_EMAIL_QUEUE_ENABLED=true` (or `config.email.queueDelivery`) on `os serve` +- Settings → Mail → **Durable queue delivery**, hot-applied without a restart + +**One retry budget, not two.** `retries` keeps its meaning — total attempts are +`retries + 1` in both modes. Inline it drives the in-process loop; queued it +becomes the queue's `maxAttempts` and the per-row loop is pinned to one attempt +per delivery. Turning the toggle on changes *where* a retry happens (durable, +backed off) and never *how many* happen, so the two layers cannot multiply. + +**Fixed in the same change: the `email.send.async` subscriber inserted a new +`sys_email` row per delivery.** It called `send()` with the message, so a job +the queue retried five times left five rows — four permanently `failed`, none +carrying the real attempt count. It now delivers the referenced row via +`deliverPersistedRow`, so one message is one row and `attempt_count` +accumulates on it. Messages published in the old shape (a bare `SendEmailInput`) +are still accepted and delivered inline for a migration window. + +Boundaries worth knowing before you switch it on: + +- **"Send test email" always sends inline**, in every mode — the button has to + report the provider's own answer (`535 …`), and "queued" is exactly the + non-answer #5087 removed from it. +- **Messages with attachments or custom headers are delivered inline**, because + `sys_email` has no columns for them and a queued copy would arrive stripped. + Queueing them is tracked separately; this ships the loss-free behaviour. +- **A declaration that cannot be honoured fails the boot.** `queueDelivery: true` + from the constructor or `OS_EMAIL_QUEUE_ENABLED` with no durable queue + registered (or with `persist: false`) throws on `kernel:ready`, naming the + fix — the #5132 judgement, applied to durability. The **settings toggle** is + the opposite trade: it logs at `error` and keeps sending inline, because one + save must not stop the mail. +- The kernel's built-in in-memory `queue` fallback does **not** count as a + durable queue: it delivers synchronously with no retry or DLQ, so publishing + to it would report `queued` for a message nothing could ever recover. Mount + `@objectstack/service-queue` over an ObjectQL engine (the `queue` capability + does this on `os serve`) to get the `sys_job_queue`-backed adapter. + +Leaving `queueDelivery` unset keeps today's behaviour byte for byte. diff --git a/packages/cli/src/commands/serve-email-capability.test.ts b/packages/cli/src/commands/serve-email-capability.test.ts index caf45134d7..a4ea73c099 100644 --- a/packages/cli/src/commands/serve-email-capability.test.ts +++ b/packages/cli/src/commands/serve-email-capability.test.ts @@ -143,6 +143,58 @@ describe('resolveEmailCapabilityArg', () => { .toThrow(/log \/ resend \/ postmark \/ smtp/); }); + // ── durable queue delivery (#5160) ─────────────────────────────────────── + + it('leaves queueDelivery unset when nothing declares it', () => { + // Absent, not `false`: the plugin's boot gate fires on an explicit `true`, + // and an option nobody wrote should not appear in what it is constructed + // with at all. + expect(resolveEmailCapabilityArg({}, {})).not.toHaveProperty('options.queueDelivery'); + }); + + it('reads OS_EMAIL_QUEUE_ENABLED as the boolean feature flag it is', () => { + // `_ENABLED` + default-off is the Prime Directive #9 shape for a boolean + // flag; a bare `OS_EMAIL_QUEUE` would read as a config value (a queue + // name), which is the naming trap that rule exists to close. + for (const on of ['1', 'true', 'TRUE', 'yes', 'on']) { + expect(resolveEmailCapabilityArg({}, { OS_EMAIL_QUEUE_ENABLED: on }).options.queueDelivery, on) + .toBe(true); + } + for (const off of ['0', 'false', 'no', '']) { + expect(resolveEmailCapabilityArg({}, { OS_EMAIL_QUEUE_ENABLED: off }).options.queueDelivery, off) + .toBe(false); + } + }); + + it('accepts the same declaration from objectstack.config.ts, with env winning', () => { + expect(resolveEmailCapabilityArg({ queueDelivery: true }, {}).options.queueDelivery).toBe(true); + expect( + resolveEmailCapabilityArg({ queueDelivery: true }, { OS_EMAIL_QUEUE_ENABLED: 'false' }) + .options.queueDelivery, + ).toBe(false); + }); + + it('does not decide here whether the declaration can be honoured', () => { + // No kernel exists at this point, so no service registry can be read. The + // plugin asserts a durable queue on `kernel:ready` — reading a registry + // that is still filling and recording the verdict is the failure mode + // AGENTS.md names. All this function does is carry the declaration. + expect(() => resolveEmailCapabilityArg({}, { + OS_EMAIL_QUEUE_ENABLED: 'true', OS_EMAIL_PROVIDER: 'log', + })).not.toThrow(); + }); + + it('does not add a second retry knob for the queue', () => { + // One "retry" concept, one config. `OS_EMAIL_RETRIES` drives the inline + // loop or becomes the queue's attempt budget — never both, so the two + // layers cannot multiply into 5x5 connections. + const { options } = resolveEmailCapabilityArg({}, { + OS_EMAIL_QUEUE_ENABLED: 'true', OS_EMAIL_RETRIES: '4', + }); + expect(options).toMatchObject({ queueDelivery: true, retries: 4 }); + expect(Object.keys(options).filter((k) => /retr|attempt/i.test(k))).toEqual(['retries']); + }); + it('still derives the fallback from-address and template context', () => { const { options } = resolveEmailCapabilityArg({}, { OS_APP_NAME: 'Acme CRM' }, 'ignored'); expect(options.defaultTemplateContext).toMatchObject({ appName: 'Acme CRM' }); diff --git a/packages/cli/src/commands/serve.ts b/packages/cli/src/commands/serve.ts index 2c99940b47..871457e580 100644 --- a/packages/cli/src/commands/serve.ts +++ b/packages/cli/src/commands/serve.ts @@ -2862,6 +2862,11 @@ export interface EmailCapabilityArg { * from `@objectstack/plugin-email` — the package that has to materialise the * transport — rather than restated here. Two literals describing one vocabulary * is how the settings dropdown and the transports drifted apart (#5094). + * + * `OS_EMAIL_QUEUE_ENABLED=true` (or `config.email.queueDelivery`) switches + * delivery from inline to the durable `sys_job_queue` path (#5160). It reuses + * `OS_EMAIL_RETRIES` as its attempt budget rather than adding a second retry + * knob — see `EmailServicePlugin.makeQueueDelivery`. */ export function resolveEmailCapabilityArg( cfgEmail: Record = {}, @@ -2882,6 +2887,15 @@ export function resolveEmailCapabilityArg( } } const retries = env.OS_EMAIL_RETRIES ? Number(env.OS_EMAIL_RETRIES) : cfgEmail.retries; + // `OS_EMAIL_QUEUE_ENABLED` — a boolean feature flag, so `_ENABLED` and + // default-off (Prime Directive #9; a bare `OS_EMAIL_QUEUE` would read as a + // config value, e.g. a queue name). Whether the declaration can be HONOURED + // is not knowable here — no kernel exists yet — so the plugin asserts it on + // `kernel:ready`, where the service registry has settled, and fails the boot + // there if no durable queue showed up. + const queueDelivery = env.OS_EMAIL_QUEUE_ENABLED != null + ? ['1', 'true', 'yes', 'on'].includes(String(env.OS_EMAIL_QUEUE_ENABLED).trim().toLowerCase()) + : cfgEmail.queueDelivery; const defaultTemplateContext = { appName: env.OS_APP_NAME || cfgEmail.appName || configAppName || 'ObjectStack', ...(cfgEmail.defaultTemplateContext || {}), @@ -2915,6 +2929,7 @@ export function resolveEmailCapabilityArg( ...(Object.keys(providerOptions).length > 0 ? { providerOptions } : {}), defaultFrom, ...(retries != null && !Number.isNaN(retries) ? { retries } : {}), + ...(queueDelivery != null ? { queueDelivery: !!queueDelivery } : {}), defaultTemplateContext, }; diff --git a/packages/plugins/plugin-email/package.json b/packages/plugins/plugin-email/package.json index 8be6be66fc..9de7155d16 100644 --- a/packages/plugins/plugin-email/package.json +++ b/packages/plugins/plugin-email/package.json @@ -25,6 +25,8 @@ "nodemailer": "^9.0.3" }, "devDependencies": { + "@objectstack/objectql": "workspace:*", + "@objectstack/service-queue": "workspace:*", "@objectstack/service-settings": "workspace:*", "@types/node": "^26.1.2", "@types/nodemailer": "^8.0.1", diff --git a/packages/plugins/plugin-email/src/email-plugin.queue-delivery.test.ts b/packages/plugins/plugin-email/src/email-plugin.queue-delivery.test.ts new file mode 100644 index 0000000000..76d5f48399 --- /dev/null +++ b/packages/plugins/plugin-email/src/email-plugin.queue-delivery.test.ts @@ -0,0 +1,498 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. +// +// EmailServicePlugin — durable queue delivery, end to end (#5160). +// +// These run the REAL `DbQueueAdapter` over a fake ObjectQL engine holding both +// `sys_email` and `sys_job_queue`, so what is asserted is the actual round +// trip: `send()` writes the row and publishes, a worker poll delivers it, and +// a failing transport is retried by the queue until the job lands in the DLQ. +// A hand-written stub queue could not show the two halves agreeing, and the +// halves disagreeing is the whole risk — the pre-#5160 subscriber called +// `send()`, so every queue retry inserted ANOTHER sys_email row. + +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import { createMemoryQueue } from '@objectstack/core'; +import { assertEngineDeleteDispatch } from '@objectstack/objectql'; +import { DbQueueAdapter } from '@objectstack/service-queue'; +import { EmailServicePlugin, resolveDurableQueue } from './email-plugin.js'; +import { EmailService, EMAIL_SEND_QUEUE } from './email-service.js'; + +// ── harness ──────────────────────────────────────────────────────────────── + +interface Resolved { value: unknown; source?: string } + +function fakeSettings(values: Record) { + const listeners: Array<() => void> = []; + const actions = new Map Promise>(); + return { + createClient: () => ({}), + getNamespace: async () => ({ values }), + subscribe: (_ns: string, cb: () => void) => { listeners.push(cb); }, + registerAction: (ns: string, id: string, fn: (a: any) => Promise) => { + actions.set(`${ns}/${id}`, fn); + }, + async save(patch: Record) { + Object.assign(values, patch); + for (const l of listeners) l(); + await new Promise((r) => setTimeout(r, 0)); + }, + action: (id: string) => actions.get(`mail/${id}`), + }; +} + +/** + * Engine mimicking objectql's `where:`-based find and + * `(table, { id, ...patch })` update — the same shape + * `service-queue`'s own adapter tests use, so the queue half is exercised + * against the signatures the real engine has. + */ +function fakeEngine() { + const tables = new Map(); + const rowsOf = (t: string) => tables.get(t) ?? []; + const matches = (row: any, where: Record) => + Object.entries(where).every(([k, v]) => row[k] === v); + return { + tables, + rows: (t: string) => [...rowsOf(t)], + async find(table: string, opts: any = {}) { + let out = opts.where ? rowsOf(table).filter((r) => matches(r, opts.where)) : [...rowsOf(table)]; + if (opts.orderBy) { + for (const ord of [...opts.orderBy].reverse()) { + out.sort((a, b) => { + const av = a[ord.field], bv = b[ord.field]; + if (av === bv) return 0; + const cmp = av > bv ? 1 : -1; + return ord.order === 'desc' ? -cmp : cmp; + }); + } + } + if (opts.offset) out = out.slice(opts.offset); + if (opts.limit) out = out.slice(0, opts.limit); + return out; + }, + async insert(table: string, data: any) { + const t = rowsOf(table); + t.push({ ...data }); + tables.set(table, t); + return { id: data.id }; + }, + async update(table: string, patch: any) { + const r = rowsOf(table).find((x) => x.id === patch.id); + if (!r) throw new Error(`row ${patch.id} not found in ${table}`); + Object.assign(r, patch); + return r; + }, + async delete(table: string, opts: any) { + // [#4550] Pinned to ObjectQL.delete's OWN dispatch predicate. A double + // looser than the engine it stands in for is how #4434 shipped a REST + // route that answered 500 to every caller with its suite green — and + // the half a hand-written mirror drops is exactly the scalar test + // (`where: { id: { $in: [...] } }` looks like an id and is not one). + const dispatch = assertEngineDeleteDispatch(opts); + if (dispatch.kind === 'multi') { + const survivors = rowsOf(table).filter((r) => !matches(r, opts?.where ?? {})); + const deleted = rowsOf(table).length - survivors.length; + tables.set(table, survivors); + return { deleted }; + } + tables.set(table, rowsOf(table).filter((r) => r.id !== dispatch.id)); + return { id: dispatch.id }; + }, + }; +} + +function fakeCtx(services: Record) { + const hooks: Record Promise | void>> = {}; + const logger = { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }; + return { + logger, + getService: (name: string): T => { + if (!(name in services)) throw new Error(`service '${name}' not registered`); + return services[name] as T; + }, + registerService: (name: string, svc: unknown) => { services[name] = svc; }, + hook: (name: string, fn: () => Promise | void) => { (hooks[name] ??= []).push(fn); }, + /** Mirrors ObjectKernel's `context.trigger`: a throwing handler fails the boot. */ + fire: async (name: string) => { for (const fn of hooks[name] ?? []) await fn(); }, + }; +} + +/** Last element — the package's tsconfig lib predates Array.prototype.at. */ +function last(arr: T[]): T | undefined { return arr[arr.length - 1]; } + +const MAIL_DEFAULTS: Record = { + provider: { value: 'log', source: 'global' }, + from_email: { value: 'no-reply@example.test', source: 'global' }, + from_name: { value: 'ObjectStack', source: 'default' }, + queue_delivery: { value: false, source: 'default' }, +}; + +/** A movable clock so queue backoff can be stepped over without real waiting. */ +function fakeClock(startMs = Date.UTC(2026, 7, 4, 12, 0, 0)) { + let t = startMs; + return { now: () => new Date(t), advance: (ms: number) => { t += ms; } }; +} + +interface BootOpts { + /** 'db' = real DbQueueAdapter, 'degraded' = the kernel's in-memory fallback, 'none' = no service. */ + queue?: 'db' | 'degraded' | 'none'; + mail?: Record; + plugin?: Record; + transport?: { send: (m: any) => Promise }; +} + +async function boot(opts: BootOpts = {}) { + const engine = fakeEngine(); + const clock = fakeClock(); + const settings = fakeSettings({ ...MAIL_DEFAULTS, ...(opts.mail ?? {}) }); + const transport = opts.transport ?? { send: vi.fn(async () => ({ messageId: '' })) }; + + const adapter = new DbQueueAdapter({ + engine: engine as never, + clock, + options: { autoStart: false, pollIntervalMs: 60_000, defaultMaxAttempts: 3 }, + }); + const services: Record = { + manifest: { register: () => {} }, + objectql: engine, + settings, + }; + const mode = opts.queue ?? 'db'; + if (mode === 'db') services.queue = adapter; + if (mode === 'degraded') services.queue = createMemoryQueue(); + + const ctx = fakeCtx(services); + const plugin = new EmailServicePlugin({ seedTemplates: false, transport, ...(opts.plugin ?? {}) }); + await plugin.init(ctx as never); + await plugin.start(ctx as never); + const ready = () => ctx.fire('kernel:ready'); + return { + plugin, ctx, engine, settings, adapter, clock, transport, ready, + service: () => services.email as EmailService, + sysEmail: () => engine.rows('sys_email'), + jobs: () => engine.rows('sys_job_queue'), + }; +} + +// ── the durable-queue discriminator ──────────────────────────────────────── + +describe('resolveDurableQueue', () => { + it('rejects the kernel in-memory fallback — it cannot carry a durable job', () => { + // ObjectKernel pre-injects this on every boot with no queue plugin. It + // delivers synchronously and un-awaited, with no durability, retry or DLQ, + // so `getService('queue') !== undefined` is not the question to ask. + const fallback = createMemoryQueue(); + expect(typeof fallback.publish).toBe('function'); + expect(resolveDurableQueue(() => fallback)).toBeUndefined(); + }); + + it('accepts a real adapter, and reports absence as absence', () => { + const adapter = new DbQueueAdapter({ engine: fakeEngine() as never, options: { autoStart: false } }); + expect(resolveDurableQueue(() => adapter)).toBe(adapter); + expect(resolveDurableQueue(() => { throw new Error('not registered'); })).toBeUndefined(); + expect(resolveDurableQueue(() => ({}))).toBeUndefined(); + }); +}); + +// ── end to end ───────────────────────────────────────────────────────────── + +describe('queue delivery — round trip', () => { + it('send() returns queued, then a worker poll advances the SAME row to sent', async () => { + const h = await boot({ plugin: { queueDelivery: true } }); + await h.ready(); + + const res = await h.service().send({ to: 'a@b.com', subject: 'Hi', text: 'hello' }); + + // 1. Answer to the caller — and it is true, not hopeful. + expect(res).toMatchObject({ status: 'queued' }); + expect(h.transport.send).not.toHaveBeenCalled(); + // 2. The row is in the database… + expect(h.sysEmail()).toHaveLength(1); + expect(h.sysEmail()[0]).toMatchObject({ id: res.id, status: 'queued', attempt_count: 0 }); + // 3. …and the job is in sys_job_queue, referencing it. A process death + // here loses neither; that is the difference from the momentary + // `queued` the inline path writes. + expect(h.jobs()).toHaveLength(1); + expect(h.jobs()[0]).toMatchObject({ queue: EMAIL_SEND_QUEUE, status: 'pending' }); + expect(JSON.parse(h.jobs()[0].payload_json)).toEqual({ rowId: res.id }); + + await h.adapter.pollOnce(); + + expect(h.sysEmail()).toHaveLength(1); // still ONE row + expect(h.sysEmail()[0]).toMatchObject({ + id: res.id, status: 'sent', message_id: '', attempt_count: 1, + }); + expect(h.jobs()[0]).toMatchObject({ status: 'completed', attempts: 1 }); + }); + + it('SMTP 535: the queue retries with backoff, exhausts, and DLQs — one row throughout', async () => { + // The regression the issue names: the old subscriber called `send()`, so + // each redelivery INSERTED a new sys_email row. Five attempts, five rows, + // four of them stuck at `failed`, none carrying the real attempt count. + const send = vi.fn(async () => { throw new Error('535 5.7.8 authentication failed'); }); + const h = await boot({ + transport: { send }, + // retries: 2 ⇒ 3 total attempts, in the queue rather than in-process. + plugin: { queueDelivery: true, retries: 2 }, + }); + await h.ready(); + + const res = await h.service().send({ to: 'a@b.com', subject: 'Hi', text: 'hello' }); + expect(h.jobs()[0].max_attempts).toBe(3); + + for (let attempt = 1; attempt <= 3; attempt++) { + await h.adapter.pollOnce(); + expect(send, `transport calls after attempt ${attempt}`).toHaveBeenCalledTimes(attempt); + // ONE row, its attempt_count accumulating on it. + expect(h.sysEmail(), `sys_email rows after attempt ${attempt}`).toHaveLength(1); + expect(h.sysEmail()[0].attempt_count).toBe(attempt); + h.clock.advance(10 * 60_000); // step over the backoff window + } + + // Terminal state: the job is dead-lettered, the row records the failure. + expect(h.jobs()[0]).toMatchObject({ status: 'dlq', attempts: 3 }); + expect(h.jobs()[0].last_error).toMatch(/535/); + expect(h.sysEmail()[0]).toMatchObject({ id: res.id, status: 'failed', attempt_count: 3 }); + expect(h.sysEmail()[0].error).toMatch(/535 5\.7\.8/); + + // …and it is reachable for an operator to replay. + const failed = await h.adapter.listFailed(EMAIL_SEND_QUEUE); + expect(failed).toHaveLength(1); + expect(failed[0].data).toEqual({ rowId: res.id }); + }); + + it('a transport that recovers on the second attempt sends once, on one row', async () => { + let calls = 0; + const send = vi.fn(async () => { + if (++calls === 1) throw new Error('421 service unavailable'); + return { messageId: '' }; + }); + const h = await boot({ transport: { send }, plugin: { queueDelivery: true, retries: 3 } }); + await h.ready(); + + const res = await h.service().send({ to: 'a@b.com', subject: 'Hi', text: 'hello' }); + await h.adapter.pollOnce(); + expect(h.sysEmail()[0]).toMatchObject({ status: 'failed', attempt_count: 1 }); + + h.clock.advance(10 * 60_000); + await h.adapter.pollOnce(); + + expect(h.sysEmail()).toHaveLength(1); + expect(h.sysEmail()[0]).toMatchObject({ + id: res.id, status: 'sent', message_id: '', attempt_count: 2, + }); + expect(h.jobs()[0]).toMatchObject({ status: 'completed' }); + }); + + it('ignores a redelivery of a row that is already sent', async () => { + const h = await boot({ plugin: { queueDelivery: true } }); + await h.ready(); + const res = await h.service().send({ to: 'a@b.com', subject: 'Hi', text: 'hello' }); + await h.adapter.pollOnce(); + expect(h.transport.send).toHaveBeenCalledTimes(1); + + // A lease expiry / a second worker puts the same job back on the wire. + await h.engine.update('sys_job_queue', { id: h.jobs()[0].id, status: 'pending' }); + await h.adapter.pollOnce(); + + expect(h.transport.send).toHaveBeenCalledTimes(1); // never sent twice + expect(h.sysEmail()[0]).toMatchObject({ id: res.id, status: 'sent', attempt_count: 1 }); + }); + + it('does not retry forever on a vanished row — it reports the loss and completes', async () => { + const h = await boot({ plugin: { queueDelivery: true } }); + await h.ready(); + const res = await h.service().send({ to: 'a@b.com', subject: 'Hi', text: 'hello' }); + await h.engine.delete('sys_email', { where: { id: res.id } }); + + await h.adapter.pollOnce(); + + // No number of retries makes a deleted row reappear. + expect(h.jobs()[0]).toMatchObject({ status: 'completed' }); + const line = String(last(h.ctx.logger.error.mock.calls)?.[0] ?? ''); + expect(line).toMatch(/no longer exists/); + expect(line).toMatch(/NEVER be delivered/); + }); + + it('still accepts the legacy payload shape, delivering it inline exactly once', async () => { + // Producers written against the pre-#5160 subscriber publish a raw + // SendEmailInput. Routing that back through `send()` while queue mode is + // on would re-publish the message being consumed — an endless loop. + const h = await boot({ plugin: { queueDelivery: true } }); + await h.ready(); + + await h.adapter.publish(EMAIL_SEND_QUEUE, { to: 'a@b.com', subject: 'Legacy', text: 'x' }); + await h.adapter.pollOnce(); + + expect(h.transport.send).toHaveBeenCalledTimes(1); + expect(h.sysEmail()).toHaveLength(1); + expect(h.sysEmail()[0]).toMatchObject({ subject: 'Legacy', status: 'sent' }); + expect(h.jobs()).toHaveLength(1); // no re-publish + expect(h.jobs()[0].status).toBe('completed'); + }); +}); + +// ── default path ─────────────────────────────────────────────────────────── + +describe('queue delivery off (default)', () => { + it('sends inline and publishes nothing, even with a durable queue mounted', async () => { + const h = await boot(); + await h.ready(); + + const res = await h.service().send({ to: 'a@b.com', subject: 'Hi', text: 'hello' }); + + expect(res).toMatchObject({ status: 'sent', messageId: '' }); + expect(h.jobs()).toHaveLength(0); + expect(h.sysEmail()[0]).toMatchObject({ status: 'sent', attempt_count: 1 }); + }); +}); + +// ── gate 1: constructor / CLI declaration ────────────────────────────────── + +describe('constructor gate — an undeliverable declaration fails the boot', () => { + it('throws when no queue service is registered', async () => { + const h = await boot({ queue: 'none', plugin: { queueDelivery: true } }); + await expect(h.ready()).rejects.toThrow(/queueDelivery is enabled but no durable queue service/); + }); + + it('throws when the only queue is the kernel in-memory fallback', async () => { + // The case a bare presence check would wave through, and the reason this + // gate reads `__serviceInfo` instead: the fallback has a `publish`, so + // `send()` would answer `queued` for a job nothing can retry or recover. + const h = await boot({ queue: 'degraded', plugin: { queueDelivery: true } }); + await expect(h.ready()).rejects.toThrow(/no durability, retry or DLQ/); + }); + + it('throws when sys_email persistence is off — nothing for a job to reference', async () => { + const h = await boot({ plugin: { queueDelivery: true, persist: false } }); + await expect(h.ready()).rejects.toThrow(/persistence is disabled/); + }); + + it('names both the consequence and the way out', async () => { + const h = await boot({ queue: 'none', plugin: { queueDelivery: true } }); + await expect(h.ready()).rejects.toThrow(/lost if it dies/); + const h2 = await boot({ queue: 'none', plugin: { queueDelivery: true } }); + await expect(h2.ready()).rejects.toThrow(/OS_EMAIL_QUEUE_ENABLED=false/); + }); + + it('boots normally when the declaration CAN be honoured', async () => { + const h = await boot({ plugin: { queueDelivery: true } }); + await expect(h.ready()).resolves.toBeUndefined(); + }); + + it('does not fire for an undeclared deployment — no queue, no queue mode, no error', async () => { + const h = await boot({ queue: 'none' }); + await expect(h.ready()).resolves.toBeUndefined(); + const res = await h.service().send({ to: 'a@b.com', subject: 'Hi', text: 'x' }); + expect(res.status).toBe('sent'); + }); +}); + +// ── gate 2: settings page ────────────────────────────────────────────────── + +describe('settings gate — a save degrades, it never breaks the boot or the mail', () => { + it('hot-enables queue delivery when the toggle is saved on', async () => { + const h = await boot(); + await h.ready(); + expect((await h.service().send({ to: 'a@b.com', subject: 'Before', text: 'x' })).status).toBe('sent'); + + await h.settings.save({ queue_delivery: { value: true, source: 'global' } }); + + const res = await h.service().send({ to: 'a@b.com', subject: 'After', text: 'x' }); + expect(res.status).toBe('queued'); + expect(h.jobs()).toHaveLength(1); + expect(JSON.parse(h.jobs()[0].payload_json)).toEqual({ rowId: res.id }); + expect(h.ctx.logger.error).not.toHaveBeenCalled(); + }); + + it('hot-disables it again', async () => { + const h = await boot({ mail: { queue_delivery: { value: true, source: 'global' } } }); + await h.ready(); + expect((await h.service().send({ to: 'a@b.com', subject: 'On', text: 'x' })).status).toBe('queued'); + + await h.settings.save({ queue_delivery: { value: false, source: 'global' } }); + + expect((await h.service().send({ to: 'a@b.com', subject: 'Off', text: 'x' })).status).toBe('sent'); + expect(h.jobs()).toHaveLength(1); // only the first send's job + }); + + it('with no durable queue: logs at error, keeps sending inline, does NOT throw', async () => { + const h = await boot({ queue: 'none' }); + await h.ready(); + + await h.settings.save({ queue_delivery: { value: true, source: 'global' } }); + + const line = String(h.ctx.logger.error.mock.calls[0][0]); + expect(line).toMatch(/durable queue delivery/i); + expect(line).toMatch(/still being SENT/); + expect(line).toMatch(/service-queue/); + // A save must not stop the mail: what degraded is durability, not delivery. + const res = await h.service().send({ to: 'a@b.com', subject: 'Hi', text: 'x' }); + expect(res.status).toBe('sent'); + expect(last(h.sysEmail())).toMatchObject({ status: 'sent' }); + }); + + it('leaves the constructor declaration alone while the toggle is merely defaulted', async () => { + // `source: 'default'` is nobody's decision. Reading it as one would let a + // settings page no operator has opened switch off a declared deployment + // mode on the next save of an unrelated field. + const h = await boot({ + plugin: { queueDelivery: true }, + mail: { queue_delivery: { value: false, source: 'default' } }, + }); + await h.ready(); + + await h.settings.save({ from_name: { value: 'Acme', source: 'global' } }); + + expect((await h.service().send({ to: 'a@b.com', subject: 'Hi', text: 'x' })).status).toBe('queued'); + }); +}); + +// ── mail/test ────────────────────────────────────────────────────────────── + +describe('mail/test is never queued', () => { + it('sends inline and reports the transport result, with queue mode on', async () => { + const h = await boot({ + plugin: { queueDelivery: true }, + mail: { queue_delivery: { value: true, source: 'global' } }, + }); + await h.ready(); + + const result = await h.settings.action('test')!({ + values: { provider: 'log', from_email: 'no-reply@example.test' }, + payload: {}, + ctx: { body: { to: 'ops@example.test' } }, + }); + + // The answer came from the TRANSPORT — it either sent or it did not — + // and never from the queue. "Queued" is exactly the non-answer #5087 + // removed from this button. + expect(result.ok).toBe(true); + expect(String(result.message)).toMatch(/Sent test email to ops@example\.test/); + expect(h.transport.send).toHaveBeenCalledTimes(1); + expect(h.jobs()).toHaveLength(0); + // The audit row is finalized, not left at `queued` for a worker. + expect(last(h.sysEmail())).toMatchObject({ status: 'sent' }); + }); + + it('reports a real SMTP failure instead of enqueueing it', async () => { + const h = await boot({ + transport: { send: vi.fn(async () => { throw new Error('535 5.7.8 authentication failed'); }) }, + plugin: { queueDelivery: true }, + mail: { queue_delivery: { value: true, source: 'global' } }, + }); + await h.ready(); + + const result = await h.settings.action('test')!({ + values: { provider: 'log', from_email: 'no-reply@example.test' }, + payload: {}, + ctx: { body: { to: 'ops@example.test' } }, + }); + + expect(result.ok).toBe(false); + expect(String(result.message)).toMatch(/535 5\.7\.8/); + expect(h.jobs()).toHaveLength(0); + }); +}); + +beforeEach(() => { vi.clearAllMocks(); }); diff --git a/packages/plugins/plugin-email/src/email-plugin.ts b/packages/plugins/plugin-email/src/email-plugin.ts index daf0b32595..89ab1c8362 100644 --- a/packages/plugins/plugin-email/src/email-plugin.ts +++ b/packages/plugins/plugin-email/src/email-plugin.ts @@ -6,9 +6,19 @@ import type { IEmailTransport, EmailAddress, IMetadataService, + IQueueService, + QueueBackoffPolicy, } from '@objectstack/spec/contracts'; import { SysEmail, SysEmailTemplate } from '@objectstack/platform-objects/audit'; -import { EmailService, LogTransport, type EmailPersistence, type TemplateLoader, type EmailTemplateRow } from './email-service.js'; +import { + EmailService, + LogTransport, + EMAIL_SEND_QUEUE, + type EmailPersistence, + type EmailQueueDelivery, + type TemplateLoader, + type EmailTemplateRow, +} from './email-service.js'; import { makeTransport, SmtpTransport, @@ -66,6 +76,60 @@ export interface EmailServicePluginOptions { seedTemplates?: boolean; /** Additional templates seeded alongside the built-ins. */ templates?: EmailTemplate[]; + /** + * Deliver through the durable `queue` service instead of inline (#5160). + * Default false — inline delivery, unchanged. + * + * When `true`, `send()` persists the `sys_email` row, publishes an + * `email.send.async` job referencing it, and returns `status: 'queued'` + * straight away; a worker delivers the row and finalizes it in place, and + * `retries` becomes the queue's attempt budget (`retries + 1` attempts, + * exponentially backed off, then DLQ) instead of an in-process loop. + * + * Declaring it `true` here — the constructor / `OS_EMAIL_QUEUE_ENABLED` + * channel — is a DEPLOYMENT declaration, so a boot that cannot honour it + * fails rather than starting half-configured (#5132 precedent). The + * settings-page toggle is the opposite trade: it degrades to inline + * delivery and says so, because one save must not stop the mail. + */ + queueDelivery?: boolean; +} + +/** + * Backoff applied to queued deliveries: 1s, 2s, 4s … capped at 5 minutes. + * + * Deliberately unlike the inline loop's 2s ceiling — an SMTP host that just + * rejected a connection is rarely ready 2s later, and the whole point of + * moving the retry into `sys_job_queue` is that waiting minutes costs nothing + * (no process is held open across it). + */ +const QUEUE_DELIVERY_BACKOFF: QueueBackoffPolicy = { + type: 'exponential', + delayMs: 1000, + maxDelayMs: 5 * 60_000, +}; + +/** + * Resolve a queue service that can actually carry a durable email job, or + * `undefined`. + * + * The presence of a service named `queue` is NOT the question. `ObjectKernel` + * pre-injects an in-memory fallback for `queue` on every boot that lacks a + * queue plugin (`createMemoryQueue`), and that fallback delivers synchronously, + * un-awaited, with no durability, no retry and no DLQ. Publishing to it would + * let `send()` answer `queued` for a message nothing can ever retry — the + * declared-but-not-delivered gap #5087 closed for transports, re-opened one + * layer over. It labels itself for exactly this purpose + * (`__serviceInfo.status === 'degraded'`, ADR-0076 D12), so read the label. + */ +export function resolveDurableQueue(getService: (name: string) => unknown): IQueueService | undefined { + let queue: any; + try { queue = getService('queue'); } catch { return undefined; } + if (!queue || typeof queue.publish !== 'function' || typeof queue.subscribe !== 'function') { + return undefined; + } + if (queue.__serviceInfo?.status === 'degraded') return undefined; + return queue as IQueueService; } /** @@ -98,6 +162,14 @@ export class EmailServicePlugin implements Plugin { private unsubscribeTemplates?: () => void; /** SMTP transport currently in use, if any — closed in dispose(). */ private liveSmtp?: SmtpTransport; + /** + * The `mail` settings page's override of `options.queueDelivery`. + * `undefined` means no operator has touched the toggle (the manifest + * default still resolves it), so the constructor declaration stands — + * the same "is this value SELECTED or merely defaulted?" reading + * `applyMailSettings` already applies to `provider`. + */ + private queueDeliveryFromSettings?: boolean; constructor(options: EmailServicePluginOptions = {}) { this.options = options; @@ -179,7 +251,7 @@ export class EmailServicePlugin implements Plugin { try { const settings = ctx.getService('settings'); if (settings && typeof settings.createClient === 'function') { - const applySettings = async () => { + const applySettings = async (phase: 'boot' | 'saved' = 'boot') => { try { const payload = await settings.getNamespace('mail'); const values: Record = {}; @@ -188,16 +260,16 @@ export class EmailServicePlugin implements Plugin { values[k] = v?.value; if (v?.source) sources[k] = String(v.source); } - this.applyMailSettings(values, sources, ctx); + this.applyMailSettings(values, sources, ctx, phase); } catch (err: any) { ctx.logger.warn('EmailServicePlugin: failed to apply mail settings: ' + (err?.message ?? err)); } }; - await applySettings(); + await applySettings('boot'); // Subscribe to namespace changes; rebuild on every update. if (typeof settings.subscribe === 'function') { settings.subscribe('mail', () => { - void applySettings(); + void applySettings('saved'); }); ctx.logger.info('EmailServicePlugin: bound to settings:changed for namespace=mail'); } @@ -303,7 +375,10 @@ export class EmailServicePlugin implements Plugin { } try { - const result = await target.send({ + // ALWAYS inline, never the queue (#5160). The operator pressed + // a button and is waiting for the SMTP server's own answer; + // "queued" would be the same non-answer #5087 removed here. + const result = await target.sendInline({ to, from: merged.from_email ? { address: String(merged.from_email), @@ -387,6 +462,12 @@ export class EmailServicePlugin implements Plugin { this.service.setTemplateLoader(templateLoader); ctx.logger.info('EmailServicePlugin: sys_email persistence + template loader enabled'); + // Re-apply the delivery mode now that persistence exists: queue + // delivery references a `sys_email` row, so it is only meaningful once + // there is somewhere to write one. (`applyMailSettings` above may + // already have set the flag; this recomputes the same answer.) + this.applyQueueDelivery(ctx); + // ── sys_email OUTBOX DRAIN (afterInsert) ───────────────────────── // Apps that can only `api.write` (e.g. sandboxed action bodies, which // expose no `api.email`) cannot reach the email service directly — the @@ -446,24 +527,101 @@ export class EmailServicePlugin implements Plugin { ctx.logger.info('EmailServicePlugin: sys_email outbox drain hook installed'); } - // Bind 'email.send.async' queue subscriber for durable, retry-on-failure delivery. - // Producers: `queue.publish('email.send.async', sendInput, { maxAttempts: 5, backoff: {...} })` - // The queue handles retry / DLQ via sys_job_queue. + // ── 'email.send.async' SUBSCRIBER ──────────────────────────────── + // The consuming half of queue delivery (#5160). The canonical payload + // is `{ rowId }` — the id of a `sys_email` row `send()` already + // persisted — and the worker finalizes THAT row in place. + // + // It used to be `svc.send(msg.data)`, which inserted a brand-new + // sys_email row on every delivery: a message the queue retried 5 times + // left 5 rows, four of them permanently `failed`, none of them carrying + // the true attempt count. One message is one row; `attempt_count` + // accumulates on it across redeliveries. try { const queue: any = ctx.getService('queue'); if (queue && typeof queue.subscribe === 'function' && this.service) { const svc = this.service; - await queue.subscribe('email.send.async', async (msg: any) => { - const result = await svc.send(msg.data); + await queue.subscribe(EMAIL_SEND_QUEUE, async (msg: any) => { + const data = msg?.data; + const rowId = typeof data?.rowId === 'string' && data.rowId ? data.rowId : ''; + // `msg.attempts` is 1-based for the CURRENT delivery, so the + // attempts already spent on this row is one fewer. + const priorAttempts = Math.max(0, Number(msg?.attempts ?? 1) - 1); + + if (!rowId) { + // Migration window: producers written against the pre-#5160 + // subscriber publish a raw SendEmailInput. Deliver it inline — + // routing it through `send()` while queue mode is on would + // re-publish the very message being consumed. + const legacy = await svc.sendInline(data); + if (legacy.status === 'failed') throw new Error(legacy.error ?? 'email send failed'); + return; + } + + const rows = await (engine as any).find('sys_email', { + where: { id: rowId }, + limit: 1, + context: SYSTEM_CTX, + }); + const row = Array.isArray(rows) ? rows[0] : (rows as any)?.data?.[0]; + if (!row) { + // Do NOT throw: no number of retries makes a deleted row + // reappear, and burning the attempt budget only moves the same + // dead job to the DLQ later. Report it — a send that was + // accepted and can never be delivered is a durability loss, and + // it will look like nothing happened at all. + ctx.logger.error( + `EmailServicePlugin: sys_email row '${rowId}' referenced by an ${EMAIL_SEND_QUEUE} job no longer ` + + 'exists — that message will NEVER be delivered and the caller was already told it was queued. ' + + 'Fix: stop deleting sys_email rows in `queued` state (it is an append-only log), or drain the ' + + 'queue before purging.', + ); + return; + } + // Already delivered — a duplicate delivery (lease expiry, a + // second worker) must not send the mail twice. + if (row.status === 'sent' || row.message_id) return; + + // maxAttempts: 1 — the QUEUE owns retrying. Letting the row loop + // retry underneath it would multiply the two budgets together. + const result = await svc.deliverPersistedRow(row, { maxAttempts: 1, priorAttempts }); if (result.status === 'failed') { // Force the queue to retry / DLQ by throwing throw new Error(result.error ?? 'email send failed'); } }); - ctx.logger.info('EmailServicePlugin: subscribed to email.send.async queue'); + ctx.logger.info(`EmailServicePlugin: subscribed to ${EMAIL_SEND_QUEUE} queue`); } } catch (err) { - ctx.logger.warn('EmailServicePlugin: email.send.async subscription failed', err as any); + ctx.logger.warn(`EmailServicePlugin: ${EMAIL_SEND_QUEUE} subscription failed`, err as any); + } + + // ── CONSTRUCTOR / CLI GATE (#5160, #5132 precedent) ────────────── + // `queueDelivery: true` from the constructor (or OS_EMAIL_QUEUE_ENABLED) + // is a deployment declaration: this server was told to make mail + // delivery survive its own restart. If it cannot, the honest answer is a + // failed boot, not a server that looks configured and silently retries + // in-process — the same judgement #5132 made for a provider that cannot + // deliver. + // + // Asserted HERE, at kernel:ready, and not in `init()`: during Phase 1 + // the queue provider may simply not have registered yet, and the + // kernel's own core-service fallbacks are injected only after that + // phase. A verdict recorded then would be contradicted by the same boot + // (AGENTS.md — "never record a verdict the boot can still contradict"). + if (this.options.queueDelivery === true) { + const blocker = this.queueDeliveryBlocker(ctx, !!persistence); + if (blocker) { + throw new Error( + `EmailServicePlugin: queueDelivery is enabled but ${blocker}, so mail would keep being delivered ` + + 'inline — a send that fails would be retried only in this process and lost if it dies, which is ' + + 'the durability this option was switched on to get. Fix: mount the queue capability backed by a ' + + 'durable adapter (@objectstack/service-queue over an ObjectQL engine, which upgrades to the ' + + 'sys_job_queue DbQueueAdapter)' + + (persistence ? '' : ' and leave sys_email persistence on (`persist` must not be false)') + + ', or set OS_EMAIL_QUEUE_ENABLED=false / omit `queueDelivery` to declare inline delivery.', + ); + } } // Seed built-in + user-provided templates (upsert by name+locale). @@ -494,6 +652,57 @@ export class EmailServicePlugin implements Plugin { }); } + /** + * Effective delivery mode: the settings toggle when an operator has set + * one, otherwise the constructor / CLI declaration. + */ + private queueDeliveryEnabled(): boolean { + return this.queueDeliveryFromSettings ?? this.options.queueDelivery === true; + } + + /** + * Build the wiring handed to {@link EmailService}. `resolve` is a thunk on + * purpose: `QueueServicePlugin` swaps its in-memory placeholder for the + * `sys_job_queue`-backed adapter during `kernel:ready`, so a handle captured + * now would publish into the discarded one for the life of the process + * (AGENTS.md — "resolve where it is used, not where you start"). + */ + private makeQueueDelivery(ctx: PluginContext): EmailQueueDelivery { + return { + resolve: () => resolveDurableQueue((name) => ctx.getService(name)), + // ONE retry budget, shared by both modes. `retries` already means + // "extra attempts after the first" for inline delivery; queued, the + // same number becomes the queue's total attempt cap. A second knob + // here would let the two layers disagree, and nesting a row-level + // loop inside a queue-level one would multiply them. + maxAttempts: Math.max(1, (this.options.retries ?? 0) + 1), + backoff: QUEUE_DELIVERY_BACKOFF, + }; + } + + /** Push the effective delivery mode onto the running service. */ + private applyQueueDelivery(ctx: PluginContext): void { + if (!this.service) return; + this.service.setQueueDelivery( + this.queueDeliveryEnabled() ? this.makeQueueDelivery(ctx) : undefined, + ); + } + + /** + * Why queue delivery cannot be honoured right now, or `undefined` when it + * can. Phrased as a sentence fragment for both gates to embed. + */ + private queueDeliveryBlocker(ctx: PluginContext, hasPersistence: boolean): string | undefined { + if (!hasPersistence) { + return 'sys_email persistence is disabled (`persist: false`), so a queued job would have no row to deliver'; + } + if (!resolveDurableQueue((name) => ctx.getService(name))) { + return 'no durable queue service is registered (the kernel\'s in-memory fallback delivers synchronously ' + + 'with no durability, retry or DLQ, so it cannot carry this)'; + } + return undefined; + } + /** * [#4509] Materialize declared `email_template` metadata, bind the provenance * stamp, and keep the rows live for runtime authoring. @@ -613,9 +822,43 @@ export class EmailServicePlugin implements Plugin { values: Record, sources: Record, ctx: PluginContext, + phase: 'boot' | 'saved' = 'boot', ): void { if (!this.service) return; + // ── Delivery mode (#5160) ──────────────────────────────────────── + // Handled before every early `return` below, so a provider that cannot + // be built does not also strand the delivery mode at its old value. + // Only an operator-SELECTED value overrides the constructor declaration; + // the manifest default (`source: 'default'`) is not a decision anyone + // made, and treating it as one would let a settings page nobody opened + // silently switch off a mode the deployment declared. + if ((sources.queue_delivery ?? 'default') !== 'default') { + this.queueDeliveryFromSettings = values.queue_delivery === true + || String(values.queue_delivery).toLowerCase() === 'true'; + } + this.applyQueueDelivery(ctx); + if (this.queueDeliveryEnabled() && phase === 'saved') { + // Reported on a SAVE, not at boot: at boot the queue provider may + // simply not have registered yet, and a verdict recorded then is one + // this very boot can contradict. After a save the registry is settled + // and this is the operator's immediate feedback. Inline delivery + // continues either way — this save must not stop the mail, so what is + // lost is durability, not delivery, and the line says which. + const blocker = this.queueDeliveryBlocker(ctx, !!this.service.options.persistence); + if (blocker) { + ctx.logger.error( + `EmailServicePlugin: "durable queue delivery" is ON but ${blocker} — mail is still being SENT, ` + + 'inline, but a failed send is retried only in this process and lost if it dies (no sys_job_queue ' + + 'job, no DLQ). Fix: mount the queue capability with a durable adapter ' + + '(@objectstack/service-queue over an ObjectQL engine), or turn the toggle off so inline delivery ' + + 'is the declared intent.', + ); + } else { + ctx.logger.info('EmailServicePlugin: durable queue delivery enabled from mail settings.'); + } + } + const fromEmail = typeof values.from_email === 'string' ? values.from_email : undefined; const fromName = typeof values.from_name === 'string' ? values.from_name : undefined; if (fromEmail) this.service.setDefaultFrom({ address: fromEmail, name: fromName }); diff --git a/packages/plugins/plugin-email/src/email-service.queue-delivery.test.ts b/packages/plugins/plugin-email/src/email-service.queue-delivery.test.ts new file mode 100644 index 0000000000..727b4c8989 --- /dev/null +++ b/packages/plugins/plugin-email/src/email-service.queue-delivery.test.ts @@ -0,0 +1,386 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. +// +// EmailService — durable queue delivery (#5160). +// +// What these pin, in one sentence each: +// - the DEFAULT is untouched: with no `queueDelivery` wiring nothing is ever +// published and the inline path runs exactly as before; +// - with it wired, `send()` persists the row, publishes a job that REFERENCES +// that row, and returns `queued` without touching the transport; +// - the retry budget exists once — `retries` either drives the inline loop or +// becomes the queue's `maxAttempts`, never both; +// - every way queue delivery can be unavailable degrades to inline delivery +// and SAYS SO at `error`, once. Mail must not stop because a queue is +// missing; the operator must not be left thinking it is durable when it is +// not. + +import { describe, it, expect, vi } from 'vitest'; +import type { IQueueService } from '@objectstack/spec/contracts'; +import { + EmailService, + EMAIL_SEND_QUEUE, + type EmailPersistence, + type EmailQueueDelivery, +} from './email-service.js'; + +interface Published { + queue: string; + data: any; + options: any; +} + +function makePersistence() { + const rows = new Map>(); + const p: EmailPersistence = { + async insert(row) { rows.set(row.id, { ...row }); return { id: row.id }; }, + async update(id, patch) { + const cur = rows.get(id); + if (cur) rows.set(id, { ...cur, ...patch }); + }, + }; + return { p, rows }; +} + +/** A queue that records publishes and never delivers on its own. */ +function makeQueue() { + const published: Published[] = []; + const queue = { + published, + async publish(queue: string, data: any, options: any) { + published.push({ queue, data, options }); + return `msg-${published.length}`; + }, + async subscribe() { /* the worker half lives in the plugin */ }, + async unsubscribe() { /* noop */ }, + }; + return queue as typeof queue & IQueueService; +} + +function makeLogger() { + return { info: vi.fn(), warn: vi.fn(), error: vi.fn() }; +} + +function wiring(queue: IQueueService | undefined, maxAttempts = 1): EmailQueueDelivery { + return { + resolve: () => queue, + maxAttempts, + backoff: { type: 'exponential', delayMs: 1000, maxDelayMs: 300_000 }, + }; +} + +const MSG = { to: 'a@b.com', subject: 'Hi', text: 'hello' }; + +describe('EmailService — queue delivery off (default)', () => { + it('never publishes and delivers inline, exactly as before', async () => { + const transport = { send: vi.fn(async () => ({ messageId: '' })) }; + const queue = makeQueue(); + const { p, rows } = makePersistence(); + const svc = new EmailService({ transport, defaultFrom: 'no@reply.com', persistence: p }); + + const res = await svc.send(MSG); + + expect(res).toMatchObject({ status: 'sent', messageId: '' }); + expect(queue.published).toHaveLength(0); + expect(transport.send).toHaveBeenCalledTimes(1); + expect(rows.get(res.id)).toMatchObject({ status: 'sent', attempt_count: 1 }); + }); +}); + +describe('EmailService — queue delivery on', () => { + it('persists the row, publishes a job REFERENCING it, and returns queued without sending', async () => { + const transport = { send: vi.fn(async () => ({ messageId: '' })) }; + const queue = makeQueue(); + const { p, rows } = makePersistence(); + const svc = new EmailService({ + transport, defaultFrom: 'no@reply.com', persistence: p, queueDelivery: wiring(queue), + }); + + const res = await svc.send({ ...MSG, relatedObject: 'lead', relatedId: 'L1' }); + + // The claim `send()` makes to its caller. + expect(res).toMatchObject({ status: 'queued' }); + expect(res.messageId).toBeUndefined(); + // …and the two facts that make it true. + expect(rows.get(res.id)).toMatchObject({ + status: 'queued', + to_addresses: 'a@b.com', + subject: 'Hi', + related_object: 'lead', + attempt_count: 0, + }); + expect(queue.published).toHaveLength(1); + expect(queue.published[0].queue).toBe(EMAIL_SEND_QUEUE); + // The payload is the ROW ID, never the message — that is what keeps N + // retries on one row instead of inserting N rows. + expect(queue.published[0].data).toEqual({ rowId: res.id }); + // Nothing was sent yet: a worker does that. + expect(transport.send).not.toHaveBeenCalled(); + }); + + it('publishes with the retry budget derived from `retries`, plus backoff and idempotency', async () => { + const transport = { send: vi.fn(async () => ({ messageId: '' })) }; + const queue = makeQueue(); + const { p } = makePersistence(); + const svc = new EmailService({ + transport, defaultFrom: 'no@reply.com', persistence: p, + // `retries: 4` ⇒ 5 total attempts, whichever mode delivers them. + retries: 4, queueDelivery: wiring(queue, 5), + }); + + const res = await svc.send(MSG); + + expect(queue.published[0].options).toMatchObject({ + maxAttempts: 5, + // Both spellings, consistently: `maxAttempts` is canonical, `retries` is + // the legacy field MemoryQueueAdapter reads. Publishing only one leaves + // some adapters silently doing a single attempt. + retries: 4, + backoff: { type: 'exponential', delayMs: 1000, maxDelayMs: 300_000 }, + idempotencyKey: `sys_email:${res.id}`, + }); + }); + + it('routes sendTemplate through the queue too', async () => { + const transport = { send: vi.fn(async () => ({ messageId: '' })) }; + const queue = makeQueue(); + const { p } = makePersistence(); + const svc = new EmailService({ + transport, defaultFrom: 'no@reply.com', persistence: p, queueDelivery: wiring(queue), + templateLoader: { + async load() { + return { name: 'welcome', locale: 'en-US', subject: 'Welcome {{name}}', body_html: '

Hi {{name}}

' }; + }, + }, + }); + + const res = await svc.sendTemplate({ template: 'welcome', to: 'a@b.com', data: { name: 'Ada' } }); + + expect(res.status).toBe('queued'); + expect(queue.published).toHaveLength(1); + expect(queue.published[0].data).toEqual({ rowId: res.id }); + expect(transport.send).not.toHaveBeenCalled(); + }); + + it('still marks its row managed during the insert, so the drain hook skips it', async () => { + // The afterInsert outbox drain fires inside persistence.insert. In queue + // mode the row IS still `queued` when it fires, so without the managed + // flag the hook and the queue worker would both deliver it. + let managedAtInsert: boolean | undefined; + const queue = makeQueue(); + const transport = { send: vi.fn(async () => ({ messageId: '' })) }; + let svc!: EmailService; + const persistence: EmailPersistence = { + async insert(row) { + managedAtInsert = svc.isServiceManaged(String(row.id)); + return { id: row.id }; + }, + async update() { /* noop */ }, + }; + svc = new EmailService({ + transport, defaultFrom: 'no@reply.com', persistence, queueDelivery: wiring(queue), + }); + + const res = await svc.send(MSG); + + expect(res.status).toBe('queued'); + expect(managedAtInsert).toBe(true); + expect(svc.isServiceManaged(res.id)).toBe(false); + }); + + it('sendInline() bypasses the queue — the mail/test path', async () => { + const transport = { send: vi.fn(async () => ({ messageId: '' })) }; + const queue = makeQueue(); + const { p } = makePersistence(); + const svc = new EmailService({ + transport, defaultFrom: 'no@reply.com', persistence: p, queueDelivery: wiring(queue), + }); + + const res = await svc.sendInline(MSG); + + expect(res).toMatchObject({ status: 'sent', messageId: '' }); + expect(queue.published).toHaveLength(0); + }); + + it('delivers a message with attachments inline rather than queueing it stripped', async () => { + // sys_email has no attachment / header columns, so a row cannot rebuild + // them. Queueing such a message would deliver it WITHOUT the attachment — + // silent data loss wearing durability's clothes. + const transport = { send: vi.fn(async () => ({ messageId: '' })) }; + const queue = makeQueue(); + const { p } = makePersistence(); + const logger = makeLogger(); + const svc = new EmailService({ + transport, defaultFrom: 'no@reply.com', persistence: p, logger, queueDelivery: wiring(queue), + }); + + const res = await svc.send({ ...MSG, attachments: [{ filename: 'a.txt', content: 'hi' }] }); + + expect(res.status).toBe('sent'); + expect(queue.published).toHaveLength(0); + expect(transport.send).toHaveBeenCalledWith(expect.objectContaining({ + attachments: [{ filename: 'a.txt', content: 'hi' }], + })); + expect(logger.info).toHaveBeenCalledWith(expect.stringContaining('attachments')); + // A capability gap, not a failure: nothing is logged at error. + expect(logger.error).not.toHaveBeenCalled(); + }); + + it('does the same for custom headers', async () => { + const transport = { send: vi.fn(async () => ({ messageId: '' })) }; + const queue = makeQueue(); + const { p } = makePersistence(); + const svc = new EmailService({ + transport, defaultFrom: 'no@reply.com', persistence: p, queueDelivery: wiring(queue), + }); + + const res = await svc.send({ ...MSG, headers: { 'X-Campaign': 'spring' } }); + + expect(res.status).toBe('sent'); + expect(queue.published).toHaveLength(0); + }); +}); + +describe('EmailService — queue delivery enabled but unavailable', () => { + it('falls back to inline delivery and reports the lost durability ONCE, at error', async () => { + const transport = { send: vi.fn(async () => ({ messageId: '' })) }; + const { p, rows } = makePersistence(); + const logger = makeLogger(); + const svc = new EmailService({ + transport, defaultFrom: 'no@reply.com', persistence: p, logger, + queueDelivery: wiring(undefined), + }); + + const first = await svc.send(MSG); + const second = await svc.send(MSG); + + // Degrading persistence must never degrade DELIVERY. + expect(first).toMatchObject({ status: 'sent' }); + expect(second).toMatchObject({ status: 'sent' }); + expect(rows.get(first.id)).toMatchObject({ status: 'sent' }); + + // Said once, not once per send. + expect(logger.error).toHaveBeenCalledTimes(1); + const line = String(logger.error.mock.calls[0][0]); + expect(line).toMatch(/no durable queue service/); + // Consequence AND fix, in the first line (AGENTS.md degradation-log-level). + expect(line).toMatch(/lost if it dies/); + expect(line).toMatch(/service-queue/); + }); + + it('reports the same way when persistence is off — no row, nothing to reference', async () => { + const transport = { send: vi.fn(async () => ({ messageId: '' })) }; + const queue = makeQueue(); + const logger = makeLogger(); + const svc = new EmailService({ + transport, defaultFrom: 'no@reply.com', logger, queueDelivery: wiring(queue), + }); + + const res = await svc.send(MSG); + + expect(res.status).toBe('sent'); + expect(queue.published).toHaveLength(0); + expect(String(logger.error.mock.calls[0][0])).toMatch(/persistence is disabled/); + }); + + it('delivers inline when publish() throws, rather than stranding a committed row', async () => { + const transport = { send: vi.fn(async () => ({ messageId: '' })) }; + const { p, rows } = makePersistence(); + const logger = makeLogger(); + const broken = { + async publish() { throw new Error('sys_job_queue write failed'); }, + async subscribe() { /* noop */ }, + async unsubscribe() { /* noop */ }, + } as unknown as IQueueService; + const svc = new EmailService({ + transport, defaultFrom: 'no@reply.com', persistence: p, logger, queueDelivery: wiring(broken), + }); + + const res = await svc.send(MSG); + + // The row was already committed at `queued`; leaving it for a job that was + // never created is precisely the stuck-forever state this feature exists + // to remove. + expect(res).toMatchObject({ status: 'sent', messageId: '' }); + expect(rows.get(res.id)).toMatchObject({ status: 'sent' }); + expect(String(logger.error.mock.calls[0][0])).toMatch(/sys_job_queue write failed/); + }); + + it('speaks again after the toggle is switched off and back on', async () => { + const transport = { send: vi.fn(async () => ({ messageId: '' })) }; + const { p } = makePersistence(); + const logger = makeLogger(); + const svc = new EmailService({ + transport, defaultFrom: 'no@reply.com', persistence: p, logger, + queueDelivery: wiring(undefined), + }); + + await svc.send(MSG); + expect(logger.error).toHaveBeenCalledTimes(1); + + svc.setQueueDelivery(undefined); + await svc.send(MSG); + expect(logger.error).toHaveBeenCalledTimes(1); // off ⇒ nothing to report + + svc.setQueueDelivery(wiring(undefined)); + await svc.send(MSG); + expect(logger.error).toHaveBeenCalledTimes(2); // re-enabled ⇒ re-armed + }); +}); + +describe('EmailService.deliverPersistedRow — attempt accounting', () => { + const row = () => ({ + id: 'row-1', status: 'queued', from_address: 'no@reply.com', + to_addresses: 'a@b.com', subject: 'Hi', body_text: 'hello', + }); + + it('defaults are unchanged: retries+1 attempts, attempt_count from 1', async () => { + const transport = { send: vi.fn(async () => { throw new Error('smtp 421'); }) }; + const { p, rows } = makePersistence(); + rows.set('row-1', row()); + const svc = new EmailService({ transport, persistence: p, retries: 1 }); + + const res = await svc.deliverPersistedRow(row()); + + expect(transport.send).toHaveBeenCalledTimes(2); + expect(res.status).toBe('failed'); + expect(rows.get('row-1')).toMatchObject({ status: 'failed', attempt_count: 2 }); + }); + + it('honours maxAttempts:1 so the queue owns the retry (no 5x5 multiplication)', async () => { + const transport = { send: vi.fn(async () => { throw new Error('smtp 535'); }) }; + const { p, rows } = makePersistence(); + rows.set('row-1', row()); + // retries: 4 would be 5 inline attempts. Under the queue it must be one. + const svc = new EmailService({ transport, persistence: p, retries: 4 }); + + await svc.deliverPersistedRow(row(), { maxAttempts: 1, priorAttempts: 0 }); + + expect(transport.send).toHaveBeenCalledTimes(1); + expect(rows.get('row-1')).toMatchObject({ attempt_count: 1 }); + }); + + it('accumulates attempt_count across redeliveries of the SAME row', async () => { + const transport = { send: vi.fn(async () => { throw new Error('smtp 535'); }) }; + const { p, rows } = makePersistence(); + rows.set('row-1', row()); + const svc = new EmailService({ transport, persistence: p }); + + for (let attempt = 1; attempt <= 3; attempt++) { + await svc.deliverPersistedRow(row(), { maxAttempts: 1, priorAttempts: attempt - 1 }); + expect(rows.get('row-1')).toMatchObject({ attempt_count: attempt }); + } + expect(rows.size).toBe(1); // one message, one row + }); + + it('carries the prior attempts onto a success too', async () => { + const transport = { send: vi.fn(async () => ({ messageId: '' })) }; + const { p, rows } = makePersistence(); + rows.set('row-1', row()); + const svc = new EmailService({ transport, persistence: p }); + + const res = await svc.deliverPersistedRow(row(), { maxAttempts: 1, priorAttempts: 2 }); + + expect(res).toMatchObject({ status: 'sent', messageId: '' }); + expect(rows.get('row-1')).toMatchObject({ status: 'sent', attempt_count: 3 }); + }); +}); diff --git a/packages/plugins/plugin-email/src/email-service.ts b/packages/plugins/plugin-email/src/email-service.ts index 1602f435f6..c026f00872 100644 --- a/packages/plugins/plugin-email/src/email-service.ts +++ b/packages/plugins/plugin-email/src/email-service.ts @@ -10,9 +10,76 @@ import type { EmailAddress, EmailDeliveryStatus, TransportSendResult, + IQueueService, + QueueBackoffPolicy, } from '@objectstack/spec/contracts'; import { renderTemplate, requireVars, htmlToText } from './template-engine.js'; +/** + * Queue topic durable email delivery is published to and consumed from + * (#5160). Exported so producer (`EmailService.send` in queue mode) and + * consumer (`EmailServicePlugin`'s subscriber) name it once. + */ +export const EMAIL_SEND_QUEUE = 'email.send.async'; + +/** + * Payload published to {@link EMAIL_SEND_QUEUE}. + * + * It carries the **id of an already-persisted `sys_email` row**, never the + * message itself. That is the whole point of the shape: the worker delivers + * THAT row and finalizes it in place, so N queue retries of one message stay + * one row with a cumulative `attempt_count`. The pre-#5160 subscriber called + * `send()` with the raw input, which inserted a fresh row per retry. + */ +export interface EmailSendQueuePayload { + /** `sys_email.id` of the row to deliver. */ + rowId: string; +} + +/** + * Queue-delivery wiring handed to {@link EmailService} by + * `EmailServicePlugin`. Present ⇒ `send()` enqueues instead of delivering + * inline; absent ⇒ today's inline path, byte for byte. + */ +export interface EmailQueueDelivery { + /** + * Resolve the live queue service, or `undefined` when none can carry a + * durable job. Resolved **per send**, never captured once: the queue + * service is replaced during boot (`QueueServicePlugin` upgrades its + * in-memory placeholder to the DB adapter on `kernel:ready`), and a + * handle captured early would publish into the discarded one. + */ + resolve(): IQueueService | undefined; + /** + * Total delivery attempts the QUEUE makes before the message goes to the + * DLQ. In queue mode this is the ONLY retry budget: the per-row loop in + * {@link EmailService.deliverPersistedRow} is pinned to a single attempt + * per delivery, so the two layers cannot multiply (#5160). + */ + maxAttempts: number; + /** Backoff between queue attempts. */ + backoff: QueueBackoffPolicy; +} + +/** + * Per-call attempt accounting for {@link EmailService.deliverPersistedRow}. + * + * Both fields default to today's behaviour (`priorAttempts: 0`, + * `maxAttempts: retries + 1`), so every existing caller is unaffected. + */ +export interface DeliverAttemptOptions { + /** + * Transport attempts to make in THIS call. The queue worker passes `1` — + * it owns retrying, the row loop must not retry underneath it. + */ + maxAttempts?: number; + /** + * Attempts already spent on this row by earlier calls, so `attempt_count` + * accumulates across queue redeliveries instead of resetting to 1. + */ + priorAttempts?: number; +} + /** * Internal persistence shim — typed loosely so the service can run * without an ObjectQL engine wired (e.g. unit tests, serverless). @@ -217,18 +284,37 @@ export interface EmailServiceOptions { logger?: { info: (msg: string, meta?: any) => void; warn: (msg: string, meta?: any) => void; error?: (msg: string, meta?: any) => void }; /** Default render context merged into every sendTemplate call (e.g. `{ appName }`). */ defaultTemplateContext?: Record; + /** + * Durable delivery through the `queue` service (#5160). Set ⇒ `send()` + * persists the `sys_email` row, publishes {@link EMAIL_SEND_QUEUE} + * referencing it, and returns `status: 'queued'` immediately. Unset ⇒ + * inline delivery (the default, unchanged). + */ + queueDelivery?: EmailQueueDelivery; } /** * Concrete IEmailService implementation. * - * Flow: + * Inline flow (the default): * 1. Validate + normalize input (throws on bad input). * 2. Persist queued row to sys_email (best-effort; failures logged). * 3. Call transport.send(); on success, update row to sent + * timestamp + messageId. On failure, mark failed + error. * 4. Return SendEmailResult with the persisted row id (or a fresh * id when persistence is disabled). + * + * Queue flow (`options.queueDelivery` wired, #5160) — steps 1-2 identical, + * then publish {@link EMAIL_SEND_QUEUE} referencing the row and return + * `status: 'queued'`. A worker later calls {@link deliverPersistedRow} on + * that same row, so step 3 happens out of process and survives a restart. + * + * **Retries live in exactly one layer.** Inline, `options.retries` drives the + * loop in `deliverNormalized`. Queued, the SAME number becomes the queue's + * `maxAttempts` (`retries + 1` total attempts) and the row loop is pinned to + * one attempt per delivery. Flipping the mode changes WHERE a retry happens + * — durable and backed off, instead of in-process and capped at 2s — never + * HOW MANY happen, and the two layers can never multiply into 5x5. */ export class EmailService implements IEmailService { /** @@ -240,6 +326,13 @@ export class EmailService implements IEmailService { */ private readonly managedRowIds = new Set(); + /** + * Set once queue delivery has been reported as unavailable, so the + * degradation is stated at the FIRST send and not on every one + * (AGENTS.md — "say it once, at the first degradation"). + */ + private queueDegradationReported = false; + constructor(public options: EmailServiceOptions) { if (!options.transport) throw new Error('EmailService: transport is required'); } @@ -274,7 +367,42 @@ export class EmailService implements IEmailService { this.options.defaultFrom = from; } + /** + * Turn durable queue delivery on (pass the wiring) or off (pass + * `undefined`) on a running service — the `mail` settings toggle path. + * Re-arms the one-shot degradation report so a re-enable is allowed to + * speak again. + */ + setQueueDelivery(queueDelivery: EmailQueueDelivery | undefined): void { + this.options.queueDelivery = queueDelivery; + this.queueDegradationReported = false; + } + + /** + * Send through the configured delivery mode: durable queue when + * {@link EmailServiceOptions.queueDelivery} is wired AND usable, + * inline otherwise. + */ async send(input: SendEmailInput): Promise { + return this.sendInternal(input, true); + } + + /** + * Send **synchronously through the transport**, never the queue. + * + * Two callers need this regardless of the configured mode: + * - `mail/test` — the "Send test email" button must answer with the SMTP + * server's own words ("535 authentication failed"), and "queued" is + * exactly the kind of non-answer #5087 removed from that button; + * - the `email.send.async` subscriber's legacy arm, where the payload is a + * raw `SendEmailInput` — routing that back through `send()` in queue + * mode would re-publish the message it is currently consuming. + */ + async sendInline(input: SendEmailInput): Promise { + return this.sendInternal(input, false); + } + + private async sendInternal(input: SendEmailInput, allowQueue: boolean): Promise { let normalized: NormalizedEmailMessage; try { normalized = normalizeMessage(input, this.options.defaultFrom); @@ -283,6 +411,9 @@ export class EmailService implements IEmailService { throw err; } + // `undefined` ⇒ every statement below is the pre-#5160 inline path. + const queue = allowQueue ? this.resolveQueueForSend(input) : undefined; + const id = newId(); const baseRow: Record = { id, @@ -317,12 +448,123 @@ export class EmailService implements IEmailService { } } const rowId = persistedId ?? id; + if (queue) { + // Queue mode delivers the ROW, so a row that never landed leaves the + // job with nothing to reference. Deliver inline instead of publishing + // a job that can only fail — the insert failure was already reported + // above, and dropping the message would be the worse answer. + if (persistedId === undefined) { + this.reportQueueDegradation( + 'the sys_email row could not be persisted, so a queued job would have nothing to deliver', + ); + } else if (await this.publishRow(queue, rowId)) { + // The row is in the database at `queued` and the job is in the + // queue's own store: a process death here loses neither. + return { id: rowId, status: 'queued' }; + } + } return await this.deliverNormalized(rowId, normalized); } finally { this.managedRowIds.delete(id); } } + /** + * Resolve the queue to publish THIS message to, or `undefined` to deliver + * it inline. + * + * Returning `undefined` is never silent when queue delivery was asked for: + * the first time it happens the service reports at `error` what is no + * longer durable and how to restore it, then stays quiet. Mail keeps + * flowing either way — what degrades is persistence of the retry, not + * delivery, so a missing queue must not become a missing email. + */ + private resolveQueueForSend(input: SendEmailInput): IQueueService | undefined { + const wiring = this.options.queueDelivery; + if (!wiring) return undefined; + + // `sys_email` carries no attachment or header columns, so a row cannot + // reconstruct them (`rowToNormalized`). Queueing such a message would + // deliver it stripped — silent data loss dressed as durability. Deliver + // it inline, where the in-memory message is still intact. Tracked for a + // real fix (attachment storage) rather than papered over. + if (input.attachments?.length || (input.headers && Object.keys(input.headers).length > 0)) { + this.options.logger?.info( + 'EmailService: queue delivery skipped for one message — sys_email cannot carry attachments or custom ' + + 'headers, so the message was delivered inline (in-process retries only) rather than stripped.', + ); + return undefined; + } + + if (!this.options.persistence) { + this.reportQueueDegradation( + 'sys_email persistence is disabled, so there is no row for a queued job to reference', + ); + return undefined; + } + const queue = wiring.resolve(); + if (!queue) { + this.reportQueueDegradation('no durable queue service is available'); + return undefined; + } + return queue; + } + + /** + * Publish the job that owns this row's delivery. Returns false when the + * publish failed, in which case the caller delivers inline — the row is + * already committed at `queued`, and leaving it for nobody to pick up is + * the durability gap this feature exists to close. + */ + private async publishRow(queue: IQueueService, rowId: string): Promise { + const wiring = this.options.queueDelivery; + if (!wiring) return false; + try { + await queue.publish(EMAIL_SEND_QUEUE, { rowId }, { + maxAttempts: wiring.maxAttempts, + // `maxAttempts` is the canonical field; `retries` is its legacy + // spelling and is the ONLY one `MemoryQueueAdapter` reads. Both are + // declared on QueuePublishOptions, so filling both consistently is + // completing the contract, not tolerating a dialect — publishing one + // budget that some adapters silently read as 1 would be the lie. + retries: Math.max(0, wiring.maxAttempts - 1), + backoff: wiring.backoff, + // One row, one job: a re-publish for the same row (a retried caller, + // a redelivered upstream event) collapses onto the existing message + // instead of racing a second worker onto the same sys_email row. + idempotencyKey: `sys_email:${rowId}`, + metadata: { object: 'sys_email', rowId }, + }); + return true; + } catch (err: any) { + this.reportQueueDegradation( + `publishing to '${EMAIL_SEND_QUEUE}' failed (${String(err?.message ?? err)})`, + ); + return false; + } + } + + /** + * State a queue-delivery degradation once, at `error`. + * + * `error` and not `warn`: from the outside everything still looks normal — + * `send()` returns, the row is written, the mail goes out — while the + * durable-retry guarantee the operator switched on is not in force. That is + * the durability class AGENTS.md pins at `error`, and the line owes both the + * consequence and the fix. + */ + private reportQueueDegradation(reason: string): void { + if (this.queueDegradationReported) return; + this.queueDegradationReported = true; + this.options.logger?.error?.( + `EmailService: queue delivery is enabled but ${reason} — mail is still being SENT, inline, but a send ` + + 'that fails is retried only in this process and is lost if it dies (no sys_job_queue job, no DLQ). ' + + 'Fix: mount the queue capability with a durable adapter (@objectstack/service-queue with an ObjectQL ' + + 'engine, so it upgrades to the sys_job_queue-backed DbQueueAdapter), or turn Settings → Mail → ' + + '"Durable queue delivery" off to make inline delivery the declared intent.', + ); + } + /** * Deliver a normalized message through the transport (with retry) and * finalize the persisted `sys_email` row (`sent` + message_id + sent_at, @@ -331,8 +573,11 @@ export class EmailService implements IEmailService { private async deliverNormalized( rowId: string, normalized: NormalizedEmailMessage, + opts?: DeliverAttemptOptions, ): Promise { - const maxAttempts = (this.options.retries ?? 0) + 1; + // Defaults reproduce the pre-#5160 loop exactly. + const maxAttempts = Math.max(1, opts?.maxAttempts ?? (this.options.retries ?? 0) + 1); + const priorAttempts = Math.max(0, opts?.priorAttempts ?? 0); let lastError: any; for (let attempt = 1; attempt <= maxAttempts; attempt++) { try { @@ -343,7 +588,7 @@ export class EmailService implements IEmailService { status, message_id: messageId, sent_at: new Date().toISOString(), - attempt_count: attempt, + attempt_count: priorAttempts + attempt, }); return { id: rowId, status, messageId }; } catch (err: any) { @@ -358,7 +603,7 @@ export class EmailService implements IEmailService { await this.updateRow(rowId, { status: 'failed', error: errMessage, - attempt_count: maxAttempts, + attempt_count: priorAttempts + maxAttempts, }); return { id: rowId, status: 'failed', error: errMessage }; } @@ -371,8 +616,15 @@ export class EmailService implements IEmailService { * plugin's afterInsert hook calls this to actually transmit it. Unlike * `send()`, this does NOT insert a new row — it reconstructs the message * from the row columns and finalizes that same row in place. + * + * Also the queue worker's entry point (#5160): it passes + * `{ maxAttempts: 1, priorAttempts }` so the queue owns retrying and + * `attempt_count` accumulates on the one row across redeliveries. */ - async deliverPersistedRow(row: Record): Promise { + async deliverPersistedRow( + row: Record, + opts?: DeliverAttemptOptions, + ): Promise { const rowId = String(row?.id ?? ''); if (!rowId) throw new Error('deliverPersistedRow: row.id is required'); let normalized: NormalizedEmailMessage; @@ -380,10 +632,14 @@ export class EmailService implements IEmailService { normalized = rowToNormalized(row); } catch (err: any) { const errMessage = String(err?.message ?? err ?? 'invalid row').slice(0, 1000); - await this.updateRow(rowId, { status: 'failed', error: errMessage, attempt_count: 0 }); + await this.updateRow(rowId, { + status: 'failed', + error: errMessage, + attempt_count: Math.max(0, opts?.priorAttempts ?? 0), + }); return { id: rowId, status: 'failed', error: errMessage }; } - return this.deliverNormalized(rowId, normalized); + return this.deliverNormalized(rowId, normalized, opts); } private async updateRow(id: string, patch: Record): Promise { diff --git a/packages/plugins/plugin-email/src/index.ts b/packages/plugins/plugin-email/src/index.ts index 4cb7c00fc0..90825c0037 100644 --- a/packages/plugins/plugin-email/src/index.ts +++ b/packages/plugins/plugin-email/src/index.ts @@ -14,10 +14,18 @@ * endpoints. `EMAIL_TRANSPORT_PROVIDERS` is the machine-readable form. */ -export { EmailServicePlugin } from './email-plugin.js'; +export { EmailServicePlugin, resolveDurableQueue } from './email-plugin.js'; export type { EmailServicePluginOptions } from './email-plugin.js'; -export { LogTransport, normalizeMessage, formatAddress } from './email-service.js'; -export type { EmailServiceOptions, TemplateLoader, EmailTemplateRow, EmailPersistence } from './email-service.js'; +export { LogTransport, normalizeMessage, formatAddress, EMAIL_SEND_QUEUE } from './email-service.js'; +export type { + EmailServiceOptions, + TemplateLoader, + EmailTemplateRow, + EmailPersistence, + EmailQueueDelivery, + EmailSendQueuePayload, + DeliverAttemptOptions, +} from './email-service.js'; export { renderTemplate, requireVars, htmlToText } from './template-engine.js'; export { ResendTransport, diff --git a/packages/services/service-settings/src/manifests/mail.manifest.test.ts b/packages/services/service-settings/src/manifests/mail.manifest.test.ts index 0632688410..63c9a27fa9 100644 --- a/packages/services/service-settings/src/manifests/mail.manifest.test.ts +++ b/packages/services/service-settings/src/manifests/mail.manifest.test.ts @@ -83,6 +83,31 @@ describe('mailSettingsManifest', () => { } }); + it('offers durable queue delivery as an opt-in toggle, visible for every provider', () => { + // framework#5160. Off by default because turning it on changes what + // `send()` RETURNS (`queued` instead of `sent`/`failed`) — that is an + // opt-in, never something a workspace should acquire silently on upgrade. + const queue = spec('queue_delivery'); + expect(queue.type).toBe('toggle'); + expect(queue.default).toBe(false); + expect(queue.required).not.toBe(true); + // Delivery mode is orthogonal to the provider — no `visible` expression, + // so it does not disappear when the provider changes. + expect(queue.visible).toBeUndefined(); + expect(group('delivery')).toBeDefined(); + }); + + it('tells the operator what the toggle costs and what it needs', () => { + const description = String(spec('queue_delivery').description); + // The requirement, so "I turned it on and nothing changed" is answerable. + expect(description).toMatch(/queue capability/i); + // The behaviour change the caller sees. + expect(description).toMatch(/retried/i); + // And the exception, because the button right below it is the one thing + // that keeps sending inline (#5087 — it must report the provider's answer). + expect(description).toMatch(/test email/i); + }); + it('exposes a test action that POSTs to /api/settings/mail/test', () => { const test = specs().find((s) => s.type === 'action_button' && s.id === 'test'); expect(test).toBeDefined(); diff --git a/packages/services/service-settings/src/manifests/mail.manifest.ts b/packages/services/service-settings/src/manifests/mail.manifest.ts index 447bc127ac..98540f0fa7 100644 --- a/packages/services/service-settings/src/manifests/mail.manifest.ts +++ b/packages/services/service-settings/src/manifests/mail.manifest.ts @@ -94,6 +94,20 @@ const manifest = { description: 'Example: no-reply@example.com' }, { type: 'text', key: 'from_name', label: 'From name', required: false, default: 'ObjectStack' }, + // Delivery mode (framework#5160). Off by default: turning it on changes + // what `send()` returns (`queued` instead of `sent`/`failed`), so it is + // an opt-in, never a silent upgrade. + { type: 'group', id: 'delivery', label: 'Delivery', required: false, + description: 'How outbound mail is handed to the provider.' }, + { type: 'toggle', key: 'queue_delivery', label: 'Durable queue delivery', required: false, + default: false, + description: 'Hand each message to the job queue (sys_job_queue) instead of sending it inline. ' + + 'Sends return as soon as the message is recorded, and a delivery that fails is retried with ' + + 'backoff by a worker — so it survives a restart — reaching the dead-letter queue only after the ' + + 'attempts are exhausted. Requires the queue capability with a durable adapter; without one, mail ' + + 'is still sent inline and the server logs why. "Send test email" below always sends inline, so it ' + + 'can still report the provider\'s own answer.' }, + { type: 'action_button', id: 'test', label: 'Send test email', required: false, icon: 'Send', handler: { kind: 'http', method: 'POST', url: '/api/settings/mail/test' } }, ], diff --git a/packages/services/service-settings/src/translations/en.ts b/packages/services/service-settings/src/translations/en.ts index 86bc453cde..d363f8717d 100644 --- a/packages/services/service-settings/src/translations/en.ts +++ b/packages/services/service-settings/src/translations/en.ts @@ -28,6 +28,7 @@ export const en: TranslationData = { smtp: { title: 'SMTP' }, api_key: { title: 'API key' }, from_address: { title: 'From address' }, + delivery: { title: 'Delivery', description: 'How outbound mail is handed to the provider.' }, }, keys: { provider: { @@ -49,6 +50,12 @@ export const en: TranslationData = { api_key: { label: 'API key' }, from_email: { label: 'From email', help: 'Example: no-reply@example.com' }, from_name: { label: 'From name' }, + queue_delivery: { + label: 'Durable queue delivery', + help: 'Hand each message to the job queue instead of sending it inline, so a failed delivery is ' + + 'retried with backoff by a worker and survives a restart. Requires the queue capability with a ' + + 'durable adapter. "Send test email" always sends inline.', + }, }, actions: { test: { label: 'Send test email' }, diff --git a/packages/services/service-settings/src/translations/es-ES.ts b/packages/services/service-settings/src/translations/es-ES.ts index e8b80e3830..7b3e56197d 100644 --- a/packages/services/service-settings/src/translations/es-ES.ts +++ b/packages/services/service-settings/src/translations/es-ES.ts @@ -24,6 +24,7 @@ export const esES: TranslationData = { smtp: { title: 'SMTP' }, api_key: { title: 'Clave de API' }, from_address: { title: 'Dirección de remitente' }, + delivery: { title: 'Entrega', description: 'Cómo se entrega el correo saliente al proveedor.' }, }, keys: { provider: { @@ -45,6 +46,12 @@ export const esES: TranslationData = { api_key: { label: 'Clave de API' }, from_email: { label: 'Correo del remitente', help: 'Ejemplo: no-reply@example.com' }, from_name: { label: 'Nombre del remitente' }, + queue_delivery: { + label: 'Entrega mediante cola duradera', + help: 'Entrega cada mensaje a la cola de trabajos en lugar de enviarlo en línea, de modo que una ' + + 'entrega fallida se reintenta con retroceso y sobrevive a un reinicio. Requiere la capacidad de ' + + 'cola con un adaptador duradero. "Enviar correo de prueba" siempre envía en línea.', + }, }, actions: { test: { label: 'Enviar correo de prueba' }, diff --git a/packages/services/service-settings/src/translations/ja-JP.ts b/packages/services/service-settings/src/translations/ja-JP.ts index b9c33d79cf..702c616170 100644 --- a/packages/services/service-settings/src/translations/ja-JP.ts +++ b/packages/services/service-settings/src/translations/ja-JP.ts @@ -24,6 +24,7 @@ export const jaJP: TranslationData = { smtp: { title: 'SMTP' }, api_key: { title: 'API キー' }, from_address: { title: '差出人アドレス' }, + delivery: { title: '配信方法', description: '送信メールをプロバイダーに渡す方法。' }, }, keys: { provider: { @@ -45,6 +46,12 @@ export const jaJP: TranslationData = { api_key: { label: 'API キー' }, from_email: { label: '差出人アドレス', help: '例: no-reply@example.com' }, from_name: { label: '差出人名' }, + queue_delivery: { + label: '永続キュー配信', + help: '各メッセージをインライン送信ではなくジョブキューに渡します。配信に失敗してもワーカーが' + + 'バックオフ付きで再試行し、再起動後も失われません。永続アダプターを備えたキュー機能が必要です。' + + '「テストメール送信」は常にインラインで送信します。', + }, }, actions: { test: { label: 'テストメール送信' }, diff --git a/packages/services/service-settings/src/translations/zh-CN.ts b/packages/services/service-settings/src/translations/zh-CN.ts index 23dc48b9f7..8ebea4d00a 100644 --- a/packages/services/service-settings/src/translations/zh-CN.ts +++ b/packages/services/service-settings/src/translations/zh-CN.ts @@ -24,6 +24,7 @@ export const zhCN: TranslationData = { smtp: { title: 'SMTP' }, api_key: { title: 'API 密钥' }, from_address: { title: '发件地址' }, + delivery: { title: '投递方式', description: '邮件以何种方式交给服务商。' }, }, keys: { provider: { @@ -44,6 +45,11 @@ export const zhCN: TranslationData = { api_key: { label: 'API 密钥' }, from_email: { label: '发件地址', help: '示例:no-reply@example.com' }, from_name: { label: '发件人名称' }, + queue_delivery: { + label: '持久化队列投递', + help: '把每封邮件交给任务队列而不是内联发送:投递失败由 worker 按退避重试,进程重启也不会丢失。' + + '需要挂载具备持久化适配器的队列能力。「发送测试邮件」始终走内联发送。', + }, }, actions: { test: { label: '发送测试邮件' }, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 54117488c0..523ea07444 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1472,6 +1472,12 @@ importers: specifier: ^9.0.3 version: 9.0.3 devDependencies: + '@objectstack/objectql': + specifier: workspace:* + version: link:../../objectql + '@objectstack/service-queue': + specifier: workspace:* + version: link:../../services/service-queue '@objectstack/service-settings': specifier: workspace:* version: link:../../services/service-settings