From 2ea5307d3638c80284ef2a3be092494759975d7a Mon Sep 17 00:00:00 2001 From: TJ Baker <1617679+zaridan@users.noreply.github.com> Date: Thu, 16 Jul 2026 13:20:57 -0700 Subject: [PATCH 1/3] feat(mail): persist inbound attachment bytes to BlobStore (HT-46) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Writes each attachment's bytes to the BlobStore under a mailbox-namespaced key (//) BEFORE the ingest pipeline's step-5 transaction opens, then persists only the blob-key reference inside that transaction (new thread_attachments table, migration 015). A step-5 abort after a successful blob write leaves that blob orphaned and unreferenced — the failure mode spec §4 already blessed — and a retry writes a fresh blob rather than reusing or repairing the orphan. Also wires an optional attachment read path into the Agent Inbox API: GET /api/v1/conversations/{id}'s ThreadView now carries `attachments` (metadata + a BlobStore signed URL), absent-by-default like open tracking so no existing deployment or test is affected unless the composition root opts in (this ticket wires it in for the RIQ dogfood). Follow-up not built here: a GC sweep for orphaned blobs left behind by aborted ingest attempts (tolerable per the ticket's design, cross- referenced against thread_attachments in a future pass). Co-Authored-By: Claude Fable 5 --- specs/api/agent-inbox-v1.md | 22 +++- specs/mail/inbound-ingestion.md | 28 ++++- src/api/conversations.ts | 100 +++++++++++++++- src/api/index.test.ts | 125 ++++++++++++++++++++ src/api/index.ts | 17 ++- src/composition/root.ts | 3 + src/db/migrate.test.ts | 2 + src/db/migrate.ts | 45 +++++++ src/db/postgres.test.ts | 1 + src/mail/ingest.test.ts | 202 ++++++++++++++++++++++++++++++++ src/mail/ingest.ts | 140 +++++++++++++++++++--- src/store/attachments.test.ts | 180 ++++++++++++++++++++++++++++ src/store/attachments.ts | 144 +++++++++++++++++++++++ src/store/index.ts | 11 +- 14 files changed, 990 insertions(+), 30 deletions(-) create mode 100644 src/store/attachments.test.ts create mode 100644 src/store/attachments.ts diff --git a/specs/api/agent-inbox-v1.md b/specs/api/agent-inbox-v1.md index a3f69e6..25617fe 100644 --- a/specs/api/agent-inbox-v1.md +++ b/specs/api/agent-inbox-v1.md @@ -72,8 +72,21 @@ interface ThreadView { // v1.1: outbound only, and only when open tracking is // enabled (§4g) — first time the customer viewed the reply; // null until then, always null for inbound and notes + attachments: AttachmentView[] + // HT-46: inbound attachments this thread carries. [] when + // there are none, OR when the deployment hasn't wired the + // attachment read-path deps (config-gated, absent by default + // — same posture as open tracking, §4g) createdAt: string // ISO-8601 } + +interface AttachmentView { + id: string // uuid + filename: string | null // null when the attachment arrived with no filename + contentType: string + size: number // bytes + url: string // a time-limited signed URL (never a stable/public path) +} ``` **Status semantics (v1.1, HT-26).** `active` is the working state — inbound mail creates @@ -366,12 +379,19 @@ above. - No customer-side / self-service surface (a separate future API, designed native when there are customers to serve). - No mailbox management, no search, no realtime, no webhooks-out, no tag-filtered listing. -- No attachment upload on reply yet (the blob seam exists; wiring is later). +- No attachment upload on reply yet (HT-46 wired the READ side — inbound attachments + surfaced via `ThreadView.attachments` — but an Agent still cannot attach a file to an + outbound reply). - Framework-agnostic by construction: handlers are `Request → Response`; a Vercel/Next adapter is a thin deploy-time wrapper, not part of this spec. ## 7. Changelog +- **v1.1 (2026-07-16, HT-46).** `ThreadView.attachments`: inbound attachment metadata + + a signed `BlobStore` URL, `[]` by default and config-gated (absent `attachments` deps + at the composition root, §4's `InboxApiDeps`, same posture as open tracking) — a + deployment that hasn't wired a `ThreadAttachmentStore` + `BlobStore` never surfaces + attachments. No attachment upload on reply (§6, unchanged). - **v1.1 (2026-07-11, HT-25).** Adopted the contract the Agent Inbox UI was designed against (the Claude Design prototype's `mock-api.js`, whose additions were each marked `CONTRACT ADDITION`), after review of the drift between the designed surface and v1.0. diff --git a/specs/mail/inbound-ingestion.md b/specs/mail/inbound-ingestion.md index 98bbea9..b991984 100644 --- a/specs/mail/inbound-ingestion.md +++ b/specs/mail/inbound-ingestion.md @@ -85,10 +85,23 @@ Ordered, applied to each received message. Idempotent by step 1, so a whole re-r - The store write **and** the ledger row's `received → stored` transition (recording the resulting `threadId`) commit in **one transaction** — see §4. -**Attachments belong to the pipeline, not the transport** (§2). After the parse (step 2), -attachment bytes are written to the `BlobStore` under a **mailbox-namespaced** key -(`src/providers/blob.ts` makes namespacing the caller's responsibility) as part of the -step-5 store, and the stored thread carries blob references, never inline bytes. +**Attachments belong to the pipeline, not the transport** (§2). After the parse (step 2) +and the loop guard (step 3) — a suppressed reflection never writes attachment blobs it +would then have nothing to reference — each attachment's bytes are written to the +`BlobStore` under a **mailbox-namespaced** key, `//` +(`src/providers/blob.ts` makes namespacing the caller's responsibility; `attachmentId` is +a freshly minted UUID, formable before any row id exists). This write happens **before** +step 5's transaction opens — `BlobStore.put` is a non-transactional external side effect, +so it cannot be undone if that transaction later aborts — and only the resulting blob-key +**reference** (`thread_attachments`, migration 015) is persisted inside the transaction, +stamped with the thread id that same transaction mints. HT-46 implements this: a +step-5 abort after a successful blob write leaves that blob orphaned (unreferenced by any +`thread_attachments` row, since the insert never committed) — exactly the partial-failure +mode this section's next paragraph already blesses, and a retry re-parses, re-decides, and +writes fresh blobs under fresh attachment ids rather than reusing or cleaning up the +orphan. Orphaned blobs are tolerable and GC-able (a future sweep cross-referencing +`thread_attachments` against the bucket — not built here) but never a correctness +problem: an orphan is simply never referenced, so it is never served. ## 4. Idempotency, the delivery ledger, and retries @@ -268,7 +281,12 @@ engine's existing store/keyring fakes — no cloud required: - Two concurrent deliveries of the same key → exactly one conversation (the §3-step-1 atomic claim; the second returns the first's outcome). - A simulated partial failure (transaction aborts after a blob write) → ledger `failed`, - retried to `stored`, no orphaned/duplicate conversation. + retried to `stored`, no orphaned/duplicate conversation. HT-46: the ORIGINAL blob write + is left orphaned (never referenced), and the successful retry's `thread_attachments` rows + point at a FRESH blob write, not the orphan. +- A message with multiple attachments → one `thread_attachments` row per attachment, each + with its own blob key, all inserted in the same step-5 transaction as the thread they + belong to (HT-46). - A verifiable own-message loop → `suppressed`, nothing created; a message that merely *claims* our `From` without a verifiable correlation → **ingested**, not dropped. - `append→deleted` → falls back to a fresh conversation, mail never lost. diff --git a/src/api/conversations.ts b/src/api/conversations.ts index fea9dd0..8e4ac73 100644 --- a/src/api/conversations.ts +++ b/src/api/conversations.ts @@ -17,7 +17,8 @@ import type { Keyring } from '../mail/reply-token.js' import { sendReply } from '../mail/send.js' -import type { EmailSender } from '../providers/index.js' +import type { BlobStore, EmailSender } from '../providers/index.js' +import type { StoredThreadAttachment, ThreadAttachmentStore } from '../store/attachments.js' import { type ConversationFolder, type ConversationStatus, @@ -50,6 +51,19 @@ const MAX_REPLY_TEXT_LENGTH = 5000 */ const MAX_IDEMPOTENCY_KEY_LENGTH = 255 +/** + * The wire shape of one attachment on a `ThreadView` (specs/api/agent-inbox-v1.md + * §2, HT-46): attachment METADATA plus a time-limited signed URL — never a + * stable/public path (`BlobStore.getSignedUrl`'s contract, `src/providers/blob.ts`). + */ +interface AttachmentViewJson { + id: string + filename: string | null + contentType: string + size: number + url: string +} + /** The wire shape of one `ThreadView` (specs/api/agent-inbox-v1.md §2) — `StoredThread` with `Date` fields as ISO strings and `fromAddress` renamed to `from`. */ interface ThreadViewJson { id: string @@ -60,6 +74,8 @@ interface ThreadViewJson { deliveryStatus: 'pending' | 'sent' | 'failed' | null /** Open tracking (spec §4g, v1.1): first customer view of this outbound reply; null until then, always null for inbound/notes or with the feature off. */ customerViewedAt: string | null + /** HT-46: `[]` unless this thread has stored attachment references AND the deployment wired `attachments` deps (see {@link handleGetConversation}) — absent-by-default, like `openTracking`. */ + attachments: AttachmentViewJson[] createdAt: string } @@ -178,9 +194,28 @@ export async function handleListConversations( * (see uuid.ts); and the store is asked to exclude deleted rows at the * lookup so no threads are loaded for one (no latency side-channel, §5). */ +/** + * How long a minted attachment signed URL stays valid (`BlobStore.getSignedUrl`'s + * `expiresInSeconds`, HT-46). One hour: long enough to cover an Agent opening + * the conversation and viewing/downloading an attachment in one sitting, + * short enough that a URL copied out of a stale API response doesn't stay a + * standing credential. Not tuned against any measured usage — a reasonable + * default, re-minted fresh on every `GET` since nothing here caches it. + */ +const ATTACHMENT_SIGNED_URL_EXPIRY_SECONDS = 3600 + export async function handleGetConversation( id: string, - deps: { store: ConversationStore }, + deps: { + store: ConversationStore + /** + * Attachment read-path deps (HT-46) — ABSENT BY DEFAULT, the same posture + * `InboxApiDeps.openTracking` uses: a deployment that hasn't wired a + * `ThreadAttachmentStore` + `BlobStore` here simply never surfaces + * attachments, and every `ThreadView.attachments` is `[]`. + */ + attachments?: { store: ThreadAttachmentStore; blobStore: BlobStore } + }, ): Promise { if (!isUuid(id)) { return apiError(404, 'not_found', 'No conversation with that id.') @@ -196,6 +231,11 @@ export async function handleGetConversation( return apiError(404, 'not_found', 'No conversation with that id.') } + const attachmentsByThreadId = + deps.attachments !== undefined + ? await attachmentViewsByThreadId(conversation.id, deps.attachments) + : new Map() + const body: ConversationDetailJson = { id: conversation.id, number: conversation.number, @@ -208,12 +248,53 @@ export async function handleGetConversation( assignee: conversation.assignee, createdAt: conversation.createdAt.toISOString(), updatedAt: conversation.updatedAt.toISOString(), - threads: conversation.threads.map(toThreadViewJson), + threads: conversation.threads.map((thread) => + toThreadViewJson(thread, attachmentsByThreadId.get(thread.id)), + ), } return json(200, body) } +/** + * Fetch every attachment reference for `conversationId` in one round trip + * (`ThreadAttachmentStore.listByConversationId`) and mint each one's signed + * URL, grouped by the thread id it belongs to. Signing happens here, not in + * the store, so `ThreadAttachmentStore` stays a plain persistence seam with + * no `BlobStore` dependency of its own (mirroring how `ConversationStore` + * never touches a provider either). + */ +async function attachmentViewsByThreadId( + conversationId: string, + attachments: { store: ThreadAttachmentStore; blobStore: BlobStore }, +): Promise> { + const rows = await attachments.store.listByConversationId(conversationId) + const byThreadId = new Map() + for (const row of rows) { + const view = await toAttachmentViewJson(row, attachments.blobStore) + const existing = byThreadId.get(row.threadId) + if (existing === undefined) { + byThreadId.set(row.threadId, [view]) + } else { + existing.push(view) + } + } + return byThreadId +} + +async function toAttachmentViewJson( + row: StoredThreadAttachment, + blobStore: BlobStore, +): Promise { + return { + id: row.id, + filename: row.filename, + contentType: row.contentType, + size: row.size, + url: await blobStore.getSignedUrl(row.blobKey, ATTACHMENT_SIGNED_URL_EXPIRY_SECONDS), + } +} + /** * Derive a detail response's `preview` from the threads it already carries — * the SAME rule the store applies for list summaries (`derivePreview`, spec @@ -774,7 +855,17 @@ function toConversationSummaryJson(row: { } } -function toThreadViewJson(thread: StoredThread): ThreadViewJson { +/** + * Map one `StoredThread` to its wire shape. `attachments` defaults to `[]` — + * every caller EXCEPT {@link handleGetConversation} passes none, because a + * thread this API just created (a reply or a note) cannot yet have any + * (HT-46: attachments are inbound-only, and only `handleGetConversation`'s + * deps carry the `ThreadAttachmentStore`/`BlobStore` needed to look them up). + */ +function toThreadViewJson( + thread: StoredThread, + attachments: AttachmentViewJson[] = [], +): ThreadViewJson { return { id: thread.id, direction: thread.direction, @@ -784,6 +875,7 @@ function toThreadViewJson(thread: StoredThread): ThreadViewJson { deliveryStatus: thread.deliveryStatus, customerViewedAt: thread.customerViewedAt === null ? null : thread.customerViewedAt.toISOString(), + attachments, createdAt: thread.createdAt.toISOString(), } } diff --git a/src/api/index.test.ts b/src/api/index.test.ts index 5fc4699..f96b851 100644 --- a/src/api/index.test.ts +++ b/src/api/index.test.ts @@ -6,11 +6,13 @@ import { createGmailConnectService } from '../mail/gmail-connect.js' import type { Keyring } from '../mail/reply-token.js' import type { GmailWatchClient } from '../providers/adapters/gmail/index.js' import type { + BlobStore, EmailSender, EnqueueOptions, OutboundEmail, QueueProvider, } from '../providers/index.js' +import { createThreadAttachmentStore, insertThreadAttachmentsInTx } from '../store/attachments.js' import { type ConversationStore, createConversationStore, @@ -56,6 +58,30 @@ function createThrowingSender(): EmailSender { } } +/** An in-memory `BlobStore` fake, matching `src/mail/ingest.test.ts`'s — `getSignedUrl` returns a deterministic, inspectable URL. */ +function fakeBlobStore(initial: Record = {}): BlobStore { + const store = new Map(Object.entries(initial)) + return { + async put(key, data) { + store.set(key, data) + }, + async get(key) { + const data = store.get(key) + if (data === undefined) throw new Error(`fakeBlobStore: no object at key ${key}`) + return data + }, + async getSignedUrl(key, expiresInSeconds) { + return `https://blob.example.test/${key}?expires=${expiresInSeconds}` + }, + async delete(key) { + store.delete(key) + }, + async exists(key) { + return store.has(key) + }, + } +} + function newConversation(overrides: Partial = {}): NewConversation { return { subject: 'Help with my order', @@ -180,6 +206,8 @@ describe('createInboxApi', () => { openTracking?: { publicBaseUrl: string } gmailPush?: InboxApiDeps['gmailPush'] gmailConnect?: InboxApiDeps['gmailConnect'] + /** When given, wires `attachments: { store: createThreadAttachmentStore(db), blobStore }` — this fake's `db` doesn't exist until this function creates it, so the `ThreadAttachmentStore` is built HERE rather than by the caller. */ + attachmentsBlobStore?: BlobStore } = {}, ): Promise<{ db: Db @@ -202,6 +230,14 @@ describe('createInboxApi', () => { ...(overrides.openTracking !== undefined ? { openTracking: overrides.openTracking } : {}), ...(overrides.gmailPush !== undefined ? { gmailPush: overrides.gmailPush } : {}), ...(overrides.gmailConnect !== undefined ? { gmailConnect: overrides.gmailConnect } : {}), + ...(overrides.attachmentsBlobStore !== undefined + ? { + attachments: { + store: createThreadAttachmentStore(db), + blobStore: overrides.attachmentsBlobStore, + }, + } + : {}), }) return { db, store, api, sent } } @@ -480,6 +516,95 @@ describe('createInboxApi', () => { const res = await api(get(`/api/v1/conversations/${conversationId}`)) expect(res.status).toBe(404) }) + + // --- attachments (HT-46) ------------------------------------------------- + + it('every thread carries attachments: [] when the deployment has no `attachments` deps wired (absent-by-default, like openTracking)', async () => { + const { store, api } = await freshApi() + const { conversationId } = await store.createConversation(newConversation()) + + const res = await api(get(`/api/v1/conversations/${conversationId}`)) + const body = (await res.json()) as { threads: Array<{ attachments: unknown[] }> } + expect(body.threads).toHaveLength(1) + expect(body.threads[0].attachments).toEqual([]) + }) + + it('surfaces attachment metadata + a signed URL when `attachments` deps ARE wired', async () => { + const blobStore = fakeBlobStore() + const { db, store, api } = await freshApi({ attachmentsBlobStore: blobStore }) + const { conversationId, threadId } = await store.createConversation(newConversation()) + await db.transaction((tx) => + insertThreadAttachmentsInTx(tx, [ + { + threadId, + filename: 'invoice.pdf', + contentType: 'application/pdf', + size: 1234, + blobKey: `mbox-1/attach-1/invoice.pdf`, + }, + ]), + ) + + const res = await api(get(`/api/v1/conversations/${conversationId}`)) + expect(res.status).toBe(200) + const body = (await res.json()) as { + threads: Array<{ + id: string + attachments: Array<{ + id: string + filename: string | null + contentType: string + size: number + url: string + }> + }> + } + expect(body.threads).toHaveLength(1) + expect(body.threads[0].attachments).toHaveLength(1) + expect(body.threads[0].attachments[0]).toMatchObject({ + filename: 'invoice.pdf', + contentType: 'application/pdf', + size: 1234, + url: 'https://blob.example.test/mbox-1/attach-1/invoice.pdf?expires=3600', + }) + }) + + it('scopes attachments to the right thread when a conversation has multiple threads', async () => { + const blobStore = fakeBlobStore() + const { db, store, api } = await freshApi({ attachmentsBlobStore: blobStore }) + const { conversationId, threadId: firstThreadId } = await store.createConversation( + newConversation(), + ) + const appendResult = await store.appendThread(conversationId, { + direction: 'outbound', + messageId: '', + fromAddress: 'support@example.test', + bodyText: 'Looking into it!', + }) + if (!appendResult.ok) throw new Error('unreachable') + + await db.transaction((tx) => + insertThreadAttachmentsInTx(tx, [ + { + threadId: firstThreadId, + filename: 'first.txt', + contentType: 'text/plain', + size: 1, + blobKey: 'mbox-1/a/first.txt', + }, + ]), + ) + + const res = await api(get(`/api/v1/conversations/${conversationId}`)) + const body = (await res.json()) as { + threads: Array<{ id: string; attachments: Array<{ filename: string | null }> }> + } + expect(body.threads).toHaveLength(2) + const first = body.threads.find((t) => t.id === firstThreadId) + const second = body.threads.find((t) => t.id === appendResult.threadId) + expect(first?.attachments.map((a) => a.filename)).toEqual(['first.txt']) + expect(second?.attachments).toEqual([]) + }) }) // --- reply ------------------------------------------------------------------- diff --git a/src/api/index.ts b/src/api/index.ts index da1e331..830bc10 100644 --- a/src/api/index.ts +++ b/src/api/index.ts @@ -31,7 +31,8 @@ import { TRANSPARENT_GIF, verifyViewToken } from '../mail/open-tracking.js' import type { Keyring } from '../mail/reply-token.js' -import type { EmailSender } from '../providers/index.js' +import type { BlobStore, EmailSender } from '../providers/index.js' +import type { ThreadAttachmentStore } from '../store/attachments.js' import type { ConversationStore } from '../store/conversations.js' import { authenticateRequest } from './auth.js' import { @@ -124,6 +125,15 @@ export interface InboxApiDeps { * ordinary "no such route" answer either way). */ gmailConnect?: GmailConnectDeps + /** + * Attachment read-path deps (HT-46; specs/api/agent-inbox-v1.md §2's + * `ThreadView.attachments`): ABSENT BY DEFAULT — a deployment that hasn't + * wired a `ThreadAttachmentStore` + `BlobStore` here simply never surfaces + * attachments, and `GET /api/v1/conversations/{id}` returns `[]` for every + * thread's `attachments`, exactly like `openTracking`'s absent-by-default + * posture above. + */ + attachments?: { store: ThreadAttachmentStore; blobStore: BlobStore } } /** @@ -256,7 +266,10 @@ export function createInboxApi(deps: InboxApiDeps): (request: Request) => Promis return await handleListConversations(request, { store: deps.store }) case 'conversation-item': - return await handleGetConversation(route.id, { store: deps.store }) + return await handleGetConversation(route.id, { + store: deps.store, + ...(deps.attachments !== undefined ? { attachments: deps.attachments } : {}), + }) case 'conversation-patch': return await handlePatchConversation(route.id, request, { store: deps.store }) diff --git a/src/composition/root.ts b/src/composition/root.ts index b94e6ee..f56a490 100644 --- a/src/composition/root.ts +++ b/src/composition/root.ts @@ -73,6 +73,7 @@ import { createInboundDeliveryStore, createMailboxStore, createMailboxTokenStore, + createThreadAttachmentStore, } from '../store/index.js' import { createAppHandler } from './app.js' import { type AppConfig, loadConfig } from './config.js' @@ -137,6 +138,7 @@ export async function buildApp( const tokenStore = createMailboxTokenStore(db, config.tokenEncryptionKey) const watchStateStore = createGmailWatchStateStore(db) const inboundDeliveryStore = createInboundDeliveryStore(db) + const attachmentStore = createThreadAttachmentStore(db) // --- The HMAC keyring backing reply/state/view tokens (single current key). --- const keyring: Keyring = { current: { keyId: SIGNING_KEY_ID, secret: config.signingSecret } } @@ -214,6 +216,7 @@ export async function buildApp( supportAddress: config.supportAddress, gmailPush, gmailConnect, + attachments: { store: attachmentStore, blobStore }, }) // --- The reconcile handler the queue drain dispatches to. --- diff --git a/src/db/migrate.test.ts b/src/db/migrate.test.ts index 40770ba..d0b873a 100644 --- a/src/db/migrate.test.ts +++ b/src/db/migrate.test.ts @@ -54,6 +54,7 @@ describe('migrate', () => { { id: 12, name: 'inbound_deliveries' }, { id: 13, name: 'queue_jobs' }, { id: 14, name: 'inbound_delivery_lease' }, + { id: 15, name: 'thread_attachments' }, ]) }) @@ -78,6 +79,7 @@ describe('migrate', () => { { id: 12 }, { id: 13 }, { id: 14 }, + { id: 15 }, ]) }) diff --git a/src/db/migrate.ts b/src/db/migrate.ts index b88d22b..c8cf3ff 100644 --- a/src/db/migrate.ts +++ b/src/db/migrate.ts @@ -714,6 +714,46 @@ const MIGRATION_014_INBOUND_DELIVERY_LEASE = ` ALTER TABLE inbound_deliveries ADD COLUMN claimed_until timestamptz; ` +/** + * Migration 015 — `thread_attachments` (HT-46): blob-reference rows for + * inbound attachment bytes. + * + * `src/mail/parse.ts`'s `ParsedEmail.attachments` carries bytes; this table + * carries the reference to where those bytes actually live once the ingest + * pipeline writes them to the `BlobStore` (specs/mail/inbound-ingestion.md + * §3's closing paragraph) — never the bytes themselves. One row per + * attachment, `thread_id` a plain FK (a thread has zero or many), `ON DELETE + * CASCADE` matching `threads.conversation_id`'s own cascade (migration 001): + * deleting a thread's row deletes its attachment references with it, the + * same "storage row lifetime tracks its parent" policy already used + * throughout this schema. This table does NOT delete the underlying blob + * object on cascade — `BlobStore` cleanup for an orphaned/cascaded key is + * left to a future GC pass (see `src/mail/ingest.ts`'s doc comment on why an + * orphaned blob from an aborted ingest attempt is tolerable), not built here. + * + * `blob_key` is the mailbox-namespaced `BlobStore` key + * (`//`, `src/mail/ingest.ts`) — an opaque + * string as far as this table and `BlobStore` itself are concerned (`src/ + * providers/blob.ts`'s key-namespacing contract). `filename` is nullable + * because `ParsedAttachment.filename` (`src/mail/parse.ts`) is: some + * attachments (e.g. an inline image referenced only by `Content-Id`) arrive + * with no `Content-Disposition` filename at all. `size` is `integer` + * (bytes) — ample headroom below Gmail's ~25MB message cap, the only inbound + * transport this engine has today. + */ +const MIGRATION_015_THREAD_ATTACHMENTS = ` +CREATE TABLE thread_attachments ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + thread_id uuid NOT NULL REFERENCES threads(id) ON DELETE CASCADE, + filename text, + content_type text NOT NULL, + size integer NOT NULL, + blob_key text NOT NULL, + created_at timestamptz NOT NULL DEFAULT now() +); +CREATE INDEX thread_attachments_thread_id_idx ON thread_attachments (thread_id); +` + /** * Every migration, in the order they must apply. `id` is the sole ordering * key (ascending) — array position is not relied upon, so re-sorting this @@ -786,6 +826,11 @@ const MIGRATIONS: Migration[] = [ name: 'inbound_delivery_lease', sql: MIGRATION_014_INBOUND_DELIVERY_LEASE, }, + { + id: 15, + name: 'thread_attachments', + sql: MIGRATION_015_THREAD_ATTACHMENTS, + }, ] /** diff --git a/src/db/postgres.test.ts b/src/db/postgres.test.ts index e9c0845..4465b6e 100644 --- a/src/db/postgres.test.ts +++ b/src/db/postgres.test.ts @@ -302,6 +302,7 @@ describe('createPostgresDb with a schema option', () => { 'mailbox_oauth_tokens', 'mailboxes', 'queue_jobs', + 'thread_attachments', 'threads', ]) diff --git a/src/mail/ingest.test.ts b/src/mail/ingest.test.ts index 90ef456..6501286 100644 --- a/src/mail/ingest.test.ts +++ b/src/mail/ingest.test.ts @@ -2,6 +2,7 @@ import { afterEach, describe, expect, it } from 'vitest' import { createPgliteDb, type Db, type Queryable } from '../db/client.js' import { migrate } from '../db/migrate.js' import type { BlobStore, RawInboundMessage } from '../providers/index.js' +import { createThreadAttachmentStore } from '../store/attachments.js' import { createInboundDeliveryStore } from '../store/inbound-deliveries.js' import { type IngestDeps, @@ -99,6 +100,75 @@ function fakeBlobStore(initial: Record = {}): BlobStore { } } +/** + * Wraps a `BlobStore` fake to record every `put` key, in call order — used by + * the attachment retry/orphan test to distinguish the FIRST (failed-attempt, + * orphaned) blob write from the SECOND (retry, referenced) one without + * needing to predict the random attachment id `src/mail/ingest.ts` mints for + * each write. + */ +function trackingBlobStore(inner: BlobStore): BlobStore & { putKeys: string[] } { + const putKeys: string[] = [] + return { + putKeys, + async put(key, data, opts) { + putKeys.push(key) + await inner.put(key, data, opts) + }, + get: (key) => inner.get(key), + getSignedUrl: (key, expiresInSeconds) => inner.getSignedUrl(key, expiresInSeconds), + delete: (key) => inner.delete(key), + exists: (key) => inner.exists(key), + } +} + +/** + * Build a raw `multipart/mixed` RFC5322 message with one plain-text body part + * plus one base64-encoded attachment per entry in `attachments` — the same + * shape as `tests/mail/fixtures/attachment.eml` (plain `\n` line endings; + * postal-mime, verified by that fixture's own test, tolerates them). + */ +function rawMessageWithAttachments( + overrides: Record = {}, + attachments: { filename: string; contentType: string; content: string }[] = [ + { filename: 'hello.txt', contentType: 'text/plain', content: 'Hello, world!' }, + ], +): Uint8Array { + const boundary = 'BOUNDARY-INGEST-TEST' + const headers: Record = { + From: 'customer@example.test', + To: 'support@example.test', + Subject: 'Message with attachment', + 'Message-ID': '', + 'MIME-Version': '1.0', + 'Content-Type': `multipart/mixed; boundary="${boundary}"`, + ...overrides, + } + const headerText = Object.entries(headers) + .map(([key, value]) => `${key}: ${value}`) + .join('\n') + + const bodyParts = [ + `--${boundary}`, + 'Content-Type: text/plain; charset=utf-8', + '', + 'See attached.', + ] + for (const attachment of attachments) { + bodyParts.push( + `--${boundary}`, + `Content-Type: ${attachment.contentType}; name="${attachment.filename}"`, + `Content-Disposition: attachment; filename="${attachment.filename}"`, + 'Content-Transfer-Encoding: base64', + '', + Buffer.from(attachment.content, 'utf-8').toString('base64'), + ) + } + bodyParts.push(`--${boundary}--`) + + return new TextEncoder().encode(`${headerText}\n\n${bodyParts.join('\n')}\n`) +} + /** Insert a `mailboxes` row directly — `inbound_deliveries.mailbox_id` is a real FK, and creating mailboxes is not this ticket's concern. */ async function createMailbox(db: Db, address = 'support@example.test'): Promise { const rows = await db.query<{ id: string }>( @@ -658,4 +728,136 @@ describe('ingestInboundMessage', () => { expect(outcome.kind).toBe('stored') expect(await countRows(db, 'conversations')).toBe(1) }) + + // --- HT-46: attachment blob persistence. ---------------------------------- + + describe('attachments (HT-46)', () => { + it('a message with one attachment writes its bytes to the BlobStore and persists exactly one blob-key reference', async () => { + const { db, deps, mailboxId } = await freshDeps() + const raw = inboundDelivery(mailboxId, 'provider-msg-1', rawMessageWithAttachments()) + + const outcome = await ingestInboundMessage(raw, deps) + + expect(outcome.kind).toBe('stored') + if (outcome.kind !== 'stored') throw new Error('unreachable') + + const attachmentStore = createThreadAttachmentStore(db) + const rows = await attachmentStore.listByConversationId(outcome.conversationId) + expect(rows).toHaveLength(1) + expect(rows[0]).toMatchObject({ + threadId: outcome.threadId, + filename: 'hello.txt', + contentType: 'text/plain', + size: 13, + }) + // The blob key is mailbox-namespaced (//). + expect(rows[0].blobKey.startsWith(`${mailboxId}/`)).toBe(true) + expect(rows[0].blobKey.endsWith('/hello.txt')).toBe(true) + + // The bytes actually landed in the BlobStore, byte-exact. + const storedBytes = await deps.blobStore.get(rows[0].blobKey) + expect(new TextDecoder().decode(storedBytes)).toBe('Hello, world!') + }) + + it('a message with multiple attachments persists one thread_attachments row per attachment, each with its own blob key', async () => { + const { db, deps, mailboxId } = await freshDeps() + const raw = inboundDelivery( + mailboxId, + 'provider-msg-1', + rawMessageWithAttachments({}, [ + { filename: 'one.txt', contentType: 'text/plain', content: 'first file' }, + { filename: 'two.txt', contentType: 'text/plain', content: 'second file, longer' }, + ]), + ) + + const outcome = await ingestInboundMessage(raw, deps) + + expect(outcome.kind).toBe('stored') + if (outcome.kind !== 'stored') throw new Error('unreachable') + + const attachmentStore = createThreadAttachmentStore(db) + const rows = await attachmentStore.listByConversationId(outcome.conversationId) + expect(rows).toHaveLength(2) + // Distinct blob keys — no collision between the two attachments. + expect(new Set(rows.map((r) => r.blobKey)).size).toBe(2) + + const byFilename = new Map(rows.map((r) => [r.filename, r])) + const one = byFilename.get('one.txt') + const two = byFilename.get('two.txt') + expect(one).toBeDefined() + expect(two).toBeDefined() + if (one === undefined || two === undefined) throw new Error('unreachable') + expect(new TextDecoder().decode(await deps.blobStore.get(one.blobKey))).toBe('first file') + expect(new TextDecoder().decode(await deps.blobStore.get(two.blobKey))).toBe( + 'second file, longer', + ) + }) + + it('a message with no attachments persists no thread_attachments rows', async () => { + const { db, deps, mailboxId } = await freshDeps() + const outcome = await ingestInboundMessage( + inboundDelivery(mailboxId, 'provider-msg-1', freshCustomerRaw()), + deps, + ) + + expect(outcome.kind).toBe('stored') + if (outcome.kind !== 'stored') throw new Error('unreachable') + expect( + await createThreadAttachmentStore(db).listByConversationId(outcome.conversationId), + ).toEqual([]) + }) + + it('the retry/orphan story: a step-5 abort after the blob write leaves that blob orphaned, and the retry writes a FRESH blob the stored reference actually points at', async () => { + const { db, mailboxId } = await freshDeps() + const blobStore = trackingBlobStore(fakeBlobStore()) + // 1st .transaction() call = the claim (succeeds); 2nd = step 5's + // store-write + ledger-mark transaction — fails exactly ONCE (see the + // module-level doc comment on dbFailingOnCall). The attachment blob + // write happens between steps 4 and 5, OUTSIDE any transaction, so it + // is NOT counted here and always runs on every attempt. + const faultyDb = dbFailingOnCall(db, 2) + const faultyDeps: IngestDeps = { + db: faultyDb, + inboundDeliveryStore: createInboundDeliveryStore(faultyDb), + blobStore, + keyring, + } + const raw = inboundDelivery(mailboxId, 'provider-msg-1', rawMessageWithAttachments()) + + const failedOutcome = await ingestInboundMessage(raw, faultyDeps) + expect(failedOutcome).toMatchObject({ kind: 'failed', attempts: 1 }) + // The blob write for this FIRST attempt already happened (it precedes + // the aborted transaction) — orphaned: written, but referenced by no + // thread_attachments row, since the transaction that would have + // inserted one rolled back along with the thread it belonged to. + expect(blobStore.putKeys).toHaveLength(1) + const orphanKey = blobStore.putKeys[0] + expect(await blobStore.exists(orphanKey)).toBe(true) + expect(await countRows(db, 'conversations')).toBe(0) + + const retried = await ingestInboundMessage(raw, faultyDeps) + expect(retried.kind).toBe('stored') + if (retried.kind !== 'stored') throw new Error('unreachable') + + // The retry wrote a SECOND, fresh blob (a new attachment id each + // attempt, per src/mail/ingest.ts's writeAttachmentBlobs) rather than + // reusing or repairing the orphan. + expect(blobStore.putKeys).toHaveLength(2) + const liveKey = blobStore.putKeys[1] + expect(liveKey).not.toBe(orphanKey) + + const rows = await createThreadAttachmentStore(db).listByConversationId( + retried.conversationId, + ) + expect(rows).toHaveLength(1) + // The persisted reference points at the SECOND (retry) blob, not the + // orphaned first one. + expect(rows[0].blobKey).toBe(liveKey) + + // The orphan is still sitting in the BlobStore, untouched and + // unreferenced — tolerable per the ticket's design, not cleaned up + // here (a future GC pass, not built by this ticket). + expect(await blobStore.exists(orphanKey)).toBe(true) + }) + }) }) diff --git a/src/mail/ingest.ts b/src/mail/ingest.ts index 0e113fe..686204a 100644 --- a/src/mail/ingest.ts +++ b/src/mail/ingest.ts @@ -28,19 +28,36 @@ * the ledger's `received → stored` transition become ONE transaction * (spec §4 — the crux this ticket exists to get right). * - * ## Attachments: deferred (HT-37 scope note) + * ## Attachments (HT-46) * - * `ParsedEmail.attachments` carries bytes (`src/mail/parse.ts`), but the - * `threads` store has no attachment column today, and writing attachment - * bytes to the `BlobStore` under a mailbox-namespaced key (spec §3's closing - * paragraph) is explicitly out of scope for this ticket. This pipeline - * stores the text/html body only; attachment bytes are silently NOT - * persisted anywhere yet. Flagged in the implementation report as a - * follow-up, not invented here. + * `ParsedEmail.attachments` carries bytes (`src/mail/parse.ts`). Between + * step 4 (decide) and step 5 (store), this pipeline writes each attachment's + * bytes to the `BlobStore` under a mailbox-namespaced key + * (`//`, spec §3's closing paragraph — + * see {@link writeAttachmentBlobs}) BEFORE the step-5 transaction opens, then + * persists only the resulting blob-key REFERENCES inside that transaction + * (`thread_attachments`, migration 015, `src/store/attachments.ts`) — never + * the bytes themselves, and never inside the transaction. This ordering is + * the ticket's design, not incidental: `BlobStore.put` is a non-transactional + * external side effect (it cannot be rolled back if the transaction that + * follows aborts), so spec §4's already-blessed failure mode — "a blob write + * that succeeds then a transaction that aborts" — is exactly what happens on + * a step-5 failure here, and it is HONEST about it: the blob is orphaned + * (unreferenced by any `thread_attachments` row, since that insert never + * committed), not corrupted or double-referenced. A retry (this pipeline's + * ordinary retry-the-whole-unit contract, spec §4) re-parses, re-decides, and + * re-writes FRESH blobs under fresh attachment ids — it never reuses or + * cleans up the orphaned ones from the failed attempt. Orphaned blobs are + * tolerable and GC-able (a future sweep keyed off `thread_attachments` + * cross-referenced against the bucket — not built here, flagged as a + * follow-up in the implementation report) but never a correctness problem: + * an orphan is simply never referenced by anything, so it is never served. */ +import { randomUUID } from 'node:crypto' import type { Db, Queryable } from '../db/client.js' import type { BlobStore, RawInboundMessage, RawMessageContent } from '../providers/index.js' +import { insertThreadAttachmentsInTx, type NewThreadAttachment } from '../store/attachments.js' import { appendThreadInTx, createConversationInTx, type NewThread } from '../store/conversations.js' import { type InboundDeliveryStore, @@ -48,7 +65,7 @@ import { markStoredInTx, type StoredInboundDelivery, } from '../store/inbound-deliveries.js' -import { type ParsedEmail, parseInboundEmail } from './parse.js' +import { type ParsedAttachment, type ParsedEmail, parseInboundEmail } from './parse.js' import { type Keyring, verifyReplyMessageId } from './reply-token.js' import { decideThreading, type ThreadingDecision } from './thread.js' @@ -376,6 +393,21 @@ async function processClaimedDelivery( // --- Step 4: decide (never re-implemented here). -------------------------- const decision = decideThreading(parsed, deps.keyring) + // --- Attachments: write bytes to the BlobStore BEFORE step 5's transaction + // (module doc's "Attachments (HT-46)" section) — a blob write is a + // non-transactional external side effect, so it must happen outside (and + // before) the transaction that references it. ----------------------------- + let attachmentRefs: Omit[] + try { + attachmentRefs = await writeAttachmentBlobs( + delivery.mailboxId, + parsed.attachments, + deps.blobStore, + ) + } catch (err) { + return recordFailure(delivery, deps, 'blob', err, base) + } + // --- Step 5: store + mark stored, ONE transaction (the crux — see // storeAndMarkDelivered's doc comment). ------------------------------------ try { @@ -384,6 +416,7 @@ async function processClaimedDelivery( delivery.id, decision, parsed, + attachmentRefs, delivery.attempts, ) logIngestEvent({ @@ -413,6 +446,62 @@ async function processClaimedDelivery( } } +/** + * Write every attachment's bytes to `blobStore` under a fresh, mailbox- + * namespaced key (spec §3's closing paragraph): `// + * `, where `attachmentId` is a freshly minted UUID — formable + * before any row id exists, exactly the ticket's design (the row this + * attachment will reference, `thread_attachments.thread_id`, doesn't exist + * until step 5's INSERT). Returns the reference each attachment resolved to + * (everything {@link insertThreadAttachmentsInTx} needs except `threadId`, + * which step 5 fills in once the thread row exists). `[]` for a message with + * no attachments — the common case, and the fast path (no blob writes at + * all). + * + * Called AFTER the loop guard (step 3) so a suppressed own-message-loop + * reflection never writes attachment blobs that would then have nothing to + * reference — see the module doc's "Attachments (HT-46)" section for why a + * write here can still end up orphaned by a LATER step-5 failure, and why + * that is tolerable. + */ +async function writeAttachmentBlobs( + mailboxId: string, + attachments: ParsedAttachment[], + blobStore: BlobStore, +): Promise[]> { + const refs: Omit[] = [] + for (const attachment of attachments) { + const blobKey = `${mailboxId}/${randomUUID()}/${sanitizeAttachmentFilename(attachment.filename)}` + await blobStore.put(blobKey, attachment.content, { + contentType: attachment.contentType, + contentLength: attachment.size, + }) + refs.push({ + filename: attachment.filename, + contentType: attachment.contentType, + size: attachment.size, + blobKey, + }) + } + return refs +} + +/** + * The filename segment of an attachment's blob key — NOT the `filename` + * column value (that stays the original, verbatim `ParsedAttachment.filename`, + * `null` included). `BlobStore` implementations (e.g. Supabase Storage, + * `src/providers/adapters/supabase-storage/`) treat `/` in a key as a path + * separator, so an attacker- or client-supplied filename containing `/` + * could otherwise nest the object under an unintended "folder" inside this + * attachment's own namespace slot; stripping it (and any other key-reserved + * character) keeps every attachment's key exactly three segments deep, + * whatever the filename contains. A missing filename falls back to a fixed + * placeholder — the key still needs SOME final segment. + */ +function sanitizeAttachmentFilename(filename: string | null): string { + return (filename ?? 'attachment').replaceAll(/[/\\]/g, '_') +} + /** * Record a caught processing failure on the ledger — `dead-letter` once * `MAX_INGEST_ATTEMPTS` would be reached, otherwise the retryable `failed` @@ -426,7 +515,7 @@ async function processClaimedDelivery( async function recordFailure( delivery: StoredInboundDelivery, deps: IngestDeps, - stage: 'parse' | 'store', + stage: 'parse' | 'blob' | 'store', err: unknown, base: IngestOutcomeBase, ): Promise { @@ -483,22 +572,41 @@ async function recordFailure( * 'not-found' }` — never resurrects a deleted conversation, never drops the * mail (threading.md §5, mirrored here for the ingest path). * - * `claimedAttempts` is threaded straight through to `markStoredInTx` as its - * fence (`src/store/inbound-deliveries.ts`'s "The fence" section): if it no - * longer matches, `markStoredInTx` throws `LeaseLostError` and `Db.transaction` - * rolls back the conversation/thread write this call just made along with it - * — a stale, lease-lost caller can never leave behind a conversation with no - * matching ledger mark. + * `attachmentRefs` (HT-46) — the blob-key references {@link + * writeAttachmentBlobs} already resolved, BEFORE this transaction opened — + * are persisted here via `insertThreadAttachmentsInTx`, stamped with the + * thread id this same transaction just minted. This is the only place a + * `thread_attachments` row is created, and it happens in the SAME commit as + * the thread it references: no reference can survive without its thread, and + * no thread can be missing a reference for bytes this call already wrote (a + * throw anywhere in this transaction rolls back the thread AND the + * references together, per this doc's rollback paragraph above — only the + * already-written blob bytes are left behind, orphaned, per the module doc). + * + * `claimedAttempts` (HT-45) is threaded straight through to `markStoredInTx` + * as its fence (`src/store/inbound-deliveries.ts`'s "The fence" section): if + * it no longer matches, `markStoredInTx` throws `LeaseLostError` and + * `Db.transaction` rolls back the conversation/thread write — and the + * attachment-reference rows — this call just made along with it. A stale, + * lease-lost caller can never leave behind a conversation with no matching + * ledger mark, or a `thread_attachments` row pointing at a thread that was + * never committed; the only trace it leaves is the orphaned blob bytes the + * paragraph above already accounts for. */ async function storeAndMarkDelivered( db: Db, deliveryId: string, decision: ThreadingDecision, parsed: ParsedEmail, + attachmentRefs: Omit[], claimedAttempts: number, ): Promise<{ conversationId: string; threadId: string }> { return db.transaction(async (tx) => { const written = await writeParsedEmail(tx, decision, parsed) + await insertThreadAttachmentsInTx( + tx, + attachmentRefs.map((ref) => ({ ...ref, threadId: written.threadId })), + ) await markStoredInTx(tx, deliveryId, written.threadId, claimedAttempts) return written }) diff --git a/src/store/attachments.test.ts b/src/store/attachments.test.ts new file mode 100644 index 0000000..10341e0 --- /dev/null +++ b/src/store/attachments.test.ts @@ -0,0 +1,180 @@ +import { afterEach, describe, expect, it } from 'vitest' +import { createPgliteDb, type Db } from '../db/client.js' +import { migrate } from '../db/migrate.js' +import { + createThreadAttachmentStore, + insertThreadAttachmentsInTx, + type ThreadAttachmentStore, +} from './attachments.js' +import { type ConversationStore, createConversationStore } from './conversations.js' + +// --- fixtures ---------------------------------------------------------------- + +/** Insert a conversation + its first (inbound) thread directly via the real store, returning both ids. */ +async function createConversationWithThread( + store: ConversationStore, +): Promise<{ conversationId: string; threadId: string }> { + return store.createConversation({ + subject: 'Test', + customerEmail: 'customer@example.test', + firstMessage: { + direction: 'inbound', + messageId: '', + fromAddress: 'customer@example.test', + bodyText: 'hi', + }, + }) +} + +// --- suite --------------------------------------------------------------------- + +describe('ThreadAttachmentStore', () => { + let db: Db | undefined + + afterEach(async () => { + await db?.close() + db = undefined + }) + + async function freshDeps(): Promise<{ + db: Db + conversationStore: ConversationStore + attachmentStore: ThreadAttachmentStore + }> { + db = await createPgliteDb() + await migrate(db) + return { + db, + conversationStore: createConversationStore(db), + attachmentStore: createThreadAttachmentStore(db), + } + } + + it('listByConversationId returns [] for a conversation with no attachments', async () => { + const { conversationStore, attachmentStore } = await freshDeps() + const { conversationId } = await createConversationWithThread(conversationStore) + + expect(await attachmentStore.listByConversationId(conversationId)).toEqual([]) + }) + + it('listByConversationId returns [] for a nonexistent conversation id', async () => { + const { attachmentStore } = await freshDeps() + expect( + await attachmentStore.listByConversationId('00000000-0000-4000-8000-000000000000'), + ).toEqual([]) + }) + + it('insertThreadAttachmentsInTx is a no-op for an empty array', async () => { + const { db, conversationStore, attachmentStore } = await freshDeps() + const { conversationId } = await createConversationWithThread(conversationStore) + + await db.transaction(async (tx) => { + await insertThreadAttachmentsInTx(tx, []) + }) + + expect(await attachmentStore.listByConversationId(conversationId)).toEqual([]) + }) + + it('persists multiple attachments for one thread and reads them back, oldest-first, scoped to their conversation', async () => { + const { db, conversationStore, attachmentStore } = await freshDeps() + const { conversationId, threadId } = await createConversationWithThread(conversationStore) + // A second, unrelated conversation whose attachments must never leak into + // the first conversation's read. + const other = await createConversationWithThread(conversationStore) + + await db.transaction(async (tx) => { + await insertThreadAttachmentsInTx(tx, [ + { + threadId, + filename: 'a.txt', + contentType: 'text/plain', + size: 3, + blobKey: 'mbox/a/a.txt', + }, + { + threadId, + filename: 'b.png', + contentType: 'image/png', + size: 100, + blobKey: 'mbox/b/b.png', + }, + ]) + await insertThreadAttachmentsInTx(tx, [ + { + threadId: other.threadId, + filename: 'other.txt', + contentType: 'text/plain', + size: 1, + blobKey: 'mbox/o/other.txt', + }, + ]) + }) + + // Both inserts above ran inside the SAME transaction, so `created_at` + // (bound to that transaction's `now()`) ties for both rows — the `id` + // tiebreak then decides order, which is not insertion order. Sort by + // filename before asserting so this test doesn't depend on that tie's + // resolution. + const rows = (await attachmentStore.listByConversationId(conversationId)).sort((a, b) => + (a.filename ?? '').localeCompare(b.filename ?? ''), + ) + expect(rows).toHaveLength(2) + expect(rows.map((r) => r.filename)).toEqual(['a.txt', 'b.png']) + expect(rows[0]).toMatchObject({ + threadId, + filename: 'a.txt', + contentType: 'text/plain', + size: 3, + blobKey: 'mbox/a/a.txt', + }) + expect(rows[0].id).toEqual(expect.any(String)) + expect(rows[0].createdAt).toBeInstanceOf(Date) + + const otherRows = await attachmentStore.listByConversationId(other.conversationId) + expect(otherRows).toHaveLength(1) + expect(otherRows[0].filename).toBe('other.txt') + }) + + it('supports a null filename (e.g. an inline image with no Content-Disposition filename)', async () => { + const { db, conversationStore, attachmentStore } = await freshDeps() + const { conversationId, threadId } = await createConversationWithThread(conversationStore) + + await db.transaction(async (tx) => { + await insertThreadAttachmentsInTx(tx, [ + { + threadId, + filename: null, + contentType: 'image/png', + size: 10, + blobKey: 'mbox/x/attachment', + }, + ]) + }) + + const rows = await attachmentStore.listByConversationId(conversationId) + expect(rows).toHaveLength(1) + expect(rows[0].filename).toBeNull() + }) + + it('a rolled-back transaction leaves no attachment row behind', async () => { + const { db, conversationStore, attachmentStore } = await freshDeps() + const { conversationId, threadId } = await createConversationWithThread(conversationStore) + + await expect( + db.transaction(async (tx) => { + await insertThreadAttachmentsInTx(tx, [ + { + threadId, + filename: 'a.txt', + contentType: 'text/plain', + size: 3, + blobKey: 'mbox/a/a.txt', + }, + ]) + throw new Error('simulated abort') + }), + ).rejects.toThrow('simulated abort') + + expect(await attachmentStore.listByConversationId(conversationId)).toEqual([]) + }) +}) diff --git a/src/store/attachments.ts b/src/store/attachments.ts new file mode 100644 index 0000000..7291c41 --- /dev/null +++ b/src/store/attachments.ts @@ -0,0 +1,144 @@ +/** + * `ThreadAttachmentStore` — persistence for inbound attachment blob- + * references (HT-46; specs/mail/inbound-ingestion.md §3's closing paragraph, + * migration 015). + * + * A `thread_attachments` row never carries attachment BYTES — only a + * reference to where the ingest pipeline (`src/mail/ingest.ts`) already + * wrote them in the `BlobStore` (`blob_key`), namespaced + * `//`. This mirrors `src/store/ + * inbound-deliveries.ts`'s style: a small, focused store next to the bigger + * `ConversationStore`, built on the same `Db`/`Queryable` seam. + * + * ## Transaction-scoped insert (the ingest write path) + * + * {@link insertThreadAttachmentsInTx} is deliberately NOT a method on + * {@link ThreadAttachmentStore} — like `markStoredInTx` + * (`src/store/inbound-deliveries.ts`) it takes an externally-supplied + * `Queryable` so `src/mail/ingest.ts` can insert these rows inside the SAME + * step-5 transaction as the thread it references (specs/mail/ + * inbound-ingestion.md §4: the store write and the ledger mark are one + * atomic unit; the attachment references join that same unit — a + * `thread_attachments` row can never exist for a thread that didn't survive + * the transaction, and never point at bytes that weren't durably written to + * the blob store BEFORE this transaction opened, per that module's doc + * comment). No-op for an empty array — every message has zero or more + * attachments, and zero is the common case. + * + * ## The read path + * + * {@link ThreadAttachmentStore.listByConversationId} is the ONLY read + * method: a single query joined through `threads` so a caller (the Agent + * Inbox API's `GET /api/v1/conversations/{id}`, `src/api/conversations.ts`) + * fetches every attachment for a whole conversation in one round trip, + * rather than one query per thread. Ordered oldest-first (`created_at, id`), + * matching every other list order in this codebase's stores. + */ + +import type { Db, Queryable } from '../db/client.js' + +/** One attachment reference to insert, before its `id`/`createdAt` exist — the ingest pipeline's write shape. */ +export interface NewThreadAttachment { + threadId: string + /** `null` when the attachment arrived with no filename (e.g. an inline image with only a `Content-Id`) — see `ParsedAttachment.filename`, `src/mail/parse.ts`. */ + filename: string | null + contentType: string + /** Size in bytes — `ParsedAttachment.size`. */ + size: number + /** The mailbox-namespaced `BlobStore` key the bytes were already written to (`src/mail/ingest.ts`). Opaque to this store — never interpreted or reconstructed. */ + blobKey: string +} + +/** One `thread_attachments` row as read back from storage — camelCase, timestamps as `Date`. */ +export interface StoredThreadAttachment extends NewThreadAttachment { + id: string + createdAt: Date +} + +/** Persistence operations for inbound attachment blob-references. See the module doc for the write path (a transaction-scoped function, not a method here). */ +export interface ThreadAttachmentStore { + /** + * List every attachment belonging to any thread of conversation + * `conversationId`, oldest-first. `[]` if the conversation has none (the + * common case, or a conversation whose threads carry no attachments) — + * never throws for a missing/empty conversation, since "no attachments" + * and "no such conversation" both correctly resolve to the same empty + * list here (the caller already resolved conversation existence itself). + */ + listByConversationId(conversationId: string): Promise +} + +/** `thread_attachments` columns, `ta.`-qualified for {@link createThreadAttachmentStore}'s joined read query. */ +const ATTACHMENT_COLUMNS = + 'ta.id, ta.thread_id, ta.filename, ta.content_type, ta.size, ta.blob_key, ta.created_at' + +/** Raw `thread_attachments` row shape, before mapping to {@link StoredThreadAttachment}. */ +interface ThreadAttachmentRow { + id: string + thread_id: string + filename: string | null + content_type: string + size: number + blob_key: string + created_at: Date | string +} + +/** + * Transaction-scoped: insert one row per `attachments` entry, all against + * the caller-supplied `tx` — see the module doc's "Transaction-scoped + * insert" section for why this composes with `src/mail/ingest.ts`'s step-5 + * transaction rather than opening its own. A no-op for `[]` (most messages + * have no attachments). + */ +export async function insertThreadAttachmentsInTx( + tx: Queryable, + attachments: NewThreadAttachment[], +): Promise { + for (const attachment of attachments) { + await tx.query( + `INSERT INTO thread_attachments (thread_id, filename, content_type, size, blob_key) + VALUES ($1, $2, $3, $4, $5)`, + [ + attachment.threadId, + attachment.filename, + attachment.contentType, + attachment.size, + attachment.blobKey, + ], + ) + } +} + +/** Create a {@link ThreadAttachmentStore} backed by `db`. Holds no state of its own. */ +export function createThreadAttachmentStore(db: Db): ThreadAttachmentStore { + return { + async listByConversationId(conversationId) { + const rows = await db.query( + `SELECT ${ATTACHMENT_COLUMNS} + FROM thread_attachments ta + JOIN threads t ON t.id = ta.thread_id + WHERE t.conversation_id = $1 + ORDER BY ta.created_at, ta.id`, + [conversationId], + ) + return rows.map(toStoredThreadAttachment) + }, + } +} + +/** Coerce a `timestamptz` column value into a `Date` — see `conversations.ts`'s `toDate` for the same defensive reasoning. */ +function toDate(value: Date | string): Date { + return value instanceof Date ? value : new Date(value) +} + +function toStoredThreadAttachment(row: ThreadAttachmentRow): StoredThreadAttachment { + return { + id: row.id, + threadId: row.thread_id, + filename: row.filename, + contentType: row.content_type, + size: row.size, + blobKey: row.blob_key, + createdAt: toDate(row.created_at), + } +} diff --git a/src/store/index.ts b/src/store/index.ts index 308a195..77cb500 100644 --- a/src/store/index.ts +++ b/src/store/index.ts @@ -2,10 +2,17 @@ * Barrel for the store layer (`src/store/**`) — persistence built on the * raw-SQL seam in `src/db/**`. See `src/store/conversations.ts` for the * `ConversationStore` contract and the storage-layer policy it implements, - * and `src/store/inbound-deliveries.ts` for the `InboundDeliveryStore` - * (inbound delivery ledger) contract. + * `src/store/inbound-deliveries.ts` for the `InboundDeliveryStore` (inbound + * delivery ledger) contract, and `src/store/attachments.ts` for the + * `ThreadAttachmentStore` (inbound attachment blob-reference) contract. */ +export type { + NewThreadAttachment, + StoredThreadAttachment, + ThreadAttachmentStore, +} from './attachments.js' +export { createThreadAttachmentStore, insertThreadAttachmentsInTx } from './attachments.js' export type { AppendResult, ConversationListCursor, From 143fab4ec667c9be3b82dafce76b58d3cbe15e99 Mon Sep 17 00:00:00 2001 From: TJ Baker <1617679+zaridan@users.noreply.github.com> Date: Thu, 16 Jul 2026 13:52:52 -0700 Subject: [PATCH 2/3] fix(mail): sanitize attachment blob-key filenames to an ASCII allowlist (HT-46) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit sanitizeAttachmentFilename only stripped '/' and '\\', so any unicode, '%', '#', quote, or control character in an inbound attachment's filename produced a Supabase Storage key the adapter's server-side validation rejects on every attempt — dead-lettering the whole delivery (body included) after MAX_INGEST_ATTEMPTS. Switch to an allowlist (letters, digits, '_', '.', '-') and treat '' the same as null, since '' ?? 'attachment' let an empty filename attribute through unchanged and produced a key with an empty final segment. Also move ATTACHMENT_SIGNED_URL_EXPIRY_SECONDS above handleGetConversation's doc comment (it had been inserted between the comment and the function, orphaning the doc), and add direct unit coverage for sanitizeAttachmentFilename plus an ingest-level test proving a hostile filename still produces a valid three-segment blob key end to end. Co-Authored-By: Claude Fable 5 --- src/api/conversations.ts | 20 +++++----- src/mail/ingest.test.ts | 83 ++++++++++++++++++++++++++++++++++++++++ src/mail/ingest.ts | 22 ++++++----- 3 files changed, 106 insertions(+), 19 deletions(-) diff --git a/src/api/conversations.ts b/src/api/conversations.ts index 8e4ac73..0821236 100644 --- a/src/api/conversations.ts +++ b/src/api/conversations.ts @@ -177,6 +177,16 @@ export async function handleListConversations( return json(200, body) } +/** + * How long a minted attachment signed URL stays valid (`BlobStore.getSignedUrl`'s + * `expiresInSeconds`, HT-46). One hour: long enough to cover an Agent opening + * the conversation and viewing/downloading an attachment in one sitting, + * short enough that a URL copied out of a stale API response doesn't stay a + * standing credential. Not tuned against any measured usage — a reasonable + * default, re-minted fresh on every `GET` since nothing here caches it. + */ +const ATTACHMENT_SIGNED_URL_EXPIRY_SECONDS = 3600 + /** * Handle `GET /api/v1/conversations/{id}` — fetch the conversation and * shape it as a `ConversationDetail` (spec §3b). `id` is whatever the @@ -194,16 +204,6 @@ export async function handleListConversations( * (see uuid.ts); and the store is asked to exclude deleted rows at the * lookup so no threads are loaded for one (no latency side-channel, §5). */ -/** - * How long a minted attachment signed URL stays valid (`BlobStore.getSignedUrl`'s - * `expiresInSeconds`, HT-46). One hour: long enough to cover an Agent opening - * the conversation and viewing/downloading an attachment in one sitting, - * short enough that a URL copied out of a stale API response doesn't stay a - * standing credential. Not tuned against any measured usage — a reasonable - * default, re-minted fresh on every `GET` since nothing here caches it. - */ -const ATTACHMENT_SIGNED_URL_EXPIRY_SECONDS = 3600 - export async function handleGetConversation( id: string, deps: { diff --git a/src/mail/ingest.test.ts b/src/mail/ingest.test.ts index 6501286..24653e2 100644 --- a/src/mail/ingest.test.ts +++ b/src/mail/ingest.test.ts @@ -10,6 +10,7 @@ import { ingestInboundMessage, isOwnMessageReflection, MAX_INGEST_ATTEMPTS, + sanitizeAttachmentFilename, } from './ingest.js' import type { ParsedEmail } from './parse.js' import { type Keyring, mintReplyMessageId, type SigningKey } from './reply-token.js' @@ -731,6 +732,52 @@ describe('ingestInboundMessage', () => { // --- HT-46: attachment blob persistence. ---------------------------------- + describe('sanitizeAttachmentFilename (pure unit)', () => { + /** Exactly the adapter-valid charset `src/providers/adapters/supabase-storage/` accepts in an object key segment. */ + const ADAPTER_SAFE = /^[A-Za-z0-9._-]+$/ + + it('leaves a plain ASCII filename untouched', () => { + expect(sanitizeAttachmentFilename('hello.txt')).toBe('hello.txt') + }) + + it('null and empty-string filenames both fall back to the fixed placeholder', () => { + // `null` is the "no filename at all" case; `''` is the "client sent an + // empty filename attribute" case — `?? 'attachment'` alone only catches + // the former, which is exactly the must-fix this test guards against. + expect(sanitizeAttachmentFilename(null)).toBe('attachment') + expect(sanitizeAttachmentFilename('')).toBe('attachment') + }) + + it('replaces "/" and "\\" so a crafted filename cannot add a segment inside the blob key', () => { + expect(sanitizeAttachmentFilename('a/b.txt')).toBe('a_b.txt') + expect(sanitizeAttachmentFilename('..\\..\\evil.txt')).toBe('.._.._evil.txt') + expect(sanitizeAttachmentFilename('a/b.txt')).not.toContain('/') + }) + + it('replaces non-ASCII and other adapter-unsafe characters (unicode, "#", "%", quotes, control chars)', () => { + expect(sanitizeAttachmentFilename('Résumé.pdf')).toBe('R_sum_.pdf') + expect(sanitizeAttachmentFilename('a#b%c".txt')).toBe('a_b_c_.txt') + expect(sanitizeAttachmentFilename('ab.txt')).toBe('a_b.txt') + }) + + it('every result matches the adapter-safe charset and is non-empty, for a battery of hostile inputs', () => { + for (const filename of [ + null, + '', + '/', + '\\', + '///', + 'Résumé.pdf', + 'a/b/../c.txt', + '文件.txt', + ]) { + const sanitized = sanitizeAttachmentFilename(filename) + expect(sanitized.length).toBeGreaterThan(0) + expect(sanitized).toMatch(ADAPTER_SAFE) + } + }) + }) + describe('attachments (HT-46)', () => { it('a message with one attachment writes its bytes to the BlobStore and persists exactly one blob-key reference', async () => { const { db, deps, mailboxId } = await freshDeps() @@ -793,6 +840,42 @@ describe('ingestInboundMessage', () => { ) }) + it('a slash-bearing, unicode attachment filename is sanitized in the actual blob key the pipeline writes (not just in the unit-tested sanitizer)', async () => { + const { db, deps, mailboxId } = await freshDeps() + const raw = inboundDelivery( + mailboxId, + 'provider-msg-1', + rawMessageWithAttachments({}, [ + { filename: 'a/../évil.pdf', contentType: 'application/pdf', content: 'bytes' }, + ]), + ) + + const outcome = await ingestInboundMessage(raw, deps) + + expect(outcome.kind).toBe('stored') + if (outcome.kind !== 'stored') throw new Error('unreachable') + + const attachmentStore = createThreadAttachmentStore(db) + const rows = await attachmentStore.listByConversationId(outcome.conversationId) + expect(rows).toHaveLength(1) + // The stored `filename` COLUMN keeps the original, verbatim filename — + // only the blob KEY segment is sanitized. + expect(rows[0].filename).toBe('a/../évil.pdf') + + // The blob key stays exactly three `/`-segments deep — the crafted + // filename cannot add a fourth segment or escape the mailbox/attachment + // namespace — and every segment is non-empty and adapter-safe ASCII. + const segments = rows[0].blobKey.split('/') + expect(segments).toHaveLength(3) + for (const segment of segments) { + expect(segment.length).toBeGreaterThan(0) + expect(segment).toMatch(/^[A-Za-z0-9._-]+$/) + } + + // The bytes are still retrievable at the sanitized key. + expect(new TextDecoder().decode(await deps.blobStore.get(rows[0].blobKey))).toBe('bytes') + }) + it('a message with no attachments persists no thread_attachments rows', async () => { const { db, deps, mailboxId } = await freshDeps() const outcome = await ingestInboundMessage( diff --git a/src/mail/ingest.ts b/src/mail/ingest.ts index 686204a..6ac46f4 100644 --- a/src/mail/ingest.ts +++ b/src/mail/ingest.ts @@ -490,16 +490,20 @@ async function writeAttachmentBlobs( * The filename segment of an attachment's blob key — NOT the `filename` * column value (that stays the original, verbatim `ParsedAttachment.filename`, * `null` included). `BlobStore` implementations (e.g. Supabase Storage, - * `src/providers/adapters/supabase-storage/`) treat `/` in a key as a path - * separator, so an attacker- or client-supplied filename containing `/` - * could otherwise nest the object under an unintended "folder" inside this - * attachment's own namespace slot; stripping it (and any other key-reserved - * character) keeps every attachment's key exactly three segments deep, - * whatever the filename contains. A missing filename falls back to a fixed - * placeholder — the key still needs SOME final segment. + * `src/providers/adapters/supabase-storage/`) reject object keys containing + * anything outside a restricted ASCII allowlist (letters, digits, `_`, `.`, + * `-`) — no unicode, no `/` (a path separator, which would otherwise let an + * attacker- or client-supplied filename nest the object under an unintended + * "folder" inside this attachment's own namespace slot), no `%`/`#`/quotes/ + * control characters. Every other character is replaced with `_` so the key + * stays exactly three segments deep and adapter-valid, whatever the filename + * contains. A missing OR empty filename (`null`, `undefined`, or `''` — a + * blank `''` is not caught by `??`) falls back to a fixed placeholder — the + * key still needs SOME non-empty final segment. */ -function sanitizeAttachmentFilename(filename: string | null): string { - return (filename ?? 'attachment').replaceAll(/[/\\]/g, '_') +export function sanitizeAttachmentFilename(filename: string | null): string { + const sanitized = (filename ?? '').replaceAll(/[^A-Za-z0-9._-]/g, '_') + return sanitized === '' ? 'attachment' : sanitized } /** From 36235b9e2bad7d3ffdc6201cffd7a4fdd85c141c Mon Sep 17 00:00:00 2001 From: TJ Baker <1617679+zaridan@users.noreply.github.com> Date: Thu, 16 Jul 2026 14:35:44 -0700 Subject: [PATCH 3/3] fix(mail,api): adversarial-review fixes on HT-46 (attachment blob persistence) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fix a stray raw NUL byte embedded in src/mail/ingest.test.ts's sanitizeAttachmentFilename control-character test — invisible in most editors/diffs, and enough to make the file read as binary to grep tools that skip binary files by default. Replaced with an explicit \x00 escape. - Parallelize signed-URL minting in attachmentViewsByThreadId (src/api/conversations.ts) — was awaiting BlobStore.getSignedUrl one attachment at a time in a loop; now Promise.all across independent calls. - Cap the sanitized attachment filename segment's length (src/mail/ingest.ts) — an attacker-controlled Content-Disposition filename has no length limit of its own, and an oversized blob-key segment would otherwise fail the same way on every retry. Co-Authored-By: Claude Fable 5 --- src/api/conversations.ts | 17 +++++++++++++---- src/mail/ingest.test.ts | 7 ++++++- src/mail/ingest.ts | 12 ++++++++++-- 3 files changed, 29 insertions(+), 7 deletions(-) diff --git a/src/api/conversations.ts b/src/api/conversations.ts index 0821236..41ab28d 100644 --- a/src/api/conversations.ts +++ b/src/api/conversations.ts @@ -269,12 +269,21 @@ async function attachmentViewsByThreadId( attachments: { store: ThreadAttachmentStore; blobStore: BlobStore }, ): Promise> { const rows = await attachments.store.listByConversationId(conversationId) + // Mint every row's signed URL concurrently (independent BlobStore calls, + // no shared state) rather than one at a time — a conversation with many + // attachments would otherwise pay one signing round trip per attachment, + // serially, on every GET. + const entries = await Promise.all( + rows.map( + async (row) => + [row.threadId, await toAttachmentViewJson(row, attachments.blobStore)] as const, + ), + ) const byThreadId = new Map() - for (const row of rows) { - const view = await toAttachmentViewJson(row, attachments.blobStore) - const existing = byThreadId.get(row.threadId) + for (const [threadId, view] of entries) { + const existing = byThreadId.get(threadId) if (existing === undefined) { - byThreadId.set(row.threadId, [view]) + byThreadId.set(threadId, [view]) } else { existing.push(view) } diff --git a/src/mail/ingest.test.ts b/src/mail/ingest.test.ts index 24653e2..0080cb0 100644 --- a/src/mail/ingest.test.ts +++ b/src/mail/ingest.test.ts @@ -757,7 +757,12 @@ describe('ingestInboundMessage', () => { it('replaces non-ASCII and other adapter-unsafe characters (unicode, "#", "%", quotes, control chars)', () => { expect(sanitizeAttachmentFilename('Résumé.pdf')).toBe('R_sum_.pdf') expect(sanitizeAttachmentFilename('a#b%c".txt')).toBe('a_b_c_.txt') - expect(sanitizeAttachmentFilename('ab.txt')).toBe('a_b.txt') + // A literal NUL (not an escaped placeholder) previously sat here by + // accident — invisible in most editors/diffs and enough to make this + // file read as binary to tools that sniff for one (e.g. `grep -I`). + // Written as an explicit escape so the control-character case this + // test's name promises is actually legible. + expect(sanitizeAttachmentFilename('a\x00b.txt')).toBe('a_b.txt') }) it('every result matches the adapter-safe charset and is non-empty, for a battery of hostile inputs', () => { diff --git a/src/mail/ingest.ts b/src/mail/ingest.ts index 6ac46f4..3a80fa6 100644 --- a/src/mail/ingest.ts +++ b/src/mail/ingest.ts @@ -499,11 +499,19 @@ async function writeAttachmentBlobs( * stays exactly three segments deep and adapter-valid, whatever the filename * contains. A missing OR empty filename (`null`, `undefined`, or `''` — a * blank `''` is not caught by `??`) falls back to a fixed placeholder — the - * key still needs SOME non-empty final segment. + * key still needs SOME non-empty final segment. Truncated to + * {@link MAX_SANITIZED_FILENAME_LENGTH} characters: `ParsedAttachment.filename` + * comes verbatim from an attacker-controlled `Content-Disposition` header + * with no length limit of its own, and an oversized key segment would + * otherwise make every retry of an ordinary message fail the SAME way at + * the `BlobStore` (most object-storage backends cap total key length). */ +const MAX_SANITIZED_FILENAME_LENGTH = 200 + export function sanitizeAttachmentFilename(filename: string | null): string { const sanitized = (filename ?? '').replaceAll(/[^A-Za-z0-9._-]/g, '_') - return sanitized === '' ? 'attachment' : sanitized + if (sanitized === '') return 'attachment' + return sanitized.slice(0, MAX_SANITIZED_FILENAME_LENGTH) } /**