fix(plugin-email)!: EmailPersistence.insert 必须回传 row.id,改 id 响亮拒绝而非双发 (#5523)

.changeset/email-persistence-insert-id-contract.md

@@ -0,0 +1,47 @@
+---
+"@objectstack/plugin-email": patch
+---
+
+fix(plugin-email)!: `EmailPersistence.insert` must return the row's own id — a substituted id is rejected instead of double-sending (#5523)
+
+**FROM** — `insert` could answer with an id of its own (a database-assigned
+primary key, an external delivery system's receipt id) and `EmailService.send()`
+adopted it: the substituted id was added to the service-managed set, used as the
+queued job's `rowId`, and returned to the caller.
+
+**TO** — `insert` must confirm the id it was handed. Returning a different id
+throws, naming the contract and the value returned, **before the message is
+delivered**.
+
+**Fix, one line:** return `{ id: row.id }` (or `row.id`) from `insert`. If your
+store assigns its own primary key, keep the service-minted id in the row's `id`
+column and record the store's key in a column of its own.
+
+Why the contract tightened rather than the service accommodating both: the id is
+minted by the service *before* the insert and is already load-bearing by the time
+`insert` is called — out-of-row attachment content has been uploaded under
+`sys_email/attachments/<row.id>/…`, so the row id is the only key that finds
+those bytes again. Re-keying the row also broke delivery exactly-once: the
+`sys_email` `afterInsert` outbox drain hook decides whether a freshly-inserted
+row is the service's to deliver by asking `isServiceManaged()` about **the
+inserted row's own id**, and that hook runs *inside* the insert — before `send()`
+had seen, let alone reserved, the substituted id. So the hook read the row as an
+application-inserted outbox entry and delivered it, while `send()` delivered it
+again down its own path: one message sent twice, two terminal updates racing on
+one row. The only thing that ever prevented it was the hook's `setTimeout(…, 0)`
+losing a race to `send()`'s inline delivery — and `transport.send` is real
+network I/O, so that race is normally lost.
+
+Scope of the check: it judges the confirmation's **value**, not its presence. An
+implementation that returns no id at all leaves nothing to disagree with (the
+drain hook reads the id off the inserted row, which is the minted one either
+way), so the mail still goes. An insert that *throws* is unchanged — that stays
+an operational condition the service rides out with a warning and inline
+delivery; only a *successful* insert that renames the row is fatal.
+
+Breaking for external `EmailPersistence` implementations that re-key the row —
+of which there are currently none: the in-repo implementation forwards the
+engine's own answer and ObjectQL honours the id it is handed. Filed at `patch`
+because the surface has no known external consumer and the declared TypeScript
+signature is unchanged; a maintainer who counts a narrowed public-interface
+contract as `minor`/`major` should relabel it.

packages/plugins/plugin-email/src/email-plugin.outbox-sweep.test.ts

@@ -31,6 +31,13 @@ const flush = () => new Promise((r) => setTimeout(r, 5));
 interface EngineOpts {
   /** Make `find` throw for the drain hook's own `where: { id }` re-read. */
   failIdLookup?: boolean;
+  /**
+   * Store every `sys_email` insert under THIS id instead of the one it was
+   * handed, and answer with it — a datasource that assigns its own primary key
+   * (#5523). The row the afterInsert hook sees then carries an id `send()` never
+   * reserved, which is the whole mechanism the insert contract closes.
+   */
+  reKeyInsert?: string;
 }
 
 /**
@@ -84,12 +91,13 @@ function fakeEngine(opts: EngineOpts = {}) {
       return out;
     },
     async insert(table: string, data: any) {
-      const row = { ...data };
+      const reKeyed = opts.reKeyInsert && table === 'sys_email';
+      const row = reKeyed ? { ...data, id: opts.reKeyInsert } : { ...data };
       const t = rowsOf(table);
       t.push(row);
       tables.set(table, t);
       for (const fn of hooks.afterInsert ?? []) await fn({ object: table, result: row });
-      return { id: data.id };
+      return { id: row.id };
     },
     async update(table: string, patch: any) {
       const r = rowsOf(table).find((x) => x.id === patch.id);
@@ -423,3 +431,50 @@ describe('normal delivery is unchanged', () => {
     expect(h.ctx.logger.error).not.toHaveBeenCalled();
   });
 });
+
+// ── the insert id contract, at the seam that motivated it (#5523) ───────────
+
+describe('a datasource that re-keys a sys_email insert', () => {
+  it('makes send() reject by name instead of double-sending through the drain hook', async () => {
+    // The full mechanism, end to end, through the REAL plugin persistence
+    // adapter (which forwards the engine's own answer rather than laundering it
+    // back into `row.id`):
+    //
+    //   send() mints id → reserves it → engine.insert stores the row under
+    //   'db-pk-7' and fires afterInsert with THAT id → the drain hook asks
+    //   isServiceManaged('db-pk-7'), gets false, and schedules a delivery of
+    //   its own → insert returns 'db-pk-7'.
+    //
+    // Before the contract check, `send()` then adopted 'db-pk-7' and delivered
+    // the message a second time down its own path: two deliveries of one mail
+    // and two terminal updates racing on one row. The only thing that ever
+    // stopped it was the drain hook's `setTimeout(0)` losing to send()'s inline
+    // delivery — so this transport is deliberately SLOWER than that timer,
+    // which is what real network I/O is. Reverting the check turns the
+    // `toHaveBeenCalledTimes(1)` below into 2.
+    const transport = {
+      send: vi.fn(async () => {
+        await new Promise((r) => setTimeout(r, 25));
+        return { messageId: '<sent@x>' };
+      }),
+    };
+    const h = await boot({ engine: { reKeyInsert: 'db-pk-7' }, transport });
+    await h.ready();
+
+    await expect(
+      h.service().send({ to: 'a@b.com', from: 'x@y.com', subject: 'Hi', text: 'hello' }),
+    ).rejects.toThrow(/EmailPersistence\.insert must return the row's own id/);
+
+    // Give the hook's already-scheduled delivery time to finish. It is NOT
+    // cancellable: the hook runs synchronously inside `engine.insert`, i.e.
+    // before `insert` has even returned to the service, so the contract check
+    // cannot pre-empt it. What the check removes is send()'s SECOND delivery —
+    // the row that did land is still delivered once, by the hook, and the
+    // caller is told loudly that its persistence is misconfigured.
+    await new Promise((r) => setTimeout(r, 120));
+
+    expect(transport.send).toHaveBeenCalledTimes(1);
+    expect(h.sysEmail()).toHaveLength(1);
+    expect(h.sysEmail()[0]).toMatchObject({ id: 'db-pk-7', status: 'sent' });
+  });
+});

packages/plugins/plugin-email/src/email-plugin.ts

@@ -486,6 +486,14 @@ export class EmailServicePlugin implements Plugin {
             const created = await (engine as any).insert('sys_email', row, {
               context: SYSTEM_CTX,
             });
+            // The engine's OWN answer is forwarded, not laundered into
+            // `row.id`. ObjectQL honours the id it is handed, so this confirms
+            // `row.id` — and on the day some driver re-keys the row instead,
+            // `EmailPersistence`'s insert contract (#5523) says so loudly here
+            // rather than letting the outbox drain hook double-send the
+            // message. `created` without a usable id means the engine reported
+            // no identity at all, which is not a disagreement: confirm the row
+            // we asked for.
             return created?.id ? { id: String(created.id) } : { id: String(row.id) };
           },
           async update(id, patch) {

packages/plugins/plugin-email/src/email-service.queue-delivery.test.ts

@@ -189,12 +189,17 @@ describe('EmailService — queue delivery on', () => {
     expect(svc.isServiceManaged(res.id)).toBe(false);
   });
 
-  it('releases an insert-assigned id once the job is published (#5169)', async () => {
-    // Queue mode returns EARLY (right after the publish), so the release of an
-    // insert-assigned id happens on that path too — and it must, because the
-    // row is now the worker's: the boot outbox sweep decides whether to requeue
-    // it by asking `isServiceManaged`, and a permanently-true answer would make
-    // a stranded row unsweepable forever.
+  it('rejects an insert that returns a different id — and publishes NO job (#5523)', async () => {
+    // Re-judged from #5169's "releases an insert-assigned id once the job is
+    // published". That test pinned the release on the queue path's EARLY return
+    // for an insert-assigned id; `EmailPersistence.insert` may no longer assign
+    // one, so the scenario is now a rejection. The release-on-early-return fact
+    // it also carried is still pinned by the test above, which asserts
+    // `isServiceManaged(res.id) === false` after a queued send.
+    //
+    // Queue mode matters on its own here: a job that referenced the substituted
+    // id would send the mail from the WORKER as well, so the rejection has to
+    // land before the publish, not just before inline delivery.
     const queue = makeQueue();
     const transport = { send: vi.fn(async () => ({ messageId: '<x>' })) };
     const persistence: EmailPersistence = {
@@ -205,11 +210,10 @@ describe('EmailService — queue delivery on', () => {
       transport, defaultFrom: 'no@reply.com', persistence, queueDelivery: wiring(queue),
     });
 
-    const res = await svc.send(MSG);
+    await expect(svc.send(MSG)).rejects.toThrow(/EmailPersistence\.insert must return the row's own id/);
 
-    // The job references the PERSISTED id, and that id is no longer managed.
-    expect(res).toMatchObject({ id: 'db-pk-9', status: 'queued' });
-    expect(queue.published[0].data).toEqual({ rowId: 'db-pk-9' });
+    expect(queue.published).toHaveLength(0);   // nothing for a worker to double-send
+    expect(transport.send).not.toHaveBeenCalled();
     expect(svc.isServiceManaged('db-pk-9')).toBe(false);
   });
 

packages/plugins/plugin-email/src/email-service.test.ts

@@ -147,6 +147,7 @@ describe('EmailService', () => {
     // afterInsert drain hook would fire — assert the row is flagged managed
     // there, and unflagged once send() resolves.
     let managedAtInsert: boolean | undefined;
+    let managedDuringDelivery: boolean | undefined;
     let insertedId: string | undefined;
     const transport = { send: vi.fn(async () => ({ messageId: '<m@x>' })) };
     let svc!: EmailService;
@@ -156,42 +157,133 @@ describe('EmailService', () => {
         managedAtInsert = svc.isServiceManaged(insertedId);
         return { id: row.id };
       },
-      async update() { /* noop */ },
+      async update(id) {
+        // The `sent` finalize runs INSIDE send(), i.e. while this row is still
+        // send()'s to deliver — the window the managed flag exists to protect
+        // must NOT shrink. (Inherited from the #5169 test this file used to
+        // carry on the insert-assigned-id path; that path is now a contract
+        // violation, so the fact is pinned here, on the compliant one.)
+        managedDuringDelivery = svc.isServiceManaged(String(id));
+      },
     };
     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(managedDuringDelivery).toBe(true);           // window kept through finalize
     expect(svc.isServiceManaged(insertedId!)).toBe(false); // cleared after send
   });
 
