From f5c04f7092f7839850d92a4421979fac6bc63582 Mon Sep 17 00:00:00 2001 From: Clifton Cunningham Date: Sat, 24 Oct 2026 13:44:10 +0100 Subject: [PATCH] inbox: split forwarded-as-attachment emails into messages of their own Every message/rfc822 part (postal-mime forceRfc822Attachments) becomes a child inbox message: own R2 prefix and parse job, the embedded From, Subject, Date and Message-ID, the parent's auth verdict and forwarded_by. The parent's .eml rows point at the child (expanded_message_id) and are kept out of consumers and knowledge. Redelivery is idempotent through a unique (tenant_id, parent_message_id, source_sha256) index; an embedded Message-ID already received is linked, not stored again. Depth cap 3, 50 children per message. Consumers gain parentMessageId, forwardedBy and forwardedMessageIds; the Inbox page links parent and children. Also removes the duplicate docs/plugins//upgrades copies that `plugin export` left inside inbox and freeagent in 3.6.0; docs/upgrades/ is the canonical home, and the README says to delete the copy after an export. Co-Authored-By: Claude Opus 5.5 --- README.md | 6 + .../docs/plugins/freeagent/upgrades/README.md | 28 - .../plugins/freeagent/upgrades/unreleased.md | 41 -- plugins/freeagent/docs/upgrades/unreleased.md | 11 +- plugins/inbox/README.md | 5 +- .../apps/web/src/plugins/inbox/CLAUDE.md | 21 +- .../apps/web/src/plugins/inbox/api/routes.ts | 34 +- .../web/src/plugins/inbox/db/schema/inbox.ts | 38 +- .../apps/web/src/plugins/inbox/plugin.json | 21 +- .../src/plugins/inbox/services/forwarded.ts | 205 ++++++++ .../src/plugins/inbox/services/knowledge.ts | 2 +- .../web/src/plugins/inbox/services/parse.ts | 106 +++- .../plugins/inbox/tests/api/forwarded.test.ts | 490 ++++++++++++++++++ .../src/plugins/inbox/tools/inbox-tools.ts | 26 +- .../src/plugins/inbox/ui/pages/InboxPage.tsx | 95 +++- .../docs/plugins/inbox/upgrades/README.md | 28 - .../docs/plugins/inbox/upgrades/unreleased.md | 37 -- plugins/inbox/docs/upgrades/unreleased.md | 29 +- .../shared/src/plugins/inbox/index.ts | 31 +- plugins/inbox/rocketflare-plugin.json | 4 +- plugins/inbox/skills/inbox/SKILL.md | 12 +- 21 files changed, 1075 insertions(+), 195 deletions(-) delete mode 100644 plugins/freeagent/docs/plugins/freeagent/upgrades/README.md delete mode 100644 plugins/freeagent/docs/plugins/freeagent/upgrades/unreleased.md create mode 100644 plugins/inbox/apps/web/src/plugins/inbox/services/forwarded.ts create mode 100644 plugins/inbox/apps/web/src/plugins/inbox/tests/api/forwarded.test.ts delete mode 100644 plugins/inbox/docs/plugins/inbox/upgrades/README.md delete mode 100644 plugins/inbox/docs/plugins/inbox/upgrades/unreleased.md diff --git a/README.md b/README.md index cf0ccdc..85c7603 100644 --- a/README.md +++ b/README.md @@ -63,6 +63,12 @@ expects: `classifyPluginFile` refuses any path outside `apps/web/src/plugins//docs/upgrades/` and nowhere else. `pnpm plugin add` +copies them to the host's `docs/plugins//upgrades/`, and `pnpm plugin export` copies that host +directory back verbatim. After an export, delete `plugins//docs/plugins/` +(`find plugins -path '*docs/plugins*'` should print nothing): that copy duplicates the notes at the +same host path, and it is how 3.6.0 shipped two of them. + ## The plugin API A plugin's compatibility with a kit is **observed, not declared**. `pnpm plugin export` derives the diff --git a/plugins/freeagent/docs/plugins/freeagent/upgrades/README.md b/plugins/freeagent/docs/plugins/freeagent/upgrades/README.md deleted file mode 100644 index e6c9535..0000000 --- a/plugins/freeagent/docs/plugins/freeagent/upgrades/README.md +++ /dev/null @@ -1,28 +0,0 @@ -# Release notes, written for an agent - -One file per release of this plugin, `X.Y.Z.md`, plus `unreleased.md`, which collects entries -between releases. `CHANGELOG.md` at the repository root is the human index. These files are the -porting instructions: `pnpm plugin upgrade freeagent` drives the kit's `scripts/upgrade.mjs`, -which reads their frontmatter. - -## The shape - -Frontmatter comes first, then four fixed headings in this order: **What changed**, **How to -apply**, **Conflicts to expect**, **Verify**. - -| Field | Means | -|---|---| -| `version` · `previous` | This release and the one before it. `previous: null` appears only on the first release. `plugin upgrade` walks this chain, so a gap is permanent | -| `date` | `YYYY-MM-DD` | -| `breaking` | Does an adopter have to change their own code? | -| `migrations` | **Descriptions of the schema change, never file names.** The host generates the DDL with `pnpm db:generate`; this plugin never ships a migration | -| `areas` · `touches_surfaces` · `requires_surfaces` | Where the change lands | -| `manual` | Is there a step no tool can do? | - -## The rules - -- **Released history is never rewritten.** Every app pins a commit of this repository. -- **A behaviour change adds an entry to `unreleased.md` in the same commit.** - `node scripts/release.mjs X.Y.Z` folds it into `X.Y.Z.md`. -- **Never rename anything across releases**, whether a table, a column, a provider id or a tool - name. Expand and contract instead. diff --git a/plugins/freeagent/docs/plugins/freeagent/upgrades/unreleased.md b/plugins/freeagent/docs/plugins/freeagent/upgrades/unreleased.md deleted file mode 100644 index 9feadf0..0000000 --- a/plugins/freeagent/docs/plugins/freeagent/upgrades/unreleased.md +++ /dev/null @@ -1,41 +0,0 @@ ---- -version: unreleased -previous: null -date: null -breaking: false -migrations: - - freeagent_connections — one row per tenant (unique tenant_id): environment, status, sealed refresh and access tokens, company, sync claim and last outcome; RLS policy - - freeagent_sync_state — one updated_since cursor per (tenant, resource), unique; RLS policy - - freeagent_bank_accounts, freeagent_bank_transactions, freeagent_explanations, freeagent_categories, freeagent_contacts, freeagent_bills, freeagent_invoices — the mirror, each unique on (tenant_id, url), money as numeric; RLS policies -areas: [api, ui, cli] -touches_surfaces: [] -requires_surfaces: [] -manual: true ---- - -## What changed - -First release of `freeagent`: an organisation connects its FreeAgent company over OAuth, and its books are mirrored every 30 minutes for pages, agents, analytics and other plugins. - -- OAuth through `POST /api/freeagent/connect` and the public callback `/api/hooks/freeagent/callback`; tokens sealed; `invalid_grant` marks the connection `needs_reconnect`. -- `freeagent.sync` (cron `*/30 * * * *`, "Sync now", and after connecting) mirrors bank accounts, transactions, explanations, categories, contacts, bills and invoices incrementally by `updated_since`, then calls every `freeagent:sync-listeners` extension. -- Every request sends `X-Api-Version: 2026-09-01` (multiple attachments per explanation). -- Service API for other plugins: `freeagentClientFor`, `explainTransaction`, `attachToExplanation`, `deleteExplanation`, `createBill`, `listUnexplainedTransactions`. -- Agent tools: five read tools over the mirror; `freeagent_explain_transaction` and `freeagent_create_bill` need approval and are offered to admin-level requesters only. -- UI: Settings → FreeAgent and a FreeAgent page; CLI `rocketflare freeagent status|sync`; drizzle-cube cubes contributed to `analytics` when it is installed. - -## How to apply - -1. Install with `pnpm plugin add https://github.com/rocketflare-dev/rocketflare-plugins.git@ --subdir plugins/freeagent --apply` on a kit at 0.21.0 or later. -2. Run `pnpm db:generate --name plugin-freeagent-`, read the SQL (nine `CREATE TABLE`s and their policies), then `pnpm db:migrate`. -3. Register a FreeAgent app and set `FREEAGENT_CLIENT_ID`, `FREEAGENT_CLIENT_SECRET` and `FREEAGENT_ENVIRONMENT` — the `freeagent` skill (`/freeagent register`, `/freeagent secrets`) walks both. `pnpm provision cloudflare ` writes the `*/30 * * * *` cron into both tomls. - -## Conflicts to expect - -None — a first release adds files only. - -## Verify - -1. `pnpm plugin check` passes. -2. `pnpm gate` passes, including `src/plugins/freeagent/tests/**`. -3. After connecting, `pnpm cli freeagent status` shows the company and a last-sync time. diff --git a/plugins/freeagent/docs/upgrades/unreleased.md b/plugins/freeagent/docs/upgrades/unreleased.md index 9d6907a..9f6b280 100644 --- a/plugins/freeagent/docs/upgrades/unreleased.md +++ b/plugins/freeagent/docs/upgrades/unreleased.md @@ -12,11 +12,18 @@ manual: false ## What changed -_Nothing yet. Add an entry here in the same pull request as the change — see `README.md` beside this -file for the fields and for what "How to apply" has to say._ +No code change to freeagent. The duplicated `docs/plugins/freeagent/upgrades/` copy inside the plugin directory, left there by `pnpm plugin export` in 3.6.0, is removed. + +- A host's `docs/plugins/freeagent/upgrades/` now comes only from the plugin's `docs/upgrades/`, its one canonical home. The duplicate could put a stale `unreleased.md` (`previous: null`) at the same host path. ## How to apply +1. `pnpm plugin upgrade freeagent --apply`. It changes the version stamps and the release notes under `docs/plugins/freeagent/upgrades/`. + ## Conflicts to expect +None. + ## Verify + +1. `pnpm plugin check` reports `freeagent` checks out at the new version. diff --git a/plugins/inbox/README.md b/plugins/inbox/README.md index 07a65cf..ebd445f 100644 --- a/plugins/inbox/README.md +++ b/plugins/inbox/README.md @@ -15,10 +15,11 @@ job. Other plugins subscribe to parsed messages through the `inbox:consumers` ex |---|---| | Email | A catch-all `emailHandlers` entry. Unknown addresses, senders that aren't allowed and failed authentication are bounced with a reason the forwarding mailbox shows | | UI | **Inbox**: messages, sender and forwarder, authentication verdicts, text body, attachments, raw/HTML downloads, reprocess and delete. **Settings → Inbox**: address, allow-list, require authentication, on/off | +| Forward as attachment | Each `message/rfc822` part of a message (Gmail's "Forward as attachment", with or without a disposition) becomes a **message of its own**, linked to the forward it came in: its own sender, subject, date and attachments, the forward's authentication verdict, and who forwarded it. An embedded email already received is linked, not stored again. Nesting stops at 3 levels and 50 emails per message | | Forwarding | Gmail and Proton forwarding confirmations are let through before anyone is allow-listed, and their **code and link** are shown at the top of the message | | Knowledge | Each parsed message (headers + text) and its PDF/Office/HTML/text attachments are indexed into the kit's knowledge base (`inbox:message`, `inbox:attachment`), so chat can search your mail. Upserted on reprocess, removed on delete, switchable in Settings | | Agent tools | `inbox_list_messages` and `inbox_get_message`, both read-only | -| Extension | `inbox:consumers`: a plugin contributes `{ id, onMessage(message, ctx) }` and is called once per parsed message, with attachment refs it reads through `readInboxAttachment` | +| Extension | `inbox:consumers`: a plugin contributes `{ id, onMessage(message, ctx) }` and is called once per parsed message — so once per embedded email of a forward — with attachment refs it reads through `readInboxAttachment`, and `parentMessageId` / `forwardedBy` / `forwardedMessageIds` for forwarded mail | | API | `GET/PUT /api/inbox/settings`, `GET /api/inbox/messages`, `GET /api/inbox/messages/:id`, `…/:id/raw`, `…/:id/html`, `GET /api/inbox/attachments/:id`, `POST …/:id/reprocess`, `DELETE …/:id` | | Data | `inbox_settings`, `inbox_messages` and `inbox_attachments`, all with row-level security. Bytes live in R2 under `tenants//inbox/` and are purged with the tenant | | CLI | `rocketflare inbox list [--status]` and `rocketflare inbox status` | @@ -73,3 +74,5 @@ Every member can read the inbox and its attachments. Owners and admins change th - No outbound replies. - The routing rule is a manual dashboard step; provisioning doesn't create it. - The HTML body is a download only and is never rendered in the app. +- A message split out of a forward carries the forward's SPF/DKIM/DMARC verdict; its own signatures can't be checked once it has been re-wrapped. +- Messages received before 3.7.0 aren't split until they are reprocessed. diff --git a/plugins/inbox/apps/web/src/plugins/inbox/CLAUDE.md b/plugins/inbox/apps/web/src/plugins/inbox/CLAUDE.md index 31950c7..241a967 100644 --- a/plugins/inbox/apps/web/src/plugins/inbox/CLAUDE.md +++ b/plugins/inbox/apps/web/src/plugins/inbox/CLAUDE.md @@ -18,9 +18,26 @@ Inbound email for an organisation: Cloudflare Email Routing → the kit's `email `nosniff`; the UI shows the text body only. These are bytes a stranger chose. - **Parse is idempotent.** Attachments are unique per (message, sha256) and upserted, so a reprocess keeps attachment ids a consumer may have stored. Consumers must be idempotent too. +- **Forwarded as attachment** (`services/forwarded.ts`): postal-mime runs with + `forceRfc822Attachments`, so every `message/rfc822` part — with or without a disposition — is + an attachment, never merged into the parent body. Each becomes a CHILD message: its own row + (`parent_message_id`, `forward_depth`, `source_sha256`), its own R2 prefix, the embedded + headers (read from the header block only), the PARENT's `auth` (the inner signatures cannot be + verified) and `forwarded_by` (the top-level envelope sender), then its own `inbox.parse`. The + parent keeps the `.eml` row with `expanded_message_id` set; such rows are excluded from consumer + `attachments`, knowledge and the parent's `attachment_count`. Idempotency is the unique + `(tenant_id, parent_message_id, source_sha256)` index plus the per-tenant Message-ID index (an + email already received is LINKED, not stored again) — never memory. Caps: + `INBOX_FORWARD_MAX_DEPTH` (3; at the cap parts stay plain `.eml`s) and + `INBOX_FORWARD_MAX_CHILDREN` (50 per message; the rest stay attachments and the message's + `error` says so). A child still `received` when its parent is re-run is re-enqueued, so a + failed enqueue cannot strand it — consumers must tolerate the rare second call. Deleting a + parent leaves its children (`parent_message_id` is `set null`); each is deleted on its own. - **Consumers** (`InboxConsumer`, exported from `index.ts`) are called for `kind = 'message'` only, - never for forwarding confirmations. A consumer's throw is logged and written to the message's - `error`; it does not stop the others and does not retry the job. + never for forwarding confirmations — once per message, so once per embedded email (a child + carries `parentMessageId`/`forwardedBy`; a parent lists `forwardedMessageIds`). A consumer's + throw is logged and written to the message's `error`; it does not stop the others and does not + retry the job. - **Knowledge** (`services/knowledge.ts`): after consumers, a parsed message (headers + text) and each non-image attachment are ingested with `source: 'inbox:message' | 'inbox:attachment'` and the row's id as `externalId` — upserts, so a reprocess is safe. Best-effort: failures land on the diff --git a/plugins/inbox/apps/web/src/plugins/inbox/api/routes.ts b/plugins/inbox/apps/web/src/plugins/inbox/api/routes.ts index 0ccca19..9d2ca27 100644 --- a/plugins/inbox/apps/web/src/plugins/inbox/api/routes.ts +++ b/plugins/inbox/apps/web/src/plugins/inbox/api/routes.ts @@ -17,13 +17,14 @@ import { type InboxAuth, type InboxConfirmation, type InboxMessage, + type InboxMessageLink, type InboxMessageListQuery, type InboxMessageSummary, inboxMessageListQuerySchema, type UpdateInboxSettings, updateInboxSettingsSchema, } from '@rocketflare/shared/plugins/inbox/index' -import { and, count, desc, eq, type SQL } from 'drizzle-orm' +import { and, count, desc, eq, inArray, type SQL } from 'drizzle-orm' import type { RequestCtx } from '@/plugins/api' import { createRouter, @@ -54,9 +55,30 @@ function toSummary(row: InboxMessageRow): InboxMessageSummary { receivedAt: row.receivedAt, attachmentCount: row.attachmentCount, rawSize: row.rawSize, + parentMessageId: row.parentMessageId, + forwardedBy: row.forwardedBy, + forwardedCount: row.forwardedCount, } } +const linkColumns = { + id: inboxMessages.id, + subject: inboxMessages.subject, + fromAddress: inboxMessages.fromAddress, + fromName: inboxMessages.fromName, +} + +/** The other messages one message links to — its wrapper, the emails split out of it. */ +async function loadLinks(ctx: RequestCtx, ids: readonly string[]): Promise { + if (ids.length === 0) return [] + const rows = await ctx.db + .select(linkColumns) + .from(inboxMessages) + .where(and(eq(inboxMessages.tenantId, ctx.tenantId), inArray(inboxMessages.id, [...ids]))) + const byId = new Map(rows.map(r => [r.id, r])) + return ids.flatMap(id => byId.get(id) ?? []) +} + function download(body: ReadableStream, contentType: string, filename: string, size?: number) { const headers = new Headers({ 'Content-Type': contentType, @@ -153,6 +175,13 @@ inboxRouter.get('/messages/:id', async c => { .from(inboxAttachments) .where(and(eq(inboxAttachments.tenantId, ctx.tenantId), eq(inboxAttachments.messageId, row.id))) .orderBy(inboxAttachments.createdAt) + const forwardedIds = [ + ...new Set(attachments.flatMap(a => (a.expandedMessageId ? [a.expandedMessageId] : []))), + ] + const [parent, forwardedMessages] = await Promise.all([ + row.parentMessageId ? loadLinks(ctx, [row.parentMessageId]) : Promise.resolve([]), + loadLinks(ctx, forwardedIds), + ]) const body: InboxMessage = { ...toSummary(row), messageIdHeader: row.messageIdHeader, @@ -169,8 +198,11 @@ inboxRouter.get('/messages/:id', async c => { size: a.size, sha256: a.sha256, inline: a.inline, + expandedMessageId: a.expandedMessageId, }) ), + parent: parent[0] ?? null, + forwardedMessages, } return c.json(body) }) diff --git a/plugins/inbox/apps/web/src/plugins/inbox/db/schema/inbox.ts b/plugins/inbox/apps/web/src/plugins/inbox/db/schema/inbox.ts index 6c84b20..f67452a 100644 --- a/plugins/inbox/apps/web/src/plugins/inbox/db/schema/inbox.ts +++ b/plugins/inbox/apps/web/src/plugins/inbox/db/schema/inbox.ts @@ -9,10 +9,18 @@ * `tenants//inbox//raw.eml` (purged with the tenant by the kit); the row carries * what the list and the consumers need. `message_id_header` is unique per tenant when present, * so a message delivered twice (two forwarding rules, a retried SMTP hand-off) is stored once. - * - `inbox_attachments` — one row per extracted part, cascading with its message, bytes in R2. + * A message that arrived as a `message/rfc822` part of another (Gmail's "Forward as + * attachment") is a row of its own — a CHILD — linked by `parent_message_id`, carrying the + * embedded email's own headers, the parent's authentication verdict and `forwarded_by`. + * `(tenant_id, parent_message_id, source_sha256)` is unique, so a redelivered parse cannot + * mint a second child for the same part. + * - `inbox_attachments` — one row per extracted part, cascading with its message, bytes in R2. An + * embedded message's row keeps the original `.eml` and points at the message it became through + * `expanded_message_id` — such a row is never handed to consumers or the knowledge base. */ import { sql } from 'drizzle-orm' import { + type AnyPgColumn, boolean, index, integer, @@ -75,6 +83,24 @@ export const inboxMessages = pgTable( attachmentCount: integer('attachment_count').notNull().default(0), error: text('error'), parsedAt: timestamp('parsed_at', { withTimezone: true }), + /** + * The message this one was forwarded inside (a `message/rfc822` part of it), else null. + * `set null`, not cascade: deleting the wrapper leaves the receipts it carried in place. + */ + parentMessageId: uuid('parent_message_id').references((): AnyPgColumn => inboxMessages.id, { + onDelete: 'set null', + }), + /** + * Who forwarded a child — the parent's authenticated envelope sender (the top-level one, at + * any depth). The child's own SPF/DKIM/DMARC cannot be checked; it inherits the parent's. + */ + forwardedBy: text('forwarded_by'), + /** 0 for mail that arrived on its own; a child is its parent's depth + 1 (capped). */ + forwardDepth: integer('forward_depth').notNull().default(0), + /** SHA-256 of the embedded part a child was made from — its idempotency key with the parent. */ + sourceSha256: text('source_sha256'), + /** How many embedded messages this one was split into (new children plus linked duplicates). */ + forwardedCount: integer('forwarded_count').notNull().default(0), ...timestamps(), }, table => [ @@ -83,6 +109,9 @@ export const inboxMessages = pgTable( uniqueIndex('inbox_messages_tenant_message_id_uidx') .on(table.tenantId, table.messageIdHeader) .where(sql`${table.messageIdHeader} IS NOT NULL`), + uniqueIndex('inbox_messages_tenant_parent_part_uidx') + .on(table.tenantId, table.parentMessageId, table.sourceSha256) + .where(sql`${table.parentMessageId} IS NOT NULL`), tenantIsolation('inbox_messages'), ] ) @@ -101,6 +130,13 @@ export const inboxAttachments = pgTable( sha256: text('sha256').notNull(), storageKey: text('storage_key').notNull(), inline: boolean('inline').notNull().default(false), + /** + * For an embedded `message/rfc822` part: the inbox message it was split into (a new child, or + * an earlier message with the same Message-ID). Null for an ordinary attachment. + */ + expandedMessageId: uuid('expanded_message_id').references(() => inboxMessages.id, { + onDelete: 'set null', + }), ...timestamps(), }, table => [ diff --git a/plugins/inbox/apps/web/src/plugins/inbox/plugin.json b/plugins/inbox/apps/web/src/plugins/inbox/plugin.json index 21b091c..1af0cec 100644 --- a/plugins/inbox/apps/web/src/plugins/inbox/plugin.json +++ b/plugins/inbox/apps/web/src/plugins/inbox/plugin.json @@ -27,15 +27,19 @@ "createRouter", "Database", "defineTool", + "deleteIngestedDocument", "EMAIL_CATCH_ALL", "emailCtx", "EmailCtx", "emailLocalPart", "emailSubaddress", + "ingestDocument", + "ingestDocumentFile", "isUniqueViolation", "jobCtx", "JobCtx", "JobHandler", + "PluginContext", "recordActivity", "requestCtx", "RequestCtx", @@ -45,7 +49,7 @@ "ToolCtx", "validate" ], - "@/plugins/api/peers": ["extensions"], + "@/plugins/api/peers": ["allTables", "extensions"], "@/plugins/api/ui-wiring": ["NavGuard", "UiPlugin"], "@/plugins/api/ui": [ "api", @@ -66,7 +70,14 @@ "timeAgo", "usePermissions" ], - "@/db/schema/kit": ["tenantIsolation", "tenantRef", "tenants", "timestamps", "users"], + "@/db/schema/kit": [ + "tenantIsolation", + "tenantRef", + "tenants", + "tenantUsers", + "timestamps", + "users" + ], "'../api' (apps/cli/src/plugins/api.ts)": [ "CliPlugin", "CommandContext", @@ -110,9 +121,9 @@ "rlsExcluded": [] }, "migrations": [ - "inbox_settings \u2014 one row per tenant (unique tenant_id) plus a GLOBALLY unique local_part, the sender allow-list and require_authentication; RLS policy", - "inbox_messages \u2014 one row per accepted message, raw bytes in R2; partial unique (tenant_id, message_id_header); RLS policy", - "inbox_attachments \u2014 one row per kept part, cascading with its message, bytes in R2; unique (tenant_id, message_id, sha256); RLS policy" + "inbox_settings — one row per tenant (unique tenant_id) plus a GLOBALLY unique local_part, the sender allow-list and require_authentication; RLS policy", + "inbox_messages — one row per accepted message, raw bytes in R2; partial unique (tenant_id, message_id_header); a message split out of a forward links its parent (parent_message_id, self-FK set null) with forwarded_by, forward_depth, source_sha256 and partial unique (tenant_id, parent_message_id, source_sha256); forwarded_count; RLS policy", + "inbox_attachments — one row per kept part, cascading with its message, bytes in R2; unique (tenant_id, message_id, sha256); expanded_message_id (FK set null) for an embedded message split into its own row; RLS policy" ], "subdir": "plugins/inbox" } diff --git a/plugins/inbox/apps/web/src/plugins/inbox/services/forwarded.ts b/plugins/inbox/apps/web/src/plugins/inbox/services/forwarded.ts new file mode 100644 index 0000000..deb4b3d --- /dev/null +++ b/plugins/inbox/apps/web/src/plugins/inbox/services/forwarded.ts @@ -0,0 +1,205 @@ +/** + * Forwarded-as-attachment mail. Gmail's "Forward as attachment" (and any client that forwards + * several messages at once) sends ONE email whose `message/rfc822` parts are the real messages — + * receipts, invoices — each a complete email of its own. Left as `.eml` attachments, no consumer + * and no knowledge document would ever see them, so each becomes a CHILD inbox message: + * + * - its raw bytes under its own `tenants//inbox//` prefix, like any message; + * - its own From, Subject, Date and Message-ID, read from the embedded headers; + * - the PARENT's authentication verdict. The outer forward passed the allow-list and SPF/DKIM/DMARC; + * the inner message's own signatures cannot be checked (a forward re-wraps it), so the trust is + * the forwarder's, and `forwarded_by` records whose; + * - the normal `inbox.parse` job, so consumers and the knowledge base see one message per email. + * + * Idempotent without memory: `(tenant_id, parent_message_id, source_sha256)` is unique, so a + * redelivered or reprocessed parent finds the child it already made, and an embedded Message-ID + * the organisation has already received links to that message instead of storing it twice. + */ +import { INBOX_MESSAGES_ENTITY, INBOX_PARSE_JOB } from '@rocketflare/shared/plugins/inbox/index' +import { and, eq } from 'drizzle-orm' +import PostalMime from 'postal-mime' +import type { JobCtx } from '@/plugins/api' +import { type InboxMessageRow, inboxMessages } from '../db/schema' +import { messagePrefix } from './receive' + +/** The MIME type of an embedded message. `message/global` (SMTPUTF8) is not split. */ +export const EMBEDDED_MESSAGE_TYPE = 'message/rfc822' + +/** Headers are read from at most this much of an embedded message. */ +const HEADER_SCAN_BYTES = 256 * 1024 + +export interface EmbeddedHeaders { + fromAddress: string | null + fromName: string | null + subject: string | null + messageIdHeader: string | null + sentAt: Date | null +} + +/** Where the header block of raw message bytes ends (the first empty line), or -1. */ +function headerEnd(bytes: Uint8Array): number { + const limit = Math.min(bytes.length, HEADER_SCAN_BYTES) + for (let i = 0; i < limit - 1; i += 1) { + if (bytes[i] !== 0x0a) continue + if (bytes[i + 1] === 0x0a) return i + 1 + if (bytes[i + 1] === 0x0d && bytes[i + 2] === 0x0a) return i + 2 + } + return -1 +} + +/** + * The embedded email's own headers, decoded. Only the header block is parsed — the body (and any + * PDF in it) is the child's own `inbox.parse` to read — so splitting fifty forwards costs fifty + * header parses, not fifty full ones. + */ +export async function readEmbeddedHeaders(bytes: ArrayBuffer): Promise { + const view = new Uint8Array(bytes) + const end = headerEnd(view) + const block = end === -1 ? view.slice(0, HEADER_SCAN_BYTES) : view.slice(0, end) + const email = await PostalMime.parse(block) + const date = email.date ? new Date(email.date) : null + return { + fromAddress: email.from?.address?.toLowerCase() || null, + fromName: email.from?.name || null, + subject: email.subject ?? null, + messageIdHeader: email.messageId?.trim() || null, + sentAt: date && !Number.isNaN(date.getTime()) ? date : null, + } +} + +/** One embedded part to split out, after its `.eml` attachment row is stored. */ +export interface EmbeddedPart { + bytes: ArrayBuffer + sha256: string +} + +type Found = Pick + +async function findChild( + ctx: JobCtx, + parent: InboxMessageRow, + sha256: string +): Promise { + const [row] = await ctx.db + .select({ + id: inboxMessages.id, + status: inboxMessages.status, + parentMessageId: inboxMessages.parentMessageId, + }) + .from(inboxMessages) + .where( + and( + eq(inboxMessages.tenantId, parent.tenantId), + eq(inboxMessages.parentMessageId, parent.id), + eq(inboxMessages.sourceSha256, sha256) + ) + ) + .limit(1) + return row +} + +async function findByMessageId( + ctx: JobCtx, + tenantId: string, + messageIdHeader: string +): Promise { + const [row] = await ctx.db + .select({ + id: inboxMessages.id, + status: inboxMessages.status, + parentMessageId: inboxMessages.parentMessageId, + }) + .from(inboxMessages) + .where( + and(eq(inboxMessages.tenantId, tenantId), eq(inboxMessages.messageIdHeader, messageIdHeader)) + ) + .limit(1) + return row +} + +export type ExpandOutcome = + | { status: 'created'; messageId: string } + | { status: 'existing'; messageId: string } + | { status: 'linked'; messageId: string } + | { status: 'skipped' } + +/** + * Split one embedded message out of `parent`. `existing` is the child this parent made on an + * earlier run; `linked` is an earlier message with the same Message-ID (stored once, linked from + * here); `skipped` means "keep it as a plain attachment" (it claims the parent's own Message-ID). + */ +export async function expandEmbeddedMessage( + ctx: JobCtx, + parent: InboxMessageRow, + part: EmbeddedPart +): Promise { + const resolveExisting = async (headers: EmbeddedHeaders): Promise => { + const own = await findChild(ctx, parent, part.sha256) + if (own) { + // Ours, but never parsed: the enqueue after the insert may have failed and this run is the + // retry. A second enqueue is harmless (parse is idempotent), a missing one strands the child. + if (own.status === 'received') { + await ctx.enqueue({ + type: INBOX_PARSE_JOB, + payload: { tenantId: parent.tenantId, messageId: own.id }, + }) + } + return { status: 'existing', messageId: own.id } + } + if (headers.messageIdHeader) { + const seen = await findByMessageId(ctx, parent.tenantId, headers.messageIdHeader) + if (seen?.id === parent.id) return { status: 'skipped' } + if (seen) return { status: 'linked', messageId: seen.id } + } + return null + } + + const headers = await readEmbeddedHeaders(part.bytes) + const before = await resolveExisting(headers) + if (before) return before + + const id = crypto.randomUUID() + const rawKey = `${messagePrefix(parent.tenantId, id)}/raw.eml` + const storage = ctx.storage() + await storage.put(rawKey, part.bytes, { contentType: EMBEDDED_MESSAGE_TYPE }) + const inserted = await ctx.db + .insert(inboxMessages) + .values({ + id, + tenantId: parent.tenantId, + kind: 'message', + // The envelope it actually arrived in; `forwardedBy` says who put it there. + envelopeFrom: parent.envelopeFrom, + fromAddress: headers.fromAddress, + fromName: headers.fromName, + toAddress: parent.toAddress, + subaddress: parent.subaddress, + subject: headers.subject, + messageIdHeader: headers.messageIdHeader, + sentAt: headers.sentAt, + receivedAt: parent.receivedAt, + rawKey, + rawSize: part.bytes.byteLength, + // The forward's verdict: the embedded message's own signatures cannot be verified. + auth: parent.auth, + parentMessageId: parent.id, + forwardedBy: parent.forwardedBy ?? parent.envelopeFrom, + forwardDepth: parent.forwardDepth + 1, + sourceSha256: part.sha256, + }) + // A concurrent run of the same parent (the part key) or an earlier copy of the same email (the + // Message-ID key) won: the unique indexes decide, and this run links to the winner. + .onConflictDoNothing() + .returning({ id: inboxMessages.id }) + if (inserted.length === 0) { + await storage.delete(rawKey) + return (await resolveExisting(headers)) ?? { status: 'skipped' } + } + + await ctx.enqueue({ + type: INBOX_PARSE_JOB, + payload: { tenantId: parent.tenantId, messageId: id }, + }) + await ctx.nudge(parent.tenantId, INBOX_MESSAGES_ENTITY, id) + return { status: 'created', messageId: id } +} diff --git a/plugins/inbox/apps/web/src/plugins/inbox/services/knowledge.ts b/plugins/inbox/apps/web/src/plugins/inbox/services/knowledge.ts index 3078069..fdf3447 100644 --- a/plugins/inbox/apps/web/src/plugins/inbox/services/knowledge.ts +++ b/plugins/inbox/apps/web/src/plugins/inbox/services/knowledge.ts @@ -54,7 +54,7 @@ export function messageDocumentText(row: InboxMessageRow): string { const from = row.fromName ? `${row.fromName} <${row.fromAddress}>` : (row.fromAddress ?? '') return [ `From: ${from}`, - `Forwarded by: ${row.envelopeFrom}`, + `Forwarded by: ${row.forwardedBy ?? row.envelopeFrom}`, row.sentAt ? `Date: ${row.sentAt.toISOString().slice(0, 10)}` : null, `Subject: ${row.subject ?? '(no subject)'}`, '', diff --git a/plugins/inbox/apps/web/src/plugins/inbox/services/parse.ts b/plugins/inbox/apps/web/src/plugins/inbox/services/parse.ts index ccb40da..8ca8eee 100644 --- a/plugins/inbox/apps/web/src/plugins/inbox/services/parse.ts +++ b/plugins/inbox/apps/web/src/plugins/inbox/services/parse.ts @@ -3,9 +3,16 @@ * every consumer plugin. Safe to re-run (the reprocess route does exactly that): attachments are * unique per (message, sha256), the row is overwritten, and consumers are required to be * idempotent too. + * + * Every `message/rfc822` part is an attachment here (`forceRfc822Attachments`) — never merged into + * this message's body — and is then split into a child message of its own (`./forwarded.ts`). Its + * `.eml` row stays, pointing at the child through `expanded_message_id`, and is left out of what + * consumers and the knowledge base are given: the child is where that email is processed. */ import { INBOX_EXTENSION_KEYS, + INBOX_FORWARD_MAX_CHILDREN, + INBOX_FORWARD_MAX_DEPTH, INBOX_MESSAGES_ENTITY, INBOX_TEXT_BODY_MAX, type InboxConfirmation, @@ -16,6 +23,7 @@ import { z } from 'zod' import type { JobCtx } from '@/plugins/api' import { extensions } from '@/plugins/api/peers' import { type InboxMessageRow, inboxAttachments, inboxMessages, inboxSettings } from '../db/schema' +import { EMBEDDED_MESSAGE_TYPE, type EmbeddedPart, expandEmbeddedMessage } from './forwarded' import { ingestMessage } from './knowledge' import { messagePrefix } from './receive' import { @@ -50,7 +58,20 @@ export interface InboxParsedMessage { receivedAt: Date textBody: string | null hasHtmlBody: boolean + /** + * The ordinary attachments. An embedded email (`.eml`) that was split into a message of its own + * is NOT here — it reaches consumers as that message, once. + */ attachments: readonly InboxAttachmentRef[] + /** + * Set when this message arrived as a `message/rfc822` part of another ("Forward as + * attachment"): the wrapper's id. Its authentication verdict is the wrapper's. + */ + parentMessageId: string | null + /** Who forwarded it, for a message with a `parentMessageId`; null otherwise. */ + forwardedBy: string | null + /** The messages split out of this one — each is delivered to consumers on its own. */ + forwardedMessageIds: readonly string[] } /** @@ -99,7 +120,8 @@ function toArrayBuffer(content: ArrayBuffer | Uint8Array | string): ArrayBuffer export function toParsedMessage( row: InboxMessageRow, - attachments: readonly InboxAttachmentRef[] + attachments: readonly InboxAttachmentRef[], + forwardedMessageIds: readonly string[] = [] ): InboxParsedMessage { return { tenantId: row.tenantId, @@ -115,6 +137,9 @@ export function toParsedMessage( textBody: row.textBody, hasHtmlBody: row.htmlKey !== null, attachments, + parentMessageId: row.parentMessageId, + forwardedBy: row.forwardedBy, + forwardedMessageIds, } } @@ -145,7 +170,9 @@ export async function parseMessage( return } const raw = await new Response(object.body).arrayBuffer() - const email = await PostalMime.parse(raw) + // Every embedded message is a part of its own, with or without a Content-Disposition: inlined, + // a forwarded receipt would be merged into this body and never seen on its own. + const email = await PostalMime.parse(raw, { forceRfc822Attachments: true }) const prefix = messagePrefix(tenantId, messageId) let htmlKey: string | null = null @@ -155,13 +182,22 @@ export async function parseMessage( } const kept: InboxAttachmentRef[] = [] + const embedded: (EmbeddedPart & { ref: InboxAttachmentRef })[] = [] + const failures: string[] = [] + // Confirmations are never split, and a message at the depth cap keeps its parts as attachments. + const mayExpand = row.kind === 'message' && row.forwardDepth < INBOX_FORWARD_MAX_DEPTH + let overflow = 0 for (const [index, part] of (email.attachments ?? []).entries()) { const bytes = toArrayBuffer(part.content) const contentType = (part.mimeType || 'application/octet-stream').toLowerCase() const inline = part.disposition === 'inline' || Boolean(part.related) - if (!keepPart({ inline, contentType, size: bytes.byteLength })) continue + const isEmbedded = contentType === EMBEDDED_MESSAGE_TYPE + if (!isEmbedded && !keepPart({ inline, contentType, size: bytes.byteLength })) continue const sha256 = await sha256Hex(bytes) - const filename = safeFilename(part.filename, `attachment-${index + 1}`) + const filename = safeFilename( + part.filename, + isEmbedded ? `forwarded-${index + 1}.eml` : `attachment-${index + 1}` + ) const attachmentId = crypto.randomUUID() const storageKey = `${prefix}/attachments/${sha256.slice(0, 16)}-${filename}` await storage.put(storageKey, bytes, { contentType }) @@ -185,16 +221,44 @@ export async function parseMessage( set: { filename, contentType, storageKey, inline }, }) .returning() - if (saved) { - kept.push({ - id: saved.id, - filename: saved.filename, - contentType: saved.contentType, - size: saved.size, - sha256: saved.sha256, - inline: saved.inline, - }) + if (!saved) continue + const ref: InboxAttachmentRef = { + id: saved.id, + filename: saved.filename, + contentType: saved.contentType, + size: saved.size, + sha256: saved.sha256, + inline: saved.inline, + } + if (isEmbedded && mayExpand) { + if (embedded.length < INBOX_FORWARD_MAX_CHILDREN) { + embedded.push({ bytes, sha256, ref }) + continue + } + overflow += 1 + } + kept.push(ref) + } + + // Split each embedded email into a message of its own, and point its `.eml` row at it. + const forwardedMessageIds: string[] = [] + for (const part of embedded) { + const outcome = await expandEmbeddedMessage(ctx, row, part) + if (outcome.status === 'skipped') { + kept.push(part.ref) + continue } + forwardedMessageIds.push(outcome.messageId) + await ctx.db + .update(inboxAttachments) + .set({ expandedMessageId: outcome.messageId }) + .where(and(eq(inboxAttachments.tenantId, tenantId), eq(inboxAttachments.id, part.ref.id))) + } + if (overflow > 0) { + ctx.logger.warn({ messageId, overflow }, 'inbox.parse: too many forwarded emails to split') + failures.push( + `${overflow} forwarded email(s) beyond the first ${INBOX_FORWARD_MAX_CHILDREN} were kept as attachments` + ) } const text = email.text ?? null @@ -226,6 +290,7 @@ export async function parseMessage( htmlKey, confirmation, attachmentCount: kept.length, + forwardedCount: forwardedMessageIds.length, error: null, parsedAt: new Date(), }) @@ -233,8 +298,7 @@ export async function parseMessage( .returning() if (updated && updated.kind === 'message') { - const failures: string[] = [] - const parsed = toParsedMessage(updated, kept) + const parsed = toParsedMessage(updated, kept, forwardedMessageIds) for (const consumer of consumers) { try { await consumer.onMessage(parsed, ctx) @@ -251,12 +315,12 @@ export async function parseMessage( if (settings?.ingestToKnowledge ?? true) { failures.push(...(await ingestMessage(ctx, updated, kept, settings))) } - if (failures.length > 0) { - await ctx.db - .update(inboxMessages) - .set({ error: failures.join('; ').slice(0, 2000) }) - .where(and(eq(inboxMessages.tenantId, tenantId), eq(inboxMessages.id, messageId))) - } + } + if (failures.length > 0) { + await ctx.db + .update(inboxMessages) + .set({ error: failures.join('; ').slice(0, 2000) }) + .where(and(eq(inboxMessages.tenantId, tenantId), eq(inboxMessages.id, messageId))) } await ctx.nudge(tenantId, INBOX_MESSAGES_ENTITY, messageId) } diff --git a/plugins/inbox/apps/web/src/plugins/inbox/tests/api/forwarded.test.ts b/plugins/inbox/apps/web/src/plugins/inbox/tests/api/forwarded.test.ts new file mode 100644 index 0000000..aa30192 --- /dev/null +++ b/plugins/inbox/apps/web/src/plugins/inbox/tests/api/forwarded.test.ts @@ -0,0 +1,490 @@ +/** + * Forwarded-as-attachment mail on real Postgres and the in-memory R2/queue: every + * `message/rfc822` part — with `Content-Disposition: attachment` (Gmail's "Forward as + * attachment") or without one (which postal-mime would otherwise merge into the body) — becomes a + * child message with the embedded email's own sender and subject, the parent's verdict and + * `forwardedBy`, and its own parse job. Redelivery creates nothing twice, an embedded Message-ID + * the organisation already has is linked rather than stored again, nesting stops at the depth + * cap, consumers see each embedded email exactly once, and no organisation sees another's. + */ + +import { + INBOX_FORWARD_MAX_CHILDREN, + INBOX_FORWARD_MAX_DEPTH, + type InboxMessage, +} from '@rocketflare/shared/plugins/inbox/index' +import { + createTestEnv, + createTestSession, + createTestTenant, + createTestUser, + json, + linkUserToTenant, + request, + sessionCookieHeader, + setupTestDatabase, + stubs, +} from '@testkit/integration' +import { makeEmailCtx, makeJobCtx } from '@testkit/unit' +import { and, eq } from 'drizzle-orm' +import { beforeAll, describe, expect, it, vi } from 'vitest' +import { inboxAttachments, inboxMessages } from '../../db/schema' +import type { InboxConsumer, InboxParsedMessage } from '../../index' +import { parseMessage } from '../../services/parse' +import { receiveMessage } from '../../services/receive' + +const db = setupTestDatabase() +const BASE = '/api/inbox' +const ORIGIN = { Origin: 'http://localhost:3000' } +const AUTH_PASS = 'mx.cloudflare.net; spf=pass smtp.mailfrom=me@gmail.com; dkim=pass; dmarc=pass' + +let tenantId: string +let otherTenantId: string +let localPart: string +let otherLocalPart: string +let memberCookie: Record +let otherCookie: Record + +/** One receipt as a complete email: a text body and a PDF. */ +function receipt(opts: { from: string; subject: string; messageId?: string | null }): string { + const boundary = `r_${crypto.randomUUID().slice(0, 8)}` + return [ + `From: ${opts.from}`, + 'To: me@gmail.com', + `Subject: ${opts.subject}`, + ...(opts.messageId === null + ? [] + : [`Message-ID: ${opts.messageId ?? `<${crypto.randomUUID()}@vendor.test>`}`]), + 'Date: Tue, 21 Oct 2026 09:30:00 +0000', + 'MIME-Version: 1.0', + `Content-Type: multipart/mixed; boundary="${boundary}"`, + '', + `--${boundary}`, + 'Content-Type: text/plain; charset=utf-8', + '', + `Receipt: ${opts.subject}`, + `--${boundary}`, + 'Content-Type: application/pdf; name="receipt.pdf"', + 'Content-Disposition: attachment; filename="receipt.pdf"', + 'Content-Transfer-Encoding: base64', + '', + Buffer.from(`%PDF-1.4 ${opts.subject}`).toString('base64'), + `--${boundary}--`, + '', + ].join('\r\n') +} + +/** + * A forward wrapping `parts` as `message/rfc822`. `disposition: 'attachment'` is Gmail's "Forward + * as attachment"; `null` is the bare part a mail client sends when it inlines the original. + */ +function forward( + parts: { raw: string; disposition: 'attachment' | null; name?: string }[], + opts: { subject?: string; messageId?: string; body?: string } = {} +): string { + const boundary = `f_${crypto.randomUUID().slice(0, 8)}` + const lines = [ + 'From: Me ', + 'To: receipts@example.test', + `Subject: ${opts.subject ?? 'Fwd: receipts'}`, + `Message-ID: ${opts.messageId ?? `<${crypto.randomUUID()}@gmail.com>`}`, + 'Date: Thu, 23 Oct 2026 10:00:00 +0000', + 'MIME-Version: 1.0', + `Content-Type: multipart/mixed; boundary="${boundary}"`, + '', + `--${boundary}`, + 'Content-Type: text/plain; charset=utf-8', + '', + opts.body ?? 'Two receipts for the books.', + ] + for (const [i, part] of parts.entries()) { + lines.push(`--${boundary}`, 'Content-Type: message/rfc822') + if (part.disposition) { + lines.push( + `Content-Disposition: attachment; filename="${part.name ?? `forwarded-${i + 1}.eml`}"` + ) + } + lines.push('', part.raw) + } + lines.push(`--${boundary}--`, '') + return lines.join('\r\n') +} + +/** The slice of `ForwardableEmailMessage` the plugin reads, built from a raw MIME string. */ +function message(raw: string, envelope: { from: string; to: string }) { + const headerBlock = raw.split('\r\n\r\n')[0] ?? '' + const headers = new Headers() + for (const line of headerBlock.split('\r\n')) { + const idx = line.indexOf(':') + if (idx > 0) headers.append(line.slice(0, idx).trim(), line.slice(idx + 1).trim()) + } + headers.set('Authentication-Results', AUTH_PASS) + return { + from: envelope.from, + to: envelope.to, + headers, + raw: new Response(raw).body as ReadableStream, + rawSize: raw.length, + canBeForwarded: true, + setReject: vi.fn(), + forward: vi.fn(), + reply: vi.fn(), + } as unknown as ForwardableEmailMessage +} + +type Env = ReturnType + +/** The `inbox.parse` jobs queued on `env` so far, as (tenant, message) pairs. */ +function parseJobs(env: Env): { tenantId: string; messageId: string }[] { + return stubs(env) + .queue.messages.map( + m => m.body as { type: string; payload: { tenantId: string; messageId: string } } + ) + .filter(b => b.type === 'inbox.parse') + .map(b => b.payload) +} + +/** + * Run every queued parse job from `from` on, including the ones those runs queue, as the queue + * consumer would. Returns the index to resume from. + */ +async function drain(env: Env, consumers: InboxConsumer[], from = 0): Promise { + let next = from + while (next < parseJobs(env).length) { + const job = parseJobs(env)[next] as { tenantId: string; messageId: string } + next += 1 + await parseMessage(makeJobCtx({ db, env }), job.tenantId, job.messageId, consumers) + } + return next +} + +async function receive(env: Env, raw: string, to = `${localPart}@example.test`): Promise { + const outcome = await receiveMessage( + message(raw, { from: 'me@gmail.com', to }), + makeEmailCtx({ db, env }) + ) + if (outcome.status !== 'stored') throw new Error(`not stored: ${outcome.status}`) + return outcome.messageId +} + +async function childrenOf(parentId: string, tenant = tenantId) { + return db + .select() + .from(inboxMessages) + .where(and(eq(inboxMessages.tenantId, tenant), eq(inboxMessages.parentMessageId, parentId))) +} + +async function row(id: string, tenant = tenantId) { + const [found] = await db + .select() + .from(inboxMessages) + .where(and(eq(inboxMessages.tenantId, tenant), eq(inboxMessages.id, id))) + return found +} + +function recorder() { + const seen: InboxParsedMessage[] = [] + const consumer: InboxConsumer = { id: 'test', onMessage: async m => void seen.push(m) } + return { seen, consumer } +} + +beforeAll(async () => { + tenantId = (await createTestTenant(db)).id + otherTenantId = (await createTestTenant(db)).id + localPart = `fwd-${crypto.randomUUID().slice(0, 8)}` + otherLocalPart = `fwd-${crypto.randomUUID().slice(0, 8)}` + + const owner = await createTestUser(db) + await linkUserToTenant(db, owner.id, tenantId, 'owner') + const ownerCookie = { + ...sessionCookieHeader(await createTestSession(db, owner.id, tenantId)), + ...ORIGIN, + } + const member = await createTestUser(db) + await linkUserToTenant(db, member.id, tenantId, 'member') + memberCookie = { + ...sessionCookieHeader(await createTestSession(db, member.id, tenantId)), + ...ORIGIN, + } + const outsider = await createTestUser(db) + await linkUserToTenant(db, outsider.id, otherTenantId, 'owner') + otherCookie = { + ...sessionCookieHeader(await createTestSession(db, outsider.id, otherTenantId)), + ...ORIGIN, + } + + for (const [cookie, part] of [ + [ownerCookie, localPart], + [otherCookie, otherLocalPart], + ] as const) { + const res = await request( + `${BASE}/settings`, + { method: 'PUT', headers: cookie }, + // Knowledge off: these tests are about splitting, and ingest needs no embeddings here. + { + json: { + enabled: true, + localPart: part, + allowedSenders: ['me@gmail.com'], + ingestToKnowledge: false, + }, + } + ) + expect(res.status).toBe(200) + } +}) + +describe('forwarded as attachment', () => { + it('splits each attached message/rfc822 into a child with its own sender and subject', async () => { + const env = createTestEnv() + const parentId = await receive( + env, + forward([ + { + raw: receipt({ from: 'Stripe ', subject: 'Your receipt #1001' }), + disposition: 'attachment', + }, + { + raw: receipt({ + from: 'AWS Billing ', + subject: 'AWS invoice October', + }), + disposition: 'attachment', + }, + ]) + ) + const { seen, consumer } = recorder() + await drain(env, [consumer]) + + const parent = await row(parentId) + expect(parent).toMatchObject({ status: 'parsed', forwardedCount: 2, attachmentCount: 0 }) + + const children = await childrenOf(parentId) + expect(children.map(c => [c.fromAddress, c.fromName, c.subject]).sort()).toEqual([ + ['billing@aws.example', 'AWS Billing', 'AWS invoice October'], + ['receipts@stripe.com', 'Stripe', 'Your receipt #1001'], + ]) + for (const child of children) { + expect(child).toMatchObject({ + tenantId, + status: 'parsed', + kind: 'message', + forwardedBy: 'me@gmail.com', + forwardDepth: 1, + attachmentCount: 1, + auth: parent?.auth, + }) + expect(child.messageIdHeader).toMatch(/@vendor\.test>$/) + expect(child.rawKey).toBe(`tenants/${tenantId}/inbox/${child.id}/raw.eml`) + expect(await stubs(env).files.get(child.rawKey)).not.toBeNull() + expect(child.textBody).toContain(`Receipt: ${child.subject}`) + } + + // The parent keeps the .eml rows, each pointing at the message it became. + const emls = await db + .select() + .from(inboxAttachments) + .where(and(eq(inboxAttachments.tenantId, tenantId), eq(inboxAttachments.messageId, parentId))) + expect(emls.map(a => a.expandedMessageId).sort()).toEqual(children.map(c => c.id).sort()) + expect(emls.every(a => a.contentType === 'message/rfc822')).toBe(true) + + // Consumers: the wrapper once (with no .eml in its attachments), each receipt once, with its PDF. + const parentCall = seen.filter(m => m.messageId === parentId) + expect(parentCall).toHaveLength(1) + expect(parentCall[0]?.attachments).toEqual([]) + expect([...(parentCall[0]?.forwardedMessageIds ?? [])].sort()).toEqual( + children.map(c => c.id).sort() + ) + for (const child of children) { + const calls = seen.filter(m => m.messageId === child.id) + expect(calls).toHaveLength(1) + expect(calls[0]).toMatchObject({ parentMessageId: parentId, forwardedBy: 'me@gmail.com' }) + expect(calls[0]?.attachments.map(a => a.filename)).toEqual(['receipt.pdf']) + } + + // The API shows the relationship both ways. + const parentRes = await json( + await request(`${BASE}/messages/${parentId}`, { headers: memberCookie }, { env }) + ) + expect(parentRes.forwardedCount).toBe(2) + expect(parentRes.forwardedMessages.map(f => f.subject).sort()).toEqual([ + 'AWS invoice October', + 'Your receipt #1001', + ]) + const childId = children[0]?.id ?? '' + const childRes = await json( + await request(`${BASE}/messages/${childId}`, { headers: memberCookie }, { env }) + ) + expect(childRes).toMatchObject({ + parentMessageId: parentId, + forwardedBy: 'me@gmail.com', + parent: { id: parentId, subject: 'Fwd: receipts' }, + }) + }) + + it('splits a message/rfc822 part with no disposition instead of merging it into the body', async () => { + const env = createTestEnv() + const parentId = await receive( + env, + forward( + [ + { + raw: receipt({ from: 'shop@vendor.test', subject: 'Order 77 shipped' }), + disposition: null, + }, + ], + { + body: 'See below.', + } + ) + ) + await drain(env, []) + const parent = await row(parentId) + expect(parent?.textBody).toContain('See below.') + expect(parent?.textBody).not.toContain('Order 77') + const children = await childrenOf(parentId) + expect(children.map(c => [c.fromAddress, c.subject])).toEqual([ + ['shop@vendor.test', 'Order 77 shipped'], + ]) + expect(children[0]?.textBody).toContain('Receipt: Order 77 shipped') + }) + + it('is idempotent across redelivery: no duplicate children, consumers called once per child', async () => { + const env = createTestEnv() + const parentId = await receive( + env, + forward([ + { + raw: receipt({ from: 'a@vendor.test', subject: 'Redelivery A' }), + disposition: 'attachment', + }, + // No Message-ID: only the (parent, part) key can recognise it again. + { + raw: receipt({ from: 'b@vendor.test', subject: 'Redelivery B', messageId: null }), + disposition: 'attachment', + }, + ]) + ) + const { seen, consumer } = recorder() + let cursor = await drain(env, [consumer]) + const first = await childrenOf(parentId) + expect(first).toHaveLength(2) + + // The parent's parse job delivered again (and reprocessed by hand), concurrently even. + await Promise.all([ + parseMessage(makeJobCtx({ db, env }), tenantId, parentId, [consumer]), + parseMessage(makeJobCtx({ db, env }), tenantId, parentId, [consumer]), + ]) + cursor = await drain(env, [consumer], cursor) + + const again = await childrenOf(parentId) + expect(again.map(c => c.id).sort()).toEqual(first.map(c => c.id).sort()) + expect(parseJobs(env).filter(j => first.some(c => c.id === j.messageId))).toHaveLength(2) + for (const child of first) { + expect(seen.filter(m => m.messageId === child.id)).toHaveLength(1) + } + expect((await row(parentId))?.forwardedCount).toBe(2) + // Only one raw object per child survives the race. + const objects = await stubs(env).files.list({ prefix: `tenants/${tenantId}/inbox/` }) + const raws = objects.objects.filter(o => o.key.endsWith('/raw.eml')) + expect(raws).toHaveLength(3) + }) + + it('links an embedded email the organisation already received instead of storing it again', async () => { + const env = createTestEnv() + const messageId = `` + const original = receipt({ from: 'once@vendor.test', subject: 'Seen before', messageId }) + const directId = await receive(env, original) + const parentId = await receive(env, forward([{ raw: original, disposition: 'attachment' }])) + await drain(env, []) + + expect(await childrenOf(parentId)).toEqual([]) + const [eml] = await db + .select() + .from(inboxAttachments) + .where(and(eq(inboxAttachments.tenantId, tenantId), eq(inboxAttachments.messageId, parentId))) + expect(eml?.expandedMessageId).toBe(directId) + expect(await row(parentId)).toMatchObject({ forwardedCount: 1, attachmentCount: 0 }) + }) + + it(`stops splitting at depth ${INBOX_FORWARD_MAX_DEPTH}`, async () => { + const env = createTestEnv() + let raw = receipt({ from: 'deep@vendor.test', subject: 'Level 5' }) + for (let level = 4; level >= 1; level -= 1) { + raw = forward([{ raw, disposition: 'attachment' }], { subject: `Level ${level}` }) + } + const topId = await receive( + env, + forward([{ raw, disposition: 'attachment' }], { subject: 'Level 0' }) + ) + await drain(env, []) + + const chain: NonNullable>>[] = [] + let current = await row(topId) + while (current) { + chain.push(current) + const [child] = await childrenOf(current.id) + current = child + } + expect(chain.map(m => m.forwardDepth)).toEqual([0, 1, 2, 3]) + expect(chain.map(m => m.subject)).toEqual(['Level 0', 'Level 1', 'Level 2', 'Level 3']) + const deepest = chain[3] + // At the cap the embedded message stays an ordinary .eml attachment. + expect(deepest).toMatchObject({ forwardedCount: 0, attachmentCount: 1 }) + const [eml] = await db + .select() + .from(inboxAttachments) + .where( + and( + eq(inboxAttachments.tenantId, tenantId), + eq(inboxAttachments.messageId, deepest?.id ?? '') + ) + ) + expect(eml).toMatchObject({ contentType: 'message/rfc822', expandedMessageId: null }) + }) + + it(`splits at most ${INBOX_FORWARD_MAX_CHILDREN} emails out of one message`, async () => { + const env = createTestEnv() + const parts = Array.from({ length: INBOX_FORWARD_MAX_CHILDREN + 2 }, (_, i) => ({ + raw: receipt({ from: `v${i}@vendor.test`, subject: `Bulk ${i}` }), + disposition: 'attachment' as const, + })) + const parentId = await receive(env, forward(parts)) + await parseMessage(makeJobCtx({ db, env }), tenantId, parentId, []) + expect(await childrenOf(parentId)).toHaveLength(INBOX_FORWARD_MAX_CHILDREN) + const parent = await row(parentId) + expect(parent).toMatchObject({ forwardedCount: INBOX_FORWARD_MAX_CHILDREN, attachmentCount: 2 }) + expect(parent?.error).toContain('2 forwarded email(s)') + }) + + it('keeps every organisation to its own children, even for the same embedded email', async () => { + const env = createTestEnv() + const shared = receipt({ + from: 'both@vendor.test', + subject: 'Sent to two workspaces', + messageId: ``, + }) + const mineId = await receive(env, forward([{ raw: shared, disposition: 'attachment' }])) + const theirsId = await receive( + env, + forward([{ raw: shared, disposition: 'attachment' }]), + `${otherLocalPart}@example.test` + ) + await drain(env, []) + + const [mine] = await childrenOf(mineId, tenantId) + const [theirs] = await childrenOf(theirsId, otherTenantId) + expect(mine?.tenantId).toBe(tenantId) + expect(theirs?.tenantId).toBe(otherTenantId) + expect(mine?.id).not.toBe(theirs?.id) + expect(theirs?.rawKey.startsWith(`tenants/${otherTenantId}/inbox/`)).toBe(true) + + for (const path of [`${BASE}/messages/${mine?.id}`, `${BASE}/messages/${mineId}`]) { + expect((await request(path, { headers: otherCookie }, { env })).status).toBe(404) + } + const theirView = await json( + await request(`${BASE}/messages/${theirsId}`, { headers: otherCookie }, { env }) + ) + expect(theirView.forwardedMessages.map(f => f.id)).toEqual([theirs?.id]) + }) +}) diff --git a/plugins/inbox/apps/web/src/plugins/inbox/tools/inbox-tools.ts b/plugins/inbox/apps/web/src/plugins/inbox/tools/inbox-tools.ts index ac4defe..9256b1f 100644 --- a/plugins/inbox/apps/web/src/plugins/inbox/tools/inbox-tools.ts +++ b/plugins/inbox/apps/web/src/plugins/inbox/tools/inbox-tools.ts @@ -77,6 +77,8 @@ export function listInboxMessagesTool(ctx: ToolCtx): Tool 0 && { forwardedEmails: r.forwardedCount }), })), }) }, @@ -128,7 +130,11 @@ export function getInboxMessageTool(ctx: ToolCtx): Tool { messageId: row.id, from: row.fromAddress, fromName: row.fromName, - forwardedBy: row.envelopeFrom, + forwardedBy: row.forwardedBy ?? row.envelopeFrom, + ...(row.parentMessageId && { + forwardedInside: row.parentMessageId, + context: 'Forwarded as an attachment of forwardedInside; read that message for context.', + }), subject: row.subject, sentAt: row.sentAt?.toISOString() ?? null, receivedAt: row.receivedAt.toISOString(), @@ -140,12 +146,18 @@ export function getInboxMessageTool(ctx: ToolCtx): Tool { ...(row.status !== 'parsed' && { note: `This message is ${row.status}; the body may be empty.`, }), - attachments: attachments.map(a => ({ - attachmentId: a.id, - filename: a.filename, - contentType: a.contentType, - size: a.size, - })), + attachments: attachments + .filter(a => !a.expandedMessageId) + .map(a => ({ + attachmentId: a.id, + filename: a.filename, + contentType: a.contentType, + size: a.size, + })), + // Embedded emails split into messages of their own — read each with this tool. + forwardedMessageIds: attachments.flatMap(a => + a.expandedMessageId ? [a.expandedMessageId] : [] + ), }) }, }) diff --git a/plugins/inbox/apps/web/src/plugins/inbox/ui/pages/InboxPage.tsx b/plugins/inbox/apps/web/src/plugins/inbox/ui/pages/InboxPage.tsx index 6bb2a9b..4e7bd26 100644 --- a/plugins/inbox/apps/web/src/plugins/inbox/ui/pages/InboxPage.tsx +++ b/plugins/inbox/apps/web/src/plugins/inbox/ui/pages/InboxPage.tsx @@ -1,13 +1,19 @@ /** * Inbox: what arrived, newest first, and one message beside it — sender, forwarding mailbox, * authentication verdicts, the text body and the attachments. A forwarding confirmation shows its - * code and link at the top, because that is the only reason anybody opens one. + * code and link at the top, because that is the only reason anybody opens one. A message + * forwarded as an attachment links to the wrapper it came in ("Forwarded inside …"), and the + * wrapper lists the emails split out of it ("Contains N forwarded emails"). * * Bodies are shown as TEXT only. The HTML body and the raw message are downloads (served as * attachments by the API), never rendered here: they are markup a stranger wrote. */ -import { EnvelopeIcon, PaperClipIcon } from '@heroicons/react/24/outline' -import type { InboxMessage, InboxMessageSummary } from '@rocketflare/shared/plugins/inbox/index' +import { ArrowUturnRightIcon, EnvelopeIcon, PaperClipIcon } from '@heroicons/react/24/outline' +import type { + InboxMessage, + InboxMessageLink, + InboxMessageSummary, +} from '@rocketflare/shared/plugins/inbox/index' import { useState } from 'react' import { ConfirmModal, @@ -85,7 +91,18 @@ export default function InboxPage() {
+ {m.parentMessageId && ( + + )} {m.subject || '(no subject)'} + {m.forwardedCount > 0 && ( + + {m.forwardedCount} forwarded + + )} {m.attachmentCount > 0 && ( @@ -106,7 +123,7 @@ export default function InboxPage() { )} {selected ? ( - setSelected(null)} /> + setSelected(null)} /> ) : (

Choose a message to read it.

@@ -117,7 +134,15 @@ export default function InboxPage() { ) } -function MessageDetail({ id, onDeleted }: { id: string; onDeleted: () => void }) { +function MessageDetail({ + id, + onSelect, + onDeleted, +}: { + id: string + onSelect: (id: string) => void + onDeleted: () => void +}) { const message = useInboxMessage(id) if (message.isPending) { return ( @@ -133,7 +158,7 @@ function MessageDetail({ id, onDeleted }: { id: string; onDeleted: () => void }) ) } - return + return } function Verdict({ label, value }: { label: string; value: string | null }) { @@ -141,7 +166,21 @@ function Verdict({ label, value }: { label: string; value: string | null }) { return {`${label} ${value ?? '—'}`} } -function MessageBody({ message, onDeleted }: { message: InboxMessage; onDeleted: () => void }) { +function linkLabel(link: InboxMessageLink): string { + const from = link.fromName || link.fromAddress + return `${link.subject || '(no subject)'}${from ? ` — ${from}` : ''}` +} + +function MessageBody({ + message, + onSelect, + onDeleted, +}: { + message: InboxMessage + onSelect: (id: string) => void + onDeleted: () => void +}) { + const attachments = message.attachments.filter(a => !a.expandedMessageId) const { can } = usePermissions() const reprocess = useReprocessInboxMessage() const remove = useDeleteInboxMessage() @@ -210,13 +249,26 @@ function MessageBody({ message, onDeleted }: { message: InboxMessage; onDeleted:
)} + {message.parent && ( +

+ + Forwarded inside{' '} + +

+ )}
From
{message.fromName ? `${message.fromName} <${message.fromAddress}>` : message.fromAddress}
Forwarded by
-
{message.envelopeFrom}
+
{message.forwardedBy ?? message.envelopeFrom}
To
{message.toAddress}
Sent
@@ -224,16 +276,37 @@ function MessageBody({ message, onDeleted }: { message: InboxMessage; onDeleted:
Received
{formatDateTime(message.receivedAt)}
Checks
-
+
+ {message.parentMessageId && ( + (of the forward it came in) + )}
+ {message.forwardedMessages.length > 0 && ( +
+

+ Contains {message.forwardedMessages.length} forwarded email + {message.forwardedMessages.length === 1 ? '' : 's'} +

+
    + {message.forwardedMessages.map(f => ( +
  • + + +
    • + ))} +
+
+ )} {message.error &&

{message.error}

} - {message.attachments.length > 0 && ( + {attachments.length > 0 && (