From 0e98b7fd641338dedb67be6b50b27db23f732658 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 14 Oct 2026 11:33:45 +0000 Subject: [PATCH 1/2] wip(service-mailbox): store an ingested message's attachments and forwarded parts through file-attachments Claude-Session: https://claude.ai/code/session_01YRp5z4kK6vgYkrdH97ztsy Co-authored-by: Claude --- .../services/service-mailbox/package.json | 1 + .../src/ingest/mail-ingest-service.test.ts | 237 +++++++++++++++++- .../src/ingest/mail-ingest-service.ts | 160 +++++++++++- .../src/ingest/messages.fixture.ts | 61 +++++ .../src/ingest/parse-message.ts | 80 ++++-- .../src/mailbox-service-plugin.ts | 24 +- .../src/sync/mailbox-sync-service.test.ts | 2 +- .../services/service-mailbox/vitest.config.ts | 9 + pnpm-lock.yaml | 3 + 9 files changed, 549 insertions(+), 28 deletions(-) diff --git a/packages/services/service-mailbox/package.json b/packages/services/service-mailbox/package.json index 2b238fab36d..23223b2634b 100644 --- a/packages/services/service-mailbox/package.json +++ b/packages/services/service-mailbox/package.json @@ -35,6 +35,7 @@ "@objectstack/driver-sql": "workspace:*", "@objectstack/lint": "workspace:*", "@objectstack/objectql": "workspace:*", + "@objectstack/service-storage": "workspace:*", "@types/node": "^26.6.3", "typescript": "^6.0.3", "vitest": "^4.1.11" diff --git a/packages/services/service-mailbox/src/ingest/mail-ingest-service.test.ts b/packages/services/service-mailbox/src/ingest/mail-ingest-service.test.ts index 973564d0d87..82124857ed3 100644 --- a/packages/services/service-mailbox/src/ingest/mail-ingest-service.test.ts +++ b/packages/services/service-mailbox/src/ingest/mail-ingest-service.test.ts @@ -13,12 +13,13 @@ * Every message is fed as a raw RFC 822 string; no IMAP is involved. */ -import { describe, it, expect, beforeAll, afterAll } from 'vitest'; +import { describe, it, expect, beforeAll, beforeEach, afterAll } from 'vitest'; import { ObjectQL } from '@objectstack/objectql'; import { SqlDriver } from '@objectstack/driver-sql'; import { ObjectSchema, Field } from '@objectstack/spec/data'; // Engine options are DECLARED, never erased to `any` (`pnpm check:query-options-erasure`). import type { DataEngineInsertOptions, EngineQueryOptions, EngineUpdateOptions } from '@objectstack/spec/data'; +import type { AttachFileInput, IFileAttachmentService } from '@objectstack/spec/contracts'; import { SysMember, SysUser } from '@objectstack/platform-objects/identity'; import { SysMailAssociation, SysMailbox, SysMailMessage } from '../objects/index.js'; import { MAILBOX_ADMIN_CAPABILITY } from '../mail-read-visibility.js'; @@ -26,10 +27,12 @@ import { MailIngestService, MailboxServiceError, type MailIngestEngine } from '. import { parseMatchAddresses } from './match-addresses.js'; import { appleMailForwardBlock, + binaryBytes, bounce, forwardAsAttachment, forwardInline, gmailForwardBlock, + messageWithParts, outlookForwardBlock, plainMessage, rfc822, @@ -74,6 +77,7 @@ describe('the ingest seam on a real engine', () => { let engine: ObjectQL; let service: MailIngestService; const warnings: string[] = []; + const errors: string[] = []; let mailboxSeq = 0; const find = (object: string, where: Row): Promise => @@ -106,7 +110,7 @@ describe('the ingest seam on a real engine', () => { await engine.syncSchemas(); service = new MailIngestService({ engine: () => engine as unknown as MailIngestEngine, - logger: { warn: (m: string) => warnings.push(m) }, + logger: { warn: (m: string) => warnings.push(m), error: (m: string) => errors.push(m) }, }); }); @@ -218,7 +222,7 @@ describe('the ingest seam on a real engine', () => { }); const racing = new MailIngestService({ engine: () => racingEngine as unknown as MailIngestEngine, - logger: { warn: (m: string) => warnings.push(m) }, + logger: { warn: (m: string) => warnings.push(m), error: (m: string) => errors.push(m) }, }); const second = await racing.ingest({ mailboxId: mailbox, raw }, SYSTEM); @@ -437,7 +441,7 @@ describe('the ingest seam on a real engine', () => { const said: string[] = []; const noAuth = new MailIngestService({ engine: () => withoutIdentity as unknown as MailIngestEngine, - logger: { warn: (m: string) => said.push(m) }, + logger: { warn: (m: string) => said.push(m), error: (m: string) => said.push(m) }, }); for (const id of ['', '']) { const block = gmailForwardBlock({ from: 'customer-noauth@acme.test', cc: null }); @@ -584,6 +588,231 @@ describe('the ingest seam on a real engine', () => { }); }); + describe('attachments: each part is handed to the attach verb as a file of the row', () => { + /** + * The verb as the producer sees it: it records each call, with the state + * of the parent row AT the call (the row must exist first), and refuses + * the file names it is told to. The verb's own behaviour is pinned on the + * real one in `mailbox-attachments-kernel.test.ts`; here what is measured + * is what the seam hands it, and when. + */ + type Call = AttachFileInput & { parentAtCall: Row | undefined }; + let calls: Call[]; + let refuse: Map; + let attaching: MailIngestService; + const said: { warn: string[]; error: string[] } = { warn: [], error: [] }; + const verb: IFileAttachmentService = { + async attachFile(input) { + const [parent] = await find(input.parentObject, { id: input.parentId }); + calls.push({ ...input, parentAtCall: parent }); + const refusal = refuse.get(input.filename); + if (refusal) throw refusal; + return { fileId: `file_${calls.length}`, attachmentId: `att_${calls.length}`, size: input.data.byteLength }; + }, + }; + const refusalOf = (code: string, status: number, message: string) => + Object.assign(new Error(message), { code, status }); + + const ORIGINAL = rfc822( + [ + ['From', 'Customer '], + ['To', 'rep@corp.test'], + ['Subject', 'What does it cost?'], + ['Message-ID', ''], + ], + 'What does it cost?', + ); + const PDF = binaryBytes(40); + /** The card's message: one forwarded message and one file. */ + const forwardWithFile = (messageId: string) => + messageWithParts({ + messageId, + parts: [ + { contentType: 'application/pdf; name="quote.pdf"', disposition: 'attachment; filename="quote.pdf"', bytes: PDF }, + { contentType: 'message/rfc822', disposition: 'attachment; filename="question.eml"', message: ORIGINAL }, + ], + }); + + beforeAll(() => { + attaching = new MailIngestService({ + engine: () => engine as unknown as MailIngestEngine, + logger: { warn: (m: string) => said.warn.push(m), error: (m: string) => said.error.push(m) }, + attachments: () => verb, + }); + }); + beforeEach(() => { + calls = []; + refuse = new Map(); + said.warn.length = 0; + said.error.length = 0; + }); + + it('stores the file and the forwarded message (.eml), in message order, on the row, in the mailbox’s organization', async () => { + const mailbox = await newMailbox(ORG_A); + const result = await attaching.ingest({ mailboxId: mailbox, raw: forwardWithFile('') }, SYSTEM); + + expect(result.outcome).toBe('stored'); + const messageId = (result as { messageId: string }).messageId; + const rows = await messagesIn(mailbox); + expect(rows.map((r) => r.id)).toEqual([messageId]); + expect(calls.map((c) => [c.parentObject, c.parentId, c.organizationId, c.filename, c.mimeType])).toEqual([ + ['sys_mail_message', messageId, ORG_A, 'quote.pdf', 'application/pdf'], + ['sys_mail_message', messageId, ORG_A, 'question.eml', 'message/rfc822'], + ]); + // The bytes as the message carries them: the file's, and the forwarded + // message exactly as it travels in its part. + expect(Buffer.isBuffer(calls[0]!.data)).toBe(true); + expect(new Uint8Array(calls[0]!.data)).toEqual(PDF); + expect(calls[1]!.data.toString('utf8')).toBe(ORIGINAL); + // The parent exists, in the mailbox's organization, before each call. + for (const call of calls) expect(call.parentAtCall).toMatchObject({ id: messageId, organization_id: ORG_A }); + expect([...said.warn, ...said.error]).toEqual([]); + }); + + it('control: a message without parts hands the verb nothing', async () => { + const mailbox = await newMailbox(ORG_A); + const result = await attaching.ingest({ mailboxId: mailbox, raw: plainMessage({ 'Message-ID': '' }) }, SYSTEM); + expect(result.outcome).toBe('stored'); + expect(calls).toEqual([]); + }); + + it('acts in the mailbox’s organization, never null: a mailbox in ORG_B stores its parts in ORG_B', async () => { + const mailbox = await newMailbox(ORG_B); + const result = await attaching.ingest({ mailboxId: mailbox, raw: forwardWithFile('') }, SYSTEM); + expect(calls.map((c) => [c.parentId, c.organizationId])).toEqual([ + [(result as { messageId: string }).messageId, ORG_B], + [(result as { messageId: string }).messageId, ORG_B], + ]); + }); + + it('a re-delivered copy stops at the dedupe: no part is handed over again', async () => { + const mailbox = await newMailbox(ORG_A); + const raw = forwardWithFile(''); + expect((await attaching.ingest({ mailboxId: mailbox, raw }, SYSTEM)).outcome).toBe('stored'); + expect(calls).toHaveLength(2); + + expect((await attaching.ingest({ mailboxId: mailbox, raw }, SYSTEM)).outcome).toBe('duplicate'); + expect(calls).toHaveLength(2); + expect(await messagesIn(mailbox)).toHaveLength(1); + }); + + it('control: the same parts under another Message-ID are another message, and are handed over again', async () => { + const mailbox = await newMailbox(ORG_A); + await attaching.ingest({ mailboxId: mailbox, raw: forwardWithFile('') }, SYSTEM); + await attaching.ingest({ mailboxId: mailbox, raw: forwardWithFile('') }, SYSTEM); + expect(calls).toHaveLength(4); + expect(new Set(calls.map((c) => c.parentId)).size).toBe(2); + }); + + it('a re-match never re-attaches, and a suppressed message attaches nothing', async () => { + const mailbox = await newMailbox(ORG_A); + await attaching.ingest({ mailboxId: mailbox, raw: forwardWithFile('') }, SYSTEM); + await insert('mailtest_contact', { organization_id: ORG_A, name: 'C', email: 'rep@corp.test' }); + expect(await attaching.rematch({ mailboxId: mailbox }, SYSTEM)).toMatchObject({ scanned: 1 }); + expect(calls).toHaveLength(2); + + const suppressed = messageWithParts({ + messageId: '', + headers: [['Auto-Submitted', 'auto-replied']], + parts: [{ contentType: 'application/pdf', disposition: 'attachment; filename="ooo.pdf"', bytes: PDF }], + }); + expect(await attaching.ingest({ mailboxId: mailbox, raw: suppressed }, SYSTEM)).toEqual({ + outcome: 'suppressed', + reason: 'auto_submitted', + }); + expect(calls).toHaveLength(2); + }); + + it('counts every part the parser lists — an inline image, an inline file, a nameless part — and names each', async () => { + const mailbox = await newMailbox(ORG_A); + const long = `${'q'.repeat(300)}.pdf`; + const raw = messageWithParts({ + messageId: '', + parts: [ + // An image the HTML body shows by `cid:` reference. + { contentType: 'image/png', disposition: 'inline; filename="logo.png"', contentId: '', bytes: binaryBytes() }, + // Apple Mail marks the files it attaches `inline`. + { contentType: 'application/pdf; name="apple.pdf"', disposition: 'inline; filename="apple.pdf"', bytes: PDF }, + { contentType: 'application/octet-stream', bytes: binaryBytes(1) }, + { contentType: 'application/pdf', disposition: `attachment; filename="${long}"`, bytes: PDF }, + { contentType: 'application/pdf', disposition: 'attachment; filename="../evil/path.pdf"', bytes: PDF }, + // Forwarded with no file name: named by its own Subject. + { contentType: 'message/rfc822', message: ORIGINAL }, + // A forwarded message whose Subject cannot name it. + { contentType: 'message/global', message: ORIGINAL.replace('Subject: What does it cost?\r\n', '') }, + ], + }); + expect((await attaching.ingest({ mailboxId: mailbox, raw }, SYSTEM)).outcome).toBe('stored'); + const names = calls.map((c) => c.filename); + expect(names).toEqual([ + 'logo.png', + 'apple.pdf', + 'attachment-3', + `${'q'.repeat(251)}.pdf`, + '.._evil_path.pdf', + 'What does it cost?.eml', + 'forwarded.eml', + ]); + expect(calls.map((c) => c.mimeType)).toEqual([ + 'image/png', 'application/pdf', 'application/octet-stream', 'application/pdf', 'application/pdf', + 'message/rfc822', 'message/global', + ]); + expect(names[3]).toHaveLength(255); + }); + + it('a refused part loses neither the message nor the other parts, and is said at error, by name', async () => { + const mailbox = await newMailbox(ORG_A); + refuse.set('quote.pdf', refusalOf('FILES_DISABLED', 403, 'Attachments are not enabled on sys_mail_message.')); + + const result = await attaching.ingest({ mailboxId: mailbox, raw: forwardWithFile('') }, SYSTEM); + + expect(result.outcome).toBe('stored'); + expect(await messagesIn(mailbox)).toHaveLength(1); + expect(calls.map((c) => c.filename)).toEqual(['quote.pdf', 'question.eml']); + expect(said.error).toHaveLength(1); + expect(said.error[0]).toContain("'quote.pdf'"); + expect(said.error[0]).toContain('FILES_DISABLED (403)'); + expect(said.error[0]).toContain('is NOT stored'); + expect(said.warn).toEqual([]); + }); + + it('control: the same message with no refusal says nothing', async () => { + const mailbox = await newMailbox(ORG_A); + await attaching.ingest({ mailboxId: mailbox, raw: forwardWithFile('') }, SYSTEM); + expect(said.error).toEqual([]); + }); + + it('a part the verb throws on without an envelope is said too, and the message is still stored', async () => { + const mailbox = await newMailbox(ORG_A); + refuse.set('question.eml', new Error('storage backend unreachable')); + const result = await attaching.ingest({ mailboxId: mailbox, raw: forwardWithFile('') }, SYSTEM); + expect(result.outcome).toBe('stored'); + expect(said.error).toHaveLength(1); + expect(said.error[0]).toContain("'question.eml'"); + expect(said.error[0]).toContain('storage backend unreachable'); + }); + + it('with no attach verb composed, the message is stored without its parts, and that is said once', async () => { + for (const attachments of [() => undefined, undefined]) { + const quiet: string[] = []; + const loud: string[] = []; + const withoutStorage = new MailIngestService({ + engine: () => engine as unknown as MailIngestEngine, + logger: { warn: (m: string) => quiet.push(m), error: (m: string) => loud.push(m) }, + ...(attachments ? { attachments } : {}), + }); + const mailbox = await newMailbox(ORG_A); + for (const id of ['', '']) { + expect((await withoutStorage.ingest({ mailboxId: mailbox, raw: forwardWithFile(id) }, SYSTEM)).outcome).toBe('stored'); + } + expect(await messagesIn(mailbox)).toHaveLength(2); + expect(quiet.filter((m) => m.includes("'file-attachments' service is composed"))).toHaveLength(1); + expect(loud).toEqual([]); + } + expect(calls).toEqual([]); + }); + }); + describe('who may call', () => { const refusal = async (run: () => Promise) => { try { diff --git a/packages/services/service-mailbox/src/ingest/mail-ingest-service.ts b/packages/services/service-mailbox/src/ingest/mail-ingest-service.ts index ecbf9870d51..f8fc21733a6 100644 --- a/packages/services/service-mailbox/src/ingest/mail-ingest-service.ts +++ b/packages/services/service-mailbox/src/ingest/mail-ingest-service.ts @@ -31,16 +31,40 @@ * organization (D5: forwarding is consent). A block the parser does not * recognise is not guessed at, and a forward from anyone else is not read: * the message is matched on its other participants, which may pool it. + * - **Attachments are stored as files of the row** (the attachment item of + * the inbound-mail record, ADR-0029 D1 "attachments by file reference"). + * Each part the message carries besides its bodies — a file, or a message + * forwarded as an attachment, stored as the `.eml` it is — is handed to the + * server-side attach verb, `IFileAttachmentService.attachFile` (the kernel + * service `file-attachments`), with the message's own `sys_mail_message` + * row as its parent and the mailbox's organization as the organization the + * call acts in, never `null`: the verb, which applies no per-user check, + * then writes the `sys_file` and `sys_attachment` rows behind the same wall + * as the message. No second representation is kept: no column of the row + * names a part, and no byte is stored in a column. * * ## Writes, in order * * 1. the message row, `unassociated`; - * 2. one link per matched record (`matched_by: 'ingest'`); - * 3. `association_state: 'associated'`, once its links exist. + * 2. one file and one attachment per part, in message order — the row exists + * first, because it is the parent; + * 3. one link per matched record (`matched_by: 'ingest'`); + * 4. `association_state: 'associated'`, once its links exist. * - * A failure between them leaves the message in the pool with the links that - * landed — visible to the mailbox's owner and administrators — and a re-match - * completes it. The failure itself reaches the caller. + * Only a message this call stored reaches step 2: a duplicate (stage 1's + * dedupe, or the unique key settling a concurrent pair) and a suppressed + * message stop before it, so a re-delivery adds no attachment, and a re-match + * never reads a part at all. + * + * A failure at steps 1, 3 or 4 leaves the message in the pool with the links + * that landed — visible to the mailbox's owner and administrators — and a + * re-match completes it. The failure itself reaches the caller. A part the + * verb refuses (over the upload size limit, the service not yet serving, any + * gate on the attachment) is not a failure of the message: the message and + * its other parts are stored, and the refused part is reported at `error`, + * once, by name — nothing else will report it, because a re-delivery is a + * duplicate. Where no `file-attachments` service is composed at all, the + * message is stored without its parts, and that is said once per instance. * * Every write is a system write that carries the mailbox's organization * (ADR-0131 D9), and every match is read inside it — the organization-user @@ -56,6 +80,7 @@ * read gate serves (Q4 A). A mailbox in another organization is not found. */ +import { FILE_ATTACHMENT_SERVICE, type IFileAttachmentService } from '@objectstack/spec/contracts'; import { keysetWalk } from '@objectstack/types'; import { MAILBOX_ADMIN_CAPABILITY, @@ -81,6 +106,7 @@ import { import { parseInboundMessage, participantAddresses, + type MailAttachmentPart, type MailParticipantHeaders, type ParsedInboundMessage, type RawMailMessage, @@ -96,9 +122,13 @@ export interface MailIngestEngine extends MailMatchEngine { update(object: string, data: Record, options: Record): Promise; } -/** The logger surface the seam reports through. */ +/** + * The logger surface the seam reports through: `error` for a part of a stored + * message that was not stored (nothing else reports it), `warn` for the rest. + */ export interface MailIngestLogger { warn(message: string): void; + error(message: string): void; } /** @@ -199,6 +229,14 @@ export const REMATCH_LINK_SCAN_LIMIT = 5_000; */ export const SENDER_USER_SCAN_LIMIT = 100; +/** The bounds of the `sys_attachment` columns a part's name and type land in (`file_name`, `mime_type`). */ +const ATTACHMENT_NAME_MAX = 255; +const ATTACHMENT_TYPE_MAX = 128; +/** RFC 2046 §4.5.1: the media type of bytes whose own type cannot be stored. */ +const OCTET_STREAM = 'application/octet-stream'; +/** The extension a forwarded message is stored under: the file it is, an RFC 822 message. */ +const EML = '.eml'; + /** The identity objects the organization-user door reads (registered by the auth plugin). */ const USER_OBJECT = 'sys_user'; const MEMBER_OBJECT = 'sys_member'; @@ -268,6 +306,51 @@ function distinctAddresses(addresses: ReadonlyArray): string[] return [...new Set(addresses.map((a) => a.address))]; } +/** + * A name a sender wrote, made a file name: control characters and path + * separators become `_`, surrounding space goes, and a name over the column's + * bound is cut in its stem so the extension — the storage key keeps it — stays. + * `null` when nothing is left. + */ +function fileNameOf(value: string | null): string | null { + if (value === null) return null; + const clean = value.replace(/[\u0000-\u001f\u007f/\\]/g, '_').trim(); + if (clean === '') return null; + if (clean.length <= ATTACHMENT_NAME_MAX) return clean; + const ext = /\.[^.\s]{1,16}$/.exec(clean)?.[0] ?? ''; + return `${fit(clean.slice(0, clean.length - ext.length), ATTACHMENT_NAME_MAX - ext.length) ?? ''}${ext}`; +} + +/** + * The file name one part is stored under. A file keeps the name the message + * gives it, or is `attachment-` (its position among the message's parts) + * when it has none. A forwarded message is an `.eml` file named by the part's + * own name, or else by the forwarded message's Subject, or else `forwarded`. + */ +function attachmentNameOf(part: MailAttachmentPart, position: number): string { + if (part.forwarded === null) return fileNameOf(part.filename) ?? `attachment-${position}`; + const base = fileNameOf(part.filename) ?? fileNameOf(part.forwarded.subject) ?? 'forwarded'; + return base.toLowerCase().endsWith(EML) ? base : (fileNameOf(`${base}${EML}`) ?? `forwarded${EML}`); +} + +/** The media type a part is stored with: its own, unless the column cannot hold it. */ +function attachmentTypeOf(part: MailAttachmentPart): string { + const type = part.mimeType.trim(); + return type === '' || type.length > ATTACHMENT_TYPE_MAX ? OCTET_STREAM : type; +} + +/** + * A refusal as the log names it: `CODE (status): message`. The verb's + * refusals carry their ADR-0112 `code` and `status`, and each message names its + * cause and what moves it (the upload size limit names its setting). + */ +function describeRefusal(err: unknown): string { + const e = (err ?? {}) as { code?: unknown; status?: unknown; message?: unknown }; + const message = typeof e.message === 'string' && e.message !== '' ? e.message : String(err); + if (typeof e.code !== 'string') return message; + return `${e.code}${typeof e.status === 'number' ? ` (${e.status})` : ''}: ${message}`; +} + /** * The ingest seam and the re-match action. The engine is resolved on each * call, never cached at construction, so a plugin may build this before the @@ -277,6 +360,8 @@ export class MailIngestService { private readonly matcher: MailRecordMatcher; /** Whether the absence of the identity objects the door reads has been said. */ private reportedNoIdentityObjects = false; + /** Whether the absence of the attach verb has been said. */ + private reportedNoAttachmentService = false; constructor( private readonly options: { @@ -285,6 +370,15 @@ export class MailIngestService { logger: MailIngestLogger; /** The ingest clock, for a message with no `receivedAt`. */ now?: () => Date; + /** + * The server-side attach verb — the kernel service `file-attachments` + * (`FILE_ATTACHMENT_SERVICE`), which the storage capability registers — + * resolved per message that carries a part, never cached, so a service + * registered after this seam is constructed is still found. Answers + * `undefined` where none is composed; absent, likewise: such a message is + * stored without its parts, and that is said once. + */ + attachments?: () => IFileAttachmentService | undefined; }, ) { this.matcher = new MailRecordMatcher(options.engine, options.logger); @@ -326,7 +420,8 @@ export class MailIngestService { if (parsed.unreadForwardedParts > 0) { this.options.logger.warn( `[mailbox] a message for mailbox '${mailbox.id}' carries ${parsed.unreadForwardedParts} forwarded part(s) that ` + - 'could not be read; their addresses take no part in matching, and the message is stored without them.', + 'could not be read; their addresses take no part in matching, and the message is matched on its other ' + + 'participants.', ); } @@ -387,6 +482,7 @@ export class MailIngestService { if (messageId === null) { throw new MailboxServiceError('INTERNAL_ERROR', `The engine stored a message for mailbox '${mailbox.id}' but returned no id.`); } + await this.attach(organizationId, mailbox.id, messageId, parsed.attachments); if (links.length === 0) return { outcome: 'stored', messageId, associationState: 'unassociated', associations: [] }; for (const link of links) await this.link(engine, organizationId, messageId, link, 'ingest'); @@ -496,6 +592,56 @@ export class MailIngestService { } } + /** + * Store each part of a message this call stored as a file of its row, in + * message order: parent `sys_mail_message` / `messageId`, in the mailbox's + * organization. A refused part is reported and the next one is stored — + * the message is already stored, and a refusal of one part loses no other. + */ + private async attach( + organizationId: string, + mailboxId: string, + messageId: string, + parts: ReadonlyArray, + ): Promise { + if (parts.length === 0) return; + const verb = this.options.attachments?.(); + if (!verb || typeof verb.attachFile !== 'function') { + if (!this.reportedNoAttachmentService) { + this.reportedNoAttachmentService = true; + this.options.logger.warn( + `[mailbox] a message for mailbox '${mailboxId}' carries ${parts.length} attachment(s), but no ` + + `'${FILE_ATTACHMENT_SERVICE}' service is composed, so no attachment of an ingested message is stored: such ` + + 'messages are stored without their attached files and forwarded messages. Compose the storage capability ' + + `(\`@objectstack/service-storage\`, which registers '${FILE_ATTACHMENT_SERVICE}') to store them.`, + ); + } + return; + } + for (const [index, part] of parts.entries()) { + const filename = attachmentNameOf(part, index + 1); + const mimeType = attachmentTypeOf(part); + try { + await verb.attachFile({ + parentObject: MAIL_MESSAGE_OBJECT, + parentId: messageId, + organizationId, + filename, + mimeType, + data: Buffer.from(part.content.buffer, part.content.byteOffset, part.content.byteLength), + }); + } catch (err) { + this.options.logger.error( + `[mailbox] attachment ${index + 1} of ${parts.length} ('${filename}', ${mimeType}, ` + + `${part.content.byteLength} bytes) of message '${messageId}' in mailbox '${mailboxId}' is NOT stored. ` + + 'The message and its other attachments are stored, and nothing else will report this one: a ' + + 're-delivery of the message is a duplicate and adds no attachment. The file-attachments service ' + + `refused it: ${describeRefusal(err)}`, + ); + } + } + } + /** Write one link; `false` when the unique key says it already exists. */ private async link( engine: MailIngestEngine, diff --git a/packages/services/service-mailbox/src/ingest/messages.fixture.ts b/packages/services/service-mailbox/src/ingest/messages.fixture.ts index 683e8955e25..dd9e4eaa04d 100644 --- a/packages/services/service-mailbox/src/ingest/messages.fixture.ts +++ b/packages/services/service-mailbox/src/ingest/messages.fixture.ts @@ -198,3 +198,64 @@ export function bounce(): string { ].join('\r\n'), ); } + +/** One part of a {@link messageWithParts} message besides its text body. */ +export interface FixturePart { + /** The Content-Type value, parameters included (`application/pdf; name="q.pdf"`). */ + contentType: string; + /** The Content-Disposition value (`attachment; filename="q.pdf"`), or none. */ + disposition?: string; + /** The Content-ID, angle brackets included. */ + contentId?: string; + /** The part's bytes, sent base64. */ + bytes?: Uint8Array; + /** A forwarded message, sent as it is (an RFC 822 text), in place of `bytes`. */ + message?: string; +} + +/** + * A `multipart/mixed` message: a text/plain note, then each part in order — + * a file sent base64, or a forwarded message sent as it is. + */ +export function messageWithParts(options: { + messageId: string; + from?: string; + subject?: string; + headers?: HeaderList; + parts: ReadonlyArray; +}): string { + const boundary = `parts-${++boundarySeq}-boundary`; + const lines: string[] = [`--${boundary}`, 'Content-Type: text/plain; charset=utf-8', '', 'See attached.']; + for (const part of options.parts) { + lines.push(`--${boundary}`, `Content-Type: ${part.contentType}`); + if (part.disposition) lines.push(`Content-Disposition: ${part.disposition}`); + if (part.contentId) lines.push(`Content-ID: ${part.contentId}`); + if (part.message !== undefined) { + lines.push('', part.message); + } else { + const base64 = Buffer.from(part.bytes ?? new Uint8Array()).toString('base64'); + lines.push('Content-Transfer-Encoding: base64', '', ...(base64.match(/.{1,76}/g) ?? [''])); + } + } + lines.push(`--${boundary}--`, ''); + return rfc822( + [ + ['From', options.from ?? 'Rep '], + ['To', 'inbox@corp.test'], + ['Subject', options.subject ?? 'Fwd: A question'], + ['Message-ID', options.messageId], + ['Date', 'Mon, 12 Oct 2026 10:00:00 +0000'], + ...(options.headers ?? []), + ['MIME-Version', '1.0'], + ['Content-Type', `multipart/mixed; boundary="${boundary}"`], + ], + lines.join('\r\n'), + ); +} + +/** Bytes that are not text: every byte value once, then `filler` more. */ +export function binaryBytes(filler = 0): Uint8Array { + const out = new Uint8Array(256 + filler); + for (let i = 0; i < out.length; i++) out[i] = i % 256; + return out; +} diff --git a/packages/services/service-mailbox/src/ingest/parse-message.ts b/packages/services/service-mailbox/src/ingest/parse-message.ts index 22cc3e186d5..1e247745518 100644 --- a/packages/services/service-mailbox/src/ingest/parse-message.ts +++ b/packages/services/service-mailbox/src/ingest/parse-message.ts @@ -24,7 +24,16 @@ * angle brackets stripped; * - the header values automatic-mail suppression reads, and the top-level * media type (`multipart/report`, RFC 6522, carries delivery reports); - * - the header block as received, for the dedupe fallback. + * - the header block as received, for the dedupe fallback; + * - the parts the message carries besides its text and HTML bodies — every + * part `postal-mime` lists as an attachment, in message order, with its + * bytes: a file attached to the message, and each forwarded-message part as + * the message it forwards. A part marked `Content-Disposition: inline` is + * one of them too: Apple Mail marks the files it attaches that way, and an + * image the HTML body shows by `cid:` reference is part of the message as + * sent, so leaving either out would drop bytes the message carries. A part + * nested inside a forwarded part is not listed on its own: it travels inside + * that part's bytes. */ import PostalMime, { addressParser } from 'postal-mime'; @@ -42,6 +51,29 @@ export interface MailParticipantHeaders { cc: string[]; } +/** + * One part of a message stored as an attachment of its row: a file the + * message carries, or a message it forwards as an attachment. + */ +export interface MailAttachmentPart { + /** The part's file name as the message gives it (Content-Disposition `filename`, or Content-Type `name`), or `null`. */ + filename: string | null; + /** The part's media type, lower-cased, without parameters. */ + mimeType: string; + /** + * The part's bytes, its transfer encoding decoded. For a forwarded-message + * part, the forwarded message itself, exactly as it travels in the part. + */ + content: Uint8Array; + /** + * Set for a forwarded-message part (`message/rfc822`, `message/global`), with + * the forwarded message's own Subject: `null` when it has none, or when the + * part was not read (refused by the parser, or past {@link MAX_FORWARDED_PARTS}). + * `null` for every other part. + */ + forwarded: { subject: string | null } | null; +} + /** One parsed inbound message. */ export interface ParsedInboundMessage { /** Message-ID, angle brackets stripped, or `null` when the message has none. */ @@ -66,6 +98,13 @@ export interface ParsedInboundMessage { * {@link forwarded}; the message itself is still stored. */ unreadForwardedParts: number; + /** + * Every part stored as an attachment of the message's row, in message order: + * the files it carries and the messages it forwards as attachments (those + * counted in {@link unreadForwardedParts} included — their bytes are kept even + * where their addresses are not read). + */ + attachments: MailAttachmentPart[]; /** Lower-cased header name → the first value, unfolded, encoded words left as sent. */ header(name: string): string | null; /** The top-level media type, lower-cased, without parameters (`text/plain` when absent). */ @@ -149,6 +188,12 @@ function isoDate(value: string | undefined): string | null { return Number.isNaN(at.getTime()) ? null : at.toISOString(); } +/** A part's content as bytes (`attachmentEncoding: 'arraybuffer'` hands an `ArrayBuffer`). */ +function bytesOf(content: ArrayBuffer | Uint8Array | string): Uint8Array { + if (typeof content === 'string') return new TextEncoder().encode(content); + return content instanceof Uint8Array ? content : new Uint8Array(content); +} + function asText(raw: RawMailMessage): string { if (typeof raw === 'string') return raw; // The WHATWG `latin1` label decodes one character per byte, so the header @@ -174,25 +219,31 @@ function mediaTypeOf(contentType: string | null): { mediaType: string; reportTyp export async function parseInboundMessage(raw: RawMailMessage): Promise { const email: Email = await PostalMime.parse(raw, PARSE_OPTIONS); const forwarded: MailParticipantHeaders[] = []; + const attachments: MailAttachmentPart[] = []; let unreadForwardedParts = 0; for (const attachment of email.attachments) { - if (!FORWARDED_MESSAGE_TYPES.has(attachment.mimeType.toLowerCase())) continue; - if (forwarded.length >= MAX_FORWARDED_PARTS) { - unreadForwardedParts++; + const mimeType = attachment.mimeType.toLowerCase(); + const content = bytesOf(attachment.content); + const filename = attachment.filename ? attachment.filename : null; + if (!FORWARDED_MESSAGE_TYPES.has(mimeType)) { + attachments.push({ filename, mimeType, content, forwarded: null }); continue; } - const content = attachment.content; - try { - const inner: Email = await PostalMime.parse( - typeof content === 'string' ? content : content instanceof Uint8Array ? content : new Uint8Array(content), - PARSE_OPTIONS, - ); - forwarded.push(participantsOf(inner.headers)); - } catch { - // A forwarded part the parser refuses names no addresses; the received - // message is still mirrored (never dropped), and the count says so. + let subject: string | null = null; + if (forwarded.length >= MAX_FORWARDED_PARTS) { unreadForwardedParts++; + } else { + try { + const inner: Email = await PostalMime.parse(content, PARSE_OPTIONS); + forwarded.push(participantsOf(inner.headers)); + subject = inner.subject ? inner.subject : null; + } catch { + // A forwarded part the parser refuses names no addresses; the received + // message is still mirrored (never dropped), and the count says so. + unreadForwardedParts++; + } } + attachments.push({ filename, mimeType, content, forwarded: { subject } }); } const headers = email.headers; @@ -210,6 +261,7 @@ export async function parseInboundMessage(raw: RawMailMessage): Promise firstHeader(headers, name.toLowerCase()), mediaType, reportType, diff --git a/packages/services/service-mailbox/src/mailbox-service-plugin.ts b/packages/services/service-mailbox/src/mailbox-service-plugin.ts index 9194d70efc8..af329fadbd3 100644 --- a/packages/services/service-mailbox/src/mailbox-service-plugin.ts +++ b/packages/services/service-mailbox/src/mailbox-service-plugin.ts @@ -1,7 +1,13 @@ // Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. import type { Plugin, PluginContext } from '@objectstack/core'; -import type { II18nService, IJobService, JobRunOutcome } from '@objectstack/spec/contracts'; +import { + FILE_ATTACHMENT_SERVICE, + type IFileAttachmentService, + type II18nService, + type IJobService, + type JobRunOutcome, +} from '@objectstack/spec/contracts'; import { SysMailAssociation, SysMailbox, SysMailMessage } from './objects/index.js'; import { installMailReadVisibility, type MailGateEngine } from './mail-read-visibility.js'; import { MAILBOX_SERVICE, MailIngestService, type MailIngestEngine } from './ingest/index.js'; @@ -59,7 +65,13 @@ export interface MailboxServicePluginOptions { * mailbox, with automatic mail (auto-replies, bounces) suppressed, and stamps * its associations from the email fields an app flags with `mailAssociation`; * `rematch` replays a mailbox's stored participant sets against the records - * that exist now and adds the missing links. + * that exist now and adds the missing links. A stored message's attachments, + * and each message it forwards as an attachment (as an `.eml` file), are + * stored as files of its `sys_mail_message` row through the kernel service + * `file-attachments` ({@link FILE_ATTACHMENT_SERVICE}), which the storage + * capability registers; the seam resolves it per message, so plugin order does + * not matter, and where none is composed it stores the message without them + * and says so once. * * With `imap` composed, it also feeds the seam: the mailbox sync * ({@link MailboxSyncService}) runs as the interval job @@ -101,6 +113,14 @@ export class MailboxServicePlugin implements Plugin { const ingest = new MailIngestService({ engine: () => ctx.getService('objectql'), logger: ctx.logger, + attachments: () => { + try { + return ctx.getService(FILE_ATTACHMENT_SERVICE); + } catch { + // Not composed: the storage capability is not mounted in this kernel. + return undefined; + } + }, }); ctx.registerService(MAILBOX_SERVICE, ingest); diff --git a/packages/services/service-mailbox/src/sync/mailbox-sync-service.test.ts b/packages/services/service-mailbox/src/sync/mailbox-sync-service.test.ts index 8507d357d17..9be100409ce 100644 --- a/packages/services/service-mailbox/src/sync/mailbox-sync-service.test.ts +++ b/packages/services/service-mailbox/src/sync/mailbox-sync-service.test.ts @@ -90,7 +90,7 @@ describe('the IMAP sync, end to end, on a real engine', () => { engine.registry.registerObject(def as any, 'mailbox-sync-test', 'mailbox-sync-test'); } await engine.syncSchemas(); - ingest = new MailIngestService({ engine: () => engine as unknown as MailIngestEngine, logger: { warn: () => {} } }); + ingest = new MailIngestService({ engine: () => engine as unknown as MailIngestEngine, logger: { warn: () => {}, error: () => {} } }); }); afterAll(async () => { diff --git a/packages/services/service-mailbox/vitest.config.ts b/packages/services/service-mailbox/vitest.config.ts index abf5f666d05..20830c6b857 100644 --- a/packages/services/service-mailbox/vitest.config.ts +++ b/packages/services/service-mailbox/vitest.config.ts @@ -29,8 +29,17 @@ export default defineConfig({ { find: /^@objectstack\/driver-sql$/, replacement: path.resolve(__dirname, '../../drivers/driver-sql/src/index.ts') }, { find: /^@objectstack\/platform-objects\/apps$/, replacement: path.resolve(__dirname, '../../platform-objects/src/apps/index.ts') }, { find: /^@objectstack\/platform-objects\/identity$/, replacement: path.resolve(__dirname, '../../platform-objects/src/identity/index.ts') }, + { find: /^@objectstack\/platform-objects\/audit$/, replacement: path.resolve(__dirname, '../../platform-objects/src/audit/index.ts') }, + { find: /^@objectstack\/platform-objects\/system$/, replacement: path.resolve(__dirname, '../../platform-objects/src/system/index.ts') }, + // The real `file-attachments` verb the attachment pins store through, and what it reads. + { find: /^@objectstack\/service-storage$/, replacement: path.resolve(__dirname, '../service-storage/src/index.ts') }, + { find: /^@objectstack\/observability$/, replacement: path.resolve(__dirname, '../../observability/src/index.ts') }, { find: /^@objectstack\/spec$/, replacement: path.resolve(__dirname, '../../spec/src/index.ts') }, + { find: /^@objectstack\/spec\/api$/, replacement: path.resolve(__dirname, '../../spec/src/api/index.ts') }, + { find: /^@objectstack\/spec\/contracts$/, replacement: path.resolve(__dirname, '../../spec/src/contracts/index.ts') }, { find: /^@objectstack\/spec\/data$/, replacement: path.resolve(__dirname, '../../spec/src/data/index.ts') }, + { find: /^@objectstack\/spec\/kernel$/, replacement: path.resolve(__dirname, '../../spec/src/kernel/index.ts') }, + { find: /^@objectstack\/spec\/system$/, replacement: path.resolve(__dirname, '../../spec/src/system/index.ts') }, { find: /^@objectstack\/spec\/security$/, replacement: path.resolve(__dirname, '../../spec/src/security/index.ts') }, { find: /^@objectstack\/spec\/shared$/, replacement: path.resolve(__dirname, '../../spec/src/shared/index.ts') }, { find: /^@objectstack\/spec\/identity$/, replacement: path.resolve(__dirname, '../../spec/src/identity/index.ts') }, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index d290d546a0a..c51e1be4840 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -2759,6 +2759,9 @@ importers: '@objectstack/objectql': specifier: workspace:* version: link:../../objectql + '@objectstack/service-storage': + specifier: workspace:* + version: link:../service-storage '@types/node': specifier: ^26.6.3 version: 26.6.3 From 3a2884edb67878f2293120e085a663d94e2b2ef8 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 14 Oct 2026 12:00:35 +0000 Subject: [PATCH 2/2] wip(service-mailbox): real-verb kernel pins, parse pins, README and changeset for mail attachments Claude-Session: https://claude.ai/code/session_01YRp5z4kK6vgYkrdH97ztsy Co-authored-by: Claude --- .changeset/23307-mail-attachments.md | 12 + packages/services/service-mailbox/README.md | 6 +- .../src/ingest/parse-message.test.ts | 49 ++- .../src/mailbox-attachments-kernel.test.ts | 289 ++++++++++++++++++ 4 files changed, 352 insertions(+), 4 deletions(-) create mode 100644 .changeset/23307-mail-attachments.md create mode 100644 packages/services/service-mailbox/src/mailbox-attachments-kernel.test.ts diff --git a/.changeset/23307-mail-attachments.md b/.changeset/23307-mail-attachments.md new file mode 100644 index 00000000000..ea130b62a7d --- /dev/null +++ b/.changeset/23307-mail-attachments.md @@ -0,0 +1,12 @@ +--- +'@objectstack/service-mailbox': minor +--- + +Inbound mail stores its attachments: each attachment of an ingested message, and each message it forwards as an attachment as an `.eml` file, is stored through the `file-attachments` service as a file of the message's `sys_mail_message` row + +Clause-②: no + +- **What is stored, and where.** Every part of a stored message besides its text and HTML bodies (inline parts included, and each forwarded `message/rfc822` or `message/global` part as the `.eml` file it travels as) becomes one `sys_file` and one `sys_attachment` whose parent is the message's own `sys_mail_message` row, written by `IFileAttachmentService.attachFile` in the mailbox's organization after the row, and only for a message the call stored, so a re-delivered copy adds no attachment. +- **Without the storage capability.** A kernel that composes no `file-attachments` service (registered by `@objectstack/service-storage`) stores each message without its attachments, and says so once in the log. +- **Refused attachments.** A part the service refuses, such as one over the `max_upload_mb` upload size limit, is logged at `error` with its file name and the refusal, and the message and its other attachments are still stored. +- **For a host that constructs `MailIngestService` itself.** It takes the service as the optional `attachments` resolver (`MailboxServicePlugin` supplies it), and its `MailIngestLogger` now needs `error`, as `MailSyncLogger` already does. diff --git a/packages/services/service-mailbox/README.md b/packages/services/service-mailbox/README.md index 4e7d7eb27e1..37d1d862522 100644 --- a/packages/services/service-mailbox/README.md +++ b/packages/services/service-mailbox/README.md @@ -4,7 +4,7 @@ Inbound mail for ObjectStack: the mirrored message, its links to an app's record > ⚠️ **Mail arrives only where an IMAP account source is composed.** The capability boot (`requires: ['mailbox']`) constructs the plugin with no options, so it mounts the objects, their read gates and the ingest seam, and syncs no mailbox — it says so once at boot. Compose the plugin yourself with `imap.accountFor` (below) to poll mailboxes. > -> ⚠️ **Attachments are not stored.** Nothing on the server writes `sys_attachment` rows today, so a message's attachments — and the body of a message forwarded as an attachment — are not kept; only its participants are read. +> ⚠️ **Attachments are stored only where the storage capability is composed.** A message's attachments are written through the `file-attachments` service that `@objectstack/service-storage` registers (the `storage` capability, on the always-on slate). A kernel that does not compose it stores each message without its attachments, and says so once. ## What this package provides @@ -18,7 +18,7 @@ Inbound mail for ObjectStack: the mirrored message, its links to an app's record - **`manage_mailboxes`** — an organization capability declared in `@objectstack/spec`. Platform administrators and organization administrators hold it by default; every other position only by explicit grant. Object grants and field-level security still apply to a holder. - **The ingest seam** — `MailIngestService`, registered as the kernel service `mailbox` (`MAILBOX_SERVICE`): - **`ingest({ mailboxId, raw, receivedAt? }, context)`** reads one raw RFC 822 message and writes at most one `sys_mail_message` row, answering `stored`, `duplicate` or `suppressed`: - - **participants** are the addresses of From, Sender, Reply-To, To and Cc, plus the same headers of a message forwarded as an attachment (a `message/rfc822` part, or its internationalized twin `message/global`, RFC 6532). At most `MAX_FORWARDED_PARTS` (20) forwarded parts of one message are read; the rest, and any part the MIME parser refuses, take no part in matching, and the message is stored without them (a warning names how many). The mailbox's own address is left out. A forward quoted inline in the body is read too, for an organization user's forward only: the "Forwarded message" block Gmail, Outlook (on the web and Microsoft 365) or Apple Mail writes into the text/plain body, with its English labels, adds its From, Reply-To, To and Cc — when the message's From is a user (`sys_user`) who is a member (`sys_member`) of the mailbox's organization. A block the parser does not recognise is not guessed at, and a forward from anyone else is not read: the message is matched on its other participants. The participants are stored on the row in `match_addresses`, each with its source (`outer`, `attachment` or `inline`); + - **participants** are the addresses of From, Sender, Reply-To, To and Cc, plus the same headers of a message forwarded as an attachment (a `message/rfc822` part, or its internationalized twin `message/global`, RFC 6532). At most `MAX_FORWARDED_PARTS` (20) forwarded parts of one message are read; the rest, and any part the MIME parser refuses, take no part in matching (a warning names how many), and are still stored as attachments. The mailbox's own address is left out. A forward quoted inline in the body is read too, for an organization user's forward only: the "Forwarded message" block Gmail, Outlook (on the web and Microsoft 365) or Apple Mail writes into the text/plain body, with its English labels, adds its From, Reply-To, To and Cc — when the message's From is a user (`sys_user`) who is a member (`sys_member`) of the mailbox's organization. A block the parser does not recognise is not guessed at, and a forward from anyone else is not read: the message is matched on its other participants. The participants are stored on the row in `match_addresses`, each with its source (`outer`, `attachment` or `inline`); - **association** — every record of every object whose email field declares `mailAssociation: true` and holds one of those addresses (ASCII case-insensitive) is linked to the message, inside the mailbox's organization only, with one `sys_mail_association` row each (`matched_by: 'ingest'`). A message no record matches waits in the pool (`association_state: 'unassociated'`); - **dedupe** — one row per message per mailbox, keyed on the Message-ID (or, for a message without one, on its header block as received), so a second fetch answers `duplicate` and writes nothing; - **threading** — `message_id`, `in_reply_to` and `reference_ids` hold the ids of RFC 5322 §3.6.4, angle brackets stripped; @@ -26,7 +26,7 @@ Inbound mail for ObjectStack: the mirrored message, its links to an app's record Only the system context may ingest: a message enters a mailbox from its mail source, never from a user. - **`rematch({ mailboxId, messageIds? }, context)`** replays each message's stored participant set against the records that exist now and adds the links that are missing (`matched_by: 'rematch'`), moving a pool message to `associated` once it has one. It never removes a link. The system context may run it, and so may the mailbox's owner and a `manage_mailboxes` holder of the mailbox's organization; anyone else is refused (`PERMISSION_DENIED`), and another organization's mailbox is not found (`RECORD_NOT_FOUND`). - - Attachments are not stored yet, so a message forwarded as an attachment is stored with the forwarder's own text, and the forwarded message's body is not kept. + - **attachments** — each part of a stored message besides its text and HTML bodies is stored as a file of its `sys_mail_message` row: one `sys_file` and one `sys_attachment` (parent `sys_mail_message`, the row's id), through `IFileAttachmentService.attachFile`, the kernel service `file-attachments` (`FILE_ATTACHMENT_SERVICE`). Both rows land in the mailbox's organization, and the deployment's upload size limit (`max_upload_mb`) applies. That covers every file the message carries, a part marked `Content-Disposition: inline` included (Apple Mail marks attached files that way, and an image the HTML body shows by `cid:` reference is part of the message), and every message forwarded as an attachment (`message/rfc822` or `message/global`), stored as the `.eml` file it travels as: named by the part's own file name, else by its Subject, else `forwarded.eml`. A part with no file name is `attachment-`, its position among the message's parts. Parts are stored only for a message this call stored, after its row: a duplicate or a suppressed message stores none, so a re-delivery adds no attachment, and a re-match never reads a part. A part the service refuses (over the upload size limit, for instance) is logged at `error` with its name and the refusal, and the message and its other parts are still stored. No column of the row names a part, and no byte is kept in a column; - **The IMAP sync** — `MailboxSyncService`, composed with `MailboxServicePlugin({ imap })` and run as the interval job `mailbox:imap-sync` on the composed job service (`requires: ['job']`): - for every mailbox your `accountFor` answers, it opens a **read-only** session on its IMAP account through `@objectstack/connector-imap` — the folder is opened with `EXAMINE` and bodies are fetched with `BODY.PEEK[]`, so the mailbox is never altered — and hands each message above the stored cursor, in UID order, to `ingest` under the system context. Dedupe, threading, association, the pool and suppression are the seam's, unchanged; - **the cursor** is kept on the mailbox row — `imap_folder`, `imap_uid_validity`, `imap_last_uid`, `imap_synced_at` — and written by the sync only. A cursor taken under another UIDVALIDITY, or in another folder, is discarded and the folder is read from its start; every message already stored then answers `duplicate`; diff --git a/packages/services/service-mailbox/src/ingest/parse-message.test.ts b/packages/services/service-mailbox/src/ingest/parse-message.test.ts index 4181e6174d2..4c8a617e48c 100644 --- a/packages/services/service-mailbox/src/ingest/parse-message.test.ts +++ b/packages/services/service-mailbox/src/ingest/parse-message.test.ts @@ -14,7 +14,7 @@ import { parseMatchAddresses, serializeMatchAddresses, } from './match-addresses.js'; -import { forwardAsAttachment, plainMessage, rfc822 } from './messages.fixture.js'; +import { binaryBytes, forwardAsAttachment, messageWithParts, plainMessage, rfc822 } from './messages.fixture.js'; const ORIGINAL = rfc822( [ @@ -137,6 +137,9 @@ describe('parseInboundMessage — a message forwarded as an attachment (source ` const parsed = await parseInboundMessage(raw); expect(parsed.forwarded).toHaveLength(MAX_FORWARDED_PARTS); expect(parsed.unreadForwardedParts).toBe(2); + // Their addresses are not read, but their bytes are kept as attachments. + expect(parsed.attachments).toHaveLength(MAX_FORWARDED_PARTS + 2); + expect(parsed.attachments.slice(-2).map((a) => a.forwarded)).toEqual([{ subject: null }, { subject: null }]); }); it('control: a forward quoted inline in the body names no forwarded participants (that is the `inline` source)', async () => { @@ -151,6 +154,50 @@ describe('parseInboundMessage — a message forwarded as an attachment (source ` }); }); +describe('parseInboundMessage — the parts stored as attachments of the row', () => { + it('lists every part besides the bodies, in message order, with its bytes as carried', async () => { + const png = binaryBytes(); + const pdf = binaryBytes(7); + const parsed = await parseInboundMessage( + messageWithParts({ + messageId: '', + parts: [ + { contentType: 'image/png', disposition: 'inline; filename="logo.png"', contentId: '', bytes: png }, + { contentType: 'application/pdf; name="apple.pdf"', disposition: 'inline; filename="apple.pdf"', bytes: pdf }, + { contentType: 'application/octet-stream', bytes: pdf }, + { contentType: 'message/rfc822', disposition: 'attachment; filename="question.eml"', message: ORIGINAL }, + { contentType: 'message/global', message: ORIGINAL }, + ], + }), + ); + expect(parsed.attachments.map((a) => [a.filename, a.mimeType, a.forwarded])).toEqual([ + ['logo.png', 'image/png', null], + ['apple.pdf', 'application/pdf', null], + [null, 'application/octet-stream', null], + ['question.eml', 'message/rfc822', { subject: 'A question' }], + [null, 'message/global', { subject: 'A question' }], + ]); + expect(parsed.attachments[0]!.content).toEqual(png); + expect(parsed.attachments[1]!.content).toEqual(pdf); + // A forwarded message is kept exactly as it travels in its part. + expect(new TextDecoder().decode(parsed.attachments[3]!.content)).toBe(ORIGINAL); + expect(parsed.text?.trim()).toBe('See attached.'); + }); + + it('control: a message with no part besides its body lists none, and a second text part joins the body', async () => { + expect((await parseInboundMessage(plainMessage())).attachments).toEqual([]); + const boundary = 'two-texts'; + const parsed = await parseInboundMessage( + rfc822( + [['From', 'rep@corp.test'], ['MIME-Version', '1.0'], ['Content-Type', `multipart/mixed; boundary="${boundary}"`]], + [`--${boundary}`, 'Content-Type: text/plain', '', 'one', `--${boundary}`, 'Content-Type: text/plain', '', 'two', `--${boundary}--`, ''].join('\r\n'), + ), + ); + expect(parsed.attachments).toEqual([]); + expect(parsed.text).toContain('two'); + }); +}); + describe('the participant set (`match_addresses`)', () => { it('normalizes ASCII letters only, and refuses what is not an addr-spec', () => { expect(normalizeMailAddress(' Jane.DOE@Acme.TEST ')).toBe('jane.doe@acme.test'); diff --git a/packages/services/service-mailbox/src/mailbox-attachments-kernel.test.ts b/packages/services/service-mailbox/src/mailbox-attachments-kernel.test.ts new file mode 100644 index 00000000000..5d07db28320 --- /dev/null +++ b/packages/services/service-mailbox/src/mailbox-attachments-kernel.test.ts @@ -0,0 +1,289 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * A message's attachments, stored through the REAL attach verb: a real + * `ObjectKernel` boot composing `ObjectQLPlugin`, `StorageServicePlugin` (its + * local adapter in a temp directory, which registers `file-attachments` and + * arms it at `kernel:bootstrapped`) and `MailboxServicePlugin`, over + * `@objectstack/driver-sql` + better-sqlite3 `:memory:`. The ingest seam is + * reached as the kernel service `mailbox`; it finds the verb on its own, as it + * does in a deployment. + * + * What is pinned, each beside its control: + * + * - one forwarded message with one file is ONE `sys_mail_message` row and TWO + * attachments — the file and the forwarded message as an `.eml` — each a + * committed `sys_file` behind a `sys_attachment` whose parent is that row, + * both in the mailbox's organization, the bytes as the message carried them + * (control: a message without parts has no attachment); + * - a re-delivered copy adds no attachment (control: the same parts under + * another Message-ID are another message's); + * - a mailbox in another organization stores its message's files in that + * organization (control: the first mailbox's, in its own); + * - a part over the deployment's upload size limit is refused by the verb + * (`PAYLOAD_TOO_LARGE`) and leaves nothing behind, while the message and its + * other parts are stored, and the refusal is said at `error`, by name; + * - without the storage capability the message is stored without its parts, + * and that is said once. + */ + +import { describe, it, expect, beforeAll, afterAll, vi } from 'vitest'; +import { promises as fs } from 'node:fs'; +import { join } from 'node:path'; +import { tmpdir } from 'node:os'; +import { ObjectKernel, type Plugin, type PluginContext } from '@objectstack/core'; +import { ObjectQLPlugin, type ObjectQL } from '@objectstack/objectql'; +import { SqlDriver } from '@objectstack/driver-sql'; +import { StorageServicePlugin } from '@objectstack/service-storage'; +import type { DataEngineInsertOptions, EngineQueryOptions } from '@objectstack/spec/data'; +import type { IStorageService } from '@objectstack/spec/contracts'; +import { MAILBOX_SERVICE, type MailIngestService, MailboxServicePlugin } from './index.js'; +import { binaryBytes, messageWithParts, plainMessage, rfc822 } from './ingest/messages.fixture.js'; + +const SYSTEM = { isSystem: true } as const; +const ORG_A = 'org_attach_a'; +const ORG_B = 'org_attach_b'; +/** `max_upload_mb` as saved: 0.001 MB ⇒ floor(0.001 × 1 MiB) = 1048 bytes. */ +const MAX_UPLOAD_MB = 0.001; +const MAX_UPLOAD_BYTES = 1048; + +type Row = Record; + +/** A settings service carrying only the saved upload size limit, as `StorageServicePlugin` reads it. */ +class UploadLimitSettingsPlugin implements Plugin { + name = 'test.upload-limit-settings'; + version = '1.0.0'; + async init(ctx: PluginContext): Promise { + ctx.registerService('settings', { + createClient: () => ({}), + getNamespace: async () => ({ values: { max_upload_mb: { value: MAX_UPLOAD_MB, source: 'global' } } }), + subscribe: () => undefined, + registerAction: () => undefined, + }); + } +} + +/** Hands the test the kernel's logger — the one every plugin context carries — so what the seam says is read. */ +class LoggerProbePlugin implements Plugin { + name = 'test.logger-probe'; + version = '1.0.0'; + logger?: PluginContext['logger']; + async init(ctx: PluginContext): Promise { + this.logger = ctx.logger; + } +} + +async function bootKernel(options: { storage: boolean }) { + const probe = new LoggerProbePlugin(); + const kernel = new ObjectKernel({ logger: { level: 'fatal' } }); + await kernel.use(probe); + await kernel.use(new ObjectQLPlugin()); + let rootDir: string | undefined; + if (options.storage) { + rootDir = await fs.mkdtemp(join(tmpdir(), 'oss-23307-')); + await kernel.use(new UploadLimitSettingsPlugin()); + await kernel.use(new StorageServicePlugin({ adapter: 'local', local: { rootDir }, registerRoutes: false })); + } + await kernel.use(new MailboxServicePlugin()); + await kernel.bootstrap(); + // What the seam says from here on: it speaks only when it ingests. + const error = vi.spyOn(probe.logger!, 'error'); + const warn = vi.spyOn(probe.logger!, 'warn'); + const ql = kernel.getService('objectql'); + const driver = new SqlDriver({ client: 'better-sqlite3', connection: { filename: ':memory:' }, useNullAsDefault: true }); + await driver.connect(); + ql.registerDriver(driver, true); + await ql.syncSchemas(); + const said = (spy: typeof error) => spy.mock.calls.map((args) => String(args[0])); + return { kernel, ql, rootDir, errors: () => said(error), warnings: () => said(warn) }; +} + +const ORIGINAL = rfc822( + [ + ['From', 'Customer '], + ['To', 'rep@corp.test'], + ['Subject', 'What does it cost?'], + ['Message-ID', ''], + ], + 'What does it cost?', +); +/** A file within the deployment's upload size limit. */ +const QUOTE = binaryBytes(100); + +/** The card's message: one forwarded message and one file. */ +const forwardWithFile = (messageId: string, extra: Parameters[0]['parts'] = []) => + messageWithParts({ + messageId, + parts: [ + { contentType: 'application/pdf; name="quote.pdf"', disposition: 'attachment; filename="quote.pdf"', bytes: QUOTE }, + ...extra, + { contentType: 'message/rfc822', disposition: 'attachment; filename="question.eml"', message: ORIGINAL }, + ], + }); + +describe('a message’s attachments, through the real file-attachments verb on a booted kernel', () => { + let booted: Awaited>; + let mailboxes: MailIngestService; + let storage: IStorageService; + + const find = (object: string, where: Row): Promise => + booted.ql.find(object, { where, limit: 500, context: SYSTEM } satisfies EngineQueryOptions) as Promise; + const insert = async (object: string, data: Row): Promise => + String(((await booted.ql.insert(object, data, { context: SYSTEM } satisfies DataEngineInsertOptions)) as Row).id); + const newMailbox = (organization: string) => + insert('sys_mailbox', { organization_id: organization, name: `Inbox ${organization}`, address: 'inbox@corp.test' }); + /** A row's attachments, by file name (the store answers in no declared order; message order is pinned on the calls, in the unit suite). */ + const attachmentsOf = async (messageId: string) => + (await find('sys_attachment', { parent_object: 'sys_mail_message', parent_id: messageId })).sort((a, b) => + String(a.file_name).localeCompare(String(b.file_name)), + ); + const bytesOf = async (fileId: string) => { + const [file] = await find('sys_file', { id: fileId }); + return new Uint8Array(await storage.download(String(file!.key))); + }; + + beforeAll(async () => { + booted = await bootKernel({ storage: true }); + mailboxes = booted.kernel.getService(MAILBOX_SERVICE); + storage = booted.kernel.getService('storage'); + }, 60_000); + + afterAll(async () => { + try { + await booted?.kernel.shutdown(); + } catch { + /* best-effort teardown */ + } + if (booted?.rootDir) await fs.rm(booted.rootDir, { recursive: true, force: true }); + }); + + it('arms: the mail message opts into attachments, and the verb is registered', () => { + expect(booted.ql.registry.getObject('sys_mail_message')?.enable?.files).toBe(true); + expect(booted.kernel.getService('file-attachments')).toBeTruthy(); + }); + + it('one forwarded message with one file: one row, two attachments on it, both in the mailbox’s organization', async () => { + const mailbox = await newMailbox(ORG_A); + const result = await mailboxes.ingest({ mailboxId: mailbox, raw: forwardWithFile('') }, SYSTEM); + + expect(result.outcome).toBe('stored'); + const rows = await find('sys_mail_message', { mailbox_id: mailbox }); + expect(rows).toHaveLength(1); + const messageId = String(rows[0]!.id); + expect((result as { messageId: string }).messageId).toBe(messageId); + + const attachments = await attachmentsOf(messageId); + expect(attachments.map((a) => [a.file_name, a.mime_type, a.size, a.organization_id])).toEqual([ + ['question.eml', 'message/rfc822', Buffer.byteLength(ORIGINAL), ORG_A], + ['quote.pdf', 'application/pdf', QUOTE.byteLength, ORG_A], + ]); + for (const attachment of attachments) { + const [file] = await find('sys_file', { id: attachment.file_id }); + expect(file).toMatchObject({ status: 'committed', organization_id: ORG_A, scope: 'attachments' }); + } + expect(Buffer.from(await bytesOf(String(attachments[0]!.file_id))).toString('utf8')).toBe(ORIGINAL); + expect(await bytesOf(String(attachments[1]!.file_id))).toEqual(QUOTE); + expect(booted.errors().filter((m) => m.includes('[mailbox]'))).toEqual([]); + }); + + it('control: a message without parts has no attachment', async () => { + const mailbox = await newMailbox(ORG_A); + const result = await mailboxes.ingest( + { mailboxId: mailbox, raw: plainMessage({ 'Message-ID': '' }) }, + SYSTEM, + ); + expect(await attachmentsOf((result as { messageId: string }).messageId)).toEqual([]); + }); + + it('a re-delivered copy adds no attachment; control: the same parts under another Message-ID are another message’s', async () => { + const mailbox = await newMailbox(ORG_A); + const raw = forwardWithFile(''); + const first = await mailboxes.ingest({ mailboxId: mailbox, raw }, SYSTEM); + const filesBefore = (await find('sys_file', {})).length; + + expect(await mailboxes.ingest({ mailboxId: mailbox, raw }, SYSTEM)).toEqual({ + outcome: 'duplicate', + messageId: (first as { messageId: string }).messageId, + }); + expect(await attachmentsOf((first as { messageId: string }).messageId)).toHaveLength(2); + expect(await find('sys_file', {})).toHaveLength(filesBefore); + + const other = await mailboxes.ingest({ mailboxId: mailbox, raw: forwardWithFile('') }, SYSTEM); + expect(other.outcome).toBe('stored'); + expect(await attachmentsOf((other as { messageId: string }).messageId)).toHaveLength(2); + expect(await find('sys_file', {})).toHaveLength(filesBefore + 2); + }); + + it('a mailbox in another organization stores its message’s files in that organization', async () => { + const mailbox = await newMailbox(ORG_B); + const result = await mailboxes.ingest({ mailboxId: mailbox, raw: forwardWithFile('') }, SYSTEM); + const attachments = await attachmentsOf((result as { messageId: string }).messageId); + expect(attachments.map((a) => a.organization_id)).toEqual([ORG_B, ORG_B]); + for (const attachment of attachments) { + const [file] = await find('sys_file', { id: attachment.file_id }); + expect(file!.organization_id).toBe(ORG_B); + } + }); + + it('a part over the upload size limit is refused and leaves nothing; the message and its other parts are stored, and it is said', async () => { + const mailbox = await newMailbox(ORG_A); + const big = binaryBytes(MAX_UPLOAD_BYTES); + expect(big.byteLength).toBeGreaterThan(MAX_UPLOAD_BYTES); + const filesBefore = (await find('sys_file', {})).length; + const raw = forwardWithFile('', [ + { contentType: 'application/octet-stream', disposition: 'attachment; filename="big.bin"', bytes: big }, + ]); + + const result = await mailboxes.ingest({ mailboxId: mailbox, raw }, SYSTEM); + + expect(result.outcome).toBe('stored'); + expect(await find('sys_mail_message', { mailbox_id: mailbox })).toHaveLength(1); + const attachments = await attachmentsOf((result as { messageId: string }).messageId); + expect(attachments.map((a) => a.file_name)).toEqual(['question.eml', 'quote.pdf']); + // The refused part left no sys_file at all, committed or pending. + expect(await find('sys_file', {})).toHaveLength(filesBefore + 2); + expect(await find('sys_file', { name: 'big.bin' })).toEqual([]); + const said = booted.errors().filter((m) => m.includes("'big.bin'")); + expect(said).toHaveLength(1); + expect(said[0]).toContain('PAYLOAD_TOO_LARGE (413)'); + expect(said[0]).toContain('is NOT stored'); + }); +}); + +describe('a message’s attachments on a kernel without the storage capability', () => { + let booted: Awaited>; + + beforeAll(async () => { + booted = await bootKernel({ storage: false }); + }, 60_000); + + afterAll(async () => { + try { + await booted?.kernel.shutdown(); + } catch { + /* best-effort teardown */ + } + }); + + it('stores the message without its parts, and says so once', async () => { + expect(() => booted.kernel.getService('file-attachments')).toThrow(); + const mailboxes = booted.kernel.getService(MAILBOX_SERVICE); + const mailbox = String( + ((await booted.ql.insert( + 'sys_mailbox', + { organization_id: ORG_A, name: 'Inbox', address: 'inbox@corp.test' }, + { context: SYSTEM } satisfies DataEngineInsertOptions, + )) as Row).id, + ); + for (const id of ['', '']) { + expect((await mailboxes.ingest({ mailboxId: mailbox, raw: forwardWithFile(id) }, SYSTEM)).outcome).toBe('stored'); + } + const rows = (await booted.ql.find('sys_mail_message', { + where: { mailbox_id: mailbox }, + context: SYSTEM, + } satisfies EngineQueryOptions)) as Row[]; + expect(rows).toHaveLength(2); + expect(booted.warnings().filter((m) => m.includes("'file-attachments' service is composed"))).toHaveLength(1); + expect(booted.errors().filter((m) => m.includes('[mailbox]'))).toEqual([]); + }); +});