From 1ac7ca6e0015ddfa4044bb655db8a4a89565f7fa Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 13:22:32 +0000 Subject: [PATCH] feat(plugin-email): large attachments go to storage, so queued delivery covers them too (#5172) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 2 of the sys_email payload work (#5177/#5211 shipped phase 1). Over the 256 KiB in-row budget, attachment content is uploaded to the `file-storage` capability and the row carries `{ filename, contentType?, size, hash, contentForm, storageKey }` instead of base64. The queue worker fetches the content back and rebuilds the message, so a signed contract or an exported report finally gets the same durability guarantee as every other message. The cut the design turns on: filename/contentType/size/hash are PERMANENT audit evidence and stay on the row forever; the bytes are a delivery artifact and are deleted a grace window after the row reaches a terminal state, at which point `storageKey` is replaced by `contentReclaimedAt`. That is what decouples an append-only mail log from unbounded binary growth. Reclamation is a delayed `email.attachment.reclaim` job whose payload carries the storage keys, not just the row id — so a row deleted in the meantime (a future declarative retention policy, a purge) reclaims its content instead of orphaning it. Every failure degrades to inline delivery of the WHOLE message and says so: no capability mounted, and an upload that failed, are distinguished in the log. On the read side nothing is swallowed — an unfetchable, truncated or substituted object fails the row rather than putting a message on the wire without an attachment it declares. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_017MCKJaEomEqg4tvz4SzdNd --- .../large-email-attachments-storage-refs.md | 12 + .../apps/translations/en.objects.generated.ts | 2 +- .../translations/es-ES.objects.generated.ts | 2 +- .../translations/ja-JP.objects.generated.ts | 2 +- .../translations/zh-CN.objects.generated.ts | 2 +- .../src/audit/sys-email.object.ts | 10 +- .../src/attachment-reclaim.test.ts | 318 +++++++++++++ .../plugin-email/src/attachment-reclaim.ts | 303 ++++++++++++ .../src/attachment-storage.test.ts | 326 +++++++++++++ .../plugin-email/src/attachment-storage.ts | 323 +++++++++++++ .../email-plugin.attachment-storage.test.ts | 359 ++++++++++++++ .../plugins/plugin-email/src/email-plugin.ts | 120 +++++ .../email-service.attachment-storage.test.ts | 436 ++++++++++++++++++ .../plugins/plugin-email/src/email-service.ts | 313 ++++++++++++- packages/plugins/plugin-email/src/index.ts | 33 +- .../src/sys-email-payload.test.ts | 10 +- .../plugin-email/src/sys-email-payload.ts | 365 ++++++++++++--- 17 files changed, 2848 insertions(+), 88 deletions(-) create mode 100644 .changeset/large-email-attachments-storage-refs.md create mode 100644 packages/plugins/plugin-email/src/attachment-reclaim.test.ts create mode 100644 packages/plugins/plugin-email/src/attachment-reclaim.ts create mode 100644 packages/plugins/plugin-email/src/attachment-storage.test.ts create mode 100644 packages/plugins/plugin-email/src/attachment-storage.ts create mode 100644 packages/plugins/plugin-email/src/email-plugin.attachment-storage.test.ts create mode 100644 packages/plugins/plugin-email/src/email-service.attachment-storage.test.ts diff --git a/.changeset/large-email-attachments-storage-refs.md b/.changeset/large-email-attachments-storage-refs.md new file mode 100644 index 0000000000..9f41fd1f5a --- /dev/null +++ b/.changeset/large-email-attachments-storage-refs.md @@ -0,0 +1,12 @@ +--- +"@objectstack/plugin-email": minor +"@objectstack/platform-objects": patch +--- + +plugin-email: large attachments (>256 KiB) now get durable queue delivery, with their content held out of the `sys_email` row + +A message whose attachments exceeded the in-row budget was pushed back onto inline delivery — whole, but with none of the durability queue delivery exists to provide, which meant the platform was weakest about exactly the mail that matters most (a signed contract, an exported report). Its content now goes to the `file-storage` capability, the row records a `storageKey` plus the audit metadata, and the queue worker fetches the content back to rebuild the message. + +- **Zero migration.** `attachments_json` declared `storageKey` from the start; this adds the producer and the reader. Attachments at or under `SYS_EMAIL_ATTACHMENT_LIMIT_BYTES` still go in the row exactly as before, and the boundary includes equality. +- **The row stays an audit log, not a blob store.** `filename` / `contentType` / `size` / `hash` stay on the row permanently; the content is a delivery artifact and is deleted a grace window (24h) after the row reaches a terminal state, at which point `storageKey` is replaced by `contentReclaimedAt`. Reclamation is a delayed `email.attachment.reclaim` queue job that carries the storage keys, so a row deleted in the meantime reclaims its content instead of orphaning it. +- **Nothing degrades silently.** No `file-storage` capability, or an upload that fails, keeps today's behaviour — inline delivery of the whole message — and says which of the two it was and how to fix it. On the way back, content that cannot be fetched (outage, missing object, no capability on the worker, truncated or substituted bytes) fails the row loudly; a message is never delivered without an attachment it declares. diff --git a/packages/platform-objects/src/apps/translations/en.objects.generated.ts b/packages/platform-objects/src/apps/translations/en.objects.generated.ts index 49f57694e9..391ee0d64b 100644 --- a/packages/platform-objects/src/apps/translations/en.objects.generated.ts +++ b/packages/platform-objects/src/apps/translations/en.objects.generated.ts @@ -2051,7 +2051,7 @@ export const enObjects: NonNullable = { }, attachments_json: { label: "Attachments (JSON)", - help: "Attachments as a JSON array of { filename, contentType?, size, hash, cid?, contentForm, inline?, storageKey? }, with content base64 in `inline`. Written only when the combined raw size is within the plugin-email budget (SYS_EMAIL_ATTACHMENT_LIMIT_BYTES, 256 KiB — ~350 KB of base64 at worst); a larger message is delivered inline and stores nothing here, so the row stays bounded. `storageKey` (out-of-row content) has no producer yet — objectstack#5172." + help: "Attachments as a JSON array of { filename, contentType?, size, hash, cid?, contentForm, inline?, storageKey?, contentReclaimedAt? }. Content up to the plugin-email budget (SYS_EMAIL_ATTACHMENT_LIMIT_BYTES, 256 KiB combined raw — ~350 KB of base64 at worst) is base64 in `inline`; larger content goes to the file-storage capability and the element carries `storageKey` instead, so the row stays bounded either way. filename/contentType/size/hash are PERMANENT audit evidence; out-of-row content is a delivery artifact and is deleted a grace window after the row reaches a terminal state, at which point `storageKey` is replaced by `contentReclaimedAt`." }, status: { label: "Status", diff --git a/packages/platform-objects/src/apps/translations/es-ES.objects.generated.ts b/packages/platform-objects/src/apps/translations/es-ES.objects.generated.ts index b5b43b58c2..30b2408e11 100644 --- a/packages/platform-objects/src/apps/translations/es-ES.objects.generated.ts +++ b/packages/platform-objects/src/apps/translations/es-ES.objects.generated.ts @@ -2051,7 +2051,7 @@ export const esESObjects: NonNullable = { }, attachments_json: { label: "Adjuntos (JSON)", - help: "Adjuntos como un array JSON de { filename, contentType?, size, hash, cid?, contentForm, inline?, storageKey? }, con el contenido en base64 en `inline`. Solo se escribe cuando el tamaño bruto combinado está dentro del presupuesto de plugin-email (SYS_EMAIL_ATTACHMENT_LIMIT_BYTES, 256 KiB — ~350 KB de base64 en el peor caso); un mensaje mayor se entrega en línea y aquí no se almacena nada, de modo que la fila permanece acotada. `storageKey` (contenido fuera de la fila) todavía no tiene productor — objectstack#5172." + help: "Adjuntos como un array JSON de { filename, contentType?, size, hash, cid?, contentForm, inline?, storageKey?, contentReclaimedAt? }. El contenido que cabe en el presupuesto de plugin-email (SYS_EMAIL_ATTACHMENT_LIMIT_BYTES, 256 KiB brutos combinados — ~350 KB de base64 en el peor caso) va en base64 en `inline`; el contenido mayor se guarda en la capacidad file-storage y el elemento lleva `storageKey` en su lugar, de modo que la fila permanece acotada en ambos casos. filename/contentType/size/hash son evidencia de auditoría PERMANENTE; el contenido fuera de la fila es un artefacto de entrega y se elimina tras un periodo de gracia una vez que la fila alcanza un estado terminal, momento en el que `storageKey` se sustituye por `contentReclaimedAt`." }, status: { label: "Estado", diff --git a/packages/platform-objects/src/apps/translations/ja-JP.objects.generated.ts b/packages/platform-objects/src/apps/translations/ja-JP.objects.generated.ts index f828fa921c..b3b0285a64 100644 --- a/packages/platform-objects/src/apps/translations/ja-JP.objects.generated.ts +++ b/packages/platform-objects/src/apps/translations/ja-JP.objects.generated.ts @@ -2051,7 +2051,7 @@ export const jaJPObjects: NonNullable = { }, attachments_json: { label: "添付ファイル（JSON）", - help: "添付ファイルの JSON 配列で、要素は { filename, contentType?, size, hash, cid?, contentForm, inline?, storageKey? } の形をとり、内容は base64 で `inline` に格納されます。添付の合計生サイズが plugin-email の予算（SYS_EMAIL_ATTACHMENT_LIMIT_BYTES、256 KiB —— 最悪でも base64 で約 350 KB）に収まる場合にのみ書き込まれます。上限を超えるメッセージはインライン配信され、この列には何も保存されないため、行のサイズは有界に保たれます。`storageKey`（行外の内容）にはまだプロデューサーがありません —— objectstack#5172。" + help: "添付ファイルの JSON 配列で、要素は { filename, contentType?, size, hash, cid?, contentForm, inline?, storageKey?, contentReclaimedAt? } の形をとります。plugin-email の予算（SYS_EMAIL_ATTACHMENT_LIMIT_BYTES、合計生サイズ 256 KiB —— 最悪でも base64 で約 350 KB）以内の内容は base64 で `inline` に格納され、それを超える内容は file-storage ケイパビリティに置かれて要素は代わりに `storageKey` を持つため、いずれの場合も行のサイズは有界に保たれます。filename/contentType/size/hash は**恒久的な**監査証跡です。行外の内容は配信用の成果物であり、行が終端状態に達してから猶予期間を過ぎると削除され、その時点で `storageKey` は `contentReclaimedAt` に置き換わります。" }, status: { label: "ステータス", diff --git a/packages/platform-objects/src/apps/translations/zh-CN.objects.generated.ts b/packages/platform-objects/src/apps/translations/zh-CN.objects.generated.ts index ec3799e80a..3f403ddb75 100644 --- a/packages/platform-objects/src/apps/translations/zh-CN.objects.generated.ts +++ b/packages/platform-objects/src/apps/translations/zh-CN.objects.generated.ts @@ -2051,7 +2051,7 @@ export const zhCNObjects: NonNullable = { }, attachments_json: { label: "附件（JSON）", - help: "附件的 JSON 数组，元素形状为 { filename, contentType?, size, hash, cid?, contentForm, inline?, storageKey? }，内容以 base64 存放在 `inline` 中。仅当附件合计原始大小在 plugin-email 的预算内（SYS_EMAIL_ATTACHMENT_LIMIT_BYTES，256 KiB —— 最坏约 350 KB 的 base64）时才写入；超出上限的邮件改走内联投递，此列不落任何内容，因此行体积有界。`storageKey`（行外内容）目前还没有生产者 —— 见 objectstack#5172。" + help: "附件的 JSON 数组，元素形状为 { filename, contentType?, size, hash, cid?, contentForm, inline?, storageKey?, contentReclaimedAt? }。在 plugin-email 预算（SYS_EMAIL_ATTACHMENT_LIMIT_BYTES，合计原始大小 256 KiB —— 最坏约 350 KB 的 base64）以内的内容以 base64 存放在 `inline`；超出的内容存入 file-storage 能力，元素改为携带 `storageKey`，因此两种情况下行体积都有界。filename/contentType/size/hash 是**永久**审计证据；行外内容属于投递工件，在行到达终态并经过宽限窗后被删除，届时 `storageKey` 由 `contentReclaimedAt` 取代。" }, status: { label: "状态", diff --git a/packages/platform-objects/src/audit/sys-email.object.ts b/packages/platform-objects/src/audit/sys-email.object.ts index e0c4ee0aeb..33114c60be 100644 --- a/packages/platform-objects/src/audit/sys-email.object.ts +++ b/packages/platform-objects/src/audit/sys-email.object.ts @@ -130,10 +130,12 @@ export const SysEmail = ObjectSchema.create({ required: false, description: 'Attachments as a JSON array of { filename, contentType?, size, hash, cid?, contentForm, ' - + 'inline?, storageKey? }, with content base64 in `inline`. Written only when the combined raw ' - + 'size is within the plugin-email budget (SYS_EMAIL_ATTACHMENT_LIMIT_BYTES, 256 KiB — ~350 KB of ' - + 'base64 at worst); a larger message is delivered inline and stores nothing here, so the row ' - + 'stays bounded. `storageKey` (out-of-row content) has no producer yet — objectstack#5172.', + + 'inline?, storageKey?, contentReclaimedAt? }. Content up to the plugin-email budget ' + + '(SYS_EMAIL_ATTACHMENT_LIMIT_BYTES, 256 KiB combined raw — ~350 KB of base64 at worst) is base64 in ' + + '`inline`; larger content goes to the file-storage capability and the element carries `storageKey` ' + + 'instead, so the row stays bounded either way. filename/contentType/size/hash are PERMANENT audit ' + + 'evidence; out-of-row content is a delivery artifact and is deleted a grace window after the row ' + + 'reaches a terminal state, at which point `storageKey` is replaced by `contentReclaimedAt`.', group: 'Content', }), diff --git a/packages/plugins/plugin-email/src/attachment-reclaim.test.ts b/packages/plugins/plugin-email/src/attachment-reclaim.test.ts new file mode 100644 index 0000000000..10803bb811 --- /dev/null +++ b/packages/plugins/plugin-email/src/attachment-reclaim.test.ts @@ -0,0 +1,318 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. +// +// Reclaiming out-of-row attachment content (objectstack#5172). +// +// The three things these exist to prove: +// +// 1. **Content dies with the delivery, metadata does not.** A terminal row +// past its grace window loses its bytes and keeps filename / contentType / +// size / hash — forever. That asymmetry is the whole feature. +// 2. **A deleted row cannot orphan content.** The job carries the storage +// keys, so it deletes them even when the `sys_email` row is gone by the +// time it fires — which is exactly what a future declarative `retention` +// policy in the #5192 shape would do to it. +// 3. **Nothing is deleted early.** A row still in flight, or one touched +// inside the grace window, re-arms the job instead of deleting the content +// a retry may still need. And when it cannot re-arm, it SAYS so at +// `error`, because bytes that were supposed to be temporary have just +// become permanent. + +import { describe, it, expect, vi } from 'vitest'; +import { createHash } from 'node:crypto'; +import { assertEngineDeleteDispatch } from '@objectstack/objectql'; +import { + reclaimAttachmentContent, + RECLAIM_OBJECT, + type AttachmentReclaimEngine, +} from './attachment-reclaim.js'; +import { + EMAIL_ATTACHMENT_RECLAIM_GRACE_MS, + type EmailAttachmentStore, +} from './attachment-storage.js'; + +const NOW = Date.UTC(2026, 7, 5, 12, 0, 0); +const ago = (ms: number) => new Date(NOW - ms).toISOString(); +const HOUR = 3600_000; +const sha = (s: string) => `sha256:${createHash('sha256').update(s).digest('hex')}`; + +const KEYS = ['sys_email/attachments/row-1/000-aaaaaaaaaaaaaaaa']; + +/** `attachments_json` as `send()` writes it for a storage-backed message. */ +const storageColumn = JSON.stringify([{ + filename: 'contract.pdf', + contentType: 'application/pdf', + size: 307200, + hash: sha('contract'), + contentForm: 'buffer', + storageKey: KEYS[0], +}]); + +function fakeEngine(rows: Array>) { + const tables = new Map([[RECLAIM_OBJECT, rows.map((r) => ({ ...r }))]]); + const rowsOf = (t: string) => tables.get(t) ?? []; + const engine = { + rows: (t = RECLAIM_OBJECT) => [...rowsOf(t)], + updates: [] as Array>, + async find(table: string, o: any = {}) { + const where = o?.where ?? {}; + let out = rowsOf(table).filter((r) => Object.entries(where).every(([k, v]) => r[k] === v)); + if (o.limit) out = out.slice(0, o.limit); + return out; + }, + async update(table: string, patch: any) { + engine.updates.push({ table, ...patch }); + 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, o: any) { + // [#4550/#5197] Pinned to ObjectQL.delete's own dispatch predicate — a + // double looser than the engine it stands in for turns a green suite + // into no suite at all. + const dispatch = assertEngineDeleteDispatch(o); + if (dispatch.kind === 'multi') { + const survivors = rowsOf(table).filter( + (r) => !Object.entries(o?.where ?? {}).every(([k, v]) => r[k] === v), + ); + 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 }; + }, + }; + return engine satisfies AttachmentReclaimEngine & Record; +} + +function fakeStore(present: string[] = KEYS, opts: { failDelete?: boolean } = {}) { + const objects = new Set(present); + return { + objects, + deleted: [] as string[], + async upload() { /* not used here */ }, + async download() { return Buffer.alloc(0); }, + async delete(key: string) { + if (opts.failDelete) throw new Error('AccessDenied'); + // Both shipped adapters delete idempotently (local swallows ENOENT, S3's + // DeleteObject is idempotent), so a retry must not fail on keys the + // previous attempt already removed. + objects.delete(key); + (this as any).deleted.push(key); + }, + } as EmailAttachmentStore & { objects: Set; deleted: string[] }; +} + +const logger = () => ({ info: vi.fn(), warn: vi.fn(), error: vi.fn() }); + +const terminalRow = (over: Record = {}) => ({ + id: 'row-1', + status: 'sent', + message_id: '', + subject: 'Your contract', + attachments_json: storageColumn, + created_at: ago(30 * HOUR), + updated_at: ago(30 * HOUR), + ...over, +}); + +describe('a terminal row past the grace window', () => { + it('deletes the content and rewrites the column, keeping every audit field', async () => { + const engine = fakeEngine([terminalRow()]); + const store = fakeStore(); + + const out = await reclaimAttachmentContent({ rowId: 'row-1', keys: KEYS }, { + engine, store, now: () => NOW, logger: logger(), + }); + + expect(out).toMatchObject({ kind: 'reclaimed', rowId: 'row-1' }); + expect(store.objects.size).toBe(0); + + const [el] = JSON.parse(engine.rows()[0].attachments_json); + // The delivery artifact is gone… + expect(el.storageKey).toBeUndefined(); + expect(el.contentReclaimedAt).toBe(new Date(NOW).toISOString()); + // …and the audit artifact is exactly as it was. + expect(el).toMatchObject({ + filename: 'contract.pdf', + contentType: 'application/pdf', + size: 307200, + hash: sha('contract'), + contentForm: 'buffer', + }); + // Nothing else about the row moved. + expect(engine.rows()[0]).toMatchObject({ status: 'sent', message_id: '', subject: 'Your contract' }); + }); + + it('treats a terminal `failed` row the same way — a message nobody will retry is done too', async () => { + const engine = fakeEngine([terminalRow({ status: 'failed', message_id: undefined, error: '550 nope' })]); + const store = fakeStore(); + + const out = await reclaimAttachmentContent({ rowId: 'row-1', keys: KEYS }, { + engine, store, now: () => NOW, + }); + + expect(out.kind).toBe('reclaimed'); + expect(store.objects.size).toBe(0); + expect(engine.rows()[0].error).toBe('550 nope'); + }); + + it('is idempotent — running again finds no storageKey and writes nothing', async () => { + const engine = fakeEngine([terminalRow()]); + const store = fakeStore(); + await reclaimAttachmentContent({ rowId: 'row-1', keys: KEYS }, { engine, store, now: () => NOW }); + const writesAfterFirst = engine.updates.length; + + const out = await reclaimAttachmentContent({ rowId: 'row-1', keys: KEYS }, { engine, store, now: () => NOW }); + + expect(out.kind).toBe('reclaimed'); + expect(engine.updates.length).toBe(writesAfterFirst); + }); +}); + +describe('a row that is gone — the orphan case #5192 warns about', () => { + it('deletes the content from the payload alone, because the job carries the keys', async () => { + // The row a declarative `retention` policy (or a purge) removed. Nothing + // anywhere still points at these bytes. + const engine = fakeEngine([]); + const store = fakeStore(); + const log = logger(); + + const out = await reclaimAttachmentContent({ rowId: 'row-1', keys: KEYS }, { + engine, store, now: () => NOW, logger: log, + }); + + expect(out).toMatchObject({ kind: 'orphan-reclaimed', rowId: 'row-1', keys: KEYS }); + expect(store.objects.size).toBe(0); + expect(String(log.info.mock.calls[0][0])).toContain('no longer exists'); + }); + + it('does not need the row to be readable at all — the keys are the authority', async () => { + const engine = { + async find() { return []; }, + async update() { throw new Error('the table is gone too'); }, + } satisfies AttachmentReclaimEngine; + const store = fakeStore(); + + const out = await reclaimAttachmentContent({ rowId: 'row-1', keys: KEYS }, { + engine, store, now: () => NOW, + }); + + expect(out.kind).toBe('orphan-reclaimed'); + expect(store.objects.size).toBe(0); + }); +}); + +describe('nothing is deleted early', () => { + it('re-arms for a row still at `queued` — that message has a delivery ahead of it', async () => { + const engine = fakeEngine([terminalRow({ status: 'queued', message_id: undefined })]); + const store = fakeStore(); + const rearm = vi.fn(async () => true); + + const out = await reclaimAttachmentContent({ rowId: 'row-1', keys: KEYS }, { + engine, store, now: () => NOW, rearm, + }); + + expect(out).toMatchObject({ kind: 'rearmed', delayMs: EMAIL_ATTACHMENT_RECLAIM_GRACE_MS }); + expect((out as any).reason).toContain("still at 'queued'"); + expect(store.objects.size).toBe(1); + expect(engine.updates).toHaveLength(0); + }); + + it('re-arms for the REMAINDER when a terminal row was touched inside the grace window', async () => { + // A queue retry re-stamped the row two hours ago; 22h of grace are left. + const engine = fakeEngine([terminalRow({ status: 'failed', updated_at: ago(2 * HOUR) })]); + const store = fakeStore(); + const rearm = vi.fn(async () => true); + + const out = await reclaimAttachmentContent({ rowId: 'row-1', keys: KEYS }, { + engine, store, now: () => NOW, rearm, + }); + + expect(out.kind).toBe('rearmed'); + expect((out as any).delayMs).toBe(EMAIL_ATTACHMENT_RECLAIM_GRACE_MS - 2 * HOUR); + expect(rearm).toHaveBeenCalledWith(EMAIL_ATTACHMENT_RECLAIM_GRACE_MS - 2 * HOUR); + expect(store.objects.size).toBe(1); + }); + + it('re-arms rather than deleting when the storage capability is not mounted here', async () => { + const engine = fakeEngine([terminalRow()]); + const rearm = vi.fn(async () => true); + + const out = await reclaimAttachmentContent({ rowId: 'row-1', keys: KEYS }, { + engine, store: undefined, now: () => NOW, rearm, + }); + + expect(out.kind).toBe('rearmed'); + expect((out as any).reason).toContain('file-storage capability is not mounted'); + }); + + it('states the consequence at `error` when it cannot re-arm — those bytes are now permanent', async () => { + const engine = fakeEngine([terminalRow({ status: 'queued', message_id: undefined })]); + const store = fakeStore(); + const log = logger(); + + const out = await reclaimAttachmentContent({ rowId: 'row-1', keys: KEYS }, { + engine, store, now: () => NOW, logger: log, rearm: async () => false, + }); + + expect(out.kind).toBe('stalled'); + expect(log.warn).not.toHaveBeenCalled(); + const line = String(log.error.mock.calls[0][0]); + expect(line).toContain('could NOT be reclaimed'); + expect(line).toContain('stay in the backend forever'); // the consequence + expect(line).toContain('@objectstack/service-storage'); // the fix + expect(store.objects.size).toBe(1); + }); +}); + +describe('failure handling', () => { + it('throws (so the job retries) when a delete does not land, and leaves the column pointing at it', async () => { + const engine = fakeEngine([terminalRow()]); + const store = fakeStore(KEYS, { failDelete: true }); + + await expect(reclaimAttachmentContent({ rowId: 'row-1', keys: KEYS }, { + engine, store, now: () => NOW, + })).rejects.toThrow(/could not delete 1 of 1 storage object\(s\)/); + + // Bytes first, row second: the column must NOT be rewritten while the + // content is still there, or the last pointer to it is gone. + expect(JSON.parse(engine.rows()[0].attachments_json)[0].storageKey).toBe(KEYS[0]); + expect(engine.updates).toHaveLength(0); + }); + + it('does nothing at all for a payload that names no keys', async () => { + const engine = fakeEngine([terminalRow()]); + const store = fakeStore(); + + const out = await reclaimAttachmentContent({ rowId: 'row-1', keys: [] }, { + engine, store, now: () => NOW, + }); + + expect(out).toEqual({ kind: 'noop', rowId: 'row-1' }); + expect(store.objects.size).toBe(1); + }); + + it('deletes each key once even if the payload repeats one', async () => { + const engine = fakeEngine([terminalRow()]); + const store = fakeStore(); + + await reclaimAttachmentContent({ rowId: 'row-1', keys: [KEYS[0], KEYS[0]] }, { + engine, store, now: () => NOW, + }); + + expect(store.deleted).toEqual(KEYS); + }); + + it('reclaims a row whose timestamps are unreadable — the job delay already bought the grace', async () => { + const engine = fakeEngine([terminalRow({ created_at: 'not-a-date', updated_at: null, sent_at: null })]); + const store = fakeStore(); + + const out = await reclaimAttachmentContent({ rowId: 'row-1', keys: KEYS }, { + engine, store, now: () => NOW, + }); + + expect(out.kind).toBe('reclaimed'); + }); +}); diff --git a/packages/plugins/plugin-email/src/attachment-reclaim.ts b/packages/plugins/plugin-email/src/attachment-reclaim.ts new file mode 100644 index 0000000000..4a6b690d56 --- /dev/null +++ b/packages/plugins/plugin-email/src/attachment-reclaim.ts @@ -0,0 +1,303 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * Reclaiming out-of-row attachment content (objectstack#5172). + * + * ## The one job this does + * + * Attachment bytes written to the `file-storage` capability by + * `attachment-storage.ts` are a **delivery artifact**: once the `sys_email` + * row they belong to is terminal and a grace window has passed, nothing will + * ever need them again. This module deletes them and rewrites the row's + * `attachments_json` so the element says `contentReclaimedAt` instead of + * `storageKey`. The audit metadata — filename, contentType, size, hash — + * is untouched, forever. That asymmetry is the entire point of #5172: an + * append-only mail log keeps growing, but it stops growing *binary*. + * + * ## Why a scheduled job and not a sweep + * + * The maintainer left the trigger open ("终态时同步删 vs 清扫式"). Neither of + * the two obvious answers survives contact with this table: + * + * - **Synchronous delete at the terminal transition** cannot honour a grace + * window at all, and the window is not decorative: the `email.send.async` + * subscriber re-reads and re-delivers a row that is sitting at `failed`, so + * deleting content the moment the row says `failed` deletes it out from + * under the queue's own retry. + * - **A sweep** (the #5161/#5191 shape) needs a query whose result set + * *shrinks*. `status IN ('sent','failed') AND old` does not: after the + * first pass, every row it returns is one it already reclaimed, so each + * boot re-scans the same page of ancient rows forever. Driving the sweep + * from the storage side instead — list the prefix, ask about each blob — + * is not portable: `LocalStorageAdapter.list` is a single-level `readdir` + * (it would not even see `…//`), while the S3 adapter's is + * recursive and unpaginated. One mechanism that silently means two + * different things on the two shipped adapters is worse than no mechanism. + * + * So the trigger is a **delayed queue job**, published at the terminal + * transition with `delay = ` {@link EMAIL_ATTACHMENT_RECLAIM_GRACE_MS} and an + * idempotency key per row. That is durable (it is a `sys_job_queue` row, the + * same substrate the delivery itself rides), exact (no scanning, no polling), + * and available by construction — content only ever goes out of row when + * durable queue delivery is in force, so the queue that must reclaim it is the + * queue that queued the message. + * + * ## Why a deleted row cannot orphan its content + * + * This is the property the issue asks to be argued rather than asserted, so: + * **the job payload carries the storage keys, not just the row id** + * ({@link EmailAttachmentReclaimPayload}). When the job fires and the row is + * gone — reaped by a future declarative `retention` policy in the #5192 shape, + * purged by an operator, deleted by anything at all — the job still knows + * exactly which bytes to delete, and deletes them. Row deletion is therefore + * not a way to *lose* the content; it is the strongest possible signal to + * reclaim it, because a row that no longer exists certainly has no delivery + * left to do. + * + * Note what this does NOT depend on: it does not require the reclaim grace to + * be shorter than the row's retention window, it does not require the row to + * carry a reclamation flag, and it does not require reading the row at all in + * the orphan case. `sys_email` is `lifecycle.class: 'record'` today, so the + * spec actively forbids declaring `retention` on it (`object.zod.ts`: a + * `record` class "is permanent business truth — retention/ttl/storage/archive + * policies are not allowed on it"); the guarantee above is what makes it safe + * for that to change later without anyone having to remember this file. + * + * ## What is deliberately left as a residual risk + * + * If the reclaim job itself is lost — a `purge()` of the queue, a DLQ entry + * nobody replays — those bytes stay in the bucket. That is a byte leak, not a + * correctness failure, it is *visible* (`listFailed` / the DLQ), and closing + * it would mean adding the second scanner this module just argued against. + * The honest trade is stated here rather than papered over with a sweep that + * only works on one adapter. + */ + +import { + EMAIL_ATTACHMENT_RECLAIM_GRACE_MS, + deleteAttachmentKeys, + type EmailAttachmentReclaimPayload, + type EmailAttachmentStore, +} from './attachment-storage.js'; +import { withContentReclaimed } from './sys-email-payload.js'; + +/** System context — reclamation is platform bookkeeping, not a user query. */ +const SYSTEM_CTX = { isSystem: true, positions: [], permissions: [] } as const; + +/** Backing object; a constant so the query and the log lines agree. */ +export const RECLAIM_OBJECT = 'sys_email'; + +/** + * `sys_email.status` values after which content is never needed again. + * + * `queued` is excluded and that is the whole guard: a row still at `queued` + * has a delivery ahead of it, whatever this job was told a day ago. + */ +export const TERMINAL_EMAIL_STATUSES: readonly string[] = ['sent', 'failed']; + +/** + * The two engine operations reclamation issues, declared structurally. + * + * Same reasoning as #5210's `LifecycleFloorRegistrar`: this is a slot lookup's + * worth of surface, and typing it is what makes a renamed or re-ordered engine + * method a build error instead of a runtime throw inside a job that retries + * silently until it dead-letters. + */ +export interface AttachmentReclaimEngine { + find(object: string, options: Record): Promise; + update(object: string, data: Record, options?: Record): Promise; +} + +/** Structural logger — same shape the outbox sweep uses. */ +interface ReclaimLogger { + info?: (msg: string, meta?: any) => void; + warn?: (msg: string, meta?: any) => void; + error?: (msg: string, meta?: any) => void; +} + +export interface ReclaimAttachmentContentOptions { + engine: AttachmentReclaimEngine; + /** The live storage capability; `undefined` ⇒ nothing can be deleted now. */ + store: EmailAttachmentStore | undefined; + logger?: ReclaimLogger; + /** Clock seam for tests. */ + now?: () => number; + /** Override {@link EMAIL_ATTACHMENT_RECLAIM_GRACE_MS} (tests). */ + graceMs?: number; + /** + * Re-publish this job to fire again in `delayMs`. `undefined`, or a `false` + * return, means the job could not be re-armed — reported loudly, because the + * content then waits for nothing. + */ + rearm?: (delayMs: number) => Promise; +} + +export type AttachmentReclaimOutcome = + /** The row is terminal: content deleted, `attachments_json` rewritten. */ + | { kind: 'reclaimed'; rowId: string; keys: string[] } + /** The row is gone: content deleted from the payload's keys alone. */ + | { kind: 'orphan-reclaimed'; rowId: string; keys: string[] } + /** Not yet — the row is still in flight. Re-armed for another grace window. */ + | { kind: 'rearmed'; rowId: string; delayMs: number; reason: string } + /** Not yet, and nothing re-armed it. The bytes are now waiting on nobody. */ + | { kind: 'stalled'; rowId: string; reason: string } + /** Nothing to do — the payload named no keys. */ + | { kind: 'noop'; rowId: string }; + +/** Normalize the two `find()` return shapes ObjectQL hands back. */ +function firstRow(raw: unknown): Record | undefined { + if (Array.isArray(raw)) return raw[0] as Record | undefined; + const data = (raw as any)?.data; + return Array.isArray(data) ? (data[0] as Record | undefined) : undefined; +} + +/** Millisecond age of a row's last recorded change, or `undefined` if unknown. */ +function msSinceLastChange(row: Record, now: number): number | undefined { + const stamp = row.updated_at ?? row.sent_at ?? row.created_at; + if (stamp == null) return undefined; + const t = new Date(String(stamp)).getTime(); + if (!Number.isFinite(t)) return undefined; + return now - t; +} + +/** + * Run one reclamation job. + * + * Throws only for failures a retry can fix — a storage delete that did not + * land, a row rewrite that did not land. Both adapters' `delete` are idempotent + * for a key that is already gone (the local one swallows `ENOENT`, S3's + * `DeleteObject` is idempotent by protocol), so the retry re-runs cleanly + * rather than getting permanently stuck on the half it already finished. + */ +export async function reclaimAttachmentContent( + payload: EmailAttachmentReclaimPayload, + opts: ReclaimAttachmentContentOptions, +): Promise { + const rowId = String(payload?.rowId ?? ''); + const keys = Array.from(new Set((payload?.keys ?? []).filter((k) => typeof k === 'string' && k !== ''))); + const now = opts.now?.() ?? Date.now(); + const graceMs = opts.graceMs ?? EMAIL_ATTACHMENT_RECLAIM_GRACE_MS; + + if (keys.length === 0) return { kind: 'noop', rowId }; + + if (!opts.store) { + // The capability that HOLDS the bytes is not mounted on this process. + // Deleting is impossible; re-arm so a process that has it can finish the + // job, and say so if nothing can. + return rearmOrStall( + opts, rowId, graceMs, + 'the file-storage capability is not mounted on the process running the reclaim job, so the content ' + + 'cannot be deleted here', + ); + } + + const found = await opts.engine.find(RECLAIM_OBJECT, { + where: { id: rowId }, + limit: 1, + context: SYSTEM_CTX, + }); + const row = firstRow(found); + + // ── The row is gone ──────────────────────────────────────────────────── + // Nothing references these bytes any more and nothing ever will. Deleting + // them from the payload's own key list is what makes row deletion incapable + // of orphaning content — see the module header. + if (!row) { + await deleteOrThrow(opts.store, keys, rowId, 'the sys_email row no longer exists'); + opts.logger?.info?.( + `EmailServicePlugin: reclaimed ${keys.length} out-of-row attachment object(s) for sys_email row ` + + `'${rowId}', which no longer exists — the content had no remaining reference.`, + ); + return { kind: 'orphan-reclaimed', rowId, keys }; + } + + // ── The row is still in flight ───────────────────────────────────────── + const status = String(row.status ?? ''); + if (!TERMINAL_EMAIL_STATUSES.includes(status)) { + return rearmOrStall( + opts, rowId, graceMs, + `the sys_email row is still at '${status || '(unset)'}', so its content may still be needed for delivery`, + ); + } + + // Belt-and-braces on top of the job's own delay: if the row changed inside + // the grace window (a retry that re-stamped it, a clock the publisher and + // the row do not share), wait out the remainder. This can only ever DELAY a + // deletion, never bring one forward, which is the only direction a tolerance + // on a destructive operation may point. + const age = msSinceLastChange(row, now); + if (age !== undefined && age < graceMs) { + return rearmOrStall( + opts, rowId, graceMs - age, + `the sys_email row was last updated ${Math.round(age / 1000)}s ago, inside the ` + + `${Math.round(graceMs / 1000)}s reclaim grace window`, + ); + } + + // ── Reclaim ──────────────────────────────────────────────────────────── + // Bytes first, row second. The other order would rewrite away the last + // pointer to content a failed delete left behind — an orphan created by the + // very function that exists to prevent them. This order's failure mode is a + // row that names a key which 404s, which is loud on read (the decoder + // verifies) and fixed by the job's next attempt. + await deleteOrThrow(opts.store, keys, rowId, `the sys_email row is terminal ('${status}')`); + + const rewritten = withContentReclaimed(row.attachments_json, new Date(now).toISOString()); + if (rewritten !== undefined) { + await opts.engine.update( + RECLAIM_OBJECT, + { id: rowId, attachments_json: rewritten }, + { context: SYSTEM_CTX }, + ); + } + + opts.logger?.info?.( + `EmailServicePlugin: reclaimed ${keys.length} out-of-row attachment object(s) for sys_email row ` + + `'${rowId}' (status '${status}'). The row keeps filename/contentType/size/hash as audit evidence; only ` + + 'the delivery content was deleted.', + ); + return { kind: 'reclaimed', rowId, keys }; +} + +/** Delete every key or throw naming what is left — the job's retry finishes it. */ +async function deleteOrThrow( + store: EmailAttachmentStore, + keys: string[], + rowId: string, + because: string, +): Promise { + const { failed } = await deleteAttachmentKeys(store, keys); + if (failed.length === 0) return; + throw new Error( + `reclaiming attachment content for sys_email row '${rowId}' (${because}) could not delete ` + + `${failed.length} of ${keys.length} storage object(s): ` + + failed.map((f) => `'${f.key}' (${f.error})`).join('; ') + + '. Those bytes are still billed and no longer needed; the reclaim job will retry.', + ); +} + +/** Re-arm the job, or report that nothing will look at this content again. */ +async function rearmOrStall( + opts: ReclaimAttachmentContentOptions, + rowId: string, + delayMs: number, + reason: string, +): Promise { + const delay = Math.max(1000, Math.round(delayMs)); + if (opts.rearm && (await opts.rearm(delay))) { + return { kind: 'rearmed', rowId, delayMs: delay, reason }; + } + // `error`, not `warn`: from the outside nothing is wrong — the mail went + // out, the row is fine — while attachment content that was supposed to be + // temporary has just become permanent, silently, in a bucket somebody pays + // for. That is the durability/consequence class AGENTS.md pins at `error`, + // and the line owes both the consequence and the fix. + opts.logger?.error?.( + `EmailServicePlugin: out-of-row attachment content for sys_email row '${rowId}' could NOT be reclaimed and ` + + `could NOT be rescheduled — ${reason}. Those storage objects will now stay in the backend forever unless ` + + 'they are deleted by hand. Fix: keep the durable queue service (@objectstack/service-queue over an ' + + 'ObjectQL engine) and the file-storage capability (@objectstack/service-storage) mounted on the process ' + + `that consumes email jobs, then delete the leftovers under the row's key prefix.`, + ); + return { kind: 'stalled', rowId, reason }; +} diff --git a/packages/plugins/plugin-email/src/attachment-storage.test.ts b/packages/plugins/plugin-email/src/attachment-storage.test.ts new file mode 100644 index 0000000000..8e4d021c3f --- /dev/null +++ b/packages/plugins/plugin-email/src/attachment-storage.test.ts @@ -0,0 +1,326 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. +// +// Out-of-row attachment content — the codec half (objectstack#5172). +// +// What these pin, in one sentence each: +// - the key scheme is deterministic, path-safe, and says which row owns it; +// - an offloaded element carries the PERMANENT audit metadata and a +// reference, and no content; +// - the round trip is byte- and type-exact, including the `string | Buffer` +// arm, because the transport emits a different Content-Type for each; +// - every way the content can fail to come back — outage, truncation, +// substitution, reclaimed, no capability at all — is a REFUSAL, never a +// message delivered without the attachment it declares; +// - a partial upload cleans up after itself, because bytes no row will ever +// reference are the one orphan class this design must not create. + +import { describe, it, expect, vi } from 'vitest'; +import { createHash } from 'node:crypto'; +import { + EMAIL_ATTACHMENT_KEY_PREFIX, + attachmentStorageKey, + deleteAttachmentKeys, + fetchAttachmentContent, + offloadAttachmentsToStorage, + type EmailAttachmentStore, +} from './attachment-storage.js'; +import { + decodeAttachmentsFromRow, + decodeAttachmentsFromRowAsync, + storageKeysInColumn, + withContentReclaimed, +} from './sys-email-payload.js'; + +const sha = (s: string | Buffer) => `sha256:${createHash('sha256').update(s).digest('hex')}`; + +/** In-memory stand-in for the `file-storage` capability. */ +function fakeStore(opts: { failUploadAt?: number; failDelete?: Set } = {}) { + const objects = new Map(); + let uploads = 0; + const store = { + objects, + uploadCalls: [] as Array<{ key: string; contentType?: string }>, + async upload(key: string, data: Buffer, options?: { contentType?: string }) { + if (opts.failUploadAt !== undefined && uploads === opts.failUploadAt) { + uploads++; + throw new Error('bucket is on fire'); + } + uploads++; + store.uploadCalls.push({ key, ...(options?.contentType ? { contentType: options.contentType } : {}) }); + objects.set(key, Buffer.from(data)); + }, + async download(key: string) { + const found = objects.get(key); + if (!found) throw new Error(`NoSuchKey: ${key}`); + return found; + }, + async delete(key: string) { + if (opts.failDelete?.has(key)) throw new Error('AccessDenied'); + objects.delete(key); + }, + }; + return store satisfies EmailAttachmentStore & Record; +} + +describe('the storage key scheme', () => { + it('names the owning object, groups by row, and orders by attachment index', () => { + const k0 = attachmentStorageKey('row-1', 0, sha('a')); + const k1 = attachmentStorageKey('row-1', 1, sha('b')); + expect(k0.startsWith(`${EMAIL_ATTACHMENT_KEY_PREFIX}/row-1/`)).toBe(true); + expect(k0).toMatch(/\/000-[0-9a-f]{16}$/); + expect(k1).toMatch(/\/001-[0-9a-f]{16}$/); + // Zero-padded so a bucket listing sorts the way the attachments do. + expect([k1, k0].sort()).toEqual([k0, k1]); + }); + + it('is deterministic, so a retried send overwrites its own bytes instead of leaving a second copy', () => { + expect(attachmentStorageKey('row-1', 0, sha('a'))).toBe(attachmentStorageKey('row-1', 0, sha('a'))); + }); + + it('separates rows even for identical content', () => { + expect(attachmentStorageKey('row-1', 0, sha('a'))).not.toBe(attachmentStorageKey('row-2', 0, sha('a'))); + }); + + it('folds a row id that could escape the prefix — a key is a path on the local adapter', () => { + const key = attachmentStorageKey('../../etc/passwd', 0, sha('a')); + expect(key.startsWith(`${EMAIL_ATTACHMENT_KEY_PREFIX}/`)).toBe(true); + expect(key).not.toContain('..'); + expect(key.split('/')).toHaveLength(4); // sys_email / attachments / / + }); + + it('never puts the filename in the key — filenames are author-supplied and live in the row', () => { + const key = attachmentStorageKey('row-1', 0, sha('a')); + expect(key).not.toContain('.'); + }); +}); + +describe('offloading content to storage', () => { + const BIG = Buffer.alloc(300 * 1024, 0x41); + + it('stores the content and records a reference plus the PERMANENT audit metadata', async () => { + const store = fakeStore(); + const res = await offloadAttachmentsToStorage( + [{ filename: 'contract.pdf', content: BIG, contentType: 'application/pdf' }], + 'row-1', + store, + ); + + expect(res.kind).toBe('storage'); + if (res.kind !== 'storage') return; + const [el] = JSON.parse(res.json); + // The audit half — this is what survives reclamation, forever. + expect(el).toMatchObject({ + filename: 'contract.pdf', + contentType: 'application/pdf', + size: BIG.byteLength, + hash: sha(BIG), + contentForm: 'buffer', + }); + // The delivery half — a reference, and NOT the content. + expect(el.storageKey).toBe(res.keys[0]); + expect(el.inline).toBeUndefined(); + expect(el.content).toBeUndefined(); + // The row is now tiny regardless of how big the attachment was. + expect(res.json.length).toBeLessThan(400); + expect(store.objects.get(res.keys[0])!.equals(BIG)).toBe(true); + expect(store.uploadCalls[0].contentType).toBe('application/pdf'); + }); + + it('round-trips through the async reader, byte- and type-exact for BOTH content arms', async () => { + const store = fakeStore(); + const text = 'x'.repeat(300 * 1024); // string arm — charset matters on the wire + const res = await offloadAttachmentsToStorage( + [ + { filename: '对账单.txt', content: text, cid: 'stmt@inline' }, + { filename: 'b.bin', content: BIG }, + ], + 'row-1', + store, + ); + expect(res.kind).toBe('storage'); + if (res.kind !== 'storage') return; + + const rebuilt = await decodeAttachmentsFromRowAsync(res.json, (k) => store.download(k)); + expect(rebuilt).toHaveLength(2); + expect(rebuilt![0]).toEqual({ filename: '对账单.txt', content: text, cid: 'stmt@inline' }); + expect(typeof rebuilt![0].content).toBe('string'); + expect(Buffer.isBuffer(rebuilt![1].content)).toBe(true); + expect((rebuilt![1].content as Buffer).equals(BIG)).toBe(true); + }); + + it('deletes what it already uploaded when a later upload fails — no bytes without a referrer', async () => { + const store = fakeStore({ failUploadAt: 1 }); + const res = await offloadAttachmentsToStorage( + [{ filename: 'a.bin', content: BIG }, { filename: 'b.bin', content: BIG }], + 'row-1', + store, + ); + + expect(res.kind).toBe('unavailable'); + if (res.kind !== 'unavailable') return; + expect(res.detail).toContain("uploading attachment 'b.bin'"); + expect(res.detail).toContain('bucket is on fire'); + // The point of the case: nothing is left behind. + expect([...store.objects.keys()]).toEqual([]); + }); + + it('reports content outside the EmailAttachment contract rather than uploading something odd', async () => { + const store = fakeStore(); + const res = await offloadAttachmentsToStorage( + [{ filename: 'odd.bin', content: new Uint8Array([1, 2, 3]) as unknown as Buffer }], + 'row-1', + store, + ); + expect(res.kind).toBe('unavailable'); + if (res.kind !== 'unavailable') return; + expect(res.detail).toContain('neither a string nor a Buffer'); + expect(store.objects.size).toBe(0); + }); +}); + +describe('reading content back — every failure is a refusal, never a stripped message', () => { + const BIG = Buffer.alloc(300 * 1024, 0x42); + + async function offloaded(store: EmailAttachmentStore) { + const res = await offloadAttachmentsToStorage([{ filename: 'a.bin', content: BIG }], 'row-1', store); + if (res.kind !== 'storage') throw new Error('fixture failed to offload'); + return res; + } + + it('propagates a storage read outage instead of pretending the attachment was not there', async () => { + const store = fakeStore(); + const res = await offloaded(store); + const fetch = vi.fn(async () => { throw new Error('S3 unreachable'); }); + await expect(decodeAttachmentsFromRowAsync(res.json, fetch)).rejects.toThrow(/S3 unreachable/); + }); + + it('rejects a key the backend no longer holds', async () => { + const store = fakeStore(); + const res = await offloaded(store); + store.objects.clear(); + await expect(decodeAttachmentsFromRowAsync(res.json, (k) => store.download(k))) + .rejects.toThrow(/NoSuchKey/); + }); + + it('rejects TRUNCATED storage content — size is verified on the way back in, not just on the way out', async () => { + const store = fakeStore(); + const res = await offloaded(store); + await expect(decodeAttachmentsFromRowAsync(res.json, async () => BIG.subarray(0, 10))) + .rejects.toThrow(/is 10 byte\(s\) but the row records size 307200/); + }); + + it('rejects SUBSTITUTED storage content of the right length — the digest is the real check', async () => { + const store = fakeStore(); + const res = await offloaded(store); + await expect(decodeAttachmentsFromRowAsync(res.json, async () => Buffer.alloc(BIG.byteLength, 0x43))) + .rejects.toThrow(/is not the content that was sent/); + }); + + it('rejects a non-Buffer answer from the capability, naming the key', async () => { + const store = fakeStore(); + const res = await offloaded(store); + const bad = { + download: async () => 'not a buffer' as unknown as Buffer, + } as unknown as EmailAttachmentStore; + await expect(fetchAttachmentContent(bad, res.keys[0])) + .rejects.toThrow(new RegExp(`non-Buffer value for attachment content key '${res.keys[0]}'`)); + }); + + it('refuses when NO storage capability is mounted, naming what to mount', async () => { + const store = fakeStore(); + const res = await offloaded(store); + await expect(decodeAttachmentsFromRowAsync(res.json, undefined)) + .rejects.toThrow(/no file-storage capability is mounted on this process/); + // …and the synchronous entry point says the same thing rather than + // silently returning a message with one fewer attachment. + expect(() => decodeAttachmentsFromRow(res.json)) + .toThrow(/no file-storage capability to fetch it from/); + }); +}); + +describe('reclaiming the column (the row keeps the audit metadata)', () => { + const column = JSON.stringify([ + { + filename: 'contract.pdf', + contentType: 'application/pdf', + size: 307200, + hash: sha('anything'), + cid: 'c@x', + contentForm: 'buffer', + storageKey: 'sys_email/attachments/row-1/000-abcdef0123456789', + }, + ]); + + it('drops storageKey, stamps contentReclaimedAt, and touches nothing else', () => { + const next = withContentReclaimed(column, '2026-08-05T00:00:00.000Z'); + expect(next).toBeDefined(); + const [el] = JSON.parse(next!); + expect(el).toEqual({ + filename: 'contract.pdf', + contentType: 'application/pdf', + size: 307200, + hash: sha('anything'), + cid: 'c@x', + contentForm: 'buffer', + contentReclaimedAt: '2026-08-05T00:00:00.000Z', + }); + }); + + it('is a no-op the second time — a reclaimed row is not re-stamped', () => { + const once = withContentReclaimed(column, '2026-08-05T00:00:00.000Z')!; + expect(withContentReclaimed(once, '2026-08-09T00:00:00.000Z')).toBeUndefined(); + }); + + it('leaves an inline (small-attachment) row alone — its content is not out of the row', () => { + const inlineCol = JSON.stringify([ + { filename: 'a.txt', size: 2, hash: sha('hi'), contentForm: 'string', inline: Buffer.from('hi').toString('base64') }, + ]); + expect(withContentReclaimed(inlineCol, '2026-08-05T00:00:00.000Z')).toBeUndefined(); + }); + + it('refuses to re-send a reclaimed row, and says the content was reclaimed rather than lost', async () => { + const next = withContentReclaimed(column, '2026-08-05T00:00:00.000Z')!; + await expect(decodeAttachmentsFromRowAsync(next, async () => Buffer.alloc(0))) + .rejects.toThrow(/had its out-of-row content reclaimed at 2026-08-05T00:00:00.000Z/); + }); +}); + +describe('storageKeysInColumn — deliberately tolerant, because it feeds DELETION', () => { + it('lists the keys of a well-formed column, in element order', () => { + const col = JSON.stringify([ + { filename: 'a', size: 1, hash: 'h', contentForm: 'buffer', storageKey: 'k0' }, + { filename: 'b', size: 1, hash: 'h', contentForm: 'buffer', storageKey: 'k1' }, + ]); + expect(storageKeysInColumn(col)).toEqual(['k0', 'k1']); + }); + + it('still names the keys of a column too damaged to deliver from', () => { + // No `hash`, no `contentForm` — the strict decoder rejects this outright. + const col = JSON.stringify([{ filename: '', storageKey: 'k0' }, { junk: true }]); + expect(() => decodeAttachmentsFromRow(col)).toThrow(); + // Refusing to parse here would strand those bytes forever; the failure + // mode has no upside, so this one path reads what it can. + expect(storageKeysInColumn(col)).toEqual(['k0']); + }); + + it('answers empty for absent, unparseable, and inline-only columns', () => { + expect(storageKeysInColumn(null)).toEqual([]); + expect(storageKeysInColumn('')).toEqual([]); + expect(storageKeysInColumn('{not json')).toEqual([]); + expect(storageKeysInColumn(JSON.stringify([{ filename: 'a', inline: 'aGk=' }]))).toEqual([]); + }); +}); + +describe('deleteAttachmentKeys', () => { + it('reports which keys survived, so a byte leak is never silent', async () => { + const store = fakeStore({ failDelete: new Set(['k1']) }); + store.objects.set('k0', Buffer.from('a')); + store.objects.set('k1', Buffer.from('b')); + + const res = await deleteAttachmentKeys(store, ['k0', 'k1']); + + expect(res.deleted).toEqual(['k0']); + expect(res.failed).toEqual([{ key: 'k1', error: 'AccessDenied' }]); + expect(store.objects.has('k1')).toBe(true); + }); +}); diff --git a/packages/plugins/plugin-email/src/attachment-storage.ts b/packages/plugins/plugin-email/src/attachment-storage.ts new file mode 100644 index 0000000000..298f9f0050 --- /dev/null +++ b/packages/plugins/plugin-email/src/attachment-storage.ts @@ -0,0 +1,323 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * Out-of-row attachment content — the `storageKey` producer phase 1 declared + * and left empty (objectstack#5172, phase 2 of #5177/#5211). + * + * ## What this buys, in one sentence + * + * A message whose attachments exceed {@link SYS_EMAIL_ATTACHMENT_LIMIT_BYTES} + * used to be pushed back onto inline delivery — whole, but with none of the + * durability queue delivery exists to provide. Its content now goes to the + * `file-storage` capability, the row records a **reference** plus the audit + * metadata, and the queue worker rebuilds the message by fetching the content + * back. The messages most likely to matter (a signed contract, an exported + * report) stop being the ones the platform is weakest about. + * + * ## The one asymmetry that shapes everything here + * + * `sys_email` is an append-only audit log; attachment content is a **delivery + * artifact**. Those two have different lifetimes and the whole design is that + * cut: + * + * - `filename` / `contentType` / `size` / `hash` stay in the row **forever**. + * They are the evidence that a particular file was sent to a particular + * recipient, and they are ~100 bytes. + * - the bytes live in storage only until the row reaches a terminal state and + * a grace window has passed ({@link EMAIL_ATTACHMENT_RECLAIM_GRACE_MS}). + * Nothing needs them after that: the message was delivered. + * + * Without the cut, an append-only log grows binary content without bound. With + * it, the log grows the way a log grows. + * + * ## Why the capability is declared here rather than imported + * + * {@link EmailAttachmentStore} names the three `IStorageService` methods this + * package calls and nothing else. Declared in the *consumer* (the #5210 + * `LifecycleFloorRegistrar` shape) so the slot lookup carries a contract type: + * `getService('file-storage')` would switch off checking on the exact + * calls whose failure is invisible — an upload that silently no-ops is a + * message whose content is gone and whose row says it is there. + * + * ## Loudness + * + * Every failure on this path degrades to **inline delivery of the whole + * message**, never to a queued message with content missing. `declared ≠ + * delivered` is the one outcome that is not allowed: the reader + * ({@link fetchAttachmentContent} and the strict decoder it feeds) throws + * rather than hand a transport a message with an attachment quietly absent. + */ + +import type { EmailAttachment } from '@objectstack/spec/contracts'; +import { + SYS_EMAIL_ATTACHMENT_LIMIT_BYTES, + collectAttachmentParts, + type PersistedEmailAttachment, +} from './sys-email-payload.js'; + +/** + * The slice of the `file-storage` capability (`IStorageService`) that + * out-of-row attachment content actually uses. + * + * Declared structurally in this package on purpose (#5210): a slot lookup that + * erases to `any` compiles against a renamed or re-ordered method and fails at + * runtime inside the very `try` that is supposed to degrade gracefully — i.e. + * large attachments would silently stop being durable while the log line said + * "storage unavailable", which is indistinguishable from the deployment simply + * not having storage. Typed, the mismatch is a build error. + * + * `list` / `getInfo` / presigned URLs are deliberately absent: nothing here + * enumerates storage. (It could not portably — `LocalStorageAdapter.list` is a + * single-level `readdir` while the S3 adapter's is a recursive, unpaginated + * `ListObjectsV2` — so a listing-driven reclaimer would silently do different + * things on the two shipped adapters. Reclamation is driven by the queue + * instead; see {@link EMAIL_ATTACHMENT_RECLAIM_QUEUE}.) + */ +export interface EmailAttachmentStore { + upload(key: string, data: Buffer, options?: { contentType?: string }): Promise; + download(key: string): Promise; + delete(key: string): Promise; +} + +/** Resolve the live store per use — never captured. See {@link EmailAttachmentStorage}. */ +export interface EmailAttachmentStorage { + /** + * The store to use for THIS message, or `undefined` when the capability is + * not mounted. + * + * A thunk for the same reason `EmailQueueDelivery.resolve` is one: the + * storage service is a `SwappableStorageService` whose adapter is replaced + * when the `storage` settings namespace changes, and it may register after + * this plugin. A handle captured at wiring time would upload into the + * adapter the operator just swapped away from. + */ + resolve(): EmailAttachmentStore | undefined; +} + +/** + * Key prefix for every byte this package writes to the storage backend. + * + * Names the owning object, so an operator looking at a bucket can tell what + * wrote these and which table decides when they die. Everything under it is + * reclaimable delivery state, never business truth — the opposite of + * `sys_file`, which is why this path deliberately does NOT go through the + * `sys_file` metadata store: a permanent, compliance-visible row per + * attachment would re-create the unbounded growth this design exists to avoid. + */ +export const EMAIL_ATTACHMENT_KEY_PREFIX = 'sys_email/attachments'; + +/** + * Queue topic the content reclaimer is published to and consumed from. + * + * Separate from `email.send.async` on purpose: a reclaim job is scheduled a + * day out and must not share a retry budget, a DLQ or a poll order with mail + * that is trying to go out now. + */ +export const EMAIL_ATTACHMENT_RECLAIM_QUEUE = 'email.attachment.reclaim'; + +/** + * How long after a `sys_email` row last changed its content may be reclaimed. + * + * 24 hours, a constant rather than a setting (same reasoning as + * {@link SYS_EMAIL_ATTACHMENT_LIMIT_BYTES}: a knob is a second place for a + * storage budget to drift). The number has to clear three things at once and + * this is the smallest round value that clears all of them: + * + * 1. **The queue's own retry span.** A row sits at `failed` between attempts + * — the `email.send.async` subscriber re-reads a `failed` row and delivers + * it again. With the backoff capped at 5 minutes, even a generous attempt + * budget is exhausted in well under an hour. Reclaiming inside that span + * would delete the content a retry is about to need. + * 2. **A duplicate delivery.** A lease that expires mid-send lets a second + * worker re-read the row; that read must still find its content. + * 3. **The operator.** "The invoice went out wrong, what exactly did we + * send?" is a same-day question. After a day it is answered from the + * audit metadata (filename / size / hash), which never goes away. + * + * It is also, deliberately, far shorter than any plausible row retention — see + * the module header of `attachment-reclaim.ts` for why that ordering is + * belt-and-braces rather than load-bearing. + */ +export const EMAIL_ATTACHMENT_RECLAIM_GRACE_MS = 24 * 60 * 60_000; + +/** + * Payload of an {@link EMAIL_ATTACHMENT_RECLAIM_QUEUE} job. + * + * `keys` is carried **in the payload** and not re-read from the row, and that + * is the single most load-bearing decision in the reclamation design: it is + * what makes orphaned content impossible. A job that finds its `sys_email` row + * gone — deleted by a future declarative retention policy, by a purge, by an + * operator — still knows exactly which bytes to delete. Had the job carried + * only `rowId`, deleting the row would have severed the last pointer to the + * content and left it in the bucket forever. + */ +export interface EmailAttachmentReclaimPayload { + /** `sys_email.id` whose content this job reclaims. */ + rowId: string; + /** Every storage key written for that row. Authoritative — see above. */ + keys: string[]; +} + +/** Characters a storage key segment may contain; everything else is folded. */ +const KEY_SAFE = /[^A-Za-z0-9_-]/g; + +/** + * The key one attachment's content is stored under. + * + * `sys_email/attachments//-` + * + * - **``** groups a message's parts under one folder, so purging one + * message by hand is one prefix and an operator reading a key can find the + * row it belongs to. It is folded to `[A-Za-z0-9_-]` first: a key is a path + * on the local adapter, and an id is not allowed to introduce a `/` or a + * `..` segment into one. + * - **``** is the element's index in `attachments_json`, zero-padded so + * keys sort the way the attachments do. Together with `` it makes + * the key **deterministic**: a retried send of the same row overwrites its + * own bytes instead of leaving a second copy nobody references. + * - **``** is the first 16 hex characters of the content digest the + * row also records. It costs nothing and means a key can never be silently + * re-pointed at different content: the reader verifies the FULL digest + * anyway, so this is for the human reading a bucket listing. + * + * Note what is NOT in the key: the filename. Filenames are author-supplied + * UTF-8 that may contain `/`, `..`, or control characters; they are audit + * metadata and they live in the row, never in a path. + */ +export function attachmentStorageKey(rowId: string, index: number, hash: string): string { + const safeRow = String(rowId).replace(KEY_SAFE, '_') || 'unknown'; + const hex = hash.startsWith('sha256:') ? hash.slice('sha256:'.length) : hash; + const short = hex.replace(KEY_SAFE, '').slice(0, 16) || 'nohash'; + return `${EMAIL_ATTACHMENT_KEY_PREFIX}/${safeRow}/${String(index).padStart(3, '0')}-${short}`; +} + +/** Outcome of {@link offloadAttachmentsToStorage}. */ +export type AttachmentOffload = + /** Content is in storage; `json` goes into `attachments_json`. */ + | { kind: 'storage'; json: string; keys: string[]; totalBytes: number } + /** + * Nothing was stored, and why. The caller's answer is always the same — + * deliver the message inline, whole — but the sentence differs and the + * operator needs it. + */ + | { kind: 'unavailable'; detail: string }; + +/** + * Upload one message's attachments and build the `attachments_json` elements + * that reference them. + * + * Never throws. Every failure becomes `{ kind: 'unavailable' }` carrying a + * sentence the caller logs, because the caller's response is to deliver the + * message inline — which is what happens today and is always whole. Failing + * the send instead would turn "this message is not durable" into "this message + * does not go out", a strictly worse answer to a storage hiccup. + * + * **Partial uploads are cleaned up.** If attachment 3 of 5 fails, the two + * already in the bucket are deleted before returning: no row will ever + * reference them, so leaving them would be exactly the orphan this design + * refuses to create. + */ +export async function offloadAttachmentsToStorage( + attachments: EmailAttachment[] | undefined, + rowId: string, + store: EmailAttachmentStore, +): Promise { + const collected = collectAttachmentParts(attachments); + if (collected.kind === 'none') { + // Unreachable through the one caller (a message is only offloaded because + // its attachments are over budget, which needs attachments). Answered + // rather than asserted so a future caller gets a verdict, not a crash. + return { kind: 'unavailable', detail: 'the message has no attachments to store' }; + } + if (collected.kind === 'unsupported') { + return { kind: 'unavailable', detail: collected.detail }; + } + const { parts, totalBytes } = collected; + + const items: PersistedEmailAttachment[] = []; + const keys: string[] = []; + for (let i = 0; i < parts.length; i++) { + const { att, bytes, hash, contentForm } = parts[i]!; + const key = attachmentStorageKey(rowId, i, hash); + try { + await store.upload(key, bytes, att.contentType ? { contentType: String(att.contentType) } : undefined); + } catch (err: any) { + await deleteAttachmentKeys(store, keys); + return { + kind: 'unavailable', + detail: + `uploading attachment '${String(att.filename ?? '(unnamed)')}' to the file-storage capability failed ` + + `(${String(err?.message ?? err)})`, + }; + } + keys.push(key); + items.push({ + filename: String(att.filename ?? ''), + ...(att.contentType ? { contentType: String(att.contentType) } : {}), + size: bytes.byteLength, + hash, + ...(att.cid ? { cid: String(att.cid) } : {}), + contentForm, + storageKey: key, + }); + } + + return { kind: 'storage', json: JSON.stringify(items), keys, totalBytes }; +} + +/** + * Delete storage keys, tolerating keys that are already gone. + * + * Returns the keys that could NOT be deleted, so the caller can say so. A + * delete that fails is a byte leak, not a correctness problem — the row is + * either gone or no longer references the key — but it is exactly the kind of + * leak that is invisible until a bucket bill arrives, so nothing here swallows + * the list. + * + * "Already gone" is not distinguishable across adapters (S3 deletes are + * idempotent, the local adapter's `unlink` throws `ENOENT`), so a failure is + * reported and reconciled by the job's own retry rather than being classified + * here on the basis of a message string. + */ +export async function deleteAttachmentKeys( + store: EmailAttachmentStore, + keys: readonly string[], +): Promise<{ deleted: string[]; failed: Array<{ key: string; error: string }> }> { + const deleted: string[] = []; + const failed: Array<{ key: string; error: string }> = []; + for (const key of keys) { + try { + await store.delete(key); + deleted.push(key); + } catch (err: any) { + failed.push({ key, error: String(err?.message ?? err) }); + } + } + return { deleted, failed }; +} + +/** + * Fetch one attachment's content back from storage. + * + * Deliberately **not** wrapped in a `try`. The storage metadata layer was made + * loud in #5216/#5232 precisely so a read outage stops looking like a miss; + * re-swallowing it one layer up would hand the transport a message with the + * attachment silently absent and mark the row `sent`. The throw propagates to + * `deliverPersistedRow`, which records it on the row as `failed` with the + * reason and lets the queue retry or dead-letter the job. + */ +export async function fetchAttachmentContent( + store: EmailAttachmentStore, + storageKey: string, +): Promise { + const bytes = await store.download(storageKey); + if (!Buffer.isBuffer(bytes)) { + // A store that answers with something else has not answered at all; the + // strict decoder downstream verifies size and digest, and this keeps the + // failure at the layer that can name the key. + throw new Error( + `the file-storage capability returned a non-Buffer value for attachment content key '${storageKey}'`, + ); + } + return bytes; +} diff --git a/packages/plugins/plugin-email/src/email-plugin.attachment-storage.test.ts b/packages/plugins/plugin-email/src/email-plugin.attachment-storage.test.ts new file mode 100644 index 0000000000..ec95473315 --- /dev/null +++ b/packages/plugins/plugin-email/src/email-plugin.attachment-storage.test.ts @@ -0,0 +1,359 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. +// +// EmailServicePlugin — large attachments end to end (objectstack#5172). +// +// Nothing here is mocked at the seams that matter: the queue is the REAL +// `DbQueueAdapter` over a `sys_job_queue` table, so what is asserted is the +// entire round trip a 300 KB invoice actually takes — +// +// send() → upload → sys_email row holding a REFERENCE → email.send.async +// job → worker poll → download → transport receives the whole message → +// row `sent` → a DELAYED email.attachment.reclaim job → (24h) → poll → +// bytes gone, audit metadata still on the row. +// +// …plus the two failure shapes the maintainer's ruling singles out: a +// deployment with no storage capability must keep delivering these messages +// whole (loudly, inline), and content whose row was deleted must still be +// reclaimed rather than orphaned. + +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import { assertEngineDeleteDispatch } from '@objectstack/objectql'; +import { DbQueueAdapter } from '@objectstack/service-queue'; +import { EmailServicePlugin } from './email-plugin.js'; +import type { EmailService } from './email-service.js'; +import { SYS_EMAIL_ATTACHMENT_LIMIT_BYTES } from './sys-email-payload.js'; +import { + EMAIL_ATTACHMENT_KEY_PREFIX, + EMAIL_ATTACHMENT_RECLAIM_QUEUE, + EMAIL_ATTACHMENT_RECLAIM_GRACE_MS, +} from './attachment-storage.js'; + +const NOW = Date.UTC(2026, 7, 5, 12, 0, 0); +const HOUR = 3600_000; +/** A 300 KB invoice — over the in-row budget, which is what makes it #5172's. */ +const INVOICE = Buffer.alloc(SYS_EMAIL_ATTACHMENT_LIMIT_BYTES + 44_000, 0x25); + +// ── harness ──────────────────────────────────────────────────────────────── + +/** ObjectQL-shaped engine: equality + `$lt`, ordering, limits, hooks. */ +function fakeEngine() { + const tables = new Map(); + const hooks: Record any>> = {}; + const rowsOf = (t: string) => tables.get(t) ?? []; + const matches = (row: any, where: Record) => + Object.entries(where).every(([k, v]) => { + if (v && typeof v === 'object' && !Array.isArray(v)) { + return Object.entries(v).every(([op, target]) => { + if (op === '$lt') return row[k] < (target as any); + throw new Error(`fakeEngine: unsupported operator ${op}`); + }); + } + return row[k] === v; + }); + const engine = { + rows: (t: string) => [...rowsOf(t)], + registerHook(event: string, fn: (ctx: any) => any, _meta?: unknown) { + (hooks[event] ??= []).push(fn); + }, + unregisterHooksByPackage(_pkg: string) { /* single-boot harness */ }, + async find(table: string, o: any = {}) { + let out = o.where ? rowsOf(table).filter((r) => matches(r, o.where)) : [...rowsOf(table)]; + for (const ord of [...(o.orderBy ?? [])].reverse()) { + out.sort((a, b) => { + const av = a[ord.field], bv = b[ord.field]; + if (av === bv) return 0; + return (av > bv ? 1 : -1) * (ord.order === 'desc' ? -1 : 1); + }); + } + if (o.offset) out = out.slice(o.offset); + if (o.limit) out = out.slice(0, o.limit); + return out; + }, + async insert(table: string, data: any) { + const row = { ...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 }; + }, + 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, o: any) { + // [#4550/#5197] Pinned to ObjectQL.delete's own dispatch predicate. + const dispatch = assertEngineDeleteDispatch(o); + if (dispatch.kind === 'multi') { + const survivors = rowsOf(table).filter((r) => !matches(r, o?.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 }; + }, + }; + return engine; +} + +/** In-memory `file-storage` capability, shaped like `IStorageService`. */ +function fakeStorage(opts: { failUpload?: boolean } = {}) { + const objects = new Map(); + return { + objects, + async upload(key: string, data: Buffer) { + if (opts.failUpload) throw new Error('bucket is on fire'); + objects.set(key, Buffer.from(data)); + }, + async download(key: string) { + const found = objects.get(key); + if (!found) throw new Error(`NoSuchKey: ${key}`); + return found; + }, + async delete(key: string) { objects.delete(key); }, + async exists(key: string) { return objects.has(key); }, + async getInfo(key: string) { + return { key, size: objects.get(key)?.byteLength ?? 0, lastModified: new Date() }; + }, + }; +} + +function fakeCtx(services: Record) { + const handlers: 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) => { (handlers[name] ??= []).push(fn); }, + fire: async (name: string) => { for (const fn of handlers[name] ?? []) await fn(); }, + }; +} + +function fakeClock(startMs = NOW) { + let t = startMs; + return { + now: () => new Date(t), + advance: (ms: number) => { t += ms; vi.setSystemTime(new Date(t)); }, + }; +} + +interface BootOpts { + storage?: ReturnType | null; + transport?: { send: (m: any) => Promise }; +} + +async function boot(opts: BootOpts = {}) { + const engine = fakeEngine(); + const clock = fakeClock(); + const storage = opts.storage === null ? undefined : (opts.storage ?? fakeStorage()); + 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, + queue: adapter, + }; + if (storage) services['file-storage'] = storage; + + const ctx = fakeCtx(services); + const plugin = new EmailServicePlugin({ + seedTemplates: false, + transport, + queueDelivery: true, + defaultFrom: 'no-reply@example.test', + }); + await plugin.init(ctx as never); + await plugin.start(ctx as never); + await ctx.fire('kernel:ready'); + await plugin.outboxSweepSettled; + + return { + plugin, ctx, engine, adapter, clock, transport, storage, + service: () => services.email as EmailService, + sysEmail: () => engine.rows('sys_email'), + jobs: (queue?: string) => engine.rows('sys_job_queue').filter((j) => !queue || j.queue === queue), + infoLines: () => ctx.logger.info.mock.calls.map((c) => String(c[0])).join('\n'), + errorLines: () => ctx.logger.error.mock.calls.map((c) => String(c[0])).join('\n'), + }; +} + +beforeEach(() => { vi.clearAllMocks(); vi.setSystemTime(new Date(NOW)); }); + +// ── the whole chain ──────────────────────────────────────────────────────── + +describe('a 300 KB attachment now gets the durability guarantee', () => { + it('goes out through the queue, whole, and its content dies with the delivery', async () => { + const h = await boot(); + + // 1. send() — the caller is told `queued` and NOTHING has been sent yet. + const res = await h.service().send({ + to: 'client@example.test', + subject: 'Invoice #42', + text: 'attached', + attachments: [{ filename: 'invoice.pdf', content: INVOICE, contentType: 'application/pdf' }], + }); + expect(res.status).toBe('queued'); + expect(h.transport.send).not.toHaveBeenCalled(); + + // 2. The row holds a REFERENCE plus the audit metadata — not 400 KB of base64. + const row = h.sysEmail()[0]; + const [el] = JSON.parse(row.attachments_json); + expect(el).toMatchObject({ + filename: 'invoice.pdf', + contentType: 'application/pdf', + size: INVOICE.byteLength, + contentForm: 'buffer', + }); + expect(el.inline).toBeUndefined(); + expect(el.storageKey).toContain(`${EMAIL_ATTACHMENT_KEY_PREFIX}/${res.id}/`); + expect(row.attachments_json.length).toBeLessThan(400); + expect(h.storage!.objects.get(el.storageKey)!.equals(INVOICE)).toBe(true); + + // 3. The worker rebuilds the message from the row + storage, and the + // transport sees the attachment byte for byte. + await h.adapter.pollOnce(); + expect(h.transport.send).toHaveBeenCalledTimes(1); + const sent = (h.transport.send as any).mock.calls[0][0]; + expect(sent.attachments).toHaveLength(1); + expect((sent.attachments[0].content as Buffer).equals(INVOICE)).toBe(true); + expect(h.sysEmail()[0]).toMatchObject({ status: 'sent', message_id: '', attempt_count: 1 }); + + // 4. Reclamation is armed, one grace window out — and is NOT due yet. + const reclaimJobs = h.jobs(EMAIL_ATTACHMENT_RECLAIM_QUEUE); + expect(reclaimJobs).toHaveLength(1); + expect(JSON.parse(reclaimJobs[0].payload_json)).toEqual({ rowId: res.id, keys: [el.storageKey] }); + expect(new Date(reclaimJobs[0].scheduled_for).getTime()) + .toBe(NOW + EMAIL_ATTACHMENT_RECLAIM_GRACE_MS); + + // 5. Inside the window the content is untouched — an operator can still + // answer "what exactly did we send?". + h.clock.advance(HOUR); + await h.adapter.pollOnce(); + expect(h.storage!.objects.size).toBe(1); + + // 6. Past the window it is reclaimed, and the row keeps the audit facts. + h.clock.advance(EMAIL_ATTACHMENT_RECLAIM_GRACE_MS); + await h.adapter.pollOnce(); + + expect(h.storage!.objects.size).toBe(0); + const [after] = JSON.parse(h.sysEmail()[0].attachments_json); + expect(after.storageKey).toBeUndefined(); + expect(after.contentReclaimedAt).toBeDefined(); + expect(after).toMatchObject({ + filename: 'invoice.pdf', + contentType: 'application/pdf', + size: INVOICE.byteLength, + hash: el.hash, + }); + // The message itself is still fully auditable. + expect(h.sysEmail()[0]).toMatchObject({ + status: 'sent', + subject: 'Invoice #42', + to_addresses: 'client@example.test', + }); + }); + + it('reclaims content whose row was deleted (the #5192 retention case) instead of orphaning it', async () => { + const h = await boot(); + const res = await h.service().send({ + to: 'client@example.test', + subject: 'Invoice #43', + text: 'attached', + attachments: [{ filename: 'invoice.pdf', content: INVOICE }], + }); + await h.adapter.pollOnce(); + expect(h.storage!.objects.size).toBe(1); + + // A declarative `retention` policy in the #5192 shape reaps the row — a + // bulk, predicate-shaped delete, exactly what LifecycleService issues. + await h.engine.delete('sys_email', { where: { status: 'sent' }, multi: true }); + expect(h.sysEmail()).toHaveLength(0); + + // The job carries the KEYS, so the deletion of the row is not the loss of + // the last pointer to the bytes — it is the clearest possible signal that + // nothing needs them. + h.clock.advance(EMAIL_ATTACHMENT_RECLAIM_GRACE_MS + HOUR); + await h.adapter.pollOnce(); + + expect(h.storage!.objects.size).toBe(0); + expect(h.infoLines()).toContain('no longer exists'); + expect(h.jobs(EMAIL_ATTACHMENT_RECLAIM_QUEUE)[0].status).toBe('completed'); + void res; + }); + + it('small attachments are untouched by all of this — still inline, still queued (#5177)', async () => { + const h = await boot(); + + const res = await h.service().send({ + to: 'client@example.test', + subject: 'Receipt', + text: 'attached', + attachments: [{ filename: 'r.txt', content: '¥1.00' }], + }); + await h.adapter.pollOnce(); + + const [el] = JSON.parse(h.sysEmail()[0].attachments_json); + expect(el.inline).toBeDefined(); + expect(el.storageKey).toBeUndefined(); + expect(h.storage!.objects.size).toBe(0); + expect(h.jobs(EMAIL_ATTACHMENT_RECLAIM_QUEUE)).toHaveLength(0); + expect(h.sysEmail()[0]).toMatchObject({ id: res.id, status: 'sent' }); + const sent = (h.transport.send as any).mock.calls[0][0]; + expect(sent.attachments[0].content).toBe('¥1.00'); + }); +}); + +// ── the deployments that cannot store out of row ─────────────────────────── + +describe('a deployment without the file-storage capability', () => { + it('still delivers the message WHOLE, inline, and says what to mount', async () => { + const h = await boot({ storage: null }); + + const res = await h.service().send({ + to: 'client@example.test', + subject: 'Invoice #44', + text: 'attached', + attachments: [{ filename: 'invoice.pdf', content: INVOICE }], + }); + + // Delivered, not dropped, not stripped — the pre-#5172 behaviour exactly. + expect(res.status).toBe('sent'); + const sent = (h.transport.send as any).mock.calls[0][0]; + expect((sent.attachments[0].content as Buffer).equals(INVOICE)).toBe(true); + // Nothing queued, nothing in the column. + expect(h.jobs()).toHaveLength(0); + expect(h.sysEmail()[0].attachments_json).toBeUndefined(); + // Loud, with the obstacle and the fix. + const info = h.infoLines(); + expect(info).toContain('no file-storage capability is mounted'); + expect(info).toContain('@objectstack/service-storage'); + }); + + it('falls back the same way when the upload itself fails, and reports the durability loss', async () => { + const h = await boot({ storage: fakeStorage({ failUpload: true }) }); + + const res = await h.service().send({ + to: 'client@example.test', + subject: 'Invoice #45', + text: 'attached', + attachments: [{ filename: 'invoice.pdf', content: INVOICE }], + }); + + expect(res.status).toBe('sent'); + expect(h.jobs()).toHaveLength(0); + expect(h.storage!.objects.size).toBe(0); + expect(h.errorLines()).toContain('could not be stored out of row'); + expect(h.errorLines()).toContain('bucket is on fire'); + }); +}); diff --git a/packages/plugins/plugin-email/src/email-plugin.ts b/packages/plugins/plugin-email/src/email-plugin.ts index a819d06a24..2635c84c9e 100644 --- a/packages/plugins/plugin-email/src/email-plugin.ts +++ b/packages/plugins/plugin-email/src/email-plugin.ts @@ -40,6 +40,16 @@ import { unbindEmailTemplateProvenanceStamp, } from './email-template-provenance.js'; import { sweepStrandedOutbox, type OutboxSweepResult } from './outbox-sweep.js'; +import { + EMAIL_ATTACHMENT_RECLAIM_QUEUE, + EMAIL_ATTACHMENT_RECLAIM_GRACE_MS, + type EmailAttachmentReclaimPayload, + type EmailAttachmentStore, +} from './attachment-storage.js'; +import { + reclaimAttachmentContent, + type AttachmentReclaimEngine, +} from './attachment-reclaim.js'; const SYSTEM_CTX = { isSystem: true, positions: [], permissions: [] } as const; @@ -123,6 +133,40 @@ const QUEUE_DELIVERY_BACKOFF: QueueBackoffPolicy = { * layer over. It labels itself for exactly this purpose * (`__serviceInfo.status === 'degraded'`, ADR-0076 D12), so read the label. */ +/** + * Resolve the `file-storage` capability that can hold out-of-row attachment + * content (#5172), or `undefined`. + * + * Typed at the lookup — `EmailAttachmentStore`, declared in this package — + * rather than erased to `any` (#4127/#4251, the #5210 shape): the three calls + * this makes are `upload` / `download` / `delete`, and every one of them fails + * *invisibly* if the slot's shape changes under us. An upload that throws + * `x.upload is not a function` inside the offload's own `try` looks exactly + * like "the operator has no storage", so mail would quietly stop being durable + * with a log line blaming the deployment. + * + * Unlike `queue`, the kernel injects no in-memory fallback for `file-storage`, + * so absence is a plain throw from `getService` and there is no `degraded` + * label to read. The structural probe is still real: `SwappableStorageService` + * forwards to whatever adapter the `storage` settings namespace names. + */ +export function resolveAttachmentStore( + getService: (name: string) => unknown, +): EmailAttachmentStore | undefined { + let storage: unknown; + try { storage = getService('file-storage'); } catch { return undefined; } + const candidate = storage as Partial | undefined; + if ( + !candidate + || typeof candidate.upload !== 'function' + || typeof candidate.download !== 'function' + || typeof candidate.delete !== 'function' + ) { + return undefined; + } + return candidate as EmailAttachmentStore; +} + export function resolveDurableQueue(getService: (name: string) => unknown): IQueueService | undefined { let queue: any; try { queue = getService('queue'); } catch { return undefined; } @@ -471,6 +515,15 @@ export class EmailServicePlugin implements Plugin { this.service.setTemplateLoader(templateLoader); ctx.logger.info('EmailServicePlugin: sys_email persistence + template loader enabled'); + // Out-of-row attachment content (#5172). A thunk for the same reason the + // queue is one: `file-storage` is a SwappableStorageService whose + // adapter changes when the `storage` settings namespace does, and it may + // register after this plugin. Resolving here once would pin the service + // to the adapter that happened to exist at boot. + this.service.setAttachmentStorage({ + resolve: () => resolveAttachmentStore((name) => ctx.getService(name)), + }); + // 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 @@ -622,6 +675,73 @@ export class EmailServicePlugin implements Plugin { ctx.logger.warn(`EmailServicePlugin: ${EMAIL_SEND_QUEUE} subscription failed`, err as any); } + // ── 'email.attachment.reclaim' SUBSCRIBER (#5172) ───────────────── + // The consuming half of out-of-row attachment content. Each job is + // published at a row's terminal transition with a one-day delay and + // carries the storage keys, so it can delete the content even when the + // row it belonged to is gone by the time it fires — which is exactly + // what keeps a future row-retention policy from orphaning bytes. + try { + const queue: IQueueService | undefined = resolveDurableQueue((name) => ctx.getService(name)); + if (queue) { + const reclaimEngine = engine as unknown as AttachmentReclaimEngine; + await queue.subscribe( + EMAIL_ATTACHMENT_RECLAIM_QUEUE, + async (msg) => { + const payload = msg?.data; + if (!payload?.rowId) return; + // A throw here is the job's retry signal, and the reclaimer + // throws only for failures a retry can fix (a delete that did + // not land). Both shipped adapters delete idempotently, so the + // retry re-runs the whole key list cleanly. + await reclaimAttachmentContent(payload, { + engine: reclaimEngine, + store: resolveAttachmentStore((name) => ctx.getService(name)), + logger: ctx.logger, + rearm: async (delayMs) => { + try { + await queue.publish( + EMAIL_ATTACHMENT_RECLAIM_QUEUE, + payload, + { + delay: delayMs, + // A DIFFERENT key from the original publish: the job + // being re-armed is the one currently running, so + // reusing its key would let the running message's own + // idempotency record suppress its successor. + idempotencyKey: `sys_email_attachments:${payload.rowId}:rearm:${Date.now()}`, + maxAttempts: 5, + retries: 4, + backoff: { type: 'exponential', delayMs: 60_000, maxDelayMs: 60 * 60_000 }, + metadata: { object: 'sys_email', rowId: payload.rowId }, + }, + ); + return true; + } catch { + return false; + } + }, + }); + }, + ); + ctx.logger.info( + `EmailServicePlugin: subscribed to ${EMAIL_ATTACHMENT_RECLAIM_QUEUE} queue ` + + `(out-of-row attachment content is reclaimed ${Math.round(EMAIL_ATTACHMENT_RECLAIM_GRACE_MS / 3600_000)}h ` + + 'after a sys_email row reaches a terminal state; its filename/size/hash stay on the row forever)', + ); + } + } catch (err) { + // `error`, not `warn`: without this subscriber, content that was + // uploaded on the promise of being temporary is never deleted, and + // nothing else in the process looks at those jobs. + ctx.logger.error( + `EmailServicePlugin: ${EMAIL_ATTACHMENT_RECLAIM_QUEUE} subscription FAILED — out-of-row attachment ` + + 'content will keep being written but never reclaimed, so the storage backend grows without bound ' + + 'while everything else looks healthy. Fix: check the durable queue service ' + + `(@objectstack/service-queue over an ObjectQL engine) and restart. Cause: ${(err as any)?.message ?? err}`, + ); + } + // ── 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 diff --git a/packages/plugins/plugin-email/src/email-service.attachment-storage.test.ts b/packages/plugins/plugin-email/src/email-service.attachment-storage.test.ts new file mode 100644 index 0000000000..4a6f0129f1 --- /dev/null +++ b/packages/plugins/plugin-email/src/email-service.attachment-storage.test.ts @@ -0,0 +1,436 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. +// +// EmailService — large attachments through the file-storage capability +// (objectstack#5172). +// +// The routing table these pin, exhaustively, because every cell of it used to +// collapse into one behaviour ("deliver inline, lose durability"): +// +// attachments ≤ 256 KiB → in the row, queued (#5177) +// > 256 KiB, queue + storage → in storage, queued (#5172) +// > 256 KiB, queue, no storage → inline, whole, said out loud +// > 256 KiB, queue, upload failed → inline, whole, said at `error` +// > 256 KiB, no queue → inline, whole, nothing uploaded +// +// And the two things that must never happen no matter which cell you are in: +// a message queued against a row that cannot rebuild it, and content uploaded +// that no row will ever reference. + +import { describe, it, expect, vi } from 'vitest'; +import { createHash } from 'node:crypto'; +import type { IQueueService } from '@objectstack/spec/contracts'; +import { + EmailService, + EMAIL_SEND_QUEUE, + type EmailPersistence, + type EmailQueueDelivery, +} from './email-service.js'; +import { SYS_EMAIL_ATTACHMENT_LIMIT_BYTES } from './sys-email-payload.js'; +import { + EMAIL_ATTACHMENT_KEY_PREFIX, + EMAIL_ATTACHMENT_RECLAIM_QUEUE, + EMAIL_ATTACHMENT_RECLAIM_GRACE_MS, + type EmailAttachmentStore, +} from './attachment-storage.js'; + +const sha = (b: Buffer) => `sha256:${createHash('sha256').update(b).digest('hex')}`; +const MSG = { to: 'a@b.com', subject: 'Hi', text: 'hello' }; +/** One byte over the in-row budget — the smallest message that needs #5172. */ +const OVER = Buffer.alloc(SYS_EMAIL_ATTACHMENT_LIMIT_BYTES + 1, 0x41); +/** Exactly at the budget — still an in-row attachment (the bound includes equality). */ +const AT_LIMIT = Buffer.alloc(SYS_EMAIL_ATTACHMENT_LIMIT_BYTES, 0x42); + +function makePersistence(opts: { failInsert?: boolean } = {}) { + const rows = new Map>(); + const p: EmailPersistence = { + async insert(row) { + if (opts.failInsert) throw new Error('sys_email is read-only right now'); + 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 }; +} + +function makeQueue() { + const published: Array<{ queue: string; data: any; options: any }> = []; + const queue = { + published, + async publish(q: string, data: any, options: any) { + published.push({ queue: q, 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 makeStore(opts: { failUpload?: boolean; failDelete?: boolean } = {}) { + const objects = new Map(); + const store = { + objects, + async upload(key: string, data: Buffer) { + if (opts.failUpload) throw new Error('bucket is on fire'); + objects.set(key, Buffer.from(data)); + }, + async download(key: string) { + const found = objects.get(key); + if (!found) throw new Error(`NoSuchKey: ${key}`); + return found; + }, + async delete(key: string) { + if (opts.failDelete) throw new Error('AccessDenied'); + objects.delete(key); + }, + }; + return store satisfies EmailAttachmentStore & Record; +} + +const logger = () => ({ info: vi.fn(), warn: vi.fn(), error: vi.fn() }); +const lines = (fn: ReturnType) => fn.mock.calls.map((c) => String(c[0])).join('\n'); + +function wiring(queue: IQueueService | undefined, maxAttempts = 1): EmailQueueDelivery { + return { resolve: () => queue, maxAttempts, backoff: { type: 'exponential', delayMs: 1000 } }; +} + +interface SvcOpts { + queue?: IQueueService; + store?: EmailAttachmentStore; + failInsert?: boolean; + transportSend?: (m: any) => Promise; +} + +function makeService(o: SvcOpts = {}) { + const log = logger(); + const transport = { send: vi.fn(o.transportSend ?? (async () => ({ messageId: '' }))) }; + const { p, rows } = makePersistence({ failInsert: o.failInsert }); + const svc = new EmailService({ + transport, + defaultFrom: 'no@reply.com', + persistence: p, + logger: log, + ...(o.queue !== undefined ? { queueDelivery: wiring(o.queue) } : {}), + ...(o.store !== undefined ? { attachmentStorage: { resolve: () => o.store } } : {}), + }); + return { svc, transport, rows, log }; +} + +// ── the boundary, from both sides ────────────────────────────────────────── + +describe('the 256 KiB boundary decides in-row vs in-storage, and includes equality', () => { + it('EXACTLY at the limit stays in the row and never touches storage', async () => { + const store = makeStore(); + const queue = makeQueue(); + const { svc, rows } = makeService({ queue, store }); + + const res = await svc.send({ ...MSG, attachments: [{ filename: 'at.bin', content: AT_LIMIT }] }); + + expect(res.status).toBe('queued'); + const [el] = JSON.parse(String(rows.get(res.id)!.attachments_json)); + expect(el.inline).toBeDefined(); + expect(el.storageKey).toBeUndefined(); + expect(store.objects.size).toBe(0); + }); + + it('ONE BYTE over goes to storage, and the row carries a reference instead of content', async () => { + const store = makeStore(); + const queue = makeQueue(); + const { svc, rows, transport } = makeService({ queue, store }); + + const res = await svc.send({ + ...MSG, + attachments: [{ filename: 'contract.pdf', content: OVER, contentType: 'application/pdf' }], + }); + + // Durable: the caller is told `queued`, the row is committed, the job + // references it. This is the cell #5172 exists to fill. + expect(res.status).toBe('queued'); + expect(transport.send).not.toHaveBeenCalled(); + const row = rows.get(res.id)!; + const [el] = JSON.parse(String(row.attachments_json)); + expect(el).toMatchObject({ + filename: 'contract.pdf', + contentType: 'application/pdf', + size: OVER.byteLength, + hash: sha(OVER), + contentForm: 'buffer', + }); + expect(el.storageKey).toContain(`${EMAIL_ATTACHMENT_KEY_PREFIX}/${res.id}/`); + expect(el.inline).toBeUndefined(); + // The row stays small no matter how big the attachment was. + expect(String(row.attachments_json).length).toBeLessThan(400); + expect(store.objects.get(el.storageKey)!.equals(OVER)).toBe(true); + expect(queue.published[0]).toMatchObject({ queue: EMAIL_SEND_QUEUE, data: { rowId: res.id } }); + }); + + it('counts the budget across ALL attachments, then stores them all out of row', async () => { + const store = makeStore(); + const queue = makeQueue(); + const { svc, rows } = makeService({ queue, store }); + const half = Buffer.alloc(SYS_EMAIL_ATTACHMENT_LIMIT_BYTES / 2 + 1, 0x43); + + const res = await svc.send({ + ...MSG, + attachments: [{ filename: 'a.bin', content: half }, { filename: 'b.bin', content: half }], + }); + + const els = JSON.parse(String(rows.get(res.id)!.attachments_json)); + expect(els.map((e: any) => e.storageKey)).toEqual([...store.objects.keys()]); + expect(store.objects.size).toBe(2); + }); +}); + +// ── the loud fallbacks ───────────────────────────────────────────────────── + +describe('when the content cannot be stored, the message still goes out WHOLE — and it is said out loud', () => { + it('no file-storage capability: delivers inline, names what to mount, stores nothing', async () => { + const queue = makeQueue(); + const { svc, rows, transport, log } = makeService({ queue }); // no store wired + + const res = await svc.send({ ...MSG, attachments: [{ filename: 'big.bin', content: OVER }] }); + + // Whole, through the transport, exactly as before #5172. + expect(res.status).toBe('sent'); + expect(transport.send).toHaveBeenCalledTimes(1); + const sent = transport.send.mock.calls[0][0]; + expect((sent.attachments[0].content as Buffer).equals(OVER)).toBe(true); + // Nothing queued, nothing written to the column. + expect(queue.published).toHaveLength(0); + expect(rows.get(res.id)!.attachments_json).toBeUndefined(); + // Loud: the one line the operator sees names the obstacle and the fix. + const info = lines(log.info); + expect(info).toContain('queue delivery skipped for one message'); + expect(info).toContain('no file-storage capability is mounted'); + expect(info).toContain('@objectstack/service-storage'); + expect(info).toContain(String(SYS_EMAIL_ATTACHMENT_LIMIT_BYTES)); + }); + + it('upload failed: delivers inline, and reports the DURABILITY loss at `error`, once', async () => { + const queue = makeQueue(); + const store = makeStore({ failUpload: true }); + const { svc, rows, transport, log } = makeService({ queue, store }); + + const first = await svc.send({ ...MSG, attachments: [{ filename: 'big.bin', content: OVER }] }); + await svc.send({ ...MSG, attachments: [{ filename: 'big2.bin', content: OVER }] }); + + expect(first.status).toBe('sent'); + expect(transport.send).toHaveBeenCalledTimes(2); + expect(queue.published).toHaveLength(0); + expect(rows.get(first.id)!.attachments_json).toBeUndefined(); + // `error`, not `warn`: everything looks normal and the durability the + // operator paid for is not in force. Said ONCE, at the first degradation. + expect(log.error).toHaveBeenCalledTimes(1); + const err = lines(log.error); + expect(err).toContain('could not be stored out of row'); + expect(err).toContain('bucket is on fire'); + expect(err).toContain('still SENT, inline and whole'); + expect(err).toContain('@objectstack/service-storage'); + // …and the per-message info line names the same cause. + expect(lines(log.info)).toContain('bucket is on fire'); + }); + + it('re-arms the one-shot report when the operator re-wires storage', async () => { + const queue = makeQueue(); + const { svc, log } = makeService({ queue, store: makeStore({ failUpload: true }) }); + await svc.send({ ...MSG, attachments: [{ filename: 'a.bin', content: OVER }] }); + expect(log.error).toHaveBeenCalledTimes(1); + + svc.setAttachmentStorage({ resolve: () => makeStore({ failUpload: true }) }); + await svc.send({ ...MSG, attachments: [{ filename: 'b.bin', content: OVER }] }); + + expect(log.error).toHaveBeenCalledTimes(2); + }); + + it('queue delivery off: never uploads at all — storage exists to make QUEUED delivery possible', async () => { + const store = makeStore(); + const { svc, transport } = makeService({ store }); // no queueDelivery + + const res = await svc.send({ ...MSG, attachments: [{ filename: 'big.bin', content: OVER }] }); + + expect(res.status).toBe('sent'); + expect(transport.send).toHaveBeenCalledTimes(1); + expect(store.objects.size).toBe(0); + }); + + it('sendInline() never uploads either — the "send test email" button must not write to a bucket', async () => { + const store = makeStore(); + const queue = makeQueue(); + const { svc } = makeService({ queue, store }); + + await svc.sendInline({ ...MSG, attachments: [{ filename: 'big.bin', content: OVER }] }); + + expect(store.objects.size).toBe(0); + expect(queue.published).toHaveLength(0); + }); + + it('deletes uploaded content when the row it belongs to cannot be persisted', async () => { + const store = makeStore(); + const queue = makeQueue(); + const { svc, log } = makeService({ queue, store, failInsert: true }); + + await svc.send({ ...MSG, attachments: [{ filename: 'big.bin', content: OVER }] }); + + // The one orphan class this design must not create: bytes in the bucket + // that no row references and no reclaim job knows about. + expect(store.objects.size).toBe(0); + expect(lines(log.warn)).toContain('sys_email persist failed'); + }); + + it('reports at `error` when orphaned content cannot even be deleted', async () => { + const store = makeStore({ failDelete: true }); + const queue = makeQueue(); + const { svc, log } = makeService({ queue, store, failInsert: true }); + + await svc.send({ ...MSG, attachments: [{ filename: 'big.bin', content: OVER }] }); + + const err = lines(log.error); + expect(err).toContain('could not be deleted again'); + expect(err).toContain('nothing will ever reclaim them'); + expect(store.objects.size).toBe(1); + }); +}); + +// ── the worker's half: rebuild from the row ──────────────────────────────── + +describe('rebuilding a stored-content row for delivery', () => { + /** Queue a big-attachment message and hand back the committed row. */ + async function queuedRow(o: SvcOpts & { store: EmailAttachmentStore }) { + const h = makeService(o); + const res = await h.svc.send({ + ...MSG, + attachments: [{ filename: 'contract.pdf', content: OVER, contentType: 'application/pdf' }], + }); + return { ...h, rowId: res.id, row: () => ({ ...h.rows.get(res.id)! }) }; + } + + it('fetches the content back and hands the transport the WHOLE message', async () => { + const store = makeStore(); + const queue = makeQueue(); + const h = await queuedRow({ queue, store }); + + const out = await h.svc.deliverPersistedRow(h.row(), { maxAttempts: 1, priorAttempts: 0 }); + + expect(out.status).toBe('sent'); + const sent = h.transport.send.mock.calls[0][0]; + expect(sent.attachments).toHaveLength(1); + expect(sent.attachments[0]).toMatchObject({ filename: 'contract.pdf', contentType: 'application/pdf' }); + expect((sent.attachments[0].content as Buffer).equals(OVER)).toBe(true); + }); + + it('FAILS the row and sends nothing when the content cannot be fetched (outage)', async () => { + const store = makeStore(); + const queue = makeQueue(); + const h = await queuedRow({ queue, store }); + store.objects.clear(); // the object is unreachable + + const out = await h.svc.deliverPersistedRow(h.row(), { maxAttempts: 1, priorAttempts: 2 }); + + // `declared ≠ delivered` is the outcome that is not allowed: no stripped + // message goes out, and the row records why. + expect(out.status).toBe('failed'); + expect(out.error).toMatch(/NoSuchKey/); + expect(h.transport.send).not.toHaveBeenCalled(); + expect(h.rows.get(h.rowId)).toMatchObject({ status: 'failed', attempt_count: 2 }); + }); + + it('FAILS the row when no storage capability is mounted on the delivering process', async () => { + const store = makeStore(); + const queue = makeQueue(); + const h = await queuedRow({ queue, store }); + const row = h.row(); + h.svc.setAttachmentStorage(undefined); // e.g. a worker process without it + + const out = await h.svc.deliverPersistedRow(row, { maxAttempts: 1 }); + + expect(out.status).toBe('failed'); + expect(out.error).toContain('no file-storage capability is mounted on this process'); + expect(h.transport.send).not.toHaveBeenCalled(); + }); + + it('FAILS the row rather than trusting truncated storage content', async () => { + const store = makeStore(); + const queue = makeQueue(); + const h = await queuedRow({ queue, store }); + const key = [...store.objects.keys()][0]; + store.objects.set(key, OVER.subarray(0, 100)); + + const out = await h.svc.deliverPersistedRow(h.row(), { maxAttempts: 1 }); + + expect(out.status).toBe('failed'); + expect(out.error).toMatch(/is 100 byte\(s\) but the row records size/); + expect(h.transport.send).not.toHaveBeenCalled(); + }); +}); + +// ── scheduling reclamation ───────────────────────────────────────────────── + +describe('scheduling the content reclaim', () => { + async function deliverBig(o: { transportSend?: (m: any) => Promise } = {}) { + const store = makeStore(); + const queue = makeQueue(); + const h = makeService({ queue, store, ...o }); + const res = await h.svc.send({ ...MSG, attachments: [{ filename: 'big.bin', content: OVER }] }); + const row = { ...h.rows.get(res.id)! }; + const out = await h.svc.deliverPersistedRow(row, { maxAttempts: 1 }); + const reclaim = queue.published.filter((p) => p.queue === EMAIL_ATTACHMENT_RECLAIM_QUEUE); + return { ...h, store, queue, rowId: res.id, out, reclaim }; + } + + it('publishes ONE delayed job per row, carrying the keys — not just the row id', async () => { + const h = await deliverBig(); + + expect(h.out.status).toBe('sent'); + expect(h.reclaim).toHaveLength(1); + expect(h.reclaim[0].data).toEqual({ rowId: h.rowId, keys: [...h.store.objects.keys()] }); + // Carrying the keys is what makes a later row deletion reclaim the content + // instead of orphaning it. + expect(h.reclaim[0].data.keys[0]).toContain(EMAIL_ATTACHMENT_KEY_PREFIX); + }); + + it('delays by the grace window and dedups per row', async () => { + const h = await deliverBig(); + expect(h.reclaim[0].options).toMatchObject({ + delay: EMAIL_ATTACHMENT_RECLAIM_GRACE_MS, + idempotencyKey: `sys_email_attachments:${h.rowId}`, + metadata: { object: 'sys_email', rowId: h.rowId }, + }); + }); + + it('schedules on a FAILED terminal transition too — the queue may still retry, the job re-checks', async () => { + const h = await deliverBig({ transportSend: async () => { throw new Error('550 mailbox full'); } }); + expect(h.out.status).toBe('failed'); + expect(h.reclaim).toHaveLength(1); + expect(h.reclaim[0].data.keys).toHaveLength(1); + }); + + it('schedules nothing for a message whose content is IN the row', async () => { + const store = makeStore(); + const queue = makeQueue(); + const { svc, rows } = makeService({ queue, store }); + const res = await svc.send({ ...MSG, attachments: [{ filename: 'small.txt', content: 'hi' }] }); + + await svc.deliverPersistedRow({ ...rows.get(res.id)! }, { maxAttempts: 1 }); + + expect(queue.published.filter((p) => p.queue === EMAIL_ATTACHMENT_RECLAIM_QUEUE)).toHaveLength(0); + }); + + it('reports at `error` when the reclaim job cannot be published — those bytes become permanent', async () => { + const store = makeStore(); + const queue = makeQueue(); + const h = makeService({ queue, store }); + const res = await h.svc.send({ ...MSG, attachments: [{ filename: 'big.bin', content: OVER }] }); + const row = { ...h.rows.get(res.id)! }; + queue.publish = async () => { throw new Error('queue table is gone'); }; + + await h.svc.deliverPersistedRow(row, { maxAttempts: 1 }); + + const err = lines(h.log.error); + expect(err).toContain('could not publish the attachment-reclaim job'); + expect(err).toContain('stay in the backend'); + }); +}); diff --git a/packages/plugins/plugin-email/src/email-service.ts b/packages/plugins/plugin-email/src/email-service.ts index 55ffcdc3fd..5098821676 100644 --- a/packages/plugins/plugin-email/src/email-service.ts +++ b/packages/plugins/plugin-email/src/email-service.ts @@ -19,9 +19,20 @@ import { encodeAttachmentsForRow, encodeHeadersForRow, decodeAttachmentsFromRow, + decodeAttachmentsFromRowAsync, decodeHeadersFromRow, + storageKeysInColumn, type EncodedAttachments, } from './sys-email-payload.js'; +import { + EMAIL_ATTACHMENT_RECLAIM_QUEUE, + EMAIL_ATTACHMENT_RECLAIM_GRACE_MS, + deleteAttachmentKeys, + fetchAttachmentContent, + offloadAttachmentsToStorage, + type EmailAttachmentReclaimPayload, + type EmailAttachmentStorage, +} from './attachment-storage.js'; /** * Queue topic durable email delivery is published to and consumed from @@ -88,6 +99,19 @@ export interface DeliverAttemptOptions { priorAttempts?: number; } +/** + * Reconstructing a `sys_email` row into a sendable message needs the + * `file-storage` capability once a row's attachments live out of it (#5172), + * and that is asynchronous — hence this async twin of {@link rowToNormalized}. + * + * `fetchContent` absent ⇒ a row that needs storage fails, loudly, exactly as + * it does through the synchronous entry point. It is never optional in the + * sense of "skip the attachment". + */ +export interface RowToNormalizedOptions { + fetchContent?: (storageKey: string) => Promise; +} + /** * Internal persistence shim — typed loosely so the service can run * without an ObjectQL engine wired (e.g. unit tests, serverless). @@ -189,6 +213,44 @@ function splitAddresses(v: unknown): string[] { * Throws when the row lacks the minimum fields needed to send. */ export function rowToNormalized(row: Record): NormalizedEmailMessage { + const msg = rowEnvelope(row); + // Headers and attachments (#5177). Both decoders return `undefined` for a + // row that simply has no such column — every row written before #5177 — and + // THROW for a column that is present but does not describe what it claims + // to. Throwing here is the point: the caller (`deliverPersistedRow`) turns + // it into a `failed` row carrying the reason, which is strictly better than + // handing the transport a message with an attachment quietly missing. + // + // A row whose attachments live in storage (#5172) also throws here: this + // entry point cannot fetch, and "cannot fetch" must never quietly become + // "send it without". Use {@link rowToNormalizedAsync} where a fetch is + // possible. + const attachments = decodeAttachmentsFromRow(row.attachments_json); + if (attachments) msg.attachments = attachments; + return msg; +} + +/** + * {@link rowToNormalized}, able to fetch out-of-row attachment content + * (#5172). + * + * The two share ONE validator (`readAttachmentColumn`) and one verifier + * (`materializeAttachment`), so a storage-backed attachment is checked against + * `size` and `hash` exactly as an in-row one is — a backend that returns a + * truncated object is as unacceptable as a truncated column. + */ +export async function rowToNormalizedAsync( + row: Record, + opts?: RowToNormalizedOptions, +): Promise { + const msg = rowEnvelope(row); + const attachments = await decodeAttachmentsFromRowAsync(row.attachments_json, opts?.fetchContent); + if (attachments) msg.attachments = attachments; + return msg; +} + +/** Everything a row says about a message except its attachments. */ +function rowEnvelope(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(); @@ -212,16 +274,8 @@ export function rowToNormalized(row: Record): NormalizedEmailMessag const bcc = splitAddresses(row.bcc_addresses); if (bcc.length > 0) msg.bcc = bcc; if (row.reply_to) msg.replyTo = String(row.reply_to); - // Headers and attachments (#5177). Both decoders return `undefined` for a - // row that simply has no such column — every row written before #5177 — and - // THROW for a column that is present but does not describe what it claims - // to. Throwing here is the point: the caller (`deliverPersistedRow`) turns - // it into a `failed` row carrying the reason, which is strictly better than - // handing the transport a message with an attachment quietly missing. const headers = decodeHeadersFromRow(row.headers_json); if (headers) msg.headers = headers; - const attachments = decodeAttachmentsFromRow(row.attachments_json); - if (attachments) msg.attachments = attachments; return msg; } @@ -309,6 +363,19 @@ export interface EmailServiceOptions { * inline delivery (the default, unchanged). */ queueDelivery?: EmailQueueDelivery; + /** + * Out-of-row attachment content through the `file-storage` capability + * (#5172). Set ⇒ a message whose attachments exceed + * {@link SYS_EMAIL_ATTACHMENT_LIMIT_BYTES} can still be queued, with its + * content in storage and a reference on the row. Unset (or unresolvable) ⇒ + * such a message keeps falling back to inline delivery, whole, and the + * fallback says why. + * + * Only ever consulted in queue mode: out-of-row content exists to make + * DURABLE delivery possible, and inline delivery already has the message in + * memory. + */ + attachmentStorage?: EmailAttachmentStorage; } /** @@ -351,6 +418,15 @@ export class EmailService implements IEmailService { */ private queueDegradationReported = false; + /** + * Set once an attachment OFFLOAD failure has been reported, so a flapping + * storage backend states the degradation at its first message instead of on + * every one. Re-armed by {@link setQueueDelivery} / + * {@link setAttachmentStorage}, i.e. whenever the operator changes the + * configuration this verdict was about. + */ + private attachmentStorageDegradationReported = false; + constructor(public options: EmailServiceOptions) { if (!options.transport) throw new Error('EmailService: transport is required'); } @@ -394,6 +470,17 @@ export class EmailService implements IEmailService { setQueueDelivery(queueDelivery: EmailQueueDelivery | undefined): void { this.options.queueDelivery = queueDelivery; this.queueDegradationReported = false; + this.attachmentStorageDegradationReported = false; + } + + /** + * Wire (or unwire) out-of-row attachment storage on a running service + * (#5172). Re-arms the one-shot degradation report, so mounting storage + * after a complaint is allowed to complain again if it still cannot work. + */ + setAttachmentStorage(attachmentStorage: EmailAttachmentStorage | undefined): void { + this.options.attachmentStorage = attachmentStorage; + this.attachmentStorageDegradationReported = false; } /** @@ -440,14 +527,30 @@ export class EmailService implements IEmailService { // reports that as its own degradation, below), so base64-ing a large // attachment for nobody is pure cost on the path that opted out of // persistence. - const encodedAttachments: EncodedAttachments = this.options.persistence + let encodedAttachments: EncodedAttachments = this.options.persistence ? encodeAttachmentsForRow(normalized.attachments) : { kind: 'none' }; + // The row id is minted BEFORE the offload because the storage key embeds + // it (`sys_email/attachments//…`) and the content must be in place + // before the row that references it is inserted. The other order — insert, + // then upload, then patch the row — would make a `queued` row observable + // while it under-describes its own message, which the boot sweep would + // then deliver stripped. + const id = newId(); + + // ── OUT-OF-ROW ATTACHMENT CONTENT (#5172) ────────────────────────────── + // Over the in-row budget, in queue mode, with the file-storage capability + // mounted: the content goes to storage and the row carries a reference, so + // this message gets the same durability as every other. Anything missing + // here leaves `over-limit` standing — inline delivery, whole, with the + // reason attached to the one line the operator sees. + if (allowQueue && encodedAttachments.kind === 'over-limit') { + encodedAttachments = await this.offloadAttachments(id, normalized, encodedAttachments); + } + // `undefined` ⇒ every statement below is the pre-#5160 inline path. const queue = allowQueue ? this.resolveQueueForSend(encodedAttachments) : undefined; - - const id = newId(); const headersJson = encodeHeadersForRow(normalized.headers); const baseRow: Record = { id, @@ -467,7 +570,9 @@ export class EmailService implements IEmailService { // prevent. Over-limit attachments store NOTHING (see the verdict // above), so the row stays bounded. ...(headersJson !== undefined ? { headers_json: headersJson } : {}), - ...(encodedAttachments.kind === 'inline' ? { attachments_json: encodedAttachments.json } : {}), + ...(encodedAttachments.kind === 'inline' || encodedAttachments.kind === 'storage' + ? { attachments_json: encodedAttachments.json } + : {}), ...(input.relatedObject ? { related_object: input.relatedObject } : {}), ...(input.relatedId ? { related_id: input.relatedId } : {}), ...(input.sentBy ? { sent_by: input.sentBy } : {}), @@ -491,6 +596,14 @@ export class EmailService implements IEmailService { } } const rowId = persistedId ?? id; + const storageKeys = encodedAttachments.kind === 'storage' ? encodedAttachments.keys : []; + if (persistedId === undefined && storageKeys.length > 0) { + // Content was uploaded for a row that does not exist. No row will ever + // reference these bytes, so nothing will ever reclaim them either — + // delete them here rather than create the one orphan class this design + // is built to make impossible. + await this.discardOrphanedAttachmentContent(storageKeys, 'the sys_email row could not be persisted'); + } 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 @@ -506,7 +619,11 @@ export class EmailService implements IEmailService { return { id: rowId, status: 'queued' }; } } - return await this.deliverNormalized(rowId, normalized); + // Inline delivery of a message whose content IS in storage (the publish + // failed, or queue mode is off for this send): the in-memory message is + // still whole, so the mail goes out — and the content still has to be + // reclaimed afterwards, which is why the keys travel with the delivery. + return await this.deliverNormalized(rowId, normalized, undefined, storageKeys); } finally { this.managedRowIds.delete(id); } @@ -548,7 +665,10 @@ export class EmailService implements IEmailService { `EmailService: queue delivery skipped for one message — its attachments total ` + `${encodedAttachments.totalBytes} bytes, over the ${SYS_EMAIL_ATTACHMENT_LIMIT_BYTES}-byte limit a ` + 'sys_email row carries, so the message was delivered inline (in-process retries only) rather than ' - + 'queued without them. Out-of-row storage for large attachments is objectstack#5172.', + + 'queued without them. Content that large is queueable through the file-storage capability (#5172), ' + + `but ${encodedAttachments.storageDetail ?? 'that path was not attempted for this message'}. ` + + 'Fix: mount the file-storage capability (@objectstack/service-storage) so large attachments are ' + + 'stored out of the row and the message can be delivered durably.', ); return undefined; } @@ -575,6 +695,88 @@ export class EmailService implements IEmailService { return queue; } + /** + * Try to move an over-budget message's attachment content out of the row and + * into the `file-storage` capability (#5172). + * + * Returns a `storage` verdict on success, and the ORIGINAL `over-limit` + * verdict — annotated with why — on every failure. That asymmetry is the + * maintainer's ruling made mechanical: a message the platform cannot store + * out of row is delivered **inline, whole, loudly**, never queued against a + * row that does not carry it. + */ + private async offloadAttachments( + rowId: string, + normalized: NormalizedEmailMessage, + overLimit: Extract, + ): Promise { + // Storage-backed content only pays for itself when there is a row to + // reference it AND a durable queue to deliver that row. Without either, + // uploading would buy nothing and still cost an upload: the message is + // about to be delivered inline from memory, and the two degradations that + // led here are already reported by `resolveQueueForSend`. + if (!this.options.persistence || !this.options.queueDelivery) return overLimit; + if (!this.options.queueDelivery.resolve()) return overLimit; + + const storage = this.options.attachmentStorage?.resolve(); + if (!storage) { + return { + ...overLimit, + storageDetail: 'no file-storage capability is mounted, so there is nowhere to put the content', + }; + } + + const offloaded = await offloadAttachmentsToStorage(normalized.attachments, rowId, storage); + if (offloaded.kind === 'storage') { + return { + kind: 'storage', + json: offloaded.json, + keys: offloaded.keys, + totalBytes: offloaded.totalBytes, + }; + } + + // Storage IS mounted and it did not work. Unlike "not configured", this is + // a fault: the durability the operator paid for is not in force while + // everything else looks normal, so it is stated at `error`, once. + this.reportAttachmentStorageDegradation(offloaded.detail); + return { ...overLimit, storageDetail: offloaded.detail }; + } + + /** + * Delete attachment content that no row will ever reference. + * + * Reported at `error` when it cannot be deleted: bytes that were meant to be + * temporary have silently become permanent, in a bucket somebody pays for, + * with nothing left pointing at them. + */ + private async discardOrphanedAttachmentContent(keys: string[], because: string): Promise { + const storage = this.options.attachmentStorage?.resolve(); + if (!storage) return; + const { failed } = await deleteAttachmentKeys(storage, keys); + if (failed.length === 0) return; + this.options.logger?.error?.( + `EmailService: ${failed.length} attachment storage object(s) were uploaded but ${because}, and they could ` + + 'not be deleted again: ' + + failed.map((f) => `'${f.key}' (${f.error})`).join('; ') + + '. Nothing references those bytes and nothing will ever reclaim them. Fix: delete them by hand under ' + + 'the sys_email/attachments/ prefix, and check why the file-storage backend is refusing deletes.', + ); + } + + /** State an attachment-offload degradation once, at `error`. See {@link reportQueueDegradation}. */ + private reportAttachmentStorageDegradation(detail: string): void { + if (this.attachmentStorageDegradationReported) return; + this.attachmentStorageDegradationReported = true; + this.options.logger?.error?.( + `EmailService: a message's attachments could not be stored out of row — ${detail}. The message was ` + + 'still SENT, inline and whole, but it did not get durable queue delivery: a failure would be retried ' + + 'only in this process and lost if it dies. Fix: check the file-storage capability ' + + '(@objectstack/service-storage — credentials, bucket, disk), or accept inline delivery for messages ' + + `with attachments over ${SYS_EMAIL_ATTACHMENT_LIMIT_BYTES} bytes.`, + ); + } + /** * 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 @@ -659,6 +861,7 @@ export class EmailService implements IEmailService { rowId: string, normalized: NormalizedEmailMessage, opts?: DeliverAttemptOptions, + reclaimKeys?: string[], ): Promise { // Defaults reproduce the pre-#5160 loop exactly. const maxAttempts = Math.max(1, opts?.maxAttempts ?? (this.options.retries ?? 0) + 1); @@ -675,6 +878,7 @@ export class EmailService implements IEmailService { sent_at: new Date().toISOString(), attempt_count: priorAttempts + attempt, }); + await this.scheduleAttachmentReclaim(rowId, reclaimKeys); return { id: rowId, status, messageId }; } catch (err: any) { lastError = err; @@ -690,9 +894,65 @@ export class EmailService implements IEmailService { error: errMessage, attempt_count: priorAttempts + maxAttempts, }); + await this.scheduleAttachmentReclaim(rowId, reclaimKeys); return { id: rowId, status: 'failed', error: errMessage }; } + /** + * Schedule reclamation of this row's out-of-row attachment content, one + * grace window from now (#5172). + * + * Published on EVERY terminal transition, `sent` and `failed` alike, with a + * per-row idempotency key so the retries of one message collapse onto the + * first job rather than each arming their own. `failed` is included because + * `failed` is not the end of anything by itself — the queue re-delivers such + * a row — and the job re-reads the row before it deletes, so an early + * schedule can only ever be re-armed, never act early. + * + * The keys ride in the PAYLOAD. That is what makes a later row deletion + * (retention, purge) reclaim the content instead of orphaning it — see + * `attachment-reclaim.ts`. + */ + private async scheduleAttachmentReclaim(rowId: string, keys: string[] | undefined): Promise { + if (!keys || keys.length === 0) return; + const queue = this.options.queueDelivery?.resolve(); + if (!queue) { + this.options.logger?.error?.( + `EmailService: ${keys.length} attachment storage object(s) for sys_email row '${rowId}' cannot be ` + + 'scheduled for reclamation — no durable queue service is available now, although one was when the ' + + 'content was stored. Those bytes will stay in the file-storage backend until they are deleted by ' + + 'hand. Fix: keep @objectstack/service-queue (over an ObjectQL engine) mounted for the life of the ' + + 'process, then delete leftovers under the sys_email/attachments/ prefix.', + ); + return; + } + try { + await queue.publish( + EMAIL_ATTACHMENT_RECLAIM_QUEUE, + { rowId, keys }, + { + delay: EMAIL_ATTACHMENT_RECLAIM_GRACE_MS, + // One row, one reclaim job: every retry of a message finalizes the + // row again, and each of those would otherwise arm its own copy. + idempotencyKey: `sys_email_attachments:${rowId}`, + // A reclaim that cannot run is a byte leak, not a lost message, so + // it gets a real budget and then the DLQ, where it is visible. + maxAttempts: 5, + retries: 4, + backoff: { type: 'exponential', delayMs: 60_000, maxDelayMs: 60 * 60_000 }, + metadata: { object: 'sys_email', rowId }, + }, + ); + } catch (err: any) { + this.options.logger?.error?.( + `EmailService: could not publish the attachment-reclaim job for sys_email row '${rowId}' ` + + `(${String(err?.message ?? err)}). Its ${keys.length} storage object(s) will stay in the backend ` + + 'until deleted by hand. Fix: check the durable queue service, then delete leftovers under the ' + + 'sys_email/attachments/ prefix.', + ); + } + } + /** * Deliver an ALREADY-PERSISTED `sys_email` row (the outbox-drain path). * @@ -712,9 +972,17 @@ export class EmailService implements IEmailService { ): Promise { const rowId = String(row?.id ?? ''); if (!rowId) throw new Error('deliverPersistedRow: row.id is required'); + // Keys read from the ROW, so a row delivered by the queue worker (which + // never saw the send) still schedules its own content's reclamation. + const reclaimKeys = storageKeysInColumn(row.attachments_json); let normalized: NormalizedEmailMessage; try { - normalized = rowToNormalized(row); + // Async because a row's attachments may live in the file-storage + // capability (#5172). A fetch that fails — outage, deleted object, no + // capability mounted at all — throws and lands the row at `failed` with + // the reason, which is the whole point: an unfetchable attachment must + // never become a message delivered without it. + normalized = await rowToNormalizedAsync(row, { fetchContent: this.attachmentFetcher() }); } catch (err: any) { const errMessage = String(err?.message ?? err ?? 'invalid row').slice(0, 1000); await this.updateRow(rowId, { @@ -724,7 +992,20 @@ export class EmailService implements IEmailService { }); return { id: rowId, status: 'failed', error: errMessage }; } - return this.deliverNormalized(rowId, normalized, opts); + return this.deliverNormalized(rowId, normalized, opts, reclaimKeys); + } + + /** + * A fetcher for out-of-row attachment content, or `undefined` when no + * storage capability is mounted. + * + * `undefined` is NOT a permission to skip the attachment: the decoder turns + * it into a refusal naming the key it could not read. + */ + private attachmentFetcher(): ((storageKey: string) => Promise) | undefined { + const storage = this.options.attachmentStorage?.resolve(); + if (!storage) return undefined; + return (storageKey: string) => fetchAttachmentContent(storage, storageKey); } 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 028771074b..28df12b7bc 100644 --- a/packages/plugins/plugin-email/src/index.ts +++ b/packages/plugins/plugin-email/src/index.ts @@ -14,7 +14,7 @@ * endpoints. `EMAIL_TRANSPORT_PROVIDERS` is the machine-readable form. */ -export { EmailServicePlugin, resolveDurableQueue } from './email-plugin.js'; +export { EmailServicePlugin, resolveDurableQueue, resolveAttachmentStore } from './email-plugin.js'; export type { EmailServicePluginOptions } from './email-plugin.js'; export { LogTransport, normalizeMessage, formatAddress, EMAIL_SEND_QUEUE } from './email-service.js'; export type { @@ -25,16 +25,47 @@ export type { EmailQueueDelivery, EmailSendQueuePayload, DeliverAttemptOptions, + RowToNormalizedOptions, } from './email-service.js'; export { SYS_EMAIL_ATTACHMENT_LIMIT_BYTES, encodeAttachmentsForRow, decodeAttachmentsFromRow, + decodeAttachmentsFromRowAsync, + readAttachmentColumn, + materializeAttachment, + collectAttachmentParts, + storageKeysInColumn, + withContentReclaimed, encodeHeadersForRow, decodeHeadersFromRow, type PersistedEmailAttachment, type EncodedAttachments, + type AttachmentSource, + type AttachmentPart, + type CollectedAttachments, } from './sys-email-payload.js'; +export { + EMAIL_ATTACHMENT_KEY_PREFIX, + EMAIL_ATTACHMENT_RECLAIM_QUEUE, + EMAIL_ATTACHMENT_RECLAIM_GRACE_MS, + attachmentStorageKey, + offloadAttachmentsToStorage, + deleteAttachmentKeys, + fetchAttachmentContent, + type EmailAttachmentStore, + type EmailAttachmentStorage, + type EmailAttachmentReclaimPayload, + type AttachmentOffload, +} from './attachment-storage.js'; +export { + reclaimAttachmentContent, + RECLAIM_OBJECT, + TERMINAL_EMAIL_STATUSES, + type AttachmentReclaimEngine, + type AttachmentReclaimOutcome, + type ReclaimAttachmentContentOptions, +} from './attachment-reclaim.js'; export { renderTemplate, requireVars, htmlToText } from './template-engine.js'; export { ResendTransport, diff --git a/packages/plugins/plugin-email/src/sys-email-payload.test.ts b/packages/plugins/plugin-email/src/sys-email-payload.test.ts index d9e11b472b..61d072dfe9 100644 --- a/packages/plugins/plugin-email/src/sys-email-payload.test.ts +++ b/packages/plugins/plugin-email/src/sys-email-payload.test.ts @@ -211,16 +211,20 @@ describe('a column that lies is rejected, never partially delivered', () => { ]))).toThrow(/carries no content/); }); - it('rejects a storageKey-only attachment and names the issue that will implement it', () => { + // #5172 gave `storageKey` a producer AND a reader, but the reader is the + // ASYNC decoder — this synchronous one has no capability to fetch with. It + // still refuses rather than dropping the attachment, and now names the + // missing capability instead of the issue that was going to add it. + it('rejects a storageKey-only attachment, naming the capability it would need', () => { expect(decodeAtt(JSON.stringify([ { filename: 'a.txt', size: 2, hash: sha('hi'), contentForm: 'buffer', storageKey: 'blob/abc' }, - ]))).toThrow(/objectstack#5172/); + ]))).toThrow(/no file-storage capability to fetch it from/); }); it('rejects truncated content (size disagrees)', () => { const [item] = JSON.parse(column([{ filename: 'a.txt', content: 'hello' }])); item.inline = Buffer.from('he').toString('base64'); - expect(decodeAtt(JSON.stringify([item]))).toThrow(/decodes to 2 byte\(s\) but the row records size 5/); + expect(decodeAtt(JSON.stringify([item]))).toThrow(/is 2 byte\(s\) but the row records size 5/); }); it('rejects rewritten content (hash disagrees)', () => { diff --git a/packages/plugins/plugin-email/src/sys-email-payload.ts b/packages/plugins/plugin-email/src/sys-email-payload.ts index 6bef1d5dd2..1e90f2d572 100644 --- a/packages/plugins/plugin-email/src/sys-email-payload.ts +++ b/packages/plugins/plugin-email/src/sys-email-payload.ts @@ -36,10 +36,14 @@ * change with a reason attached, which is what a change to a storage budget * should cost. * - * Out-of-row storage for large attachments (`storageKey`) is phase 2, tracked - * by objectstack#5172. The element shape already declares that key so phase 2 - * adds a *producer* rather than migrating data — see - * {@link PersistedEmailAttachment.storageKey}. + * Out-of-row storage for large attachments (`storageKey`) is phase 2 + * (objectstack#5172), and it landed exactly as phase 1 predicted: a + * *producer* for the key that was already declared, with no migration. Over + * the budget, the content goes to the `file-storage` capability and the row + * carries a reference plus the permanent audit metadata — see + * `attachment-storage.ts`. Over the budget with no storage capability (or an + * upload that fails), the pre-#5172 answer stands: inline delivery, whole, + * loudly. * * ## Why decoding is strict * @@ -121,16 +125,31 @@ export interface PersistedEmailAttachment { /** Base64 of the raw content, when the row carries it (phase 1's only producer). */ inline?: string; /** - * Reference to content held outside the row. + * Reference to content held outside the row, in the `file-storage` + * capability (objectstack#5172). * - * **Phase 1 has no producer for this key** — nothing in this repo writes it - * today, and {@link decodeAttachmentsFromRow} rejects a row that only has - * it. It is declared now so that objectstack#5172 (large attachments via - * storage, deferred by the maintainer) ships a producer + reader against an - * already-persisted shape instead of migrating rows. Declared-not-yet-live - * on purpose; not a liveness finding. + * Written instead of {@link inline} when the message is over + * {@link SYS_EMAIL_ATTACHMENT_LIMIT_BYTES} and queue delivery is in force — + * see `attachment-storage.ts` for the key scheme. Mutually exclusive with + * `inline` in practice, and a row carrying neither is rejected on read + * rather than delivered without the attachment. + * + * **Removed when the content is reclaimed** (the row reached a terminal + * state and the grace window passed), at which point + * {@link contentReclaimedAt} takes its place. Everything else on this + * element survives that: the metadata is the audit artifact, the bytes were + * only the delivery artifact. */ storageKey?: string; + /** + * ISO timestamp at which out-of-row content was deleted. + * + * Recorded so that "this row has no content" is a **statement** rather than + * an absence to be guessed at: a reader can tell an attachment that was + * reclaimed on schedule from one whose column was truncated, and says so in + * the rejection. Never set on an element that still has content. + */ + contentReclaimedAt?: string; } /** Outcome of {@link encodeAttachmentsForRow}. */ @@ -139,8 +158,22 @@ export type EncodedAttachments = | { kind: 'none' } /** Within budget: `json` goes into `attachments_json`. */ | { kind: 'inline'; json: string; totalBytes: number } - /** Over {@link SYS_EMAIL_ATTACHMENT_LIMIT_BYTES} — nothing is written to the row. */ - | { kind: 'over-limit'; totalBytes: number } + /** + * Over {@link SYS_EMAIL_ATTACHMENT_LIMIT_BYTES}. + * + * Nothing is written to the row **by this encoder**. The caller may still + * offload the content to the `file-storage` capability (#5172); when it + * cannot, `storageDetail` carries the sentence explaining why, so the one + * log line the operator gets names the actual obstacle instead of only the + * byte count. + */ + | { kind: 'over-limit'; totalBytes: number; storageDetail?: string } + /** + * Content held out of the row, in the `file-storage` capability (#5172). + * `json` goes into `attachments_json`; `keys` is what a later reclaim + * deletes. + */ + | { kind: 'storage'; json: string; keys: string[]; totalBytes: number } /** Content this codec cannot represent; nothing is written to the row. */ | { kind: 'unsupported'; detail: string }; @@ -149,28 +182,46 @@ function digestOf(bytes: Buffer): string { return `sha256:${createHash('sha256').update(bytes).digest('hex')}`; } +/** One attachment reduced to the facts every encoder needs. */ +export interface AttachmentPart { + att: EmailAttachment; + bytes: Buffer; + /** `sha256:` of {@link bytes}. */ + hash: string; + contentForm: 'string' | 'buffer'; +} + +/** Outcome of {@link collectAttachmentParts}. */ +export type CollectedAttachments = + | { kind: 'none' } + | { kind: 'ok'; parts: AttachmentPart[]; totalBytes: number } + | { kind: 'unsupported'; detail: string }; + /** - * Encode a message's attachments for `sys_email.attachments_json`. + * Reduce a message's attachments to bytes + digest, once. * - * Never throws: an attachment this codec cannot carry (too large, or a - * `content` value outside the declared `string | Buffer` contract) is reported - * as a verdict, because the caller's response is to fall back to inline - * delivery — which is today's behaviour and always whole — not to fail the - * send. Failing a send that works today would be a regression dressed as - * strictness. + * Shared by the in-row encoder below and the storage offloader (#5172) so the + * two producers of `attachments_json` cannot disagree about what an attachment + * IS — which arm of `content: string | Buffer` it used, how many bytes it is, + * what it hashes to. Two hand-written extractions would be two contracts, and + * the digest is the thing the reader verifies against. */ -export function encodeAttachmentsForRow( +export function collectAttachmentParts( attachments: EmailAttachment[] | undefined, -): EncodedAttachments { +): CollectedAttachments { if (!attachments || attachments.length === 0) return { kind: 'none' }; - const parts: Array<{ att: EmailAttachment; bytes: Buffer; contentForm: 'string' | 'buffer' }> = []; + const parts: AttachmentPart[] = []; for (const att of attachments) { const content = att?.content; + let bytes: Buffer; + let contentForm: 'string' | 'buffer'; if (typeof content === 'string') { - parts.push({ att, bytes: Buffer.from(content, 'utf8'), contentForm: 'string' }); + bytes = Buffer.from(content, 'utf8'); + contentForm = 'string'; } else if (Buffer.isBuffer(content)) { - parts.push({ att, bytes: content, contentForm: 'buffer' }); + bytes = content; + contentForm = 'buffer'; } else { return { kind: 'unsupported', @@ -178,16 +229,37 @@ export function encodeAttachmentsForRow( + 'a string nor a Buffer, which is the whole of the EmailAttachment contract', }; } + parts.push({ att, bytes, hash: digestOf(bytes), contentForm }); } - const totalBytes = parts.reduce((n, p) => n + p.bytes.byteLength, 0); + return { kind: 'ok', parts, totalBytes: parts.reduce((n, p) => n + p.bytes.byteLength, 0) }; +} + +/** + * Encode a message's attachments for `sys_email.attachments_json`. + * + * Never throws: an attachment this codec cannot carry (too large, or a + * `content` value outside the declared `string | Buffer` contract) is reported + * as a verdict, because the caller's response is to fall back to inline + * delivery — which is today's behaviour and always whole — not to fail the + * send. Failing a send that works today would be a regression dressed as + * strictness. + */ +export function encodeAttachmentsForRow( + attachments: EmailAttachment[] | undefined, +): EncodedAttachments { + const collected = collectAttachmentParts(attachments); + if (collected.kind === 'none') return { kind: 'none' }; + if (collected.kind === 'unsupported') return { kind: 'unsupported', detail: collected.detail }; + + const { parts, totalBytes } = collected; if (totalBytes > SYS_EMAIL_ATTACHMENT_LIMIT_BYTES) return { kind: 'over-limit', totalBytes }; - const items: PersistedEmailAttachment[] = parts.map(({ att, bytes, contentForm }) => ({ + const items: PersistedEmailAttachment[] = parts.map(({ att, bytes, hash, contentForm }) => ({ filename: String(att.filename ?? ''), ...(att.contentType ? { contentType: String(att.contentType) } : {}), size: bytes.byteLength, - hash: digestOf(bytes), + hash, ...(att.cid ? { cid: String(att.cid) } : {}), contentForm, inline: bytes.toString('base64'), @@ -215,22 +287,35 @@ function parseJson(column: string, value: unknown): unknown { } /** - * Rebuild `attachments` from `sys_email.attachments_json`. + * One validated `attachments_json` element together with **where its content + * is** — the seam between "is this row well-formed?" (synchronous, total) and + * "can we get the bytes?" (asynchronous, needs the storage capability). * - * Returns `undefined` for a row that has no such column (every row written - * before #5177) — reading an old row must stay safe. Everything else is - * verified: a column that is present but does not describe the message it - * claims to describe throws, so the row lands at `failed` with the reason - * rather than being delivered without the attachment. + * The split exists so there is still exactly ONE validator. The alternative — + * a second async decoder that re-implements the checks — is how the two halves + * of a codec drift, and here the checks ARE the feature: size and digest are + * what turn a truncated column into an error instead of a wrong email. */ -export function decodeAttachmentsFromRow(value: unknown): EmailAttachment[] | undefined { +export type AttachmentSource = + /** Content is in the row; already base64-decoded. */ + | { kind: 'inline'; element: PersistedEmailAttachment; bytes: Buffer } + /** Content is in the `file-storage` capability under `storageKey` (#5172). */ + | { kind: 'storage'; element: PersistedEmailAttachment; storageKey: string }; + +/** + * Validate `sys_email.attachments_json` and say, per element, where its + * content lives. Throws on anything that does not describe what it claims to. + * + * `undefined` for a row with no such column (every row written before #5177). + */ +export function readAttachmentColumn(value: unknown): AttachmentSource[] | undefined { if (isAbsent(value)) return undefined; const parsed = parseJson('attachments_json', value); if (parsed == null) return undefined; if (!Array.isArray(parsed)) reject('attachments_json', 'must decode to an array of attachments'); if (parsed.length === 0) return undefined; - return parsed.map((raw: any, i: number): EmailAttachment => { + return parsed.map((raw: any, i: number): AttachmentSource => { const at = `[${i}]`; if (!raw || typeof raw !== 'object') reject('attachments_json', `${at} is not an object`); const filename = raw.filename; @@ -251,42 +336,202 @@ export function decodeAttachmentsFromRow(value: unknown): EmailAttachment[] | un + 'declared charset', ); } - if (typeof raw.inline !== 'string' || raw.inline === '') { - if (typeof raw.storageKey === 'string' && raw.storageKey !== '') { - reject( - 'attachments_json', - `${at} references content by storageKey, which no producer writes and nothing reads yet — ` - + 'out-of-row attachment storage is objectstack#5172. Refusing rather than delivering the message ' - + 'without this attachment', - ); - } - reject('attachments_json', `${at} carries no content: neither 'inline' nor a readable reference`); + + const element: PersistedEmailAttachment = { + filename, + ...(typeof raw.contentType === 'string' && raw.contentType ? { contentType: raw.contentType } : {}), + size: raw.size, + hash: raw.hash, + ...(typeof raw.cid === 'string' && raw.cid ? { cid: raw.cid } : {}), + contentForm: raw.contentForm, + }; + + if (typeof raw.inline === 'string' && raw.inline !== '') { + return { kind: 'inline', element, bytes: Buffer.from(raw.inline, 'base64') }; + } + if (typeof raw.storageKey === 'string' && raw.storageKey !== '') { + return { kind: 'storage', element, storageKey: raw.storageKey }; + } + if (typeof raw.contentReclaimedAt === 'string' && raw.contentReclaimedAt !== '') { + // The one "no content" case that is not damage. Still a refusal: the + // metadata is audit evidence, not something a transport can send. + reject( + 'attachments_json', + `${at} ('${filename}') had its out-of-row content reclaimed at ${raw.contentReclaimedAt}, after this ` + + 'row reached a terminal state — the message it belonged to was already delivered, and the remaining ' + + 'filename/size/hash are audit evidence, not a payload. Refusing to re-send this row without the ' + + 'attachment it declares', + ); } + reject('attachments_json', `${at} carries no content: neither 'inline' nor a readable reference`); + }); +} - const bytes = Buffer.from(raw.inline, 'base64'); - if (bytes.byteLength !== raw.size) { +/** + * Turn one validated element plus its raw bytes into an `EmailAttachment`, + * verifying that the bytes are the ones the row describes. + * + * This is where `size` and `hash` earn their place, and it runs identically + * whether the bytes came out of the row or out of storage — a storage backend + * that returns a truncated object is exactly as unacceptable as a truncated + * column, and used to be exactly as invisible. + */ +export function materializeAttachment( + element: PersistedEmailAttachment, + bytes: Buffer, + origin: string, +): EmailAttachment { + if (bytes.byteLength !== element.size) { + reject( + 'attachments_json', + `${origin} for '${element.filename}' is ${bytes.byteLength} byte(s) but the row records size ` + + `${element.size} — the content was truncated or rewritten`, + ); + } + const actual = digestOf(bytes); + if (actual !== element.hash) { + reject( + 'attachments_json', + `${origin} for '${element.filename}' hashes to ${actual} but the row records ${element.hash} — the ` + + 'stored content is not the content that was sent', + ); + } + return { + filename: element.filename, + content: element.contentForm === 'string' ? bytes.toString('utf8') : bytes, + ...(element.contentType ? { contentType: element.contentType } : {}), + ...(element.cid ? { cid: element.cid } : {}), + }; +} + +/** + * Rebuild `attachments` from `sys_email.attachments_json`, **from the row + * alone**. + * + * Returns `undefined` for a row that has no such column (every row written + * before #5177) — reading an old row must stay safe. Everything else is + * verified: a column that is present but does not describe the message it + * claims to describe throws, so the row lands at `failed` with the reason + * rather than being delivered without the attachment. + * + * An element whose content is **out of the row** (`storageKey`, #5172) throws + * here rather than being skipped: this function has no way to fetch it, and a + * caller that reaches this path without a storage capability wired must get an + * error, not a message missing an attachment. Use + * {@link decodeAttachmentsFromRowAsync} on any path that can fetch. + */ +export function decodeAttachmentsFromRow(value: unknown): EmailAttachment[] | undefined { + const sources = readAttachmentColumn(value); + if (!sources) return undefined; + return sources.map((source, i) => { + if (source.kind === 'storage') { reject( 'attachments_json', - `${at}.inline decodes to ${bytes.byteLength} byte(s) but the row records size ${raw.size} — the ` - + 'column was truncated or rewritten', + `[${i}] ('${source.element.filename}') holds its content out of the row under storageKey ` + + `'${source.storageKey}', and this delivery path has no file-storage capability to fetch it from. ` + + 'Fix: mount the file-storage capability (@objectstack/service-storage) on the process that delivers ' + + 'sys_email rows. Refusing rather than delivering the message without this attachment', ); } - const actual = digestOf(bytes); - if (actual !== raw.hash) { + return materializeAttachment(source.element, source.bytes, `[${i}].inline`); + }); +} + +/** + * Rebuild `attachments`, fetching out-of-row content through `fetchContent`. + * + * `fetchContent` is `undefined` when no storage capability is available; a row + * that needs one then fails exactly as it does in + * {@link decodeAttachmentsFromRow}. A fetch that throws is **not** caught here + * — the storage layer was made loud in #5216/#5232 so a read outage stops + * looking like a miss, and re-swallowing it would put a message with a missing + * attachment on the wire. + */ +export async function decodeAttachmentsFromRowAsync( + value: unknown, + fetchContent?: (storageKey: string) => Promise, +): Promise { + const sources = readAttachmentColumn(value); + if (!sources) return undefined; + const out: EmailAttachment[] = []; + for (let i = 0; i < sources.length; i++) { + const source = sources[i]!; + if (source.kind === 'inline') { + out.push(materializeAttachment(source.element, source.bytes, `[${i}].inline`)); + continue; + } + if (!fetchContent) { reject( 'attachments_json', - `${at}.inline hashes to ${actual} but the row records ${String(raw.hash)} — the stored content is not ` - + 'the content that was sent', + `[${i}] ('${source.element.filename}') holds its content out of the row under storageKey ` + + `'${source.storageKey}', but no file-storage capability is mounted on this process. Fix: mount it ` + + '(@objectstack/service-storage) wherever sys_email rows are delivered. Refusing rather than ' + + 'delivering the message without this attachment', ); } + const bytes = await fetchContent(source.storageKey); + out.push(materializeAttachment( + source.element, + bytes, + `[${i}] content fetched from storageKey '${source.storageKey}'`, + )); + } + return out; +} - return { - filename, - content: raw.contentForm === 'string' ? bytes.toString('utf8') : bytes, - ...(typeof raw.contentType === 'string' && raw.contentType ? { contentType: raw.contentType } : {}), - ...(typeof raw.cid === 'string' && raw.cid ? { cid: raw.cid } : {}), - }; +/** + * The storage keys an `attachments_json` column references, in element order. + * + * Tolerant by design, and this is the one place in this module where that is + * right: it feeds **deletion**, so a column too damaged to deliver from must + * still give up whatever keys it can name. Refusing to parse would strand the + * bytes forever — the failure mode has no upside. + */ +export function storageKeysInColumn(value: unknown): string[] { + if (isAbsent(value)) return []; + let parsed: unknown; + try { + parsed = typeof value === 'string' ? JSON.parse(value) : value; + } catch { + return []; + } + if (!Array.isArray(parsed)) return []; + const keys: string[] = []; + for (const raw of parsed) { + const key = (raw as any)?.storageKey; + if (typeof key === 'string' && key !== '') keys.push(key); + } + return keys; +} + +/** + * Rewrite an `attachments_json` column for a row whose out-of-row content has + * been deleted: `storageKey` out, {@link PersistedEmailAttachment.contentReclaimedAt} + * in, **everything else untouched**. + * + * Returns `undefined` when the column references no storage content, so the + * caller can skip a pointless write (and so a second reclaim of the same row + * is a no-op rather than a re-stamp of the timestamp). + */ +export function withContentReclaimed(value: unknown, at: string): string | undefined { + if (isAbsent(value)) return undefined; + let parsed: unknown; + try { + parsed = typeof value === 'string' ? JSON.parse(value) : value; + } catch { + return undefined; + } + if (!Array.isArray(parsed)) return undefined; + let changed = false; + const next = parsed.map((raw: any) => { + if (!raw || typeof raw !== 'object') return raw; + const key = raw.storageKey; + if (typeof key !== 'string' || key === '') return raw; + changed = true; + const { storageKey: _dropped, ...rest } = raw; + return { ...rest, contentReclaimedAt: at }; }); + return changed ? JSON.stringify(next) : undefined; } /**