diff --git a/packages/plugins/plugin-email/src/email-plugin.ts b/packages/plugins/plugin-email/src/email-plugin.ts index d4cb2f1648..51cee58698 100644 --- a/packages/plugins/plugin-email/src/email-plugin.ts +++ b/packages/plugins/plugin-email/src/email-plugin.ts @@ -271,6 +271,65 @@ export class EmailServicePlugin implements Plugin { this.service.setTemplateLoader(templateLoader); ctx.logger.info('EmailServicePlugin: sys_email persistence + template loader enabled'); + // ── 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 + // only thing they CAN do is INSERT a sys_email row. Treat such a row, + // inserted as `status:'queued'` with no `message_id`, as an outbox + // entry: deliver it through the live transport, then finalize the SAME + // row in place (`sent`/`failed`). Without this, those rows sat at + // `queued` forever (declared-but-never-delivered). + // + // Rows that the service's own `send()` inserts are marked managed (see + // EmailService.isServiceManaged) and skipped here, so they are + // delivered exactly once by `send()` — never double-sent by the hook. + if (persistence && typeof (engine as any).registerHook === 'function') { + const svc = this.service; + const DRAIN_PKG = 'com.objectstack.service.email.drain'; + if (typeof (engine as any).unregisterHooksByPackage === 'function') { + (engine as any).unregisterHooksByPackage(DRAIN_PKG); + } + (engine as any).registerHook( + 'afterInsert', + async (hookCtx: any) => { + try { + if (hookCtx?.object !== 'sys_email') return; + const row = hookCtx?.result; + if (!row || typeof row !== 'object') return; + if (row.status !== 'queued' || row.message_id) return; + const rowId = row.id != null ? String(row.id) : ''; + if (!rowId || svc.isServiceManaged(rowId)) return; + // Defer past the current insert op: transport.send is network + // I/O and must not run inside the insert's transaction, and the + // row must be committed before we update it. Re-read under + // system context to get the full row + re-check it is still an + // undelivered queued entry (idempotent against concurrent drains). + setTimeout(() => { + void (async () => { + try { + const rows = await (engine as any).find('sys_email', { + where: { id: rowId }, + limit: 1, + context: SYSTEM_CTX, + }); + const fresh = Array.isArray(rows) ? rows[0] : (rows as any)?.data?.[0]; + const target = fresh ?? row; + if (target.status !== 'queued' || target.message_id) return; + await svc.deliverPersistedRow(target); + } catch (err: any) { + ctx.logger.warn(`EmailServicePlugin: outbox drain failed for ${rowId}: ${err?.message ?? err}`); + } + })(); + }, 0); + } catch (err: any) { + ctx.logger.warn(`EmailServicePlugin: outbox drain hook error: ${err?.message ?? err}`); + } + }, + { packageId: DRAIN_PKG }, + ); + 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. diff --git a/packages/plugins/plugin-email/src/email-service.test.ts b/packages/plugins/plugin-email/src/email-service.test.ts index 14b0e7071c..146238fe0a 100644 --- a/packages/plugins/plugin-email/src/email-service.test.ts +++ b/packages/plugins/plugin-email/src/email-service.test.ts @@ -6,6 +6,7 @@ import { LogTransport, formatAddress, normalizeMessage, + rowToNormalized, type EmailPersistence, } from './email-service.js'; @@ -139,4 +140,80 @@ describe('EmailService', () => { expect(res.status).toBe('sent'); expect(warn).toHaveBeenCalledWith(expect.stringContaining('persist failed'), expect.any(Object)); }); + + // ── Outbox-drain support (sys_email afterInsert) ─────────────────── + it('marks its own row managed during send() so the drain hook skips it', async () => { + // The persistence.insert callback fires at exactly the moment the + // afterInsert drain hook would fire — assert the row is flagged managed + // there, and unflagged once send() resolves. + let managedAtInsert: boolean | undefined; + let insertedId: string | undefined; + const transport = { send: vi.fn(async () => ({ messageId: '' })) }; + let svc!: EmailService; + const persistence: EmailPersistence = { + async insert(row) { + insertedId = String(row.id); + managedAtInsert = svc.isServiceManaged(insertedId); + return { id: row.id }; + }, + async update() { /* noop */ }, + }; + svc = new EmailService({ transport, defaultFrom: 'no@reply.com', persistence }); + const res = await svc.send({ to: 'a@b.com', subject: 'Hi', text: 'x' }); + expect(res.status).toBe('sent'); + expect(managedAtInsert).toBe(true); // hook would skip it + expect(svc.isServiceManaged(insertedId!)).toBe(false); // cleared after send + }); + + it('deliverPersistedRow delivers an existing row WITHOUT inserting a new one', async () => { + const transport = { send: vi.fn(async () => ({ messageId: '' })) }; + const { p, rows } = makePersistence(); + const insertSpy = vi.spyOn(p, 'insert'); + const svc = new EmailService({ transport, persistence: p }); + // Simulate an app-inserted outbox row. + const row = { + id: 'row-1', status: 'queued', from_address: 'no@reply.com', + to_addresses: 'a@b.com, c@d.com', subject: 'Drain me', body_text: 'hello', + }; + rows.set(row.id, { ...row }); + const res = await svc.deliverPersistedRow(row); + expect(res).toMatchObject({ id: 'row-1', status: 'sent', messageId: '' }); + expect(transport.send).toHaveBeenCalledTimes(1); + expect(transport.send).toHaveBeenCalledWith(expect.objectContaining({ + to: ['a@b.com', 'c@d.com'], from: 'no@reply.com', subject: 'Drain me', text: 'hello', + })); + expect(insertSpy).not.toHaveBeenCalled(); // never inserts a 2nd row + expect(rows.get('row-1')).toMatchObject({ status: 'sent', message_id: '' }); + }); + + it('deliverPersistedRow marks the row failed when it lacks a body', async () => { + const transport = { send: vi.fn(async () => ({ messageId: '' })) }; + const { p, rows } = makePersistence(); + const svc = new EmailService({ transport, persistence: p }); + rows.set('row-2', { id: 'row-2', status: 'queued', from_address: 'a@b.com', to_addresses: 'c@d.com', subject: 'No body' }); + const res = await svc.deliverPersistedRow({ id: 'row-2', from_address: 'a@b.com', to_addresses: 'c@d.com', subject: 'No body' }); + expect(res.status).toBe('failed'); + expect(transport.send).not.toHaveBeenCalled(); + expect(rows.get('row-2')).toMatchObject({ status: 'failed' }); + }); +}); + +describe('rowToNormalized', () => { + it('reconstructs a message from persisted columns', () => { + const msg = rowToNormalized({ + to_addresses: 'a@b.com, "B" ', from_address: 'no@reply.com', + subject: 'Hi', body_text: 'text', body_html: '

html

', + cc_addresses: 'cc@x.com', bcc_addresses: 'bcc@x.com', reply_to: 'r@x.com', + }); + expect(msg).toEqual({ + to: ['a@b.com', '"B" '], from: 'no@reply.com', subject: 'Hi', + text: 'text', html: '

html

', cc: ['cc@x.com'], bcc: ['bcc@x.com'], replyTo: 'r@x.com', + }); + }); + it('throws on missing to/from/subject/body', () => { + expect(() => rowToNormalized({ from_address: 'a@b.com', subject: 'x', body_text: 'y' })).toThrow(/to_addresses/); + expect(() => rowToNormalized({ to_addresses: 'a@b.com', subject: 'x', body_text: 'y' })).toThrow(/from_address/); + expect(() => rowToNormalized({ to_addresses: 'a@b.com', from_address: 'c@d.com', body_text: 'y' })).toThrow(/subject/); + expect(() => rowToNormalized({ to_addresses: 'a@b.com', from_address: 'c@d.com', subject: 'x' })).toThrow(/body/); + }); }); diff --git a/packages/plugins/plugin-email/src/email-service.ts b/packages/plugins/plugin-email/src/email-service.ts index 6504856c5e..abb46fdc06 100644 --- a/packages/plugins/plugin-email/src/email-service.ts +++ b/packages/plugins/plugin-email/src/email-service.ts @@ -95,6 +95,48 @@ export function normalizeMessage( return msg; } +/** Split a persisted comma-separated address column back into a list. */ +function splitAddresses(v: unknown): string[] { + if (v == null) return []; + return String(v) + .split(',') + .map((s) => s.trim()) + .filter(Boolean); +} + +/** + * Reconstruct a NormalizedEmailMessage from a persisted `sys_email` row + * (the outbox-drain path). Addresses are already canonicalized at insert + * time, so this re-splits the stored columns rather than re-validating. + * Throws when the row lacks the minimum fields needed to send. + */ +export function rowToNormalized(row: Record): NormalizedEmailMessage { + const to = splitAddresses(row.to_addresses); + if (to.length === 0) throw new Error('VALIDATION_FAILED: row has no to_addresses'); + const from = String(row.from_address ?? '').trim(); + if (!from) throw new Error('VALIDATION_FAILED: row has no from_address'); + const subject = String(row.subject ?? '').trim(); + if (!subject) throw new Error('VALIDATION_FAILED: row has no subject'); + const text = row.body_text; + const html = row.body_html; + if ((text == null || text === '') && (html == null || html === '')) { + throw new Error('VALIDATION_FAILED: row has neither body_text nor body_html'); + } + const msg: NormalizedEmailMessage = { + to, + from, + subject, + ...(text != null && text !== '' ? { text: String(text) } : {}), + ...(html != null && html !== '' ? { html: String(html) } : {}), + }; + const cc = splitAddresses(row.cc_addresses); + if (cc.length > 0) msg.cc = cc; + const bcc = splitAddresses(row.bcc_addresses); + if (bcc.length > 0) msg.bcc = bcc; + if (row.reply_to) msg.replyTo = String(row.reply_to); + return msg; +} + /** * Development transport — never actually sends. Logs to the provided * logger and returns a synthetic Message-ID. Useful for local dev, @@ -186,10 +228,24 @@ export interface EmailServiceOptions { * id when persistence is disabled). */ export class EmailService implements IEmailService { + /** + * Row ids the service is itself delivering (via `send()`). The + * EmailServicePlugin's sys_email afterInsert drain hook consults this + * set and skips these rows, so a service-originated `queued` insert is + * delivered exactly once (by `send()`) — never double-sent by the hook. + * App-originated raw inserts are absent here, so the hook handles them. + */ + private readonly managedRowIds = new Set(); + constructor(public options: EmailServiceOptions) { if (!options.transport) throw new Error('EmailService: transport is required'); } + /** True when this row id is currently being delivered by `send()`. */ + isServiceManaged(id: string): boolean { + return this.managedRowIds.has(id); + } + /** Wire (or replace) the template loader after construction. */ setTemplateLoader(loader: TemplateLoader): void { this.options.templateLoader = loader; @@ -242,17 +298,37 @@ export class EmailService implements IEmailService { attempt_count: 0, }; - let persistedId: string | undefined; - if (this.options.persistence) { - try { - const res = await this.options.persistence.insert(baseRow); - persistedId = typeof res === 'string' ? res : res?.id ?? id; - } catch (err: any) { - this.options.logger?.warn('EmailService: sys_email persist failed (non-fatal)', { error: err?.message }); + // Reserve the row id BEFORE persistence.insert so the drain hook + // (which fires synchronously inside that insert) sees it as managed + // and skips it — `send()` owns this row's delivery. + this.managedRowIds.add(id); + try { + let persistedId: string | undefined; + if (this.options.persistence) { + try { + const res = await this.options.persistence.insert(baseRow); + persistedId = typeof res === 'string' ? res : res?.id ?? id; + if (persistedId !== id) this.managedRowIds.add(persistedId); + } catch (err: any) { + this.options.logger?.warn('EmailService: sys_email persist failed (non-fatal)', { error: err?.message }); + } } + const rowId = persistedId ?? id; + return await this.deliverNormalized(rowId, normalized); + } finally { + this.managedRowIds.delete(id); } - const rowId = persistedId ?? id; + } + /** + * Deliver a normalized message through the transport (with retry) and + * finalize the persisted `sys_email` row (`sent` + message_id + sent_at, + * or `failed` + error). Shared by `send()` and `deliverPersistedRow()`. + */ + private async deliverNormalized( + rowId: string, + normalized: NormalizedEmailMessage, + ): Promise { const maxAttempts = (this.options.retries ?? 0) + 1; let lastError: any; for (let attempt = 1; attempt <= maxAttempts; attempt++) { @@ -284,6 +360,29 @@ export class EmailService implements IEmailService { return { id: rowId, status: 'failed', error: errMessage }; } + /** + * Deliver an ALREADY-PERSISTED `sys_email` row (the outbox-drain path). + * + * An app (e.g. a sandboxed action that can only `api.write`, never reach + * the email service) inserts a `sys_email` row as `status:'queued'`; the + * 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. + */ + async deliverPersistedRow(row: Record): Promise { + const rowId = String(row?.id ?? ''); + if (!rowId) throw new Error('deliverPersistedRow: row.id is required'); + let normalized: NormalizedEmailMessage; + try { + 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 }); + return { id: rowId, status: 'failed', error: errMessage }; + } + return this.deliverNormalized(rowId, normalized); + } + private async updateRow(id: string, patch: Record): Promise { if (!this.options.persistence?.update) return; try {