-  it('releases an insert-ASSIGNED row id from the managed set too (#5169)', async () => {
-    // `EmailPersistence` is public and its `insert` may answer with an id of
-    // its own — a database-assigned primary key, an external delivery system's
-    // receipt id. `send()` reserves that id as managed as well; the bug was
-    // that it never released it, so `isServiceManaged(persistedId)` stayed true
-    // forever: one leaked entry per message, and a "belongs to a live send()"
-    // assertion the drain hook and the boot sweep trust but nobody re-checks.
-    let managedDuringDelivery: boolean | undefined;
+  // ── EmailPersistence.insert id contract (#5523) ────────────────────
+  //
+  // Re-judged from #5169's "releases an insert-ASSIGNED row id" test. That test
+  // pinned the *leak fix* on a capability — `insert` answering with an id of its
+  // own — that is now a contract violation, so the leak has no source left to
+  // fix. Same scenario, opposite verdict: rejected, by name, before delivery.
+  it('rejects an insert that returns a DIFFERENT id, naming the contract, before delivering', async () => {
+    // The double-send this closes: the sys_email afterInsert drain hook asks
+    // `isServiceManaged(hookCtx.result.id)` — the INSERTED row's id. A
+    // persistence that re-keys the row hands the hook an id send() never
+    // reserved, the hook reads the row as an app-inserted outbox entry and
+    // delivers it, and send() delivers it again down its own path.
     const transport = { send: vi.fn(async () => ({ messageId: '<m@x>' })) };
-    let svc!: EmailService;
+    const update = vi.fn(async () => { /* noop */ });
     const persistence: EmailPersistence = {
-      // Ignores the minted id and hands back the row's real (DB) key.
+      // Ignores the minted id and hands back a DB primary key of its own.
       async insert() { return { id: 'db-pk-7' }; },
-      async update(id) {
-        // The `sent` finalize runs INSIDE send(), i.e. while this row is still
-        // send()'s to deliver — the window the managed flag exists to protect
-        // must NOT shrink to make the release possible.
-        managedDuringDelivery = svc.isServiceManaged(String(id));
-      },
+      update,
     };
-    svc = new EmailService({ transport, defaultFrom: 'no@reply.com', persistence });
+    const svc = new EmailService({ transport, defaultFrom: 'no@reply.com', persistence });
+
+    await expect(svc.send({ to: 'a@b.com', subject: 'Hi', text: 'x' }))
+      .rejects.toThrow(/EmailPersistence\.insert must return the row's own id/);
+
+    // Names the value actually returned, so the operator can find the
+    // implementation that returned it.
+    await expect(svc.send({ to: 'a@b.com', subject: 'Hi', text: 'x' }))
+      .rejects.toThrow(/'db-pk-7'/);
+
+    // BEFORE delivery — not "double-send, then complain". This is the whole
+    // point of where the check sits: nothing was sent, nothing was finalized.
+    expect(transport.send).not.toHaveBeenCalled();
+    expect(update).not.toHaveBeenCalled();
+
+    // And the throw does not leak the reservation — the `finally` still runs.
+    expect(svc.isServiceManaged('db-pk-7')).toBe(false);
+  });
+
+  it('accepts the bare-string confirmation form when it names the row id', async () => {
+    // `insert` may answer `row.id` instead of `{ id: row.id }` — same
+    // confirmation, and it must not be mistaken for a substitution.
+    const transport = { send: vi.fn(async () => ({ messageId: '<m@x>' })) };
+    const persistence: EmailPersistence = {
+      async insert(row) { return String(row.id); },
+      async update() { /* noop */ },
+    };
+    const 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(transport.send).toHaveBeenCalledTimes(1);
+    expect(svc.isServiceManaged(res.id)).toBe(false);
+  });
+
+  it('rejects a bare-string confirmation that names a different id', async () => {
+    // The string form is the same contract, not a loophole around it.
+    const transport = { send: vi.fn(async () => ({ messageId: '<m@x>' })) };
+    const persistence: EmailPersistence = {
+      async insert() { return 'db-pk-8'; },
+      async update() { /* noop */ },
+    };
+    const svc = new EmailService({ transport, defaultFrom: 'no@reply.com', persistence });
+
+    await expect(svc.send({ to: 'a@b.com', subject: 'Hi', text: 'x' }))
+      .rejects.toThrow(/EmailPersistence\.insert must return the row's own id/);
+    expect(transport.send).not.toHaveBeenCalled();
+  });
+
+  it('treats a confirmation that names NO id as confirming the row it inserted', async () => {
+    // Only reachable from an untyped JS implementation (the declared return
+    // type requires an id). There is no id to disagree with, and the drain hook
+    // reads the id off the inserted ROW — the minted one — so no double-send is
+    // reachable this way. The check judges the VALUE, not its presence, and an
+    // insert that silently returned nothing must not stop the mail.
+    const transport = { send: vi.fn(async () => ({ messageId: '<m@x>' })) };
+    const persistence = {
+      async insert() { /* returns undefined */ },
+      async update() { /* noop */ },
+    } as unknown as EmailPersistence;
+    const svc = new EmailService({ transport, defaultFrom: 'no@reply.com', persistence });
 
     const res = await svc.send({ to: 'a@b.com', subject: 'Hi', text: 'x' });
 
-    expect(res).toMatchObject({ id: 'db-pk-7', status: 'sent' });
-    expect(managedDuringDelivery).toBe(true);              // window kept
-    expect(svc.isServiceManaged('db-pk-7')).toBe(false);   // released, not leaked
+    expect(res.status).toBe('sent');
+    expect(transport.send).toHaveBeenCalledTimes(1);
+  });
+
+  it('still rides out an insert that THROWS — an operational failure, not a contract violation', async () => {
+    // The two must stay distinguishable: a failing insert keeps its non-fatal
+    // warning path (mail goes inline), while a *renaming* insert is fatal. If
+    // the contract check ever moved inside the insert's catch, this test would
+    // still pass and the rejection test above would go green for the wrong
+    // reason — hence both, side by side.
+    const transport = { send: vi.fn(async () => ({ messageId: '<m@x>' })) };
+    const warn = vi.fn();
+    const persistence: EmailPersistence = {
+      async insert() { throw new Error('db down'); },
+      async update() { /* noop */ },
+    };
+    const svc = new EmailService({
+      transport, defaultFrom: 'no@reply.com', persistence,
+      logger: { info: vi.fn(), warn },
+    });
+
+    const res = await svc.send({ to: 'a@b.com', subject: 'Hi', text: 'x' });
+
+    expect(res.status).toBe('sent');
+    expect(transport.send).toHaveBeenCalledTimes(1);
+    expect(warn).toHaveBeenCalledWith(expect.stringContaining('persist failed'), expect.any(Object));
   });
 
   it('deliverPersistedRow delivers an existing row WITHOUT inserting a new one', async () => {