feat(mail): provider-agnostic inbound ingest pipeline (HT-37) - #38
Merged
Conversation
ingestInboundMessage (src/mail/ingest.ts): spec §3's five steps — atomic claim -> parse (once) -> loop-guard -> decideThreading -> store + ledger 'received->stored' in ONE transaction. Idempotent, at-least-once, dead-letters after MAX_INGEST_ATTEMPTS. inbound-deliveries.ts: the ledger store (atomic claim + race-safe failed-reclaim, mark* transitions, tx-scoped markStoredInTx). conversations.ts: extracted createConversationInTx/appendThreadInTx (byte-identical transaction bodies) so the store write + ledger mark compose into one transaction; public methods are thin wrappers. Loop-suppression suppresses ONLY a verifiable own-Message-ID reflection (never sender identity, never In-Reply-To/References) so auto-responder replies still ingest per auto-submitted.json. 436 tests pass in isolation; typecheck + biome clean. KNOWN GAP (follow-up): a hard process crash between claim() and the step-5 commit strands a delivery at 'received' (claim reclaims 'failed', not 'received') — needs a staleness/lease sweep, mirroring outbound HT-16/HT-22. Attachments deferred. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
📝 WalkthroughWalkthroughAdds provider-agnostic inbound mail ingestion with delivery claims, transactional storage, threading, loop suppression, retries, dead-letter handling, blob resolution, and comprehensive persistence tests. ChangesInbound ingestion
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Provider
participant ingestInboundMessage
participant InboundDeliveryStore
participant BlobStore
participant ConversationStore
Provider->>ingestInboundMessage: submit RawInboundMessage
ingestInboundMessage->>InboundDeliveryStore: claim delivery
ingestInboundMessage->>BlobStore: get blob when blobRef is present
ingestInboundMessage->>ConversationStore: create or append conversation thread
ingestInboundMessage->>InboundDeliveryStore: mark stored, suppressed, failed, or dead-letter
Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
Resolves the src/store/index.ts barrel-export conflict with HT-38 (both added store exports) — kept both sets. 486 tests pass, typecheck + biome clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
src/store/inbound-deliveries.test.ts (1)
146-159: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider adding a concurrent-reclaim test for the
failedrow race.The module doc emphasizes that concurrent retries of the same
failedrow can never both win, but only the fresh-key race is tested withPromise.all(lines 84-94); thefailed-row reclaim (this test) is only exercised sequentially. Mirroring the existing concurrent test pattern here would give this documented atomicity guarantee actual regression coverage.♻️ Suggested additional test
it('two concurrent claims on the SAME failed row resolve to exactly one reclaim', async () => { const { store, mailboxId } = await freshStore() const { delivery } = await store.claim(mailboxId, 'provider-msg-1') await store.markFailed(delivery.id, 'parse: boom') const [a, b] = await Promise.all([ store.claim(mailboxId, 'provider-msg-1'), store.claim(mailboxId, 'provider-msg-1'), ]) expect([a.claimed, b.claimed].sort()).toEqual([false, true]) expect(a.delivery.id).toBe(b.delivery.id) })🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/store/inbound-deliveries.test.ts` around lines 146 - 159, Add a concurrent reclaim test alongside the existing failed-row test, using Promise.all to call store.claim twice for the same failed delivery. Assert both results reference the same delivery and that exactly one result has claimed set to true, covering the atomicity guarantee for failed-row races.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/mail/ingest.ts`:
- Around line 279-294: Wrap the step-3
`deps.inboundDeliveryStore.markSuppressed` call in the same failure-handling
pattern used for the step-5 write, ensuring errors are caught and the delivery
claim cannot remain stuck at `received`. Keep `decideThreading` outside this
handling because it is pure.
---
Nitpick comments:
In `@src/store/inbound-deliveries.test.ts`:
- Around line 146-159: Add a concurrent reclaim test alongside the existing
failed-row test, using Promise.all to call store.claim twice for the same failed
delivery. Assert both results reference the same delivery and that exactly one
result has claimed set to true, covering the atomicity guarantee for failed-row
races.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: aeba544d-cc2e-4151-b267-d426c4ebf21a
📒 Files selected for processing (6)
src/mail/ingest.test.tssrc/mail/ingest.tssrc/store/conversations.tssrc/store/inbound-deliveries.test.tssrc/store/inbound-deliveries.tssrc/store/index.ts
Comment on lines
+279
to
+294
| // --- Step 3: loop guard (spec §5). ---------------------------------------- | ||
| if (isOwnMessageReflection(parsed, deps.keyring)) { | ||
| const reason = 'own-message-loop' | ||
| const updated = await deps.inboundDeliveryStore.markSuppressed(delivery.id, reason) | ||
| logIngestEvent({ | ||
| ...base, | ||
| outcome: 'suppressed', | ||
| reason, | ||
| parseSize, | ||
| attachmentCount: parsed.attachments.length, | ||
| }) | ||
| return { ...base, kind: 'suppressed', reason: updated.lastError ?? reason } | ||
| } | ||
|
|
||
| // --- Step 4: decide (never re-implemented here). -------------------------- | ||
| const decision = decideThreading(parsed, deps.keyring) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
set -euo pipefail
# Locate relevant files
git ls-files | rg '^src/mail/'
# Show structure of ingest.ts
ast-grep outline src/mail/ingest.ts --view expanded
# Read relevant slices with line numbers
sed -n '90,360p' src/mail/ingest.ts
# Locate threading decision implementation and store interfaces/usages
rg -n "decideThreading|markSuppressed|interface InboundDeliveryStore|type InboundDeliveryStore|class .*InboundDeliveryStore|claim\(" src/mail -SRepository: Helpthread/helpthread
Length of output: 15939
🏁 Script executed:
set -euo pipefail
printf '\n### ingest.ts (selected lines)\n'
sed -n '90,360p' src/mail/ingest.ts
printf '\n### threading/store references\n'
rg -n "decideThreading|markSuppressed|interface InboundDeliveryStore|type InboundDeliveryStore|class .*InboundDeliveryStore|claim\(" src/mail -SRepository: Helpthread/helpthread
Length of output: 14640
🏁 Script executed:
set -euo pipefail
printf '\n### thread.ts outline\n'
ast-grep outline src/mail/thread.ts --view expanded
printf '\n### thread.ts selected lines\n'
sed -n '1,240p' src/mail/thread.ts
printf '\n### ingest store interface / implementation references\n'
rg -n "markSuppressed|markFailed|markDeadLetter|interface InboundDeliveryStore|type InboundDeliveryStore|class .*InboundDeliveryStore|function .*markSuppressed" src -SRepository: Helpthread/helpthread
Length of output: 10079
Guard markSuppressed in step 3 markSuppressed can fail like the step-5 write, but this path is still uncaught and can leave the claim stuck at received; decideThreading is pure and does not need to be wrapped.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/mail/ingest.ts` around lines 279 - 294, Wrap the step-3
`deps.inboundDeliveryStore.markSuppressed` call in the same failure-handling
pattern used for the step-5 write, ensuring errors are caught and the delivery
claim cannot remain stuck at `received`. Keep `decideThreading` outside this
handling because it is pure.
This was referenced Jul 16, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Implements HT-37 [D] — the provider-agnostic inbound ingest pipeline, the correctness heart of inbound. Under the HT-33 epic.
What's here
src/mail/ingest.ts—ingestInboundMessage(raw, deps): the spec §3 five-step pipeline — atomic claim → parse (once) → loop-guard →decideThreading→ store + ledger-mark in one transaction. Idempotent, at-least-once, dead-letters afterMAX_INGEST_ATTEMPTS.src/store/inbound-deliveries.ts— the delivery ledger store: atomicclaim(get-or-insert + race-safefailed-row reclaim),markSuppressed/markFailed/markDeadLetter, and the transaction-scopedmarkStoredInTx.src/store/conversations.ts— extractedcreateConversationInTx/appendThreadInTx(the existing transaction bodies) so the ingest can compose the store write + ledger mark into one transaction. The public methods are now thin wrappers — behavior byte-identical (verified in the diff; all pre-existing tests pass unchanged).The crux — store-write +
received→storedas one atomic unit (spec §4) — is solved by that extraction:ingest.tsopens onedb.transactionand runscreateConversationInTx/appendThreadInTx+markStoredInTxagainst the sametx; a crash rolls back both, never a half-written outcome.Review notes (mine)
conversations.tsrefactor is byte-identical to the original transaction bodies — the sacred store is behaviorally unchanged.Message-IDreflection and deliberately leavesIn-Reply-To/Referencesto normal threading — so a customer's auto-responder still gets ingested perauto-submitted.json/ spec §5. Never suppresses on sender identity (invariant HT-3: clean-room protocol doc #1 held).One real gap — flagged for a follow-up (not a happy-path bug)
A hard process crash (SIGKILL / OOM / redeploy) in the narrow window between
claim()committing (received) and either the step-5 transaction committing (stored) or the catch-block'smarkFailedrunning leaves the delivery stuck atreceived.claim()reclaims afailedrow but not a stuckreceivedone, so on re-delivery it returnsin-progressand the message is stranded — a never-drop-invariant edge. This mirrors the outbound side's own history: core send (HT-15) shipped before the lease + sweep recovery (HT-16/HT-22). The fix is a distinct mechanism — areceived-staleness reclaim or aclaimed_untillease column + an inbound sweep — best designed on its own ticket. A follow-up ticket will track it; it must land before the live dogfood.Attachments are parsed but not persisted (deferred per scope; separate task). Observability (
forgedTokenCountet al.) surfaced via a structured log line.🤖 Generated with Claude Code
Summary by CodeRabbit