feat(email): drain queued sys_email rows so app inserts actually deliver (#2002)

Apps that can only `api.write` (e.g. sandboxed action bodies, which expose
no `api.email`) cannot reach the email service — the only thing they can do
is INSERT a `sys_email` row. Until now such rows sat at `status:'queued'`
forever: declared-but-never-delivered. The transport (and the queued→sent
transition) only ran inside `EmailService.send()`, which a raw insert never
invokes.

Treat a `sys_email` row inserted as `status:'queued'` with no `message_id`
as an OUTBOX entry: an afterInsert hook delivers it through the live
transport and finalizes the same row in place (sent + message_id + sent_at,
or failed + error). Delivery is deferred past the insert so transport I/O
never runs inside the insert's transaction, and the row is re-read + re-checked
under system context before sending (idempotent against concurrent drains).

Rows that the service's own `send()` inserts are tagged managed
(EmailService.isServiceManaged) and skipped by the hook, so they are
delivered exactly once by send() — never double-sent.

- email-service.ts: managedRowIds guard; extract deliverNormalized() shared
  by send() and the new deliverPersistedRow(); rowToNormalized() helper.
- email-plugin.ts: register the sys_email afterInsert drain hook.
- tests: managed-guard active at insert moment; deliverPersistedRow delivers
  without inserting a 2nd row + fails cleanly on bodyless rows; rowToNormalized.

Verified empirically in HotCRM: a raw INSERT of a queued sys_email now
transitions queued→sent with a transport message_id within ~600ms; before,
the identical insert stayed queued forever.

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: os-zhuang

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.

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: '<m@x>' })) };
+    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: '<drained@x>' })) };
+    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: '<drained@x>' });
+    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: '<drained@x>' });
+  });
+
+  it('deliverPersistedRow marks the row failed when it lacks a body', async () => {
+    const transport = { send: vi.fn(async () => ({ messageId: '<x>' })) };
+    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" <b@c.com>', from_address: 'no@reply.com',
+      subject: 'Hi', body_text: 'text', body_html: '<p>html</p>',
+      cc_addresses: 'cc@x.com', bcc_addresses: 'bcc@x.com', reply_to: 'r@x.com',
+    });
+    expect(msg).toEqual({
+      to: ['a@b.com', '"B" <b@c.com>'], from: 'no@reply.com', subject: 'Hi',
+      text: 'text', html: '<p>html</p>', 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/);
+  });
 });

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<string, any>): 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<string>();
+
   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<SendEmailResult> {
     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<string, any>): Promise<SendEmailResult> {
+    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<string, any>): Promise<void> {
     if (!this.options.persistence?.update) return;
     try {