diff --git a/specs/api/agent-inbox-v1.md b/specs/api/agent-inbox-v1.md index 9403c05..c822859 100644 --- a/specs/api/agent-inbox-v1.md +++ b/specs/api/agent-inbox-v1.md @@ -1,9 +1,10 @@ # Agent Inbox API v1 -Status: accepted (HT-17 reads, HT-18 writes). Helpthread's first public API, designed -**native** — on Helpthread's own domain model, not reverse-engineered from any other -helpdesk's wire format. (It supersedes the earlier `conversations-v1.md` draft, which was -shaped for a FreeScout-consumer cutover that no longer applies — see the project history.) +Status: accepted (HT-17 reads, HT-18 writes, HT-16 send idempotency). Helpthread's first +public API, designed **native** — on Helpthread's own domain model, not reverse-engineered +from any other helpdesk's wire format. (It supersedes the earlier `conversations-v1.md` +draft, which was shaped for a FreeScout-consumer cutover that no longer applies — see the +project history.) ## 1. Purpose @@ -73,11 +74,13 @@ appears, not preemptively. interface ApiError { error: { code: string; message: string } } ``` `code` is a machine-readable slug (`unauthorized`, `not_found`, `validation_failed`, - `method_not_allowed`, `send_failed`, `server_error`); `message` is user-safe and MUST - NEVER contain an internal detail — no stack, no SQL, no upstream body, no id it wasn't - given. HTTP status pairs with `code`: 400 `validation_failed`, 401 `unauthorized`, 404 - `not_found`, 405 `method_not_allowed`, 500 `server_error`, 502 `send_failed` (§4a, the - provider rejected an outbound reply). + `method_not_allowed`, `send_failed`, `retry_in_progress`, `server_error`); `message` is + user-safe and MUST NEVER contain an internal detail — no stack, no SQL, no upstream body, + no id it wasn't given. HTTP status pairs with `code`: 400 `validation_failed`, 401 + `unauthorized`, 404 `not_found`, 405 `method_not_allowed`, 409 `retry_in_progress` (§4a, + HT-16 — a concurrent delivery attempt for the same `Idempotency-Key` already holds the + lease), 500 `server_error`, 502 `send_failed` (§4a, the provider rejected an outbound + reply). - **Unknown routes / methods:** an unmatched path is `404 not_found`; a known path with an unsupported method is `405` (with an `Allow` header). Both still require auth first — an unauthenticated request gets `401` before routing details leak. @@ -116,6 +119,18 @@ is indistinguishable from a nonexistent one to this API, on purpose). ### 4a. `POST /api/v1/conversations/{id}/replies` — the Agent replies +**Header:** `Idempotency-Key` is **REQUIRED** on every call (HT-16) — a non-empty, +caller-chosen string, scoped per-conversation. This is a deliberate breaking change from +the HT-15 shape of this endpoint; it has no external consumer yet (this API is +dogfood-only — CHARTER.md "dogfooded first"), so tightening the contract here has no +compatibility cost. The header is **trimmed of leading/trailing whitespace before any +other check**, so `" key "` and `"key"` are the same idempotency key — a caller whose +client or proxy adds incidental whitespace does not silently get a second send. The +**trimmed** value is what is validated, stored, and passed through to `sendReply`: it +must be non-empty and **at most 255 characters** after trimming. A missing header, a +header that is empty (or all whitespace) after trimming, or a trimmed value over 255 +characters is `400 validation_failed`, checked before the body is parsed. + Body: `{ text: string; html?: string }` — `text` 1–5000 chars, server-enforced; `html` optional. The Agent supplies only the message; every mail header is DERIVED server-side from the conversation, so the client never sets recipients or threading headers: @@ -131,26 +146,53 @@ from the conversation, so the client never sets recipients or threading headers: them (it is outbound-token-anchored; threading.md §2). Omitted when no prior message-id exists (e.g. an inbound message that arrived without a `Message-ID`). -The handler then calls `sendReply` (`src/mail/send.ts`), which mints the reply token into -the outbound `Message-ID`, persists the outbound thread (`delivery_status` `pending`→`sent`), -and sends via the injected `EmailSender`. +The handler then calls `sendReply` (`src/mail/send.ts`), passing the `Idempotency-Key` value +through. `sendReply` mints the reply token into the outbound `Message-ID` (on a genuinely +new send), persists the outbound thread with a snapshot of its envelope +(`send_envelope`: `to`/`cc`/`subject`/`references`, `sending.md` §3a), and sends via the +injected `EmailSender`. + +**Replay semantics: same key + same conversation = same logical send, never re-diffed +against the body.** If a call reuses a key already recorded against this conversation, the +NEW request's body is irrelevant — the response reflects the ORIGINAL attempt's outcome: + +- If the original attempt already succeeded (`delivery_status: 'sent'`), this call returns + `201` with that SAME `ThreadView` again, WITHOUT invoking the sender a second time. +- If the original attempt is `pending`/`failed`, this call attempts delivery using the + ORIGINAL row's stored `messageId` and `send_envelope` — never the replay call's own + `to`/`subject`/`references`, even if they differ (sending.md §3a's snapshot rule) — after + first claiming that row's delivery lease. +- If the lease could not be claimed (another attempt — a concurrent replay, or the delivery + worker, sending.md §3a — currently holds it), this call sends nothing and returns + `409 retry_in_progress`. Outcomes: -- **`201`** with the created `ThreadView` on success. A reply to a `closed` conversation - **reopens** it (the store's existing append policy). +- **`201`** with the created (or, on a replay after success, the ORIGINAL) `ThreadView`. A + reply to a `closed` conversation **reopens** it (the store's existing append policy) — + only on the call that actually creates the row, not on a replay. +- **`400 validation_failed`** on a missing/empty `Idempotency-Key` header, or a body that + violates the limits. - **`404 not_found`** if the conversation is missing or `deleted` — no message is sent; a - reply token minted before the append resolves is simply discarded (mirrors §3b). -- **`400 validation_failed`** on a body that violates the limits. + reply token minted before the append resolves is simply discarded (mirrors §3b). This + applies even to a KEYED REPLAY of a key whose original attempt already succeeded: if the + conversation has since been deleted, the replay call returns `404`, not the original + `201`. Replay-of-original-outcome does not survive a conversation delete — there is no + mail-safety impact, since the original send already happened regardless of what a later + replay call observes. +- **`409 retry_in_progress`** (HT-16) — the delivery lease for this `Idempotency-Key` is + currently held by another in-flight attempt; nothing was sent by this call. The caller + should retry the SAME key later, not mint a new one (a new key would create an + independent send, defeating the point of the dedup key). - **`502 send_failed`** if the provider rejects the message — nothing was delivered. `sendReply` returns a `send-failed` result (it does not throw): the outbound thread is - left `delivery_status = 'failed'` (a future delivery worker, HT-16, retries it with the - same Message-ID) — or, if even that mark fails, stuck `pending`. The response therefore - says only that the reply *could not be delivered* — never a specific persisted state, - never a raw provider error. This is the one outcome where an undelivered reply is - surfaced to the caller distinctly from an internal error. (Note the asymmetry: once the - provider ACCEPTS the message it is delivered, so a subsequent failure to record `'sent'` - is NOT a `send_failed` — it resolves to `201`, since reporting a delivered message as - failed would invite a resend.) + left `delivery_status = 'failed'` (retryable — by a replay with the same key, or the + delivery worker's sweep, sending.md §3a — with the same Message-ID) — or, if even that + mark fails, stuck `pending`. The response therefore says only that the reply *could not + be delivered* — never a specific persisted state, never a raw provider error. This is the + one outcome where an undelivered reply is surfaced to the caller distinctly from an + internal error. (Note the asymmetry: once the provider ACCEPTS the message it is + delivered, so a subsequent failure to record `'sent'` is NOT a `send_failed` — it resolves + to `201`, since reporting a delivered message as failed would invite a resend.) ### 4b. `PATCH /api/v1/conversations/{id}` — close or reopen diff --git a/specs/mail/sending.md b/specs/mail/sending.md index 942a781..62348de 100644 --- a/specs/mail/sending.md +++ b/specs/mail/sending.md @@ -1,7 +1,7 @@ # Outbound sending & the reply-token lifecycle -Status: accepted (HT-15). Companion to [threading.md](./threading.md) — that spec -decides which conversation an *inbound* message joins; this one covers how an +Status: accepted (HT-15, HT-16). Companion to [threading.md](./threading.md) — that +spec decides which conversation an *inbound* message joins; this one covers how an *outbound* reply is minted, persisted, and sent, and is where the threading model's authority actually originates. @@ -66,6 +66,89 @@ threading handles for one logical message and risk double-sends. The stable `Message-ID` is the idempotency anchor: a provider that de-dupes on `Message-ID` will not double-deliver a retried send. +## 3a. Send idempotency + delivery leasing (HT-16) + +§3's "retries reuse, never re-mint" rule describes what a retry must DO once +one is recognized; this section is how a retry gets recognized and kept safe +under concurrency, closing the increment §5 of the HT-15 version of this spec +left open. + +**Caller-supplied idempotency key, scoped per conversation.** A caller that +needs at-most-once delivery (the Agent Inbox API, `agent-inbox-v1.md` §4a) +supplies an `idempotencyKey` alongside the reply. `src/store/conversations.ts`'s +`appendThread` resolves it as an atomic **get-or-insert**: `INSERT ... ON +CONFLICT (conversation_id, idempotency_key) WHERE idempotency_key IS NOT NULL +DO NOTHING RETURNING *`, falling back to a `SELECT` of the pre-existing row on +conflict — inside the same transaction that holds the conversation row's `FOR +UPDATE` lock, so two callers racing with the identical key on the identical +conversation are serialized rather than double-inserting. Omitting the key is +still legal and unchanged from HT-15: a fresh send every call, no dedup +protection — a deliberate, permanently-tested contract for callers that don't +need it. + +**The envelope is a snapshot, never a recomputation.** Every outbound send +(keyed or not) now persists a `send_envelope` — `{ to, cc?, subject, +references? }` — verbatim at insert. A retry (whether replayed by the +original caller with the same key, or picked up by the delivery worker below) +resends EXACTLY that stored envelope, never re-derives `to`/`subject`/ +`references` from the conversation's current thread list. This matters +because time passes between an attempt and its retry, and inbound mail can +arrive in that gap: recomputing `References` at retry time could silently +absorb a message that wasn't part of the original send, changing what goes +out without anyone deciding it should (CHARTER.md invariant #5). The +persisted snapshot makes a retry byte-identical to the attempt it retries, by +construction. + +**A lease keeps at most one attempt in flight per row.** Before either a +keyed retry or the delivery worker sends a `pending`/`failed` row, it must +first claim the row's delivery lease (`claimThreadForDelivery`: an atomic +`UPDATE ... WHERE claimed_until IS NULL OR claimed_until < now()`). A failed +claim means someone else already holds it; the caller does not send and +reports back accordingly rather than retrying the claim itself. A successful +attempt releases the lease as it marks `sent`/`failed`. This is what makes +"exactly one send in flight per row" hold even when a caller retries the +same key concurrently with the delivery worker sweeping the same row — +**but only if the lease strictly outlives the send it is protecting.** The +lease duration (`DEFAULT_LEASE_MS`, `src/mail/send.ts`) MUST strictly exceed +the worst-case duration of the configured `EmailSender`'s `send()` call; a +send that outlives its own lease can be re-claimed and retried by another +attempt while the original call is still in flight — a genuine concurrent +double-send, not merely a race over which of two callers marks the outcome. +Every `EmailSender` used behind these retry paths must therefore bound its +own call time well below this lease (see §4). + +**Delivery is at-least-once, not at-most-once — and nothing above changes +that.** The idempotency key, the envelope snapshot, and the lease all close +off *spurious* re-sends — a retry racing another retry, or a caller +deliberately replaying — but none of them lets the engine observe what the +provider actually did with a send it already accepted. The residual case +(§3's "sent but unmarked" asymmetry, sharpened): the provider accepts the +message — the customer's mailbox already has it — and then the write that +marks the row `'sent'` fails, so the row remains `pending` with a live, +already-delivered envelope on it. If nothing revisits that row for a while, +it goes stale; once it is stale (and its lease has freed), the delivery +worker's sweep or a keyed replay's claim will find it eligible and re-send +an already-delivered message. The engine has no way to distinguish "crashed +before the provider was ever called" from "the provider was called and +succeeded, but the mark-sent write failed" — both leave the identical +stale `pending` row with a stored envelope, and both are, correctly, +retried. So: **at-least-once is the actual guarantee this system provides. +At-most-once is not something the engine can produce on its own — it holds +only to the extent the `EmailSender` provider de-duplicates on the outbound +`Message-ID`** (§4). + +**The delivery worker (`src/mail/delivery-worker.ts`) is a plain, invocable +sweep function** — `runDeliveryWorker(deps, options?)` — not built on a +queue or scheduler provider (no such adapter exists yet; see §5). One call +selects a bounded batch of eligible rows (`delivery_status = 'failed'`, or +`'pending'` older than a staleness threshold, with a free lease and a stored +envelope — pre-HT-16 rows with no envelope are left for manual handling +rather than guessed at), claims each in turn, and retries it via the exact +same "rebuild `OutboundEmail` from the row, send, mark" helper a keyed +`sendReply` retry uses. Wiring a real schedule around it (Vercel Cron, or a +future `SchedulerProvider` adapter) is deferred — at that point it is a +one-line call to this function, not a rewrite of it. + ## 4. What a sender provider must guarantee The `EmailSender` provider (`src/providers/`) is handed a fully-formed outbound @@ -85,19 +168,49 @@ accept raw MIME; reject any that will not carry `Message-ID` unaltered. The in-repo fake used by the engine tests proves only that `sendReply` *passes* the value to the seam — not that any given adapter preserves it on the wire. -## 5. Scope of the first increment (HT-15) +**Recommended: a provider SHOULD de-duplicate on `Message-ID` (HT-16).** This +is not a precondition the engine requires — at-least-once delivery (§3a) holds +with or without it — but it is not an aside either: it is the one thing +standing between this system's structural at-least-once delivery (§3a) and +true at-most-once delivery from the operator's point of view. Where a +provider does not de-duplicate on the `Message-ID` it is handed verbatim, +the operator is knowingly accepting +at-least-once delivery: the residual "accepted, then unmarked, then +re-sent" case (§3a) will occasionally reach the customer's mailbox twice, +identical down to the `Message-ID`, and nothing in the engine can prevent +that without provider-side dedup. A provider adapter's wire-level contract +test (above) should note whether the provider is known to de-dupe, so this +gap is a documented, deliberate property of a given deployment rather than +a surprise discovered in production. + +**A lease that outlives the provider's `send()` call is a precondition +too.** §3a's lease only holds "at most one attempt in flight per row" if the +provider's `send()` reliably returns well inside the lease window — an +adapter whose HTTP call has no timeout (or one comparable to or longer than +the lease) can outlive its own claim and collide with a re-claimed retry. +See each adapter's own timeout documentation for its bound. + +## 5. Scope Deliberately narrow; each deferral below has a named later home: -- **Synchronous send only** — the persist→send→mark flow runs inline. No queue - or retry worker yet; the `failed` status plus the stable id/`Message-ID` are - the seam a later delivery worker (queue provider, already interfaced) picks up. +- **Synchronous send only** — the persist→send→mark flow runs inline within + one `sendReply` call. Retrying a stuck row is now covered (§3a: a keyed + replay, or the delivery worker's sweep) — what's still deferred is wiring a + real *schedule* around that sweep (Vercel Cron, or a future + `SchedulerProvider` adapter, CHARTER.md §4) — today it is only invoked + directly (e.g. from a test or a manual trigger), never on a timer. - **Reply to an existing conversation only.** Agent-*initiated* brand-new conversations are a separate later flow. - **`In-Reply-To`/`References` are caller-supplied** (from the inbound message being answered). Deriving the full `References` chain from stored threads is a - later refinement. + later refinement. Once persisted into `send_envelope` (§3a) that snapshot is + authoritative for every retry regardless of how it was originally derived. - **A missing or deleted conversation is refused** — the token is minted first (before `appendThread` resolves) and then discarded on refusal; only persistence and sending are skipped, and the sender is never called (mirrors the store's `appendThread` policy; threading.md §5). +- **No cross-conversation or cross-Agent idempotency-key reuse policy.** A key + is scoped to one conversation (§3a); reusing the same string across + different conversations is unrelated and creates independent rows, by + design — there is no global key registry. diff --git a/src/api/conversations.ts b/src/api/conversations.ts index 7a3f1ea..3180a1c 100644 --- a/src/api/conversations.ts +++ b/src/api/conversations.ts @@ -35,6 +35,15 @@ const MIN_REPLY_TEXT_LENGTH = 1 /** Maximum length of a reply's `text` field, server-enforced (spec §4a). */ const MAX_REPLY_TEXT_LENGTH = 5000 +/** + * Maximum length (after trimming) of the `Idempotency-Key` header, server- + * enforced (spec §4a). The key is stored in a DB column and used as half of + * a unique index (`(conversation_id, idempotency_key)`, migration 003) — an + * unbounded caller-supplied string is an unnecessary storage/index-bloat + * surface for a value that only ever needs to be a short opaque token. + */ +const MAX_IDEMPOTENCY_KEY_LENGTH = 255 + /** 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 @@ -193,18 +202,39 @@ export async function handleGetConversation( * derived server-side from the conversation (see {@link deriveReplyHeaders}) * so the client can never set recipients or threading headers. * + * ## `Idempotency-Key` is REQUIRED (HT-16, a deliberate breaking change) + * + * Every call MUST carry a non-empty `Idempotency-Key` header — its absence + * is `400 validation_failed`, checked before the body is even parsed. This + * endpoint is dogfood-only today (CHARTER.md's "dogfooded first"), so + * tightening its contract has no external consumer to break. The header is + * TRIMMED before every other check or use: leading/trailing whitespace never + * makes two callers' "same" key look different, and the TRIMMED value is + * what is checked for emptiness, checked against + * {@link MAX_IDEMPOTENCY_KEY_LENGTH} (255 chars — `400 validation_failed` if + * exceeded), stored, and passed to `sendReply`. A replay of the SAME + * (trimmed) key on the SAME conversation is treated as the SAME logical + * send — never re-diffed against the body — and returns the ORIGINAL + * outcome (`sendReply`'s own replay handling, `src/mail/send.ts`): `201` + * with the original `ThreadView` if that attempt already succeeded, without + * touching the sender again. + * * Outcomes (spec §4a): `201` with the created `ThreadView` on success (a * reply to a `closed` conversation reopens it, via `sendReply` → * `ConversationStore.appendThread`'s existing policy); `404 not_found` if * the conversation is missing or `deleted` (checked BEFORE minting/sending, * and again as a race check on `sendReply`'s own result — see below); - * `400 validation_failed` on a body that violates the limits; `502 - * send_failed` if the provider rejects the message — `sendReply` returns a - * `send-failed` result (it does not throw), the outbound thread is left - * `failed` OR, if even that mark failed, stuck `pending` (`persistedStatus`), - * and nothing was delivered — so the response says only that the reply could - * not be delivered, never a specific persisted state and never a raw provider - * error (spec §4a, §5's user-safe-message rule). + * `400 validation_failed` on a missing `Idempotency-Key` header or a body + * that violates the limits; `409 retry_in_progress` if another attempt with + * the SAME key is already in flight and holds the delivery lease (HT-16; + * `sendReply`'s `retry-in-progress` result) — nothing was sent by THIS + * request; `502 send_failed` if the provider rejects the message — + * `sendReply` returns a `send-failed` result (it does not throw), the + * outbound thread is left `failed` OR, if even that mark failed, stuck + * `pending` (`persistedStatus`), and nothing was delivered — so the response + * says only that the reply could not be delivered, never a specific + * persisted state and never a raw provider error (spec §4a, §5's + * user-safe-message rule). */ export async function handleReply( id: string, @@ -221,6 +251,19 @@ export async function handleReply( return apiError(404, 'not_found', 'No conversation with that id.') } + const rawIdempotencyKey = request.headers.get('Idempotency-Key') + const idempotencyKey = rawIdempotencyKey?.trim() ?? '' + if (idempotencyKey === '') { + return apiError(400, 'validation_failed', 'Idempotency-Key header is required.') + } + if (idempotencyKey.length > MAX_IDEMPOTENCY_KEY_LENGTH) { + return apiError( + 400, + 'validation_failed', + `Idempotency-Key must be at most ${MAX_IDEMPOTENCY_KEY_LENGTH} characters.`, + ) + } + const parsedBody = await parseJsonBody(request) if (!parsedBody.ok) { return apiError(400, 'validation_failed', 'Request body must be valid JSON.') @@ -263,6 +306,7 @@ export async function handleReply( html: replyBody.html, inReplyTo, references, + idempotencyKey, }, { store: deps.store, @@ -281,6 +325,17 @@ export async function handleReply( // this message claims only what is always true: it wasn't delivered. return apiError(502, 'send_failed', 'The reply could not be delivered.') } + if (result.reason === 'retry-in-progress') { + // Another attempt with the SAME Idempotency-Key already holds the + // delivery lease (HT-16) — nothing was sent by THIS request. The + // in-flight attempt is expected to resolve the row on its own; the + // caller should retry the SAME key again later, not mint a new one. + return apiError( + 409, + 'retry_in_progress', + 'A delivery attempt for this Idempotency-Key is already in progress.', + ) + } // conversation-not-found / conversation-deleted — a race: the conversation // went missing/deleted between the header-fetch above and appendThread's // own check. Nothing was sent — mirrors §3b's generic not-found. diff --git a/src/api/index.test.ts b/src/api/index.test.ts index 8419b00..910112b 100644 --- a/src/api/index.test.ts +++ b/src/api/index.test.ts @@ -114,6 +114,37 @@ function patchRaw(path: string, rawBody: string, ...tokenArg: [string | undefine return withJsonBody('PATCH', path, rawBody, tokenArg) } +/** The `Idempotency-Key` most `replyPost` calls use, unless a test overrides it. */ +const DEFAULT_IDEMPOTENCY_KEY = 'test-idempotency-key' + +/** + * Like {@link post}, but for `POST .../replies` (HT-16 requires an + * `Idempotency-Key` header on every call to that route). Defaults to + * {@link DEFAULT_IDEMPOTENCY_KEY}; pass `idempotencyKey: null` to omit the + * header entirely (for exercising the "missing header" 400), or a specific + * string to control replay/collision scenarios. + */ +function replyPost( + path: string, + body: unknown, + options: { idempotencyKey?: string | null } = {}, +): Request { + const key = + options.idempotencyKey === undefined ? DEFAULT_IDEMPOTENCY_KEY : options.idempotencyKey + const headers: Record = { + 'Content-Type': 'application/json', + Authorization: `Bearer ${TOKEN}`, + } + if (key !== null) { + headers['Idempotency-Key'] = key + } + return new Request(`https://x.example.test${path}`, { + method: 'POST', + headers, + body: JSON.stringify(body), + }) +} + describe('createInboxApi', () => { let db: Db | undefined @@ -408,7 +439,7 @@ describe('createInboxApi', () => { const { conversationId } = await store.createConversation(newConversation()) const res = await api( - post(`/api/v1/conversations/${conversationId}/replies`, { text: 'On it!' }), + replyPost(`/api/v1/conversations/${conversationId}/replies`, { text: 'On it!' }), ) expect(res.status).toBe(201) expect(res.headers.get('Cache-Control')).toBe('no-store') @@ -466,7 +497,7 @@ describe('createInboxApi', () => { }) const res = await api( - post(`/api/v1/conversations/${conversationId}/replies`, { text: 'On it!' }), + replyPost(`/api/v1/conversations/${conversationId}/replies`, { text: 'On it!' }), ) expect(res.status).toBe(201) // delivered → success, NOT a 502 that would invite a resend expect(sent).toHaveLength(1) // the email really went out @@ -480,7 +511,7 @@ describe('createInboxApi', () => { ) const res = await api( - post(`/api/v1/conversations/${conversationId}/replies`, { text: 'Following up.' }), + replyPost(`/api/v1/conversations/${conversationId}/replies`, { text: 'Following up.' }), ) expect(res.status).toBe(201) expect(sent[0].subject).toBe('Re: Already replied') @@ -492,7 +523,7 @@ describe('createInboxApi', () => { await setStatus(db, conversationId, 'closed') const res = await api( - post(`/api/v1/conversations/${conversationId}/replies`, { text: 'Reopening.' }), + replyPost(`/api/v1/conversations/${conversationId}/replies`, { text: 'Reopening.' }), ) expect(res.status).toBe(201) @@ -502,7 +533,9 @@ describe('createInboxApi', () => { it('404s for a missing conversation id; the sender is never called', async () => { const { api, sent } = await freshApi() - const res = await api(post(`/api/v1/conversations/${RANDOM_UUID}/replies`, { text: 'Hi' })) + const res = await api( + replyPost(`/api/v1/conversations/${RANDOM_UUID}/replies`, { text: 'Hi' }), + ) expect(res.status).toBe(404) expect(await res.json()).toEqual({ error: { code: 'not_found', message: expect.any(String) }, @@ -515,14 +548,16 @@ describe('createInboxApi', () => { const { conversationId } = await store.createConversation(newConversation()) await setStatus(db, conversationId, 'deleted') - const res = await api(post(`/api/v1/conversations/${conversationId}/replies`, { text: 'Hi' })) + const res = await api( + replyPost(`/api/v1/conversations/${conversationId}/replies`, { text: 'Hi' }), + ) expect(res.status).toBe(404) expect(sent).toHaveLength(0) }) it('404s for a non-UUID-shaped id — never reaches the uuid column', async () => { const { api, sent } = await freshApi() - const res = await api(post('/api/v1/conversations/not-a-uuid/replies', { text: 'Hi' })) + const res = await api(replyPost('/api/v1/conversations/not-a-uuid/replies', { text: 'Hi' })) expect(res.status).toBe(404) expect(sent).toHaveLength(0) }) @@ -551,7 +586,9 @@ describe('createInboxApi', () => { supportAddress: SUPPORT_ADDRESS, }) - const res = await api(post(`/api/v1/conversations/${conversationId}/replies`, { text: 'Hi' })) + const res = await api( + replyPost(`/api/v1/conversations/${conversationId}/replies`, { text: 'Hi' }), + ) expect(res.status).toBe(404) expect(sent).toHaveLength(0) @@ -562,7 +599,7 @@ describe('createInboxApi', () => { const { store, api, sent } = await freshApi() const { conversationId } = await store.createConversation(newConversation()) - const res = await api(post(`/api/v1/conversations/${conversationId}/replies`, {})) + const res = await api(replyPost(`/api/v1/conversations/${conversationId}/replies`, {})) expect(res.status).toBe(400) expect(await res.json()).toEqual({ error: { code: 'validation_failed', message: expect.any(String) }, @@ -575,7 +612,7 @@ describe('createInboxApi', () => { const { conversationId } = await store.createConversation(newConversation()) const res = await api( - post(`/api/v1/conversations/${conversationId}/replies`, { text: 'a'.repeat(5001) }), + replyPost(`/api/v1/conversations/${conversationId}/replies`, { text: 'a'.repeat(5001) }), ) expect(res.status).toBe(400) expect(sent).toHaveLength(0) @@ -585,7 +622,62 @@ describe('createInboxApi', () => { const { store, api, sent } = await freshApi() const { conversationId } = await store.createConversation(newConversation()) - const res = await api(postRaw(`/api/v1/conversations/${conversationId}/replies`, 'not json{')) + // A valid Idempotency-Key header is present so this test isolates the + // JSON-parse failure specifically, not the header check. + const req = postRaw(`/api/v1/conversations/${conversationId}/replies`, 'not json{') + req.headers.set('Idempotency-Key', DEFAULT_IDEMPOTENCY_KEY) + const res = await api(req) + expect(res.status).toBe(400) + expect(await res.json()).toEqual({ + error: { code: 'validation_failed', message: expect.any(String) }, + }) + expect(sent).toHaveLength(0) + }) + + it('missing Idempotency-Key header is 400 validation_failed; the sender is never called', async () => { + const { store, api, sent } = await freshApi() + const { conversationId } = await store.createConversation(newConversation()) + + const res = await api( + replyPost( + `/api/v1/conversations/${conversationId}/replies`, + { text: 'Hi' }, + { idempotencyKey: null }, + ), + ) + expect(res.status).toBe(400) + expect(await res.json()).toEqual({ + error: { code: 'validation_failed', message: expect.any(String) }, + }) + expect(sent).toHaveLength(0) + }) + + it('an empty Idempotency-Key header is also 400 validation_failed', async () => { + const { store, api, sent } = await freshApi() + const { conversationId } = await store.createConversation(newConversation()) + + const res = await api( + replyPost( + `/api/v1/conversations/${conversationId}/replies`, + { text: 'Hi' }, + { idempotencyKey: '' }, + ), + ) + expect(res.status).toBe(400) + expect(sent).toHaveLength(0) + }) + + it('an Idempotency-Key over 255 characters (after trimming) is 400 validation_failed; the sender is never called', async () => { + const { store, api, sent } = await freshApi() + const { conversationId } = await store.createConversation(newConversation()) + + const res = await api( + replyPost( + `/api/v1/conversations/${conversationId}/replies`, + { text: 'Hi' }, + { idempotencyKey: ` ${'a'.repeat(256)} ` }, + ), + ) expect(res.status).toBe(400) expect(await res.json()).toEqual({ error: { code: 'validation_failed', message: expect.any(String) }, @@ -593,12 +685,131 @@ describe('createInboxApi', () => { expect(sent).toHaveLength(0) }) + it('leading/trailing whitespace in Idempotency-Key is trimmed before comparison — a whitespace-padded key and its trimmed twin replay the SAME send (one send only)', async () => { + const { store, api, sent } = await freshApi() + const { conversationId } = await store.createConversation(newConversation()) + const trimmedKey = 'padded-key-replay' + // U+00A0 (NO-BREAK SPACE), not a plain ASCII space/tab: the WHATWG + // `Headers` implementation already strips ORDINARY HTTP optional + // whitespace (space/tab) from a header value before this handler ever + // sees it, so padding with plain spaces would pass even without our + // own `.trim()`. NBSP is whitespace to JS's `String.prototype.trim()` + // but NOT stripped by `Headers`, so this specifically exercises the + // application-level trim this fix adds. + const paddedKey = `\u00A0${trimmedKey}\u00A0` + + const first = await api( + replyPost( + `/api/v1/conversations/${conversationId}/replies`, + { text: 'On it!' }, + { idempotencyKey: paddedKey }, + ), + ) + expect(first.status).toBe(201) + const firstBody = await first.json() + + // The replay supplies the SAME logical key with no padding at all — it + // must be recognized as the identical key, not a distinct one, so the + // sender is not invoked a second time. + const second = await api( + replyPost( + `/api/v1/conversations/${conversationId}/replies`, + { text: 'A completely different message' }, + { idempotencyKey: trimmedKey }, + ), + ) + expect(second.status).toBe(201) + const secondBody = await second.json() + + expect(secondBody).toEqual(firstBody) + expect(sent).toHaveLength(1) // the sender was invoked exactly once, for the FIRST call + }) + + it('replay of a sent reply: SAME key on the SAME conversation returns 201 with the ORIGINAL ThreadView, sender not re-invoked', async () => { + const { store, api, sent } = await freshApi() + const { conversationId } = await store.createConversation(newConversation()) + const key = 'reply-replay-key' + + const first = await api( + replyPost( + `/api/v1/conversations/${conversationId}/replies`, + { text: 'On it!' }, + { idempotencyKey: key }, + ), + ) + expect(first.status).toBe(201) + const firstBody = await first.json() + + // The replay deliberately supplies a DIFFERENT body — same key, same + // conversation is treated as the SAME logical send; the body is never + // re-diffed. + const second = await api( + replyPost( + `/api/v1/conversations/${conversationId}/replies`, + { text: 'A completely different message' }, + { idempotencyKey: key }, + ), + ) + expect(second.status).toBe(201) + const secondBody = await second.json() + + expect(secondBody).toEqual(firstBody) + expect(sent).toHaveLength(1) // the sender was invoked exactly once, for the FIRST call + }) + + it('replay while a delivery attempt for the same key is in progress is 409 retry_in_progress', async () => { + db = await createPgliteDb() + await migrate(db) + const realStore = createConversationStore(db) + const { conversationId } = await realStore.createConversation(newConversation()) + const key = 'leased-key' + + // Seed a 'failed' row under this key directly via the store, then hold + // its delivery lease — simulating another in-flight attempt (a worker + // sweep, or a concurrent request) currently sending it. + const seeded = await realStore.appendThread(conversationId, { + id: '11111111-1111-4111-8111-111111111111', + direction: 'outbound', + messageId: '', + fromAddress: SUPPORT_ADDRESS, + bodyText: 'On it!', + deliveryStatus: 'failed', + idempotencyKey: key, + sendEnvelope: { to: ['customer@example.test'], subject: 'Re: Help with my order' }, + }) + if (!seeded.ok) throw new Error('unreachable') + await realStore.claimThreadForDelivery(seeded.threadId, 30_000) + + const { sender, sent } = createFakeSender() + const api = createInboxApi({ + store: realStore, + apiToken: TOKEN, + sender, + keyring: KEYRING, + mailDomain: MAIL_DOMAIN, + supportAddress: SUPPORT_ADDRESS, + }) + + const res = await api( + replyPost( + `/api/v1/conversations/${conversationId}/replies`, + { text: 'On it!' }, + { idempotencyKey: key }, + ), + ) + expect(res.status).toBe(409) + expect(await res.json()).toEqual({ + error: { code: 'retry_in_progress', message: expect.any(String) }, + }) + expect(sent).toHaveLength(0) + }) + it('502s when the EmailSender throws; the outbound thread persists with deliveryStatus "failed"', async () => { const { store, api } = await freshApi({ sender: createThrowingSender() }) const { conversationId } = await store.createConversation(newConversation()) const res = await api( - post(`/api/v1/conversations/${conversationId}/replies`, { text: 'On it!' }), + replyPost(`/api/v1/conversations/${conversationId}/replies`, { text: 'On it!' }), ) expect(res.status).toBe(502) const body = await res.json() diff --git a/src/db/migrate.test.ts b/src/db/migrate.test.ts index 3ae121d..39461a8 100644 --- a/src/db/migrate.test.ts +++ b/src/db/migrate.test.ts @@ -42,6 +42,7 @@ describe('migrate', () => { expect(rows).toEqual([ { id: 1, name: 'conversations_and_threads' }, { id: 2, name: 'add_thread_delivery_status' }, + { id: 3, name: 'add_thread_send_idempotency' }, ]) }) @@ -51,7 +52,7 @@ describe('migrate', () => { await migrate(db) // must not throw (e.g. "relation already exists") const rows = await db.query<{ id: number }>('SELECT id FROM _migrations ORDER BY id') - expect(rows).toEqual([{ id: 1 }, { id: 2 }]) + expect(rows).toEqual([{ id: 1 }, { id: 2 }, { id: 3 }]) }) it('migration 002 ties delivery_status to direction: inbound must be NULL, outbound must be pending/sent/failed', async () => { @@ -142,4 +143,135 @@ describe('migrate', () => { ), ).rejects.toThrow() }) + + it('migration 003 ties idempotency_key and send_envelope to direction: inbound must be NULL, outbound may carry either', async () => { + db = await createPgliteDb() + await migrate(db) + + const [conversation] = await db.query<{ id: string }>( + 'INSERT INTO conversations (customer_email) VALUES ($1) RETURNING id', + ['customer@example.test'], + ) + + // Outbound with both columns set is legal. + const [outboundRow] = await db.query<{ + idempotency_key: string | null + send_envelope: { to: string[]; subject: string } | null + claimed_until: string | null + }>( + `INSERT INTO threads (conversation_id, direction, from_address, delivery_status, idempotency_key, send_envelope) + VALUES ($1, 'outbound', $2, 'pending', $3, $4) + RETURNING idempotency_key, send_envelope, claimed_until`, + [ + conversation.id, + 'support@example.test', + 'retry-key-1', + JSON.stringify({ to: ['customer@example.test'], subject: 'Re: Help' }), + ], + ) + expect(outboundRow.idempotency_key).toBe('retry-key-1') + expect(outboundRow.send_envelope).toEqual({ + to: ['customer@example.test'], + subject: 'Re: Help', + }) + expect(outboundRow.claimed_until).toBeNull() + + // Outbound with neither column set (the no-key path) is also legal. + const [outboundNoKeyRow] = await db.query<{ idempotency_key: string | null }>( + `INSERT INTO threads (conversation_id, direction, from_address, delivery_status) + VALUES ($1, 'outbound', $2, 'pending') RETURNING idempotency_key`, + [conversation.id, 'support@example.test'], + ) + expect(outboundNoKeyRow.idempotency_key).toBeNull() + + // Inbound may NOT carry an idempotency_key... + await expect( + db.query( + `INSERT INTO threads (conversation_id, direction, from_address, idempotency_key) + VALUES ($1, 'inbound', $2, $3)`, + [conversation.id, 'customer@example.test', 'some-key'], + ), + ).rejects.toThrow() + + // ...nor a send_envelope. + await expect( + db.query( + `INSERT INTO threads (conversation_id, direction, from_address, send_envelope) + VALUES ($1, 'inbound', $2, $3)`, + [conversation.id, 'customer@example.test', JSON.stringify({ to: [], subject: '' })], + ), + ).rejects.toThrow() + }) + + it('migration 003 enforces one idempotency_key per conversation via the partial unique index, but never collides on NULL', async () => { + db = await createPgliteDb() + await migrate(db) + + const [conversation] = await db.query<{ id: string }>( + 'INSERT INTO conversations (customer_email) VALUES ($1) RETURNING id', + ['customer@example.test'], + ) + + await db.query( + `INSERT INTO threads (conversation_id, direction, from_address, delivery_status, idempotency_key) + VALUES ($1, 'outbound', $2, 'pending', 'dup-key')`, + [conversation.id, 'support@example.test'], + ) + + // A second outbound row in the SAME conversation with the SAME key collides. + await expect( + db.query( + `INSERT INTO threads (conversation_id, direction, from_address, delivery_status, idempotency_key) + VALUES ($1, 'outbound', $2, 'pending', 'dup-key')`, + [conversation.id, 'support@example.test'], + ), + ).rejects.toThrow() + + // Two NULL-key outbound rows in the same conversation never collide (the + // partial index excludes NULL keys entirely) — this is the "no key ⇒ no + // dedup protection" contract, enforced at the schema level too. + await db.query( + `INSERT INTO threads (conversation_id, direction, from_address, delivery_status) + VALUES ($1, 'outbound', $2, 'pending')`, + [conversation.id, 'support@example.test'], + ) + await expect( + db.query( + `INSERT INTO threads (conversation_id, direction, from_address, delivery_status) + VALUES ($1, 'outbound', $2, 'pending')`, + [conversation.id, 'support@example.test'], + ), + ).resolves.toBeDefined() + }) + + it('migration 003 upgrades a NON-fresh 002 database with preexisting outbound rows (no backfill needed, does not fail)', async () => { + db = await createPgliteDb() + + // Apply only through migration 002, then write an outbound thread the way + // a pre-003 deployment would have — no idempotency/envelope/lease columns + // yet. + await migrate(db, { throughId: 2 }) + const [conversation] = await db.query<{ id: string }>( + 'INSERT INTO conversations (customer_email) VALUES ($1) RETURNING id', + ['customer@example.test'], + ) + const [outbound] = await db.query<{ id: string }>( + `INSERT INTO threads (conversation_id, direction, from_address, delivery_status) + VALUES ($1, 'outbound', $2, 'pending') RETURNING id`, + [conversation.id, 'support@example.test'], + ) + + // Applying 003 over that existing data must not fail — the new columns + // default to NULL, which satisfies both new CHECK constraints as-is. + await expect(migrate(db)).resolves.toBeUndefined() + + const [row] = await db.query<{ + idempotency_key: string | null + send_envelope: unknown + claimed_until: string | null + }>('SELECT idempotency_key, send_envelope, claimed_until FROM threads WHERE id = $1', [ + outbound.id, + ]) + expect(row).toEqual({ idempotency_key: null, send_envelope: null, claimed_until: null }) + }) }) diff --git a/src/db/migrate.ts b/src/db/migrate.ts index 5d2c4ad..f70a5e1 100644 --- a/src/db/migrate.ts +++ b/src/db/migrate.ts @@ -102,6 +102,76 @@ ALTER TABLE threads ADD CONSTRAINT threads_delivery_status_by_direction CHECK ( ); ` +/** + * Migration 003 — send idempotency + delivery leasing (HT-16). + * + * Three new nullable columns on `threads`, all outbound-only: + * + * - `idempotency_key` — the caller-supplied dedup key (`SendReplyInput.idempotencyKey`, + * `src/mail/send.ts`). A retry that supplies the SAME key on the SAME + * conversation must find the row `appendThread` already created for the + * first attempt — never mint a second thread/`Message-ID` for one logical + * send. `threads_conversation_idempotency_key_idx` is what makes that + * lookup atomic: a PARTIAL unique index (predicate `idempotency_key IS NOT + * NULL`) so it only constrains rows that opted into dedup — every row + * with a `NULL` key (every inbound thread, and any outbound thread sent + * without a key) is invisible to it and never collides with another + * `NULL`. `src/store/conversations.ts`'s `appendThread` targets this exact + * index with `INSERT ... ON CONFLICT (conversation_id, idempotency_key) + * WHERE idempotency_key IS NOT NULL DO NOTHING RETURNING *`, then falls + * back to a `SELECT` of the pre-existing row on a conflict (0 rows + * returned) — the "atomic get-or-insert" the store module doc describes. + * - `send_envelope` — a `jsonb` snapshot of `{ to, cc?, subject, references? + * }`, written ONCE at insert and read back verbatim on every retry + * (worker-driven or caller-replayed). **This is deliberately a snapshot, + * not a recomputation.** A retry must resend the EXACT envelope the first + * attempt would have sent — recomputing `references` from the + * conversation's CURRENT thread list would let mail that arrived *between* + * the original attempt and the retry silently change what the retry + * sends, which is exactly the kind of silent mail-semantics drift + * CHARTER.md invariant #5 forbids. Persisted for every outbound send + * (keyed or not) so the delivery worker (`src/mail/delivery-worker.ts`) + * can rebuild any eligible row's `OutboundEmail` uniformly, without caring + * whether the original call carried a dedup key. + * - `claimed_until` — a lease: a worker or a keyed-retry `sendReply` call + * "claims" a row by setting this to a near-future timestamp (`UPDATE ... + * WHERE claimed_until IS NULL OR claimed_until < now()`, an ordinary + * Postgres row-level-locked `UPDATE`, so two concurrent claimants can + * never both win), attempts delivery, then clears it back to `NULL` when + * marking `sent`/`failed`. Kept as its own nullable column, separate from + * `delivery_status`, precisely so the existing three-value + * `delivery_status` contract (`StoredThread`, the wire `ThreadView`, + * specs/api/agent-inbox-v1.md §2) is untouched — a lease is a NEW axis + * ("is anyone attempting this right now"), not a fourth delivery state. + * + * No backfill step is needed here (unlike migration 002): all three columns + * are nullable with no `NOT NULL`/CHECK that a pre-existing row could + * violate by defaulting to `NULL` — an inbound row and a pre-HT-16 outbound + * row both get `NULL` for all three and satisfy every constraint below + * as-is. + * + * The two CHECK constraints below mirror migration 002's cross-column style + * and its NULL-semantics care: `(direction = 'outbound') OR ( IS + * NULL)` is TRUE for every inbound row with a NULL column (the only legal + * inbound state) and for every outbound row regardless of the column's value + * (outbound may or may not carry one) — and, critically, is a plain boolean + * OR of two independently-evaluable booleans, so there is no "NULL makes the + * whole CHECK vacuously pass" trap the way an un-guarded `IN (...)` has + * (migration 002's comment explains that trap in full). + */ +const MIGRATION_003_SEND_IDEMPOTENCY = ` +ALTER TABLE threads ADD COLUMN idempotency_key text; +ALTER TABLE threads ADD COLUMN send_envelope jsonb; +ALTER TABLE threads ADD COLUMN claimed_until timestamptz; +ALTER TABLE threads ADD CONSTRAINT threads_idempotency_key_outbound_only CHECK ( + (direction = 'outbound') OR (idempotency_key IS NULL) +); +ALTER TABLE threads ADD CONSTRAINT threads_send_envelope_outbound_only CHECK ( + (direction = 'outbound') OR (send_envelope IS NULL) +); +CREATE UNIQUE INDEX threads_conversation_idempotency_key_idx ON threads (conversation_id, idempotency_key) WHERE idempotency_key IS NOT NULL; +` + /** * 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 @@ -114,6 +184,11 @@ const MIGRATIONS: Migration[] = [ name: 'add_thread_delivery_status', sql: MIGRATION_002_ADD_THREAD_DELIVERY_STATUS, }, + { + id: 3, + name: 'add_thread_send_idempotency', + sql: MIGRATION_003_SEND_IDEMPOTENCY, + }, ] /** diff --git a/src/mail/delivery-worker.test.ts b/src/mail/delivery-worker.test.ts new file mode 100644 index 0000000..bbabba1 --- /dev/null +++ b/src/mail/delivery-worker.test.ts @@ -0,0 +1,366 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { createPgliteDb, type Db } from '../db/client.js' +import { migrate } from '../db/migrate.js' +import type { EmailSender, OutboundEmail } from '../providers/index.js' +import { + type ConversationStore, + createConversationStore, + type SendEnvelope, +} from '../store/conversations.js' +import { runDeliveryWorker } from './delivery-worker.js' +import type { Keyring, SigningKey } from './reply-token.js' +import { type SendReplyDeps, sendReply } from './send.js' + +// --- fixtures ---------------------------------------------------------------- + +const KEY_A: SigningKey = { keyId: 'k1', secret: 'secret-A-high-entropy-0123456789abcdef' } +const keyring: Keyring = { current: KEY_A } +const mailDomain = 'mail.example.test' + +/** Records every `OutboundEmail` it is asked to send; never fails. */ +function fakeSender(): EmailSender & { sent: OutboundEmail[] } { + const sent: OutboundEmail[] = [] + return { + sent, + async send(email) { + sent.push(email) + return { providerMessageId: 'provider-1' } + }, + } +} + +function envelope(overrides: Partial = {}): SendEnvelope { + return { + to: ['customer@example.test'], + subject: 'Re: Help with my order', + ...overrides, + } +} + +describe('runDeliveryWorker', () => { + let db: Db | undefined + + afterEach(async () => { + await db?.close() + db = undefined + }) + + async function freshStore(): Promise<{ db: Db; store: ConversationStore }> { + db = await createPgliteDb() + await migrate(db) + return { db, store: createConversationStore(db) } + } + + async function seedConversation(store: ConversationStore) { + return store.createConversation({ + subject: 'Help with my order', + customerEmail: 'customer@example.test', + firstMessage: { + direction: 'inbound', + messageId: '', + fromAddress: 'customer@example.test', + bodyText: 'Where is my order?', + }, + }) + } + + async function setCreatedAt(rawDb: Db, threadId: string, createdAt: Date) { + await rawDb.query('UPDATE threads SET created_at = $1 WHERE id = $2', [createdAt, threadId]) + } + + it('sweeps one stale pending row and one failed row: both retried with the ORIGINAL messageId, both end sent', async () => { + const { db: rawDb, store } = await freshStore() + const { conversationId } = await seedConversation(store) + + const stalePending = await store.appendThread(conversationId, { + direction: 'outbound', + messageId: '', + fromAddress: 'support@example.test', + bodyText: 'Looking into it!', + deliveryStatus: 'pending', + sendEnvelope: envelope(), + }) + const failedOne = await store.appendThread(conversationId, { + direction: 'outbound', + messageId: '', + fromAddress: 'support@example.test', + bodyText: 'Second try coming', + deliveryStatus: 'failed', + sendEnvelope: envelope(), + }) + if (!stalePending.ok || !failedOne.ok) throw new Error('unreachable') + await setCreatedAt(rawDb, stalePending.threadId, new Date(Date.now() - 10 * 60_000)) + + const sender = fakeSender() + const report = await runDeliveryWorker( + { store, sender }, + { staleAfterMs: 5 * 60_000, batchSize: 50 }, + ) + + expect(report).toEqual({ attempted: 2, sent: 2, failed: 0, skipped: 0 }) + expect(sender.sent.map((e) => e.messageId).sort()).toEqual( + ['', ''].sort(), + ) + + const conversation = await store.getConversation(conversationId) + const stale = conversation?.threads.find((t) => t.id === stalePending.threadId) + const failed = conversation?.threads.find((t) => t.id === failedOne.threadId) + expect(stale).toMatchObject({ + deliveryStatus: 'sent', + messageId: '', + }) + expect(failed).toMatchObject({ + deliveryStatus: 'sent', + messageId: '', + }) + }) + + it('does not retry a fresh pending row (younger than staleAfterMs)', async () => { + const { store } = await freshStore() + const { conversationId } = await seedConversation(store) + await store.appendThread(conversationId, { + direction: 'outbound', + messageId: '', + fromAddress: 'support@example.test', + bodyText: 'Just sent', + deliveryStatus: 'pending', + sendEnvelope: envelope(), + }) + + const sender = fakeSender() + const report = await runDeliveryWorker({ store, sender }, { staleAfterMs: 5 * 60_000 }) + + expect(report).toEqual({ attempted: 0, sent: 0, failed: 0, skipped: 0 }) + expect(sender.sent).toHaveLength(0) + }) + + it('a row already leased BEFORE the sweep starts is excluded from the listing entirely — never even attempted', async () => { + const { store } = await freshStore() + const { conversationId } = await seedConversation(store) + const failedOne = await store.appendThread(conversationId, { + direction: 'outbound', + messageId: '', + fromAddress: 'support@example.test', + bodyText: 'Second try coming', + deliveryStatus: 'failed', + sendEnvelope: envelope(), + }) + if (!failedOne.ok) throw new Error('unreachable') + + // Someone else (a concurrent keyed sendReply retry, or another worker) + // already holds this row's lease before this sweep's own listing query + // runs — listDeliverableThreads' own WHERE clause excludes it, so it + // never reaches this worker's per-row claim step at all. + await store.claimThreadForDelivery(failedOne.threadId, 30_000) + + const sender = fakeSender() + const report = await runDeliveryWorker({ store, sender }) + + expect(report).toEqual({ attempted: 0, sent: 0, failed: 0, skipped: 0 }) + expect(sender.sent).toHaveLength(0) + }) + + it("skips a candidate claimed by someone else BETWEEN the listing and this worker's own claim attempt (TOCTOU), counting it as skipped rather than attempted", async () => { + const { store: realStore } = await freshStore() + const { conversationId } = await seedConversation(realStore) + const raced = await realStore.appendThread(conversationId, { + direction: 'outbound', + messageId: '', + fromAddress: 'support@example.test', + bodyText: 'retry me', + deliveryStatus: 'failed', + sendEnvelope: envelope(), + }) + if (!raced.ok) throw new Error('unreachable') + + // A store double that reports the row as eligible (as a real listing + // would, since it was still unleased when the sweep started) but whose + // claim always loses — simulating another process winning the race in + // the gap between listDeliverableThreads and claimThreadForDelivery. + const store: ConversationStore = { + ...realStore, + async claimThreadForDelivery() { + return null + }, + } + + const sender = fakeSender() + const report = await runDeliveryWorker({ store, sender }) + + expect(report).toEqual({ attempted: 0, sent: 0, failed: 0, skipped: 1 }) + expect(sender.sent).toHaveLength(0) + }) + + // HT-16 CodeRabbit fix: claimThreadForDelivery now re-checks delivery_status + // (src/store/conversations.ts), not just the lease — this is the worker-level + // regression for that fix, using the REAL claim (not a mocked always-null + // one, unlike the TOCTOU test above) so it actually exercises the store's + // WHERE clause. + it('a row eligible at listing time but delivered (marked "sent") by a concurrent keyed retry before this worker claims it is skipped, never re-sent', async () => { + const { db: rawDb, store: realStore } = await freshStore() + const { conversationId } = await seedConversation(realStore) + const raced = await realStore.appendThread(conversationId, { + direction: 'outbound', + messageId: '', + fromAddress: 'support@example.test', + bodyText: 'retry me', + deliveryStatus: 'failed', + sendEnvelope: envelope(), + }) + if (!raced.ok) throw new Error('unreachable') + + // Listing behaves normally (the row IS eligible — still 'failed' and + // unleased when the sweep starts). Immediately after, simulate a + // concurrent keyed sendReply replay completing delivery of this exact + // row before this worker gets to its own claim call. claimThreadForDelivery + // is the REAL implementation here — the fix, not a mock, is what must + // make this row unclaimable now that it is 'sent'. + const store: ConversationStore = { + ...realStore, + async listDeliverableThreads(options) { + const rows = await realStore.listDeliverableThreads(options) + await rawDb.query( + "UPDATE threads SET delivery_status = 'sent', claimed_until = NULL WHERE id = $1", + [raced.threadId], + ) + return rows + }, + } + + const sender = fakeSender() + const report = await runDeliveryWorker({ store, sender }) + + expect(report).toEqual({ attempted: 0, sent: 0, failed: 0, skipped: 1 }) + expect(sender.sent).toHaveLength(0) + }) + + it('respects batchSize', async () => { + const { store } = await freshStore() + const { conversationId } = await seedConversation(store) + for (let i = 0; i < 3; i++) { + await store.appendThread(conversationId, { + direction: 'outbound', + messageId: ``, + fromAddress: 'support@example.test', + bodyText: 'retry me', + deliveryStatus: 'failed', + sendEnvelope: envelope(), + }) + } + + const sender = fakeSender() + const report = await runDeliveryWorker({ store, sender }, { batchSize: 2 }) + + expect(report).toEqual({ attempted: 2, sent: 2, failed: 0, skipped: 0 }) + }) + + it('a send that still fails on retry is counted as failed and left retryable', async () => { + const { store } = await freshStore() + const { conversationId } = await seedConversation(store) + await store.appendThread(conversationId, { + direction: 'outbound', + messageId: '', + fromAddress: 'support@example.test', + bodyText: 'retry me', + deliveryStatus: 'failed', + sendEnvelope: envelope(), + }) + + const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {}) + const sender: EmailSender = { + async send() { + throw new Error('still down') + }, + } + const report = await runDeliveryWorker({ store, sender }) + + expect(report).toEqual({ attempted: 1, sent: 0, failed: 1, skipped: 0 }) + + const conversation = await store.getConversation(conversationId) + const thread = conversation?.threads.find((t) => t.direction === 'outbound') + expect(thread?.deliveryStatus).toBe('failed') + expect(thread?.claimedUntil).toBeNull() + errorSpy.mockRestore() + }) + + // --- cross-path race: worker vs. a keyed sendReply replay ------------------- + + // (Runs against the single-connection, in-process PGlite used in tests — + // the sender gate below deterministically interleaves the worker sweep and + // the keyed replay at the application level, but both still execute their + // `claimThreadForDelivery` UPDATE against the same single DB connection, + // never two genuinely concurrent ones. This proves the sequential + // claim-while-held logic — whichever caller claims second sees the lease + // and backs off — but NOT true multi-connection atomicity of the + // row-locked `UPDATE`. Real-race coverage waits for a multi-connection + // backend, same caveat as migrate.ts's advisory-lock note.) + it('cross-path race: a stale pending row is contended between runDeliveryWorker and a keyed sendReply replay — exactly one of them sends, the other observes the lease', async () => { + const { db: rawDb, store } = await freshStore() + const { conversationId } = await seedConversation(store) + + // Seed the stale row via a keyed sendReply call whose sender fails, so + // the row ends up 'failed' with a real stored envelope/messageId — the + // shape a genuine retry candidate has in production. + const seedDeps: SendReplyDeps = { + store, + sender: { + async send() { + throw new Error('boom') + }, + }, + keyring, + mailDomain, + } + const seedInput = { + conversationId, + from: 'support@example.test', + to: ['customer@example.test'], + subject: 'Re: Help with my order', + text: "We're looking into it!", + idempotencyKey: 'race-key', + } + const seeded = await sendReply(seedInput, seedDeps) + expect(seeded).toMatchObject({ ok: false, reason: 'send-failed', persistedStatus: 'failed' }) + if (seeded.ok || seeded.reason !== 'send-failed') throw new Error('unreachable') + await setCreatedAt(rawDb, seeded.threadId, new Date(Date.now() - 10 * 60_000)) + + // Now race a worker sweep against a keyed replay. The worker's sender is + // gated so we can force the interleaving deterministically: the worker + // claims the row and is blocked mid-send when the replay call is made. + let releaseWorkerSend: () => void = () => {} + const workerSendGate = new Promise((resolve) => { + releaseWorkerSend = resolve + }) + let workerSendCalls = 0 + const workerSender: EmailSender = { + async send() { + workerSendCalls++ + await workerSendGate + return {} + }, + } + + const workerPromise = runDeliveryWorker( + { store, sender: workerSender }, + { staleAfterMs: 5 * 60_000 }, + ) + await vi.waitFor(() => expect(workerSendCalls).toBe(1)) + + // The worker now holds the lease. A concurrent keyed replay must observe + // it and refuse to send again. + const replaySender = fakeSender() + const replayDeps: SendReplyDeps = { store, sender: replaySender, keyring, mailDomain } + const replayResult = await sendReply(seedInput, replayDeps) + expect(replayResult).toEqual({ ok: false, reason: 'retry-in-progress' }) + expect(replaySender.sent).toHaveLength(0) + + releaseWorkerSend() + const workerReport = await workerPromise + expect(workerReport).toEqual({ attempted: 1, sent: 1, failed: 0, skipped: 0 }) + + const conversation = await store.getConversation(conversationId) + const thread = conversation?.threads.find((t) => t.id === seeded.threadId) + expect(thread?.deliveryStatus).toBe('sent') + expect(thread?.messageId).toBe(seeded.messageId) + }) +}) diff --git a/src/mail/delivery-worker.ts b/src/mail/delivery-worker.ts new file mode 100644 index 0000000..24f45a5 --- /dev/null +++ b/src/mail/delivery-worker.ts @@ -0,0 +1,119 @@ +/** + * The delivery worker (HT-16) — a periodic sweep that retries outbound + * threads still stuck `pending` or `failed`, using the SAME `threadId`/ + * `Message-ID` each row already has (never re-minted; specs/mail/sending.md + * §3, `src/mail/send.ts`'s module doc). + * + * ## A plain sweep function, not a queue/cron adapter + * + * `runDeliveryWorker` is deliberately a plain `async function`, not built on + * `QueueProvider`/`SchedulerProvider` (`src/providers/`) — no such adapter + * exists yet, and CHARTER.md §4's provider-seam discipline is exactly why + * this stays a pure function of its dependencies rather than reaching for a + * platform primitive that isn't wired up. Wiring a real schedule (Vercel Cron + * calling this on an interval, or a future `SchedulerProvider` adapter) is + * deferred to whenever that seam is built — at that point it is a one-line + * call to this function, not a rewrite of it. + * + * ## What one sweep does + * + * 1. `ConversationStore.listDeliverableThreads` selects a batch of eligible + * outbound rows: `delivery_status = 'failed'`, OR `'pending'` older than + * `staleAfterMs` (a young `'pending'` row may just be a normal send still + * in flight elsewhere) — see that method's doc comment for the full + * eligibility rule, including why a row with no stored `send_envelope` + * (pre-HT-16 data) is never included. + * 2. For each candidate, `ConversationStore.claimThreadForDelivery` attempts + * to take its delivery lease. A row can be eligible in the LISTING + * snapshot but already claimed by the time this worker gets to it — by a + * concurrent keyed `sendReply` retry, or another worker sweep — in which + * case the claim returns `null` and this sweep simply skips it; there is + * no retry-the-claim loop here, the next sweep will see it again if it's + * still eligible then. + * 3. A successful claim is handed to `attemptDeliveryOfClaimedThread` + * (`src/mail/send.ts`) — the SAME helper `sendReply`'s own keyed-retry + * path uses — which rebuilds the exact `OutboundEmail` from the row + * (`messageId`, `fromAddress`, `bodyText`/`bodyHtml`, `inReplyTo`, + * `sendEnvelope`), calls the sender, and marks `sent`/`failed` while + * releasing the lease. + */ + +import type { EmailSender } from '../providers/index.js' +import type { ConversationStore } from '../store/conversations.js' +import { attemptDeliveryOfClaimedThread, DEFAULT_LEASE_MS } from './send.js' + +/** Default age a `'pending'` row must reach before this worker considers it stuck rather than merely in flight. */ +const DEFAULT_STALE_AFTER_MS = 5 * 60_000 + +/** Default cap on how many rows one sweep will attempt — a bound on a single invocation's work and blast radius, not a pagination scheme. */ +const DEFAULT_BATCH_SIZE = 50 + +/** Dependencies `runDeliveryWorker` needs, injected so it stays testable against fakes/in-memory stores. */ +export interface DeliveryWorkerDeps { + store: ConversationStore + sender: EmailSender +} + +/** Tuning knobs for one sweep; every field defaults, so `runDeliveryWorker(deps)` alone is a complete, reasonable call. */ +export interface DeliveryWorkerOptions { + /** How old a `'pending'` row must be before it's a retry candidate (default {@link DEFAULT_STALE_AFTER_MS}). */ + staleAfterMs?: number + /** Lease duration held while a candidate is being attempted (default {@link DEFAULT_LEASE_MS}, shared with `sendReply`'s own retry-claim). */ + leaseMs?: number + /** Hard cap on rows attempted in this one call (default {@link DEFAULT_BATCH_SIZE}). */ + batchSize?: number +} + +/** What one `runDeliveryWorker` call did, for logging/observability by whatever schedules it. */ +export interface DeliveryWorkerReport { + /** Candidates for which a delivery was actually attempted (claimed successfully) — `sent + failed`. */ + attempted: number + /** Attempts that ended `delivery_status = 'sent'`. */ + sent: number + /** Attempts that ended `delivery_status = 'failed'` (or, rarely, left `'pending'` because even the mark-failed write failed). */ + failed: number + /** Eligible candidates whose lease could not be claimed (already held by a concurrent attempt) — left for a later sweep. */ + skipped: number +} + +/** + * Run one delivery-retry sweep. See the module doc for the full behavior. + * Never throws for an individual candidate's send failure (that is an + * expected, counted outcome — see {@link DeliveryWorkerReport}); a genuinely + * unexpected fault (e.g. `listDeliverableThreads` itself failing) propagates + * to the caller, same as any other unexpected store error in this codebase. + */ +export async function runDeliveryWorker( + deps: DeliveryWorkerDeps, + options?: DeliveryWorkerOptions, +): Promise { + const staleAfterMs = options?.staleAfterMs ?? DEFAULT_STALE_AFTER_MS + const leaseMs = options?.leaseMs ?? DEFAULT_LEASE_MS + const batchSize = options?.batchSize ?? DEFAULT_BATCH_SIZE + + const candidates = await deps.store.listDeliverableThreads({ staleAfterMs, batchSize }) + + let sent = 0 + let failed = 0 + let skipped = 0 + + for (const candidate of candidates) { + const claimed = await deps.store.claimThreadForDelivery(candidate.id, leaseMs) + if (claimed === null) { + skipped++ + continue + } + + const result = await attemptDeliveryOfClaimedThread(claimed, { + store: deps.store, + sender: deps.sender, + }) + if (result.ok) { + sent++ + } else { + failed++ + } + } + + return { attempted: sent + failed, sent, failed, skipped } +} diff --git a/src/mail/send.test.ts b/src/mail/send.test.ts index e75f788..fd00fcb 100644 --- a/src/mail/send.test.ts +++ b/src/mail/send.test.ts @@ -297,3 +297,269 @@ describe('sendReply', () => { expect(sendSpy).not.toHaveBeenCalled() }) }) + +// --- idempotency (HT-16) ----------------------------------------------------- + +describe('sendReply idempotency (HT-16)', () => { + let db: Db | undefined + + afterEach(async () => { + await db?.close() + db = undefined + }) + + async function freshStore(): Promise<{ db: Db; store: ConversationStore }> { + db = await createPgliteDb() + await migrate(db) + return { db, store: createConversationStore(db) } + } + + async function seedConversation(store: ConversationStore) { + return store.createConversation({ + subject: 'Help with my order', + customerEmail: 'customer@example.test', + firstMessage: { + direction: 'inbound', + messageId: '', + fromAddress: 'customer@example.test', + bodyText: 'Where is my order?', + }, + }) + } + + it('regression pin: two sendReply calls with NO idempotencyKey are two independent sends with distinct threadIds — by design, permanent', async () => { + const { store } = await freshStore() + const { conversationId } = await seedConversation(store) + const sender = fakeSender() + const deps: SendReplyDeps = { store, sender, keyring, mailDomain } + const input = { + conversationId, + from: 'support@example.test', + to: ['customer@example.test'], + subject: 'Re: Help with my order', + text: "We're looking into it!", + } + + const first = await sendReply(input, deps) + const second = await sendReply(input, deps) + + expect(first.ok).toBe(true) + expect(second.ok).toBe(true) + if (!first.ok || !second.ok) throw new Error('unreachable') + expect(second.threadId).not.toBe(first.threadId) + expect(second.messageId).not.toBe(first.messageId) + expect(sender.sent).toHaveLength(2) + }) + + it('two sendReply calls with the SAME idempotencyKey result in exactly ONE sender.send() call', async () => { + const { store } = await freshStore() + const { conversationId } = await seedConversation(store) + const sender = fakeSender() + const deps: SendReplyDeps = { store, sender, keyring, mailDomain } + const input = { + conversationId, + from: 'support@example.test', + to: ['customer@example.test'], + subject: 'Re: Help with my order', + text: "We're looking into it!", + idempotencyKey: 'same-key', + } + + await sendReply(input, deps) + await sendReply(input, deps) + + expect(sender.sent).toHaveLength(1) + }) + + it('replay after success: the SAME threadId/messageId is returned and the sender is not re-invoked', async () => { + const { store } = await freshStore() + const { conversationId } = await seedConversation(store) + const sender = fakeSender() + const deps: SendReplyDeps = { store, sender, keyring, mailDomain } + const input = { + conversationId, + from: 'support@example.test', + to: ['customer@example.test'], + subject: 'Re: Help with my order', + text: "We're looking into it!", + idempotencyKey: 'replay-key', + } + + const first = await sendReply(input, deps) + const second = await sendReply(input, deps) + + expect(first).toEqual(second) + expect(sender.sent).toHaveLength(1) + }) + + it('replay after failure: failed → sent, messageId byte-identical, and the RESENT envelope matches the ORIGINAL attempt even when the retry call supplies different to/subject/references and a new inbound message arrived in between', async () => { + const { store } = await freshStore() + const { conversationId } = await seedConversation(store) + const deps1: SendReplyDeps = { store, sender: failingSender(), keyring, mailDomain } + + const first = await sendReply( + { + conversationId, + from: 'support@example.test', + to: ['customer@example.test'], + subject: 'Re: Help with my order', + text: "We're looking into it!", + references: [''], + idempotencyKey: 'retry-key-1', + }, + deps1, + ) + expect(first).toMatchObject({ ok: false, reason: 'send-failed', persistedStatus: 'failed' }) + if (first.ok || first.reason !== 'send-failed') throw new Error('unreachable') + + // A new inbound message lands on the conversation BETWEEN the failed + // attempt and the retry — a caller that recomputed References from the + // conversation's current state would now see a longer chain. + await store.appendThread(conversationId, { + direction: 'inbound', + messageId: '', + fromAddress: 'customer@example.test', + bodyText: 'Any update?', + }) + + const sender = fakeSender() + const deps2: SendReplyDeps = { store, sender, keyring, mailDomain } + // The retry deliberately supplies DIFFERENT to/subject/references — this + // must be ignored in favor of the stored snapshot from the first attempt. + const second = await sendReply( + { + conversationId, + from: 'support@example.test', + to: ['someone-else@example.test'], + subject: 'A totally different subject', + text: 'different body', + references: ['', ''], + idempotencyKey: 'retry-key-1', + }, + deps2, + ) + expect(second).toMatchObject({ ok: true, delivery: 'sent' }) + if (!second.ok) throw new Error('unreachable') + expect(second.messageId).toBe(first.messageId) + expect(second.threadId).toBe(first.threadId) + + expect(sender.sent).toHaveLength(1) + expect(sender.sent[0]).toMatchObject({ + messageId: first.messageId, + to: ['customer@example.test'], + subject: 'Re: Help with my order', + references: [''], + }) + + const conversation = await store.getConversation(conversationId) + const outbound = conversation?.threads.find((t) => t.id === first.threadId) + expect(outbound?.deliveryStatus).toBe('sent') + }) + + // (Runs against the single-connection, in-process PGlite used in tests — + // the sender gate below deterministically interleaves the two `sendReply` + // calls at the application level, but the underlying `claimThreadForDelivery` + // UPDATE is still executed by a single DB connection, never by two + // genuinely concurrent ones. This proves the sequential claim-while-held + // logic — a second caller sees the first's lease and backs off — but NOT + // true multi-connection atomicity of the row-locked `UPDATE`. Real-race + // coverage waits for a multi-connection backend, same caveat as + // migrate.ts's advisory-lock note.) + it('concurrency: a second same-key sendReply call made while the first is still in flight observes the lease and never sends', async () => { + const { store } = await freshStore() + const { conversationId } = await seedConversation(store) + + let releaseSend: () => void = () => {} + const sendGate = new Promise((resolve) => { + releaseSend = resolve + }) + let sendCallCount = 0 + const sender: EmailSender = { + async send() { + sendCallCount++ + await sendGate + return {} + }, + } + const deps: SendReplyDeps = { store, sender, keyring, mailDomain } + const input = { + conversationId, + from: 'support@example.test', + to: ['customer@example.test'], + subject: 'Re: Help with my order', + text: "We're looking into it!", + idempotencyKey: 'concurrent-key', + } + + const firstPromise = sendReply(input, deps) + // Wait until the first call has actually reached the (blocked) sender — + // i.e. it has persisted, claimed the lease, and is mid-send. + await vi.waitFor(() => expect(sendCallCount).toBe(1)) + + const second = await sendReply(input, deps) + expect(second).toEqual({ ok: false, reason: 'retry-in-progress' }) + expect(sendCallCount).toBe(1) // the second call never reached the sender + + releaseSend() + const first = await firstPromise + expect(first).toMatchObject({ ok: true, delivery: 'sent' }) + + const conversation = await store.getConversation(conversationId) + expect(conversation?.threads.filter((t) => t.direction === 'outbound')).toHaveLength(1) + }) + + // --- HT-16 CodeRabbit fix: sent-row reclaim double-send --------------------- + // + // CodeRabbit (Major): claimThreadForDelivery's WHERE clause checked only the + // lease, not delivery_status. Interleaving: a keyed sendReply's get-or-insert + // snapshot observes a row as 'pending'/'failed', but by the time it calls + // claimThreadForDelivery, a concurrent attempt has already delivered the + // message and released the lease with 'sent' — the lease is free, so the + // (unfixed) claim would succeed again and attemptDeliveryOfClaimedThread + // would resend an already-delivered message. The fix adds `AND + // delivery_status IN ('pending', 'failed')` to the claim's WHERE clause + // (src/store/conversations.ts) and, on the client side, re-reads the row on + // a failed claim so a genuinely-'sent' row resolves to the same + // success-replay result as the early 'sent' check, not 'retry-in-progress'. + it('reclaim-after-sent: a keyed row that turns "sent" between the get-or-insert snapshot and the claim call resolves as a success replay, not a resend', async () => { + const { store: realStore, db: rawDb } = await freshStore() + const { conversationId } = await seedConversation(realStore) + const sender = fakeSender() + + // A store double whose appendThread behaves exactly like the real one, + // except that — simulating a concurrent same-key attempt (or the + // delivery worker) completing delivery in the gap between this + // get-or-insert snapshot and sendReply's later claim call — it flips the + // row to 'sent' (lease already free) immediately after returning the + // ORIGINAL (still 'pending') snapshot to the caller. + const store: ConversationStore = { + ...realStore, + async appendThread(convId, thread) { + const result = await realStore.appendThread(convId, thread) + if (result.ok) { + await rawDb.query( + "UPDATE threads SET delivery_status = 'sent', claimed_until = NULL WHERE id = $1", + [result.threadId], + ) + } + return result + }, + } + const deps: SendReplyDeps = { store, sender, keyring, mailDomain } + const input = { + conversationId, + from: 'support@example.test', + to: ['customer@example.test'], + subject: 'Re: Help with my order', + text: "We're looking into it!", + idempotencyKey: 'toctou-key', + } + + const result = await sendReply(input, deps) + + expect(result.ok).toBe(true) + if (!result.ok) throw new Error('unreachable') + expect(result.delivery).toBe('sent') + expect(sender.sent).toHaveLength(0) // never re-sent — already delivered + }) +}) diff --git a/src/mail/send.ts b/src/mail/send.ts index 3c0953e..72859d5 100644 --- a/src/mail/send.ts +++ b/src/mail/send.ts @@ -30,30 +30,66 @@ * ## Retries reuse, never re-mint (specs/mail/sending.md §3) * * When the provider `send()` call fails, this function marks the thread - * `'failed'` and returns a `{ reason: 'send-failed' }` result (it does NOT + * `'failed'` and returns a `{ reason: 'send-failed' }` result (it does not * throw — a rejected send is an expected outcome the caller must handle, not * an exception) — it does not swallow the failure, retry inline, or mint a * fresh token. A `failed` (or crash-orphaned `pending`) thread is meant to be - * retried later by a queue worker (not built in this increment — - * specs/mail/sending.md §5) using the SAME `threadId`/`messageId` already on - * the row. Minting a new token per attempt would spray multiple valid - * threading handles for one logical message and risk a provider that - * de-dupes on `Message-ID` failing to catch a double-send. + * retried later using the SAME `threadId`/`messageId` already on the row — + * either by a caller replaying the SAME `Idempotency-Key` (below), or by the + * delivery worker's sweep (`src/mail/delivery-worker.ts`). Minting a fresh + * token per attempt would spray multiple valid threading handles for one + * logical message and risk a provider that de-dupes on `Message-ID` failing + * to catch a double-send. * * Conversely, once the provider ACCEPTS the message, the delivery has * happened — so a subsequent failure to record `'sent'` resolves to a * SUCCESS result, not a failure. Reporting an already-delivered message as * failed would be worse than a stale status row: it would invite a resend. * - * ## Caller responsibility: idempotency is NOT yet handled here (HT-16) + * ## Send idempotency (HT-16) * - * This increment has no idempotency key and no "retry an existing pending/ - * failed thread" path: each `sendReply` call mints a FRESH `threadId`/ - * `Message-ID` and sends. So a caller that retries the same logical reply - * (an HTTP timeout, a double-clicked UI, a queue redelivery) will send a - * SECOND email. Until the delivery-worker increment adds a real dedup key - * (HT-16), callers MUST guarantee at-most-once invocation themselves — - * `sendReply` must not be wired directly behind a retrying transport. + * `SendReplyInput.idempotencyKey` is an OPTIONAL caller-supplied dedup key, + * scoped per-conversation (`ConversationStore.appendThread`'s partial-unique- + * index get-or-insert — see its doc comment and migration 003's). What + * happens next depends on whether one was given and what it finds: + * + * 1. **No key.** The original, pre-HT-16 flow, UNCHANGED: mint, persist + * fresh, send, mark via `setThreadDeliveryStatus`. Two calls with no key + * are two independent sends — this is a deliberate "no key ⇒ no dedup + * protection" contract (see the regression-pinning test in + * `send.test.ts`), not an oversight; callers that need at-most-once + * semantics must supply a key. + * 2. **Key matches a row already `delivery_status: 'sent'`.** A replay after + * success: return that row's original `threadId`/`messageId` as a SUCCESS + * result, WITHOUT calling the sender again. + * 3. **Key matches a `pending`/`failed` row** (freshly inserted by THIS call, + * or found pre-existing from an earlier attempt — both cases converge + * here). The row is CLAIMED (`ConversationStore.claimThreadForDelivery`) + * before any send is attempted, so a concurrent duplicate call with the + * SAME key — or the delivery worker sweeping the same row — cannot also + * send it while this attempt is in flight. If the claim fails, the row is + * re-read to tell WHY: if it is now `'sent'` (someone else's concurrent + * attempt delivered it between this call's get-or-insert snapshot and the + * claim — the same TOCTOU `claimThreadForDelivery`'s `delivery_status` + * re-check closes at the store layer), this resolves to the same + * success-replay result as case 2 above, never a resend. Otherwise + * (someone else genuinely still holds the lease) this resolves to + * `{ reason: 'retry-in-progress' }` — nothing is sent, nothing is + * re-attempted here. If the claim succeeds, delivery is attempted using + * the row's ALREADY-PERSISTED `messageId` and `sendEnvelope` (never + * re-minted, never recomputed — see below), via + * {@link attemptDeliveryOfClaimedThread}, which is the exact helper the + * delivery worker also calls. + * + * The `sendEnvelope` snapshot (`{ to, cc?, subject, references? }`, + * persisted once at insert, `src/store/conversations.ts`'s `SendEnvelope`) + * is what makes a retry's mail byte-identical to the original attempt: it is + * READ BACK verbatim, never recomputed from the conversation's current + * thread list. Recomputing `references` on a retry could silently absorb an + * inbound message that arrived between the original attempt and the retry — + * exactly the kind of undocumented mail-semantics drift CHARTER.md invariant + * #5 forbids. See migration 003's doc comment (`src/db/migrate.ts`) for the + * full argument. * * ## Assumption: ids are canonical * @@ -68,9 +104,40 @@ import { randomUUID } from 'node:crypto' import type { EmailSender } from '../providers/index.js' -import type { ConversationStore } from '../store/conversations.js' +import type { ConversationStore, SendEnvelope, StoredThread } from '../store/conversations.js' import { type Keyring, mintReplyMessageId } from './reply-token.js' +/** + * Default lease duration for a delivery attempt (claim → send → mark). + * Shared as the default for both `sendReply`'s own inline retry-claim and + * `runDeliveryWorker`'s `leaseMs` option (`src/mail/delivery-worker.ts`) — + * one number, one place, rather than two independently-tuned constants for + * what is conceptually the same lease. + * + * ## The invariant this number exists to hold + * + * The lease MUST strictly exceed the worst-case duration of whatever + * `EmailSender.send()` call it is protecting (specs/mail/sending.md §3a, + * §4). A send that outlives its own lease can have its row re-claimed and + * retried by a concurrent caller — a keyed replay, or the delivery worker — + * while the original call is STILL in flight: a genuine double-send, with + * no DB write, crash, or failure anywhere in the picture. This is a + * different (and worse) hole than the "mark-sent write fails" case §3 + * already documents — that one is a single already-delivered send racing a + * *later* retry of a row gone stale; this one is two live `send()` calls + * for the same row overlapping in real time. + * + * `120_000` is chosen to comfortably clear a real provider HTTP call + * (seconds, not minutes) with a wide margin — not tuned against any + * measured worst case, because none has been measured here. Any + * `EmailSender` used behind these retry paths (§4) MUST bound its own + * `send()` call well below this lease — via its own request timeout — so + * this margin is never actually spent. Raising this constant without also + * checking every adapter's timeout against it re-opens the hole it exists + * to close. + */ +export const DEFAULT_LEASE_MS = 120_000 + /** Dependencies `sendReply` needs, injected so it stays testable against fakes/in-memory stores. */ export interface SendReplyDeps { store: ConversationStore @@ -93,6 +160,12 @@ export interface SendReplyInput { inReplyTo?: string /** `References` chain of the inbound message being answered — caller-supplied (specs/mail/sending.md §5). */ references?: string[] + /** + * Optional caller-supplied dedup key (HT-16), scoped per-conversation. See + * the module doc's "Send idempotency" section for the full contract. + * Omitted entirely means no dedup protection — a fresh send every call. + */ + idempotencyKey?: string } /** @@ -103,10 +176,19 @@ export interface SendReplyInput { * initial `appendThread` DB write itself failing), which a caller should let * surface as an internal error. * - * Critically, the three failure shapes are DISTINCT so the caller does not + * Critically, the failure shapes are DISTINCT so the caller does not * conflate them: * - `conversation-not-found` / `conversation-deleted` — refused; nothing was * minted, persisted, or sent. + * - `retry-in-progress` (HT-16) — a keyed call found a `pending`/`failed` row + * but could not claim its delivery lease, AND, on re-reading the row, it is + * genuinely still `pending`/`failed` (someone else already holds the + * lease — another concurrent call with the same key, or the delivery + * worker). Nothing was sent by THIS call; the in-flight attempt is + * expected to resolve the row on its own. If the re-read instead finds the + * row `'sent'`, that is NOT this reason — it resolves to `ok: true` + * instead (see {@link sendReply}'s claim-failure handling), because the + * message already went out. * - `send-failed` — the outbound thread was persisted (`pending`) but the * provider rejected the message, so nothing was delivered. `persistedStatus` * says whether the row was successfully moved to `'failed'` (retryable by a @@ -121,6 +203,7 @@ export interface SendReplyInput { export type SendReplyResult = | { ok: true; threadId: string; messageId: string; delivery: 'sent' } | { ok: false; reason: 'conversation-not-found' | 'conversation-deleted' } + | { ok: false; reason: 'retry-in-progress' } | { ok: false reason: 'send-failed' @@ -131,8 +214,8 @@ export type SendReplyResult = /** * Send a reply to an existing conversation, per the persist→send→mark - * ordering in the module doc. See there for the full ordering and retry - * rationale. + * ordering in the module doc. See there for the full ordering, retry, and + * idempotency-key rationale. * * Refusal (missing or deleted conversation): the token is minted before the * `appendThread` call resolves, then discarded when refusal is detected — @@ -151,6 +234,18 @@ export async function sendReply( keyring, ) + // The envelope snapshot is built from THIS call's inputs and persisted + // verbatim on insert, keyed or not — persisting it unconditionally (not + // only when idempotencyKey is set) is what lets the delivery worker + // reconstruct ANY eligible outbound row later, regardless of whether its + // original send carried a dedup key. + const sendEnvelope: SendEnvelope = { + to: input.to, + ...(input.cc !== undefined ? { cc: input.cc } : {}), + subject: input.subject, + ...(input.references !== undefined ? { references: input.references } : {}), + } + const appended = await store.appendThread(input.conversationId, { id: threadId, direction: 'outbound', @@ -160,6 +255,8 @@ export async function sendReply( bodyText: input.text ?? null, bodyHtml: input.html ?? null, deliveryStatus: 'pending', + idempotencyKey: input.idempotencyKey, + sendEnvelope, }) if (!appended.ok) { @@ -170,6 +267,86 @@ export async function sendReply( } } + if (input.idempotencyKey === undefined) { + // No key: byte-identical to the pre-HT-16 flow. `appended.created` is + // always `true` here (a NULL key can never conflict — see + // ConversationStore.appendThread's doc comment), so there is no + // existing-row case to handle; send fresh and mark via + // setThreadDeliveryStatus, exactly as before this feature existed. + return sendFreshAndMark(threadId, messageId, input, deps) + } + + const { thread } = appended + + if (thread.deliveryStatus === 'sent') { + // Replay after success: return the ORIGINAL outcome. The sender is never + // touched — the message already went out. + return { + ok: true, + threadId: thread.id, + messageId: thread.messageId as string, + delivery: 'sent', + } + } + + // `pending` or `failed` — whether just-created by THIS call or found + // pre-existing from an earlier attempt, both converge here: claim the + // delivery lease before sending, so a concurrent duplicate call (same key) + // or the delivery worker cannot also be sending this row right now. + const claimed = await store.claimThreadForDelivery(thread.id, DEFAULT_LEASE_MS) + if (claimed === null) { + // The claim can fail for two different reasons, and conflating them + // would resurrect the double-send hole the claim's `delivery_status` + // re-check (`ConversationStore.claimThreadForDelivery`'s doc comment) + // exists to close: + // + // (a) someone else genuinely holds the lease right now — the row is + // still `pending`/`failed`, `claimed_until` is in the future. This + // IS `retry-in-progress`. + // (b) the row reached `'sent'` between the snapshot captured above (this + // call's own `appended.thread`) and this claim call — e.g. a + // concurrent same-key call, or the delivery worker, already + // delivered it. The lease is free, but the claim's status re-check + // correctly refuses it. This is NOT "in progress" — it already + // succeeded — so reporting `retry-in-progress` would be a lie that + // could prompt a caller to retry a message that already went out. + // + // Re-reading the thread is the only way to tell these apart; a `'sent'` + // reading resolves to the same success-replay result the early check + // above returns. + const current = await store.getConversation(input.conversationId) + const currentThread = current?.threads.find((t) => t.id === thread.id) + if (currentThread?.deliveryStatus === 'sent') { + return { + ok: true, + threadId: currentThread.id, + messageId: currentThread.messageId as string, + delivery: 'sent', + } + } + return { ok: false, reason: 'retry-in-progress' } + } + + return attemptDeliveryOfClaimedThread(claimed, { store, sender }) +} + +/** + * The original (pre-HT-16) fresh-send flow: send via the provider, then mark + * `sent`/`failed` via `setThreadDeliveryStatus`. Used ONLY for the no-key + * path — kept as its own function (rather than folded into the claimed-row + * helper below) specifically so this code path, and the store method it + * calls, stay untouched: `send.test.ts`'s pre-HT-16 tests override + * `store.setThreadDeliveryStatus` directly to exercise the mark-failed and + * sent-but-mark-fails cases, and must keep working unedited. + */ +async function sendFreshAndMark( + threadId: string, + messageId: string, + input: SendReplyInput, + deps: SendReplyDeps, +): Promise { + const { store, sender } = deps + try { await sender.send({ messageId, @@ -184,11 +361,11 @@ export async function sendReply( }) } catch { // The provider REJECTED the message — nothing was delivered. Move the - // thread to 'failed' so a delivery worker (HT-16) can retry it with the - // SAME threadId/messageId (never re-mint). If even that mark fails, the - // row is stuck 'pending'; report which, so the caller doesn't claim a - // durable 'failed' state that isn't there. Either way delivery did not - // happen, so a caller retry is safe. + // thread to 'failed' so a later retry (a delivery worker, or a keyed + // caller) can retry it with the SAME threadId/messageId (never re-mint). + // If even that mark fails, the row is stuck 'pending'; report which, so + // the caller doesn't claim a durable 'failed' state that isn't there. + // Either way delivery did not happen, so a caller retry is safe. let persistedStatus: 'failed' | 'pending' = 'pending' try { await store.setThreadDeliveryStatus(threadId, 'failed') @@ -206,8 +383,8 @@ export async function sendReply( // best-effort from here: if the mark throws, the email still went out, so we // MUST NOT report a delivery failure (that would prompt a resend of an // already-delivered message — the double-send hole). The row stays 'pending'; - // reconciling that stale status is a delivery-worker concern (HT-16), which - // treats the stable Message-ID as the idempotency anchor rather than blindly + // reconciling that stale status is a delivery-worker concern, which treats + // the stable Message-ID as the idempotency anchor rather than blindly // re-sending a 'pending' row. try { await store.setThreadDeliveryStatus(threadId, 'sent') @@ -219,3 +396,94 @@ export async function sendReply( } return { ok: true, threadId, messageId, delivery: 'sent' } } + +/** + * Attempt delivery of an ALREADY-CLAIMED outbound row, then mark + * `sent`/`failed` and release its lease. Shared by {@link sendReply}'s + * keyed-retry path and `runDeliveryWorker`'s sweep + * (`src/mail/delivery-worker.ts`) — the one place either caller rebuilds an + * `OutboundEmail` from a stored row and calls the sender. + * + * `thread` must already be claimed (`ConversationStore.claimThreadForDelivery` + * having returned it) — this function does not claim it itself, since the + * two callers need to distinguish "claim failed" (report `retry-in-progress` + * / skip this row) from "claim succeeded, now attempt delivery" differently. + * + * Throws if `thread.messageId` or `thread.sendEnvelope` is missing — both are + * set unconditionally by every `sendReply` insert (keyed or not), so a + * legitimately eligible row always has both; a row missing either is not + * something this function should guess how to send (a `listDeliverableThreads` + * caller already filters out `send_envelope IS NULL` rows for the same + * reason — see that store method's doc comment — so this is a defensive + * invariant check, not a path either current caller can hit in practice). + */ +export async function attemptDeliveryOfClaimedThread( + thread: StoredThread, + deps: { store: ConversationStore; sender: EmailSender }, +): Promise< + | { ok: true; threadId: string; messageId: string; delivery: 'sent' } + | { + ok: false + reason: 'send-failed' + threadId: string + messageId: string + persistedStatus: 'failed' | 'pending' + } +> { + const { store, sender } = deps + + if (thread.messageId === null || thread.sendEnvelope === null) { + throw new Error( + `attemptDeliveryOfClaimedThread: outbound thread ${thread.id} is missing messageId or sendEnvelope — cannot rebuild its OutboundEmail`, + ) + } + const messageId = thread.messageId + const envelope = thread.sendEnvelope + + try { + await sender.send({ + messageId, + inReplyTo: thread.inReplyTo ?? undefined, + references: envelope.references, + from: thread.fromAddress, + to: envelope.to, + cc: envelope.cc, + subject: envelope.subject, + text: thread.bodyText ?? undefined, + html: thread.bodyHtml ?? undefined, + }) + } catch { + let persistedStatus: 'failed' | 'pending' = 'pending' + try { + await store.releaseThreadLease(thread.id, 'failed') + persistedStatus = 'failed' + } catch (markErr) { + console.error( + '[attemptDeliveryOfClaimedThread] provider send failed AND marking the thread failed also failed; row left claimed', + markErr, + ) + } + return { ok: false, reason: 'send-failed', threadId: thread.id, messageId, persistedStatus } + } + + try { + await store.releaseThreadLease(thread.id, 'sent') + } catch (markErr) { + // The row stays claimed (lease held) rather than released, but that is + // NOT meaningful protection against a resend — the lease is a fraction + // of `staleAfterMs` (delivery-worker.ts's default: 5 minutes vs. + // `DEFAULT_LEASE_MS`'s 2), so it will have expired long before the + // delivery worker would otherwise reconsider this stale-`pending` row + // anyway. Staying claimed buys, at best, a small head start. The actual + // backstop against double-delivering an already-sent message is the + // `EmailSender` provider de-duplicating on `Message-ID` + // (specs/mail/sending.md §3a, §4) — this log line exists purely so the + // "sent but unmarked" case is observable, not because the claimed state + // meaningfully delays anything. + console.error( + '[attemptDeliveryOfClaimedThread] message was sent but marking it sent failed; row left claimed (delivery still happened; see comment above — this is not a meaningful resend delay)', + markErr, + ) + } + return { ok: true, threadId: thread.id, messageId, delivery: 'sent' } +} diff --git a/src/store/conversations.test.ts b/src/store/conversations.test.ts index 6ec9578..e26155d 100644 --- a/src/store/conversations.test.ts +++ b/src/store/conversations.test.ts @@ -1,7 +1,12 @@ import { afterEach, describe, expect, it } from 'vitest' import { createPgliteDb, type Db } from '../db/client.js' import { migrate } from '../db/migrate.js' -import { createConversationStore, type NewConversation, type NewThread } from './conversations.js' +import { + createConversationStore, + type NewConversation, + type NewThread, + type SendEnvelope, +} from './conversations.js' // --- fixtures ---------------------------------------------------------------- @@ -103,7 +108,7 @@ describe('createConversationStore', () => { conversationId, newThread({ messageId: '' }), ) - expect(result).toEqual({ ok: true, threadId: expect.any(String) }) + expect(result).toMatchObject({ ok: true, threadId: expect.any(String), created: true }) const conversation = await store.getConversation(conversationId) expect(conversation?.threads).toHaveLength(2) @@ -412,4 +417,402 @@ describe('createConversationStore', () => { expect(walked).toEqual(expectedOrder) }) }) + + // --- send idempotency + delivery leasing (HT-16) --------------------------- + + function newEnvelope(overrides: Partial = {}): SendEnvelope { + return { + to: ['customer@example.test'], + subject: 'Re: Help with my order', + ...overrides, + } + } + + /** Directly rewinds a thread's claimed_until into the past — for exercising lease-expiry without a real sleep. */ + async function expireLease(db: Db, threadId: string) { + await db.query("UPDATE threads SET claimed_until = now() - interval '1 second' WHERE id = $1", [ + threadId, + ]) + } + + /** Directly rewinds a thread's created_at — for exercising the delivery worker's "stale pending" window without a real sleep. */ + async function setCreatedAt(db: Db, threadId: string, createdAt: Date) { + await db.query('UPDATE threads SET created_at = $1 WHERE id = $2', [createdAt, threadId]) + } + + describe('appendThread idempotency key (get-or-insert)', () => { + it('a fresh idempotencyKey inserts a new row (created: true) and persists the envelope', async () => { + const { store } = await freshStore() + const { conversationId } = await store.createConversation(newConversation()) + + const result = await store.appendThread( + conversationId, + newThread({ idempotencyKey: 'send-key-1', sendEnvelope: newEnvelope() }), + ) + expect(result).toMatchObject({ ok: true, created: true }) + if (!result.ok) throw new Error('unreachable') + expect(result.thread.idempotencyKey).toBe('send-key-1') + expect(result.thread.sendEnvelope).toEqual(newEnvelope()) + expect(result.thread.claimedUntil).toBeNull() + }) + + it('a repeated idempotencyKey on the SAME conversation finds the existing row (created: false); inserts nothing new', async () => { + const { store } = await freshStore() + const { conversationId } = await store.createConversation(newConversation()) + + const first = await store.appendThread( + conversationId, + newThread({ + messageId: '', + idempotencyKey: 'send-key-1', + sendEnvelope: newEnvelope(), + }), + ) + expect(first).toMatchObject({ ok: true, created: true }) + + // A "retry": same key, deliberately DIFFERENT messageId/envelope to + // prove the store returns the ORIGINAL row rather than the retry's + // (a real caller would never actually vary these, but this is the + // sharpest way to prove get-or-insert never re-inserts or overwrites). + const second = await store.appendThread( + conversationId, + newThread({ + messageId: '', + idempotencyKey: 'send-key-1', + sendEnvelope: newEnvelope({ subject: 'A different subject entirely' }), + }), + ) + expect(second).toMatchObject({ ok: true, created: false }) + if (!first.ok || !second.ok) throw new Error('unreachable') + expect(second.threadId).toBe(first.threadId) + expect(second.thread.messageId).toBe('') + expect(second.thread.sendEnvelope).toEqual(newEnvelope()) + + const conversation = await store.getConversation(conversationId) + const outboundThreads = conversation?.threads.filter((t) => t.direction === 'outbound') + expect(outboundThreads).toHaveLength(1) + }) + + it('a replay (created: false) does not bump updated_at or reopen a closed conversation', async () => { + const { db, store } = await freshStore() + const { conversationId } = await store.createConversation(newConversation()) + await store.appendThread( + conversationId, + newThread({ idempotencyKey: 'send-key-1', sendEnvelope: newEnvelope() }), + ) + await setStatus(db, conversationId, 'closed') + await setUpdatedAt(db, conversationId, new Date('2020-01-01T00:00:00.000Z')) + + const replay = await store.appendThread( + conversationId, + newThread({ idempotencyKey: 'send-key-1', sendEnvelope: newEnvelope() }), + ) + expect(replay).toMatchObject({ ok: true, created: false }) + + const conversation = await store.getConversation(conversationId) + // Still closed, still the old updated_at — a replay is not new activity. + expect(conversation?.status).toBe('closed') + expect(conversation?.updatedAt.getTime()).toBe(new Date('2020-01-01T00:00:00.000Z').getTime()) + }) + + it('DIFFERENT idempotencyKeys on the same conversation each insert their own row', async () => { + const { store } = await freshStore() + const { conversationId } = await store.createConversation(newConversation()) + + const a = await store.appendThread( + conversationId, + newThread({ + messageId: '', + idempotencyKey: 'key-A', + sendEnvelope: newEnvelope(), + }), + ) + const b = await store.appendThread( + conversationId, + newThread({ + messageId: '', + idempotencyKey: 'key-B', + sendEnvelope: newEnvelope(), + }), + ) + expect(a).toMatchObject({ created: true }) + expect(b).toMatchObject({ created: true }) + if (!a.ok || !b.ok) throw new Error('unreachable') + expect(a.threadId).not.toBe(b.threadId) + }) + + // (Runs against the single-connection, in-process PGlite used in tests — + // see createPgliteDb above. A single connection serializes the two + // `appendThread` transactions below rather than truly overlapping them, + // so this proves the sequential claim-while-held logic — the later + // `Promise.all` caller observes the earlier one's conflict correctly — + // but NOT true multi-connection atomicity of the underlying `INSERT ... + // ON CONFLICT`. Real-race coverage (two genuinely concurrent Postgres + // connections racing the same unique index) waits for a multi-connection + // backend in tests, same caveat as migrate.ts's advisory-lock note.) + it('concurrent appendThread calls with the SAME key resolve to exactly one created row', async () => { + const { store } = await freshStore() + const { conversationId } = await store.createConversation(newConversation()) + + const [a, b] = await Promise.all([ + store.appendThread( + conversationId, + newThread({ + messageId: '', + idempotencyKey: 'concurrent-key', + sendEnvelope: newEnvelope(), + }), + ), + store.appendThread( + conversationId, + newThread({ + messageId: '', + idempotencyKey: 'concurrent-key', + sendEnvelope: newEnvelope(), + }), + ), + ]) + expect(a.ok && b.ok).toBe(true) + if (!a.ok || !b.ok) throw new Error('unreachable') + expect(a.threadId).toBe(b.threadId) + // Exactly one of the two calls actually created the row. + expect([a.created, b.created].sort()).toEqual([false, true]) + + const conversation = await store.getConversation(conversationId) + expect(conversation?.threads.filter((t) => t.direction === 'outbound')).toHaveLength(1) + }) + + it('an idempotencyKey is scoped PER CONVERSATION — the same key on a different conversation inserts its own row', async () => { + const { store } = await freshStore() + const { conversationId: convA } = await store.createConversation(newConversation()) + const { conversationId: convB } = await store.createConversation(newConversation()) + + const a = await store.appendThread( + convA, + newThread({ + messageId: '', + idempotencyKey: 'shared-key', + sendEnvelope: newEnvelope(), + }), + ) + const b = await store.appendThread( + convB, + newThread({ + messageId: '', + idempotencyKey: 'shared-key', + sendEnvelope: newEnvelope(), + }), + ) + expect(a).toMatchObject({ created: true }) + expect(b).toMatchObject({ created: true }) + }) + }) + + describe('claimThreadForDelivery / releaseThreadLease', () => { + it('claims an unclaimed outbound thread, setting claimedUntil in the future', async () => { + const { store } = await freshStore() + const { conversationId } = await store.createConversation(newConversation()) + const appended = await store.appendThread(conversationId, newThread()) + if (!appended.ok) throw new Error('unreachable') + + const before = new Date() + const claimed = await store.claimThreadForDelivery(appended.threadId, 30_000) + expect(claimed).not.toBeNull() + if (claimed === null) throw new Error('unreachable') + expect(claimed.claimedUntil).not.toBeNull() + expect((claimed.claimedUntil as Date).getTime()).toBeGreaterThan(before.getTime()) + }) + + it('a second claim attempt while the lease is held returns null', async () => { + const { store } = await freshStore() + const { conversationId } = await store.createConversation(newConversation()) + const appended = await store.appendThread(conversationId, newThread()) + if (!appended.ok) throw new Error('unreachable') + + const first = await store.claimThreadForDelivery(appended.threadId, 30_000) + expect(first).not.toBeNull() + + const second = await store.claimThreadForDelivery(appended.threadId, 30_000) + expect(second).toBeNull() + }) + + it('claiming succeeds again once the previous lease has expired', async () => { + const { db, store } = await freshStore() + const { conversationId } = await store.createConversation(newConversation()) + const appended = await store.appendThread(conversationId, newThread()) + if (!appended.ok) throw new Error('unreachable') + + const first = await store.claimThreadForDelivery(appended.threadId, 30_000) + expect(first).not.toBeNull() + await expireLease(db, appended.threadId) + + const second = await store.claimThreadForDelivery(appended.threadId, 30_000) + expect(second).not.toBeNull() + }) + + it('releaseThreadLease sets delivery_status and clears claimedUntil', async () => { + const { store } = await freshStore() + const { conversationId } = await store.createConversation(newConversation()) + const appended = await store.appendThread(conversationId, newThread()) + if (!appended.ok) throw new Error('unreachable') + await store.claimThreadForDelivery(appended.threadId, 30_000) + + await store.releaseThreadLease(appended.threadId, 'sent') + + const conversation = await store.getConversation(conversationId) + const thread = conversation?.threads.find((t) => t.id === appended.threadId) + expect(thread?.deliveryStatus).toBe('sent') + expect(thread?.claimedUntil).toBeNull() + }) + + it('releaseThreadLease throws for a nonexistent thread id', async () => { + const { store } = await freshStore() + await expect(store.releaseThreadLease(RANDOM_UUID, 'sent')).rejects.toThrow() + }) + + it('claimThreadForDelivery returns null for an inbound thread id (direction-scoped)', async () => { + const { store } = await freshStore() + const { threadId } = await store.createConversation(newConversation()) + expect(await store.claimThreadForDelivery(threadId, 30_000)).toBeNull() + }) + + // --- HT-16 CodeRabbit fix: claim re-checks delivery_status, not just the lease --- + + it('claimThreadForDelivery returns null for a row already marked "sent", even with a free lease (closes the sent-row reclaim double-send)', async () => { + const { store } = await freshStore() + const { conversationId } = await store.createConversation(newConversation()) + const appended = await store.appendThread(conversationId, newThread()) + if (!appended.ok) throw new Error('unreachable') + await store.claimThreadForDelivery(appended.threadId, 30_000) + // releaseThreadLease clears claimed_until in the SAME write that + // records 'sent' — the lease is free, but the row is delivered. + await store.releaseThreadLease(appended.threadId, 'sent') + + const reclaimed = await store.claimThreadForDelivery(appended.threadId, 30_000) + expect(reclaimed).toBeNull() + }) + + it('claimThreadForDelivery still succeeds for a "failed" row with a free lease (retries remain claimable)', async () => { + const { store } = await freshStore() + const { conversationId } = await store.createConversation(newConversation()) + const appended = await store.appendThread(conversationId, newThread()) + if (!appended.ok) throw new Error('unreachable') + await store.claimThreadForDelivery(appended.threadId, 30_000) + await store.releaseThreadLease(appended.threadId, 'failed') + + const reclaimed = await store.claimThreadForDelivery(appended.threadId, 30_000) + expect(reclaimed).not.toBeNull() + }) + + it('claimThreadForDelivery still succeeds for a "pending" row with a free lease (a fresh, never-claimed row remains claimable)', async () => { + const { store } = await freshStore() + const { conversationId } = await store.createConversation(newConversation()) + const appended = await store.appendThread(conversationId, newThread()) + if (!appended.ok) throw new Error('unreachable') + + const claimed = await store.claimThreadForDelivery(appended.threadId, 30_000) + expect(claimed).not.toBeNull() + expect(claimed?.deliveryStatus).toBe('pending') + }) + }) + + describe('listDeliverableThreads', () => { + it('returns a failed row regardless of age', async () => { + const { store } = await freshStore() + const { conversationId } = await store.createConversation(newConversation()) + const appended = await store.appendThread( + conversationId, + newThread({ deliveryStatus: 'failed', sendEnvelope: newEnvelope() }), + ) + if (!appended.ok) throw new Error('unreachable') + + const eligible = await store.listDeliverableThreads({ + staleAfterMs: 5 * 60_000, + batchSize: 50, + }) + expect(eligible.map((t) => t.id)).toContain(appended.threadId) + }) + + it('excludes a fresh pending row but includes a STALE pending row', async () => { + const { db, store } = await freshStore() + const { conversationId } = await store.createConversation(newConversation()) + const fresh = await store.appendThread( + conversationId, + newThread({ messageId: '', sendEnvelope: newEnvelope() }), + ) + const stale = await store.appendThread( + conversationId, + newThread({ messageId: '', sendEnvelope: newEnvelope() }), + ) + if (!fresh.ok || !stale.ok) throw new Error('unreachable') + await setCreatedAt(db, stale.threadId, new Date(Date.now() - 10 * 60_000)) + + const eligible = await store.listDeliverableThreads({ + staleAfterMs: 5 * 60_000, + batchSize: 50, + }) + const ids = eligible.map((t) => t.id) + expect(ids).toContain(stale.threadId) + expect(ids).not.toContain(fresh.threadId) + }) + + it('excludes a row whose lease is currently held', async () => { + const { store } = await freshStore() + const { conversationId } = await store.createConversation(newConversation()) + const appended = await store.appendThread( + conversationId, + newThread({ deliveryStatus: 'failed', sendEnvelope: newEnvelope() }), + ) + if (!appended.ok) throw new Error('unreachable') + await store.claimThreadForDelivery(appended.threadId, 30_000) + + const eligible = await store.listDeliverableThreads({ + staleAfterMs: 5 * 60_000, + batchSize: 50, + }) + expect(eligible.map((t) => t.id)).not.toContain(appended.threadId) + }) + + it('excludes a row with no stored send_envelope (pre-HT-16 data) even if otherwise eligible', async () => { + const { store } = await freshStore() + const { conversationId } = await store.createConversation(newConversation()) + const appended = await store.appendThread( + conversationId, + newThread({ deliveryStatus: 'failed' }), // no sendEnvelope + ) + if (!appended.ok) throw new Error('unreachable') + + const eligible = await store.listDeliverableThreads({ + staleAfterMs: 5 * 60_000, + batchSize: 50, + }) + expect(eligible.map((t) => t.id)).not.toContain(appended.threadId) + }) + + it('respects batchSize as a hard cap, ordered oldest-created_at-first', async () => { + const { db, store } = await freshStore() + const { conversationId } = await store.createConversation(newConversation()) + const ids: string[] = [] + for (let i = 0; i < 3; i++) { + const appended = await store.appendThread( + conversationId, + newThread({ + messageId: ``, + deliveryStatus: 'failed', + sendEnvelope: newEnvelope(), + }), + ) + if (!appended.ok) throw new Error('unreachable') + await setCreatedAt(db, appended.threadId, new Date(2026, 0, i + 1)) + ids.push(appended.threadId) + } + + const eligible = await store.listDeliverableThreads({ + staleAfterMs: 5 * 60_000, + batchSize: 2, + }) + expect(eligible).toHaveLength(2) + expect(eligible.map((t) => t.id)).toEqual([ids[0], ids[1]]) + }) + }) }) diff --git a/src/store/conversations.ts b/src/store/conversations.ts index 581fab4..28dfa66 100644 --- a/src/store/conversations.ts +++ b/src/store/conversations.ts @@ -54,10 +54,51 @@ * * All three are enforced inside a single transaction per {@link appendThread} * call — see its doc comment for the concurrency reasoning. + * + * ## Send idempotency + delivery leasing (HT-16) + * + * Migration 003 adds three outbound-only columns this module now exposes: + * `idempotency_key`, `send_envelope`, and `claimed_until` (see the migration's + * doc comment, `src/db/migrate.ts`, for the full schema-level rationale). + * {@link appendThread} implements the "atomic get-or-insert" a caller-supplied + * idempotency key needs: `INSERT ... ON CONFLICT (conversation_id, + * idempotency_key) WHERE idempotency_key IS NOT NULL DO NOTHING RETURNING + * *`, falling back to a `SELECT` of the pre-existing row when the insert is + * skipped — both inside the SAME transaction that already takes the `FOR + * UPDATE` lock on the conversation row, so a concurrent retry with the same + * key is fully serialized against the original attempt rather than racing + * it. {@link AppendResult}'s `created` flag tells the caller (`src/mail/ + * send.ts`) which case happened: `true` for a fresh insert (no key, or a key + * never seen before), `false` when an existing row was found instead — at + * which point `thread` carries that row's ALREADY-PERSISTED `messageId` and + * `sendEnvelope`, which a retry must reuse verbatim rather than re-minting or + * recomputing (see migration 003's doc comment on why the envelope is a + * snapshot). + * + * {@link ConversationStore.claimThreadForDelivery} and + * {@link ConversationStore.releaseThreadLease} are the lease pair a keyed + * retry or the delivery worker (`src/mail/delivery-worker.ts`) uses to make + * sure at most one in-flight attempt is ever sending a given outbound thread + * at a time — see their own doc comments below. */ import type { Db, Queryable, SqlValue } from '../db/client.js' +/** + * A snapshot of the mail headers an outbound reply was sent with: + * recipients, subject, and the `References` chain. Persisted VERBATIM into + * `threads.send_envelope` at insert and read back unchanged on every retry — + * never recomputed from the conversation's current state (migration 003's + * doc comment explains why: recomputing `references` could silently absorb + * inbound mail that arrived between the original attempt and the retry). + */ +export interface SendEnvelope { + to: string[] + cc?: string[] + subject: string + references?: string[] +} + /** One message to be persisted as a new thread — inbound customer mail, or outbound agent/assistant mail. */ export interface NewThread { /** @@ -93,6 +134,21 @@ export interface NewThread { * received, and the column stays `NULL` for those rows. */ deliveryStatus?: 'pending' | 'sent' | 'failed' | null + /** + * Caller-supplied dedup key for an OUTBOUND thread (HT-16; + * `SendReplyInput.idempotencyKey`, `src/mail/send.ts`). Omitted (or + * `undefined`) means "no dedup protection for this send" — see + * {@link ConversationStore.appendThread}'s doc comment for what that + * means at the storage layer. Never set for an inbound thread — migration + * 003's CHECK constraint rejects that. + */ + idempotencyKey?: string + /** + * A snapshot of this OUTBOUND thread's mail envelope, written once at + * insert (see {@link SendEnvelope}'s doc comment for why it is a snapshot, + * not a live derivation). Never set for an inbound thread. + */ + sendEnvelope?: SendEnvelope } /** Input to {@link ConversationStore.createConversation}: a new conversation plus its first thread. */ @@ -114,6 +170,17 @@ export interface StoredThread { bodyHtml: string | null /** Outbox status — `null` for inbound threads, `'pending'|'sent'|'failed'` for outbound ones. See {@link NewThread.deliveryStatus}. */ deliveryStatus: 'pending' | 'sent' | 'failed' | null + /** Dedup key this OUTBOUND thread was sent with, or `null` if it was sent (or received) without one. See {@link NewThread.idempotencyKey}. */ + idempotencyKey: string | null + /** This OUTBOUND thread's persisted envelope snapshot, or `null` for an inbound thread. See {@link SendEnvelope}. */ + sendEnvelope: SendEnvelope | null + /** + * The delivery lease: non-`null` while a `sendReply` retry or the + * delivery worker is actively attempting this OUTBOUND thread, `null` + * otherwise (never attempted, or the last attempt already released it). + * See {@link ConversationStore.claimThreadForDelivery}. + */ + claimedUntil: Date | null createdAt: Date } @@ -136,7 +203,7 @@ export interface StoredConversation { * callers should handle it as ordinary control flow. */ export type AppendResult = - | { ok: true; threadId: string } + | { ok: true; threadId: string; created: boolean; thread: StoredThread } | { ok: false; reason: 'not-found' | 'deleted' } /** Persistence operations for conversations and their threads. See the module doc for the storage-layer policy this implements. */ @@ -154,11 +221,106 @@ export interface ConversationStore { * closed/deleted/missing policy documented at the top of this module. * See that doc for the full behavior; summarized: missing → `not-found`, * deleted → `deleted` (nothing inserted), closed → inserted AND - * reopened, open → inserted. Any successful insert also bumps the - * conversation's `updated_at`. + * reopened, open → inserted. A genuinely NEW row (`created: true`) also + * bumps the conversation's `updated_at` (and reopens a closed one, per + * the above); a REPLAY that found an existing row instead (`created: + * false`) touches the conversation row not at all — nothing new + * happened, so nothing about the conversation should look like it did. + * + * ## The `idempotencyKey` case: atomic get-or-insert + * + * When `thread.idempotencyKey` is set, this is NOT a plain insert: it is + * `INSERT ... ON CONFLICT (conversation_id, idempotency_key) WHERE + * idempotency_key IS NOT NULL DO NOTHING RETURNING *`, and on a conflict + * (0 rows — this exact key already exists on this conversation) a + * `SELECT` of that pre-existing row, all inside the same transaction that + * takes the `FOR UPDATE` lock on the conversation row above. That lock is + * what makes this safe under concurrency: two callers racing with the + * SAME key on the SAME conversation are serialized by it, so the second + * one's `INSERT ... ON CONFLICT` always sees the first one's already-committed + * row rather than racing its own insert against it. `created` tells the + * caller which happened; `thread` is the row either way — for a replay, + * `thread.messageId` and `thread.sendEnvelope` are the ORIGINAL attempt's, + * never regenerated (see the module doc and migration 003's doc comment). + * + * When `thread.idempotencyKey` is omitted, this behaves exactly as before + * HT-16: a plain insert, `created` is always `true`. This is the "no key ⇒ + * no dedup protection" contract `src/mail/send.ts`'s module doc names + * explicitly — deliberate, and covered by a permanent regression test. */ appendThread(conversationId: string, thread: NewThread): Promise + /** + * Claim `threadId` for delivery: an atomic `UPDATE ... SET claimed_until = + * now() + leaseMs WHERE id = $1 AND (claimed_until IS NULL OR + * claimed_until < now()) AND delivery_status IN ('pending', 'failed') + * RETURNING *`, scoped to outbound rows. Ordinary Postgres row-level + * locking on the `UPDATE` is what makes "at most one claimant wins" hold + * even under true concurrency (two overlapping calls for the same + * `threadId`, from two processes or two `Promise.all`-ed calls in one) — + * no advisory lock or explicit transaction is needed here, a single + * `UPDATE` is already atomic with respect to itself. + * + * The `delivery_status` re-check is not redundant with the lease check: it + * closes a TOCTOU where a row reaches `'sent'` (via `releaseThreadLease`, + * which clears `claimed_until` in the SAME write that records the + * outcome) between whenever a caller last observed it as `'pending'`/ + * `'failed'` and this claim call. Without it, that now-`'sent'` row still + * has a free lease and would be claimed again, and the caller (a keyed + * `sendReply` replay, or `src/mail/delivery-worker.ts`'s sweep) would + * re-send an already-delivered message. Because both checks ride the same + * row-locked `UPDATE`, a row can never be claimed once it is `'sent'` — + * there is no window where the lease is free but the status check hasn't + * "caught up" yet. + * + * Returns the freshly-claimed {@link StoredThread} (with the new + * `claimedUntil`) on success, or `null` if the row is missing, not + * outbound, already `'sent'`, or already claimed by someone else whose + * lease hasn't expired — the caller (`src/mail/send.ts`'s retry path, or + * `src/mail/delivery-worker.ts`'s sweep) must treat `null` as "don't send + * this row right now" and, if it needs to distinguish "already delivered" + * from "genuinely in flight," re-read the row's `delivery_status` itself + * (see `sendReply`'s honest-409 handling). + */ + claimThreadForDelivery(threadId: string, leaseMs: number): Promise + + /** + * Release `threadId`'s delivery lease and record the outcome in one + * write: `UPDATE ... SET delivery_status = status, claimed_until = NULL + * WHERE id = $1 AND direction = 'outbound' RETURNING id`, scoped and + * throwing-on-zero-rows exactly like {@link setThreadDeliveryStatus} (see + * its doc comment for why a silent no-op would be worse than a throw). + * Kept as a SEPARATE method from `setThreadDeliveryStatus` — not a + * parameter that also clears the lease — so the ORIGINAL (pre-HT-16, + * no-idempotency-key) `sendReply` flow keeps calling + * `setThreadDeliveryStatus` completely unchanged, byte-identical to + * before this feature existed. + */ + releaseThreadLease(threadId: string, status: 'sent' | 'failed'): Promise + + /** + * List OUTBOUND threads eligible for a delivery-worker retry sweep + * (`src/mail/delivery-worker.ts`): `delivery_status = 'failed'`, OR + * `delivery_status = 'pending'` AND `created_at` older than + * `options.staleAfterMs` (a `'pending'` row younger than that may simply + * be a normal send still in flight — not yet a candidate); AND the lease + * is free (`claimed_until IS NULL OR claimed_until < now()`); AND + * `send_envelope IS NOT NULL` — a row with no stored envelope (only + * possible for a `threads` row written before migration 003 shipped) + * cannot be safely retried: rebuilding its `to`/`subject`/`references` + * from the conversation's CURRENT state would be exactly the silent + * mail-semantics drift migration 003's envelope snapshot exists to + * prevent, so such a row is left for manual/administrative handling + * instead of a worker guessing at it. Ordered oldest-`created_at`-first, + * capped at `options.batchSize` — the worker's own batch limit, not an + * over-fetch-by-one pagination trick (there is no pagination here; a + * skipped row is simply picked up on the NEXT sweep). + */ + listDeliverableThreads(options: { + staleAfterMs: number + batchSize: number + }): Promise + /** * Read one conversation with all of its threads, ordered oldest-first * (`created_at, id` — the `id` tiebreak makes ordering stable even for @@ -307,7 +469,15 @@ interface ConversationSummaryRow extends ConversationRow { thread_count: number } -/** Raw `threads` row shape, before mapping to {@link StoredThread}. */ +/** + * Raw `threads` row shape, before mapping to {@link StoredThread}. `send_envelope` + * is typed `unknown` at this layer (not `SendEnvelope | null`) because it + * arrives already-parsed from a `jsonb` column (PGlite, verified against the + * installed 0.5.4, decodes `jsonb` to a plain JS value automatically — no + * `JSON.parse` needed on read), but nothing here has actually validated its + * shape; {@link toStoredThread} does the one authoritative cast, since this + * codebase controls every writer of the column (see {@link insertThread}). + */ interface ThreadRow { id: string conversation_id: string @@ -318,11 +488,14 @@ interface ThreadRow { body_text: string | null body_html: string | null delivery_status: string | null + idempotency_key: string | null + send_envelope: unknown + claimed_until: Date | string | null created_at: Date | string } const THREAD_COLUMNS = - 'id, conversation_id, direction, message_id, in_reply_to, from_address, body_text, body_html, delivery_status, created_at' + 'id, conversation_id, direction, message_id, in_reply_to, from_address, body_text, body_html, delivery_status, idempotency_key, send_envelope, claimed_until, created_at' /** * Create a {@link ConversationStore} backed by `db`. Every operation opens @@ -337,7 +510,7 @@ export function createConversationStore(db: Db): ConversationStore { 'INSERT INTO conversations (subject, customer_email) VALUES ($1, $2) RETURNING id', [input.subject, input.customerEmail], ) - const threadId = await insertThread(tx, conversation.id, input.firstMessage) + const { threadId } = await insertThread(tx, conversation.id, input.firstMessage) return { conversationId: conversation.id, threadId } }) }, @@ -349,7 +522,9 @@ export function createConversationStore(db: Db): ConversationStore { // conversation can't race between this status check and the insert // below (e.g. two replies arriving for the same closed conversation // at once should both observe-and-reopen deterministically, not - // interleave into an inconsistent status). + // interleave into an inconsistent status). This same lock is what + // makes the idempotency-key get-or-insert below safe under + // concurrency — see the interface doc comment above. const rows = await tx.query<{ status: string }>( 'SELECT status FROM conversations WHERE id = $1 FOR UPDATE', [conversationId], @@ -362,20 +537,25 @@ export function createConversationStore(db: Db): ConversationStore { return { ok: false, reason: 'deleted' } } - const threadId = await insertThread(tx, conversationId, thread) - - if (row.status === 'closed') { - await tx.query( - "UPDATE conversations SET status = 'open', updated_at = now() WHERE id = $1", - [conversationId], - ) - } else { - await tx.query('UPDATE conversations SET updated_at = now() WHERE id = $1', [ - conversationId, - ]) + const { threadId, created, row: threadRow } = await insertThread(tx, conversationId, thread) + + // A REPLAY (an existing row was found, nothing new inserted) touches + // the conversation not at all — no reopen, no updated_at bump. Only + // a genuinely new row counts as new activity on the conversation. + if (created) { + if (row.status === 'closed') { + await tx.query( + "UPDATE conversations SET status = 'open', updated_at = now() WHERE id = $1", + [conversationId], + ) + } else { + await tx.query('UPDATE conversations SET updated_at = now() WHERE id = $1', [ + conversationId, + ]) + } } - return { ok: true, threadId } + return { ok: true, threadId, created, thread: toStoredThread(threadRow) } }) }, @@ -424,6 +604,69 @@ export function createConversationStore(db: Db): ConversationStore { } }, + async claimThreadForDelivery(threadId, leaseMs) { + // A single UPDATE is already atomic with respect to itself under + // Postgres row-level locking: two overlapping calls for the same + // threadId serialize on the row, and the second one's WHERE clause is + // re-evaluated against the FIRST call's committed result — so at most + // one of them ever sees `claimed_until IS NULL OR claimed_until < + // now()` as true and gets a row back. No explicit transaction needed. + // + // `delivery_status IN ('pending', 'failed')` re-checks the OUTCOME on + // the same locked row, not just the lease: a row that reached 'sent' + // (via releaseThreadLease, which clears claimed_until in the same + // write that records the status) between a caller last observing it + // as pending/failed and this claim call must never be reclaimed — + // that would resend an already-delivered message. See this method's + // doc comment on the interface for the full TOCTOU it closes. + const rows = await db.query( + `UPDATE threads + SET claimed_until = now() + ($2::double precision * interval '1 millisecond') + WHERE id = $1 AND direction = 'outbound' + AND (claimed_until IS NULL OR claimed_until < now()) + AND delivery_status IN ('pending', 'failed') + RETURNING ${THREAD_COLUMNS}`, + [threadId, leaseMs], + ) + return rows.length === 0 ? null : toStoredThread(rows[0]) + }, + + async releaseThreadLease(threadId, status) { + // Same scoping and throw-on-zero-rows contract as setThreadDeliveryStatus + // (see its doc comment) — kept as a separate method rather than a + // parameter there so the pre-HT-16 no-idempotency-key send path keeps + // calling setThreadDeliveryStatus completely unchanged. + const updated = await db.query<{ id: string }>( + "UPDATE threads SET delivery_status = $1, claimed_until = NULL WHERE id = $2 AND direction = 'outbound' RETURNING id", + [status, threadId], + ) + if (updated.length === 0) { + throw new Error( + `releaseThreadLease: no outbound thread with id ${threadId} (wrong id, an inbound thread, or the row was deleted)`, + ) + } + }, + + async listDeliverableThreads(options) { + const rows = await db.query( + `SELECT ${THREAD_COLUMNS} FROM threads + WHERE direction = 'outbound' + AND send_envelope IS NOT NULL + AND ( + delivery_status = 'failed' + OR ( + delivery_status = 'pending' + AND created_at < now() - ($1::double precision * interval '1 millisecond') + ) + ) + AND (claimed_until IS NULL OR claimed_until < now()) + ORDER BY created_at + LIMIT $2`, + [options.staleAfterMs, options.batchSize], + ) + return rows.map(toStoredThread) + }, + async listConversations(options) { // Built up as parameterized fragments — never string-interpolated // values, only structure (which fragment appears) is decided in JS. @@ -496,12 +739,27 @@ export function createConversationStore(db: Db): ConversationStore { * generate one (defeating the point of a DB default) or special-case a * `null`/`undefined` id column value, which is not what "no id supplied" * means here. + * + * ## The idempotency-key get-or-insert (HT-16) + * + * The INSERT always carries `ON CONFLICT (conversation_id, idempotency_key) + * WHERE idempotency_key IS NOT NULL DO NOTHING RETURNING ` — this is + * harmless and never triggers when `thread.idempotencyKey` is omitted (a + * `NULL` key can never collide with the partial unique index; see migration + * 003's doc comment), which is exactly why the no-key path needs no separate + * code path here to stay byte-identical to pre-HT-16 behavior. When a key IS + * given and the insert is skipped because that `(conversation_id, + * idempotency_key)` pair already exists, the `RETURNING` clause comes back + * empty and this function falls back to a `SELECT` of that pre-existing row. + * The caller (`appendThread`) is what wraps this in the transaction holding + * the conversation row's `FOR UPDATE` lock, which is what makes the + * conflict-then-select sequence race-free — see that method's doc comment. */ async function insertThread( tx: Queryable, conversationId: string, thread: NewThread, -): Promise { +): Promise<{ threadId: string; created: boolean; row: ThreadRow }> { // Derive delivery_status from direction so the row always satisfies the // schema's direction↔status CHECK (migration 002): an outbound thread // defaults to 'pending' (its outbox starting state) unless the caller set a @@ -509,43 +767,74 @@ async function insertThread( // passed, since delivery status is meaningless for received mail. const deliveryStatus = thread.direction === 'outbound' ? (thread.deliveryStatus ?? 'pending') : null + const idempotencyKey = thread.idempotencyKey ?? null + // jsonb columns take a caller-serialized string, per src/db/client.ts's + // module doc — `SqlValue` deliberately has no "plain object" member, so + // this is the one place a `SendEnvelope` is turned into JSON text. + const sendEnvelopeJson = + thread.sendEnvelope !== undefined ? JSON.stringify(thread.sendEnvelope) : null + + const rows = + thread.id !== undefined + ? await tx.query( + `INSERT INTO threads (id, conversation_id, direction, message_id, in_reply_to, from_address, body_text, body_html, delivery_status, idempotency_key, send_envelope) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) + ON CONFLICT (conversation_id, idempotency_key) WHERE idempotency_key IS NOT NULL DO NOTHING + RETURNING ${THREAD_COLUMNS}`, + [ + thread.id, + conversationId, + thread.direction, + thread.messageId, + thread.inReplyTo ?? null, + thread.fromAddress, + thread.bodyText ?? null, + thread.bodyHtml ?? null, + deliveryStatus, + idempotencyKey, + sendEnvelopeJson, + ], + ) + : await tx.query( + `INSERT INTO threads (conversation_id, direction, message_id, in_reply_to, from_address, body_text, body_html, delivery_status, idempotency_key, send_envelope) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) + ON CONFLICT (conversation_id, idempotency_key) WHERE idempotency_key IS NOT NULL DO NOTHING + RETURNING ${THREAD_COLUMNS}`, + [ + conversationId, + thread.direction, + thread.messageId, + thread.inReplyTo ?? null, + thread.fromAddress, + thread.bodyText ?? null, + thread.bodyHtml ?? null, + deliveryStatus, + idempotencyKey, + sendEnvelopeJson, + ], + ) - if (thread.id !== undefined) { - const [row] = await tx.query<{ id: string }>( - `INSERT INTO threads (id, conversation_id, direction, message_id, in_reply_to, from_address, body_text, body_html, delivery_status) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) - RETURNING id`, - [ - thread.id, - conversationId, - thread.direction, - thread.messageId, - thread.inReplyTo ?? null, - thread.fromAddress, - thread.bodyText ?? null, - thread.bodyHtml ?? null, - deliveryStatus, - ], - ) - return row.id + if (rows.length === 1) { + return { threadId: rows[0].id, created: true, row: rows[0] } } - const [row] = await tx.query<{ id: string }>( - `INSERT INTO threads (conversation_id, direction, message_id, in_reply_to, from_address, body_text, body_html, delivery_status) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8) - RETURNING id`, - [ - conversationId, - thread.direction, - thread.messageId, - thread.inReplyTo ?? null, - thread.fromAddress, - thread.bodyText ?? null, - thread.bodyHtml ?? null, - deliveryStatus, - ], + // Conflict: DO NOTHING skipped the insert, which is only possible when + // idempotencyKey is non-null (see the doc comment above) — fetch the row + // that already holds this (conversationId, idempotencyKey) pair. + const existing = await tx.query( + `SELECT ${THREAD_COLUMNS} FROM threads WHERE conversation_id = $1 AND idempotency_key = $2`, + [conversationId, idempotencyKey], ) - return row.id + const existingRow = existing[0] + if (existingRow === undefined) { + // Structurally unreachable: ON CONFLICT only fires against a row that + // satisfies this exact WHERE, inside the same transaction. Thrown rather + // than silently returning a made-up result if it ever did happen. + throw new Error( + `insertThread: ON CONFLICT DO NOTHING skipped the insert but no existing row was found for conversation ${conversationId}, idempotency key ${idempotencyKey}`, + ) + } + return { threadId: existingRow.id, created: false, row: existingRow } } /** @@ -604,6 +893,12 @@ function toStoredThread(row: ThreadRow): StoredThread { bodyText: row.body_text, bodyHtml: row.body_html, deliveryStatus: row.delivery_status as StoredThread['deliveryStatus'], + idempotencyKey: row.idempotency_key, + // Cast, not parsed: this codebase is the only writer of send_envelope + // (insertThread, always via JSON.stringify of a SendEnvelope), and the + // jsonb column already arrives decoded (see ThreadRow's doc comment). + sendEnvelope: row.send_envelope as SendEnvelope | null, + claimedUntil: row.claimed_until === null ? null : toDate(row.claimed_until), createdAt: toDate(row.created_at), } }