From 816389a185b7f5084617ed74825938958c36ce45 Mon Sep 17 00:00:00 2001 From: xcs Date: Wed, 26 Aug 2026 03:47:48 +0800 Subject: [PATCH] =?UTF-8?q?=E9=82=AE=E4=BB=B6=E6=94=B6=E5=8F=91=E7=BB=88?= =?UTF-8?q?=E5=AE=A1=E4=BF=AE=E5=A4=8D=E8=BD=AE:token=20=E6=8C=89=E5=87=AD?= =?UTF-8?q?=E8=AF=81=E6=8C=87=E7=BA=B9=E3=80=81=E6=AF=92=E4=BF=A1=E6=9C=89?= =?UTF-8?q?=E9=80=80=E8=B7=AF=E3=80=81=E5=8F=AA=E6=94=B9=E5=8F=91=E4=BF=A1?= =?UTF-8?q?=E8=AE=BE=E7=BD=AE=E4=B8=8D=E4=B8=A2=E4=BF=A1=E3=80=81=E9=87=8D?= =?UTF-8?q?=E5=8F=91=E5=85=88=E6=8A=A2=E3=80=81=E5=88=97=E8=A1=A8=E4=B8=8D?= =?UTF-8?q?=E6=90=AC=E5=8E=9F=E4=BB=B6=E3=80=81save=20=E4=B8=8D=E7=9B=96?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=E5=88=97=E3=80=81=E9=87=8D=E9=94=9A=E5=87=BA?= =?UTF-8?q?=E5=A3=B0=E3=80=81=E8=84=8F=E6=B8=B8=E6=A0=87=E4=B8=8D=E7=82=B8?= =?UTF-8?q?=E9=A1=B5=E3=80=81=E7=BA=BF=E7=A8=8B=E5=8F=A3=E5=BE=84=E5=BD=92?= =?UTF-8?q?=E4=B8=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 全分支终审(跨任务视角)逮出的十条 P0 + 八条 P1,一次修完。每条都配了测试。 P0-1 企业微信 token 缓存改按**凭证指纹**(sha256(corpId,corpSecret),secret 不进 Map 的键): 按 account.id 存有两条真实串号路径 —— 草稿探测共用 'draft'(探完 A 公司再探 B 公司会命中 A 的 token,回的是 A 的应用邮箱地址,凭证填错的 B 也显示成功);已存账号换了凭证 id 不变, 下一轮轮询继续拿旧 token 拉旧 corp 的收件箱,分发给新账号绑的通道(窗口 ≤ 7200s)。 P0-2 一封「毒信」不许卡死整条路:①domain 的 buildInboundRecord 把落 text/jsonb 的字段一律剥 NUL (PG 拒收它,MIME 解出二进制垃圾时真会出现);②单封失败先落**降级骨架**(原件原样存、 主题从原始字节扒、正文写明失败原因)→ 落成了就推进游标并把「第 N 封(providerRef)解析失败, 已降级存档」写进 last_poll_error;骨架也落不下才停在这一封,那时 error **点名是哪一封**。 降级行不自动分发(正文是失败原因,不是客户的话)。 P0-3 cursorReset 只比**收侧**键(imapHost/imapPort/imapTls/login/folder;corpId/apiBase): 从前比全部 settings,把 465 改成 587 也会清游标 → IMAP 重锚 → 上次游标到重锚之间的来信永远不拉。 P0-4 复用旧行重发**先抢**(OutboundStore.claimForResend,一条条件更新;真 store 与内存件同款): 两个后端连同一个库,两台同时重放同一次放行都判「超时可复用」→ 两封真邮件,破掉「同键不发两封」。 P0-5 列表与只要元信息的读**不取 raw**(一封上限 16MB × 200 行);rawAvailable 由 `raw is not null` 投影,不靠取回字节再判;只有「下载原件」走全列的 get()。InboundRecord 新增 rawAvailable。 P0-6 账号 save() 只写**配置列**:运行列(cursor/last_polled_at/last_poll_error/last_inbound_at/ polling_until)只经 savePoll/claimPolling/releasePolling 改 —— 整行 upsert 会把另一台正持着的租约 与刚推进的游标盖回读那一刻的快照(08-24 loop_state 整份覆盖同款指纹)。清游标改为显式单写那一列。 P0-7 UIDVALIDITY 变化不再静默:provider 回 note,用例落进 last_poll_error(「已重锚在当下; 期间来信未拉取」)—— 丢了一段信而账号页一片绿是界面撒谎。 P0-8 parseCursor 连取值范围一起查(非负安全整数 + 界内:UID ≤ 0xFFFFFFFF、秒 ≤ Date 上限)—— 1e13 正是安全整数,乘 1000 出界,一行脏游标能让 listAccounts 整页 500。debt ⑨ 划线销账。 P0-9 线程口径归一:threadIdOf 改成**取根**(references[0] ?? inReplyTo,ADR-013 决策 3), mail-wiring 调它、不再内联一份。原测试改断言,不删。 P0-10 文档说老实话:mail.send 工具路发出的信**不登记 outbound_ref**(recordOutbound 的唯一调用方 channel.sendMail 在 src 里零调用方)—— API 文档 §4 与 ADR-013 决策 6 各补一段,根治记 debt。 P1:imap anchor 的 Math.min(...uids) 改 reduce(十万级 UID 会栈溢出)· wecom 翻页加页数上限 (服务端一直回 has_more:1 会死循环)· MAIL_TOOL 导出并对**真契约**断言(原 v1/v2 是手抄夹具、 自证,且 v1 版本号写错)· reset-runs 清运行数据时把邮箱账号的游标与拉取状态一并清(留着游标而信没了, wecom 十分钟重叠窗会把最近的信当新信再投一次)· due.test 的 leaseMsFor(30) 断言改写死 300_000(原为自证)· cursor.test 删掉指向评审报告的注释 · 设计稿稿头标注三处已被 ADR-013 取代 · ADR-013 补记 wecom compose_send 不带 inReplyTo/references(它发出的信在对方那里不串线程)。 契约变更(§12.4):port.ts 的 InboundStore 加 getMeta、OutboundStore 加 claimForResend、 savePoll 的 cursor 允许显式 null、MailProvider.pull 回参可带 note —— 实现方只有本模块 adapter 与 __tests__/fakes.ts 的内存件(两边语义一致,已同改),CONTRACT.md 已同步,api-report 已刷新。 测试:1922 → 1942(mail 模块 102 → 113,含真 PG 的 save 不盖运行列 / 不搬原件 / 两条真连接抢同一行)。 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01TM4kUBvekALu9qqmp7D3wr --- .../apps/api/src/modules/mail/port.d.ts | 44 +++++- .../api/src/app/__tests__/mail-routes.test.ts | 5 +- .../api/src/app/__tests__/mail-wiring.test.ts | 48 +++++++ apps/api/src/app/mail-wiring.ts | 12 +- apps/api/src/app/reset-runs.ts | 13 +- apps/api/src/app/seed-mvp.ts | 2 +- apps/api/src/modules/mail/CONTRACT.md | 25 +++- .../mail/__tests__/account-view.test.ts | 37 +++++ .../src/modules/mail/__tests__/cursor.test.ts | 21 ++- .../src/modules/mail/__tests__/due.test.ts | 5 +- apps/api/src/modules/mail/__tests__/fakes.ts | 54 +++++++- .../mail/__tests__/inbound-record.test.ts | 83 ++++++++++- .../__tests__/mail-service-guards.test.ts | 36 +++++ .../mail/__tests__/mail-service-poll.test.ts | 130 +++++++++++++++++- .../mail/__tests__/mail-store.it.test.ts | 102 +++++++++++++- .../mail/__tests__/wecom-provider.test.ts | 57 +++++++- .../internal/adapter/imap-smtp-provider.ts | 14 +- .../mail/internal/adapter/mail-store.ts | 113 ++++++++++++++- .../mail/internal/adapter/wecom-provider.ts | 37 ++++- .../mail/internal/domain/account-view.ts | 24 +++- .../modules/mail/internal/domain/cursor.ts | 23 +++- .../mail/internal/domain/inbound-record.ts | 82 ++++++++--- .../modules/mail/internal/domain/records.ts | 7 + .../mail/internal/usecase/mail-service.ts | 123 +++++++++++++++-- apps/api/src/modules/mail/port.ts | 39 +++++- .../toolbox/__tests__/tool-contract.test.ts | 17 --- docs/API文档-邮件收发.md | 8 ++ docs/adr/ADR-013-mail-gateway.md | 22 ++- docs/debt.md | 26 +++- .../specs/2026-08-25-mail-design.md | 10 +- 30 files changed, 1105 insertions(+), 114 deletions(-) diff --git a/api-report/apps/api/src/modules/mail/port.d.ts b/api-report/apps/api/src/modules/mail/port.d.ts index 3beaee0..82f99e2 100644 --- a/api-report/apps/api/src/modules/mail/port.d.ts +++ b/api-report/apps/api/src/modules/mail/port.d.ts @@ -20,7 +20,10 @@ export type MailAccountStore = { list(): Promise; get(id: string): Promise; findByName(name: string): Promise; - /** upsert;`isDefault:true` 时先把别的置 false(同一事务) */ + /** + * upsert **只写配置列**;`isDefault:true` 时先把别的置 false(同一事务)。 + * 运行列(cursor / lastPolledAt / lastPollError / lastInboundAt / pollingUntil)一概不动 —— 见 `savePoll`。 + */ save(record: MailAccountRecord): Promise; remove(id: string): Promise; /** enabled 且(从没拉过 或 last_polled_at + interval <= now)且租约空 / 过期 */ @@ -28,9 +31,16 @@ export type MailAccountStore = { /** 一条 SQL 判 + 占(设计稿 §9.1);抢到 true */ claimPolling(id: string, leaseMs: number): Promise; releasePolling(id: string): Promise; - /** 拉取过程的进度:游标 / 时刻 / 错误 / 最近来信 / 探到的地址;行不存在 = no-op */ + /** + * 拉取过程的进度:游标 / 时刻 / 错误 / 最近来信 / 探到的地址;行不存在 = no-op。 + * + * **运行列只走这条路**(与 `claimPolling` / `releasePolling` 一起)——`save()` 一个字都不碰它们: + * 运营面 PUT 的同时另一台正拉着,整行 upsert 会把租约与游标一并覆盖回读那一刻的快照 + * (`sys.loop_state` 整份覆盖那一课的同款指纹)。 + * `cursor: null` 是**显式清游标**(改了收侧配置要重锚时用)。 + */ savePoll(id: string, patch: { - cursor?: MailCursor; + cursor?: MailCursor | null; polledAt?: Date; error?: string | null; lastInboundAt?: Date; @@ -43,6 +53,10 @@ export type InboundStore = { inserted: boolean; }>; knownProviderRefs(accountId: string, refs: readonly string[]): Promise>; + /** + * 列表 —— **不取 `raw` 原件字节**(一封上限 16MB,50 行就是几百 MB,只为算 200 字预览); + * `rawAvailable` 由 `raw is not null` 投影出来,不靠把字节搬回来再判。 + */ list(query: { accountId?: string | undefined; limit: number; @@ -50,6 +64,12 @@ export type InboundStore = { record: InboundRecord; deliveries: DeliveryRecord[]; }>>; + /** 只要元信息的读(分发 / 详情 / 附件 / 重投都走它)—— 同样不取 `raw` 字节 */ + getMeta(mailId: string): Promise<{ + record: InboundRecord; + deliveries: DeliveryRecord[]; + } | undefined>; + /** 全列读 —— 只有「下载原件」那一条路要它 */ get(mailId: string): Promise<{ record: InboundRecord; deliveries: DeliveryRecord[]; @@ -60,6 +80,18 @@ export type InboundStore = { export type OutboundStore = { insert(record: OutboundRecord): Promise; byIdempotencyKey(key: string): Promise; + /** + * 复用旧行重发之前**先抢**(一条条件更新,抢到才回 true): + * `failed` 的行、或 `created_at <= staleBefore` 的 `sending` 行才抢得到;抢到即置 `sending` 并把 + * `created_at` 推到 `now` —— 这一手让那一行**离开可抢的集合**,第二个人当场抢不到。 + * + * 本仓是两个后端连同一个库:没有这一抢,两台同时重放同一次放行都会判「超时可复用」, + * 发出**两封真邮件**,直接破掉契约白纸黑字的「同键不发两封」。 + */ + claimForResend(id: string, opts: { + readonly staleBefore: Date; + readonly now: Date; + }): Promise; setState(id: string, state: 'sent' | 'failed', patch: { error?: string; providerRef?: string; @@ -108,6 +140,11 @@ export type MailProvider = { probe(account: ProviderAccount): Promise; /** 无游标时锚定初始游标(不回填;设了 backfillFrom 才从那天起) */ anchor(account: ProviderAccount, backfillFrom: Date | null, now: Date): Promise; + /** + * 拉一批。回参的 `note` 是这一轮**不算失败、但人必须知道**的事 + * (当前唯一一种:IMAP 的 UIDVALIDITY 变了、已重锚在当下、期间来信不回填)—— + * 用例把它落进 `last_poll_error`。丢了一段信而界面一片绿,那是界面撒谎。 + */ pull(account: ProviderAccount, cursor: MailCursor, opts: { readonly limit: number; readonly now: Date; @@ -118,6 +155,7 @@ export type MailProvider = { messages: RawMessage[]; cursor: MailCursor; more: boolean; + note?: string | undefined; }>; send(account: ProviderAccount, message: OutboundMessage): Promise<{ providerRef?: string | undefined; diff --git a/apps/api/src/app/__tests__/mail-routes.test.ts b/apps/api/src/app/__tests__/mail-routes.test.ts index 3deabb3..ee2578b 100644 --- a/apps/api/src/app/__tests__/mail-routes.test.ts +++ b/apps/api/src/app/__tests__/mail-routes.test.ts @@ -69,7 +69,10 @@ describe('账号路由', () => { expect(await (await h.json('POST', '/api/gov/mail/accounts/acc-1/poll')).json()).toMatchObject({ pulled: 1, stored: 1 }); await h.accounts.save(account({ enabled: false })); expect((await h.json('POST', '/api/gov/mail/accounts/acc-1/poll')).status).toBe(412); - await h.accounts.save(account({ pollingUntil: new Date('2027-01-01T00:00:00Z') })); + // 租约是**运行列**:`save()` 只写配置列(整行 upsert 会盖掉另一台正持着的租约), + // 所以按真路径占它 —— claimPolling 就是另一台机器那一下 + await h.accounts.save(account()); + expect(await h.accounts.claimPolling('acc-1', 60_000)).toBe(true); expect((await h.json('POST', '/api/gov/mail/accounts/acc-1/poll')).status).toBe(409); }); diff --git a/apps/api/src/app/__tests__/mail-wiring.test.ts b/apps/api/src/app/__tests__/mail-wiring.test.ts index 9459d66..4b5b64c 100644 --- a/apps/api/src/app/__tests__/mail-wiring.test.ts +++ b/apps/api/src/app/__tests__/mail-wiring.test.ts @@ -2,7 +2,11 @@ import { describe, expect, it } from 'vitest'; import type { MailInboundDetail, MailSendInput, MailSendResult } from '@contracts'; import type { InboundMail } from '@contracts'; +import { auditPassed, checkEvolution, staticAudit } from '@modules/toolbox'; +import type { ToolContract } from '@modules/toolbox'; +import { threadIdOf } from '@modules/mail'; import { channelSubscriber, channelTransport, mailSendTool, toInboundMail } from '../mail-wiring.ts'; +import { MAIL_TOOL } from '../seed-mvp.ts'; const detail: MailInboundDetail = { id: 'in-1', @@ -53,6 +57,17 @@ describe('toInboundMail', () => { expect(mail.attachments).toEqual([{ name: 'bom.xlsx', contentType: 'application/vnd.ms-excel', size: 10, ref: 'up-1' }]); expect(mail.receivedAt).toBe('2026-08-25T09:30:00.000Z'); }); + + it('线程口径只有一处:这里就是 mail 模块的 threadIdOf,不是另抄的一份', () => { + // 从前这里内联了一份同义实现,而模块导出的 threadIdOf 取的是**最近一跳** —— + // 同一个模块两套相反的语义,下一个人照导出的那个接线就会接反 + expect(toInboundMail(detail).threadId).toBe(threadIdOf(detail)); + const onlyHop: MailInboundDetail = { ...detail, references: [], inReplyTo: '' }; + expect(toInboundMail(onlyHop).threadId).toBe(threadIdOf(onlyHop)); + expect(toInboundMail(onlyHop).threadId).toBe(''); + const none: MailInboundDetail = { ...detail, references: [], inReplyTo: null }; + expect(toInboundMail(none).threadId).toBeUndefined(); + }); }); describe('channelSubscriber', () => { @@ -122,3 +137,36 @@ describe('mailSendTool', () => { expect(invalid.error?.code).toBe('invalid'); }); }); + +describe('mail.send 2.0.0 的演进', () => { + /* + * **对真契约断言,不对手抄的夹具**。上一版这里的 v1/v2 都是照着 `seed-mvp.ts` 手抄的两份 JSON: + * 改真的那一份它不会红(自证),而且抄的 v1 版本号还写成了 `1.0.0` —— + * 真前一版是 `1.1.0`(MAIL_TOOL 展开 HISTORY_TOOL,版本跟着它)。 + * v1 是从 git 里那一版逐字取回来的**历史快照**(只有它必须手写:那份代码已经不在仓里了)。 + */ + const v1: ToolContract = { + ...MAIL_TOOL, + version: '1.1.0', + description: '发出邮件(报价单 / 询价函 / 催办):写类动作,放行时才真正触达世界', + input: { type: 'object', required: ['to', 'subject'], properties: { to: { type: 'string' }, subject: { type: 'string' }, body: { type: 'string' } } }, + output: { type: 'object', properties: { queued: { type: 'boolean' } } }, + connectionKeys: ['endpoint'], + }; + + it('真契约就是 2.0.0,且过得了静态审', () => { + expect(MAIL_TOOL.name).toBe('mail.send'); + expect(MAIL_TOOL.version).toBe('2.0.0'); + expect(auditPassed(staticAudit(MAIL_TOOL))).toBe(true); + }); + + it('保留 v1 的 body / queued、不新增必填 —— 演进检查零违规', () => { + expect(checkEvolution(v1, MAIL_TOOL)).toEqual([]); + // 反证这条断言真的在看东西:真把 v1 的 body 删掉就该报违规 + const dropped: ToolContract = { + ...MAIL_TOOL, + input: { type: 'object', required: ['to', 'subject'], properties: { to: { type: 'array' }, subject: { type: 'string' } } }, + }; + expect(checkEvolution(v1, dropped).map((f) => f.field)).toContain('input.body'); + }); +}); diff --git a/apps/api/src/app/mail-wiring.ts b/apps/api/src/app/mail-wiring.ts index 8dc3772..0cde742 100644 --- a/apps/api/src/app/mail-wiring.ts +++ b/apps/api/src/app/mail-wiring.ts @@ -15,21 +15,19 @@ import { MailNoDefaultAccountError, MailSecretMissingError, MailSendInvalidError, + threadIdOf, } from '@modules/mail'; import type { ToolHandler } from '@modules/toolbox'; /** * mail 的入站详情 → channel 的 InboundMail:只带存下来的附件。 * - * **线程取的是根,不是最近一跳**:我方出站登记的 threadId 恒是这一串的**首封** - * (`channelTransport` 回的是 `input.threadId ?? messageId`,channel 据此 `recordOutbound`), - * 而归属解析①级是精确等值比对。若这里取 `inReplyTo`(最近一跳),跑一个来回就漂: - * 我方 M1 → 客户回 M1(命中)→ 我方再回(实际 M2、仍记在 M1 名下)→ 客户回 M2 → - * 取到 M2,而库里从没记过 M2 ⇒ ①级落空,退到③地址+时间窗。 - * `references[0]` 才稳定:客户端续链时首项恒是首封;客户端只给 In-Reply-To 时退回它。 + * **线程口径只有一处**:`threadIdOf`(mail 模块的 domain,取这一串的根;缘由与反例见那里)。 + * 这里曾经内联过一份同义实现,而模块里那一份当时取的是最近一跳 —— 同一个模块两套相反的语义, + * 下一个人照导出的那个接线就会接反。现在调它。 */ export function toInboundMail(mail: MailInboundDetail): InboundMail { - const threadId = mail.references[0] ?? mail.inReplyTo ?? undefined; + const threadId = threadIdOf(mail); return { messageId: mail.messageId, ...(threadId === undefined ? {} : { threadId }), diff --git a/apps/api/src/app/reset-runs.ts b/apps/api/src/app/reset-runs.ts index e668814..a3b8ed9 100644 --- a/apps/api/src/app/reset-runs.ts +++ b/apps/api/src/app/reset-runs.ts @@ -5,7 +5,7 @@ * * **清的是「跑出来的」**:任务、工作流、卡、任务记忆、事件、监听、台账行、活动记忆… * **留的是「配置好的」**:员工与岗位授权、会话、工作组、流程包、活动库、工具、 - * 触发器、通道、模型实例、纪要。 + * 触发器、通道、模型实例、纪要、**邮箱账号**(它身上的游标与拉取状态是运行痕,跟着清)。 * * 直接走 SQL 而不绕模块:模块出口里没有、也不该有「删掉所有运行痕迹」这种动作 —— * 留痕只增不改是系统的规矩,清库是**开发期的外科手术**,不是系统的一项能力。 @@ -89,6 +89,17 @@ export async function resetRuns(): Promise { ); const all = [...RUN_TABLES, ...ledger.rows.map((r) => r.name)]; await pool.query(`truncate table ${all.join(', ')} restart identity cascade`); + /* + * 邮件账号是**配置**(留着),但它身上挂着的**游标是运行痕** —— 清了 `sys.mail_inbound` + * 却留着游标,下一轮就从游标处往后拉:wecom 那十分钟重叠窗会把最近的信当新信**再投一次** + * (库里已经没有那一行了,唯一键挡不住),可能凭空再起一单。 + * 游标清空 = 下一轮重锚在当下、不回填,与「第一次配上账号」同一个起点。 + * 上次拉取 / 错误 / 最近来信也一并清:它们说的都是刚被清掉的那些信。 + */ + await pool.query( + `update sys.mail_account + set cursor = null, last_polled_at = null, last_poll_error = null, last_inbound_at = null, polling_until = null`, + ); const kept = await pool.query<{ n: string }>( `select count(*)::text as n from ledger.__object_meta where kind = 'master'`, ); diff --git a/apps/api/src/app/seed-mvp.ts b/apps/api/src/app/seed-mvp.ts index e0b8b55..cad8758 100644 --- a/apps/api/src/app/seed-mvp.ts +++ b/apps/api/src/app/seed-mvp.ts @@ -56,7 +56,7 @@ const HISTORY_TOOL = { * `connectionKeys: ['account']` —— 从哪个信箱发:'default' = 默认账号,或某个账号 id。 * `dryRun`:注册的试运行会**真调处理器**,样例带 dryRun 只校验不发。 */ -const MAIL_TOOL = { +export const MAIL_TOOL = { ...HISTORY_TOOL, name: 'mail.send', displayName: '发送邮件', diff --git a/apps/api/src/modules/mail/CONTRACT.md b/apps/api/src/modules/mail/CONTRACT.md index d52b07e..6fe320d 100644 --- a/apps/api/src/modules/mail/CONTRACT.md +++ b/apps/api/src/modules/mail/CONTRACT.md @@ -35,6 +35,22 @@ 不变量(ADR-013): - **入站先持久后分发**:`sys.mail_inbound` 两把唯一键 `(account_id, message_id)` / `(account_id, provider_ref)`, 重复拉到一律 `inserted:false` 静默跳过、不分发;游标逐封(IMAP)/ 逐批(wecom)推进,失败停在最后一封成功处 + - **一封解析不动的信不许卡死整条路**(2026-08-25 终审):落 text/jsonb 的字段一律剥 NUL(`stripNul`); + 单封 `storeOne` 抛错时先落**降级骨架**(原件原样存、主题从原始字节扒、正文写明失败原因)—— + 落成了就推进游标并把「第 N 封(providerRef)解析失败,已降级存档」写进 `last_poll_error`; + 骨架也落不下才停在这一封,那时 `last_poll_error` **必须点名是哪一封**。降级行**不自动分发** + (正文是失败原因不是客户的话),人可按「重新投递」 + - **线程取根**:`threadIdOf` = `references[0] ?? inReplyTo`(ADR-013 决策 3),全仓唯一口径 —— + app 的 `mail-wiring.toInboundMail` 调它,不另抄一份 + - **provider 的 `note` 要上界面**:`pull` 回的 `note`(当前唯一一种:UIDVALIDITY 变化已重锚、期间来信未拉取) + 落进 `last_poll_error`;丢了一段信而账号页一片绿是界面撒谎 + - **读取分两档**:`list` / `getMeta` **不取 `raw` 原件字节**(一封上限 16MB),`rawAvailable` 由 + `raw is not null` 投影;只有「下载原件」那一条路走全列的 `get` + - **账号的配置列与运行列分家**:`save()` 只写配置列;运行列(cursor / last_polled_at / last_poll_error / + last_inbound_at / polling_until)只经 `savePoll` / `claimPolling` / `releasePolling` 改 —— + 整行 upsert 会把另一台正持着的租约与刚推进的游标盖回读那一刻的快照。清游标是显式的 `savePoll({cursor:null})` + - **重锚只看收侧**:`cursorReset` 只比 kind 与收侧 settings 键(imap-smtp:`imapHost`/`imapPort`/`imapTls`/ + `login`/`folder`;wecom:`corpId`/`apiBase`)—— 改 SMTP 端口不该把来信丢掉 - **分发自动只投一次**:先落 `dispatching` 再调,`done / skipped / failed(原话脱敏)`;不自动重试 —— channel → event 按 `cause.ref` 不幂等,重投会起两单;「重新投递」是人按的钮,那一刻起是至少一次; 一个订阅方失败不影响其它;订阅是代码级装配,不落库、不回放老信 @@ -42,6 +58,9 @@ - **跨进程租约**:`polling_until` 一条 SQL 判 + 占,过期 max(5min, 3×间隔);抢不到 → `MailPollBusyError` - **出站先落行**:`sending` → provider → `sent` / `failed(原话)`;带 `idempotencyKey` 且已 `sent` 的不发第二封; 厂商失败**不抛**,以 `state='failed'` 返回(调用方各自翻译) + - **复用旧行重发要先抢**:`OutboundStore.claimForResend` 一条条件更新(`failed`、或停够 5 分钟的 `sending` + 才抢得到;抢到即置 `sending` 并把 `created_at` 推到 now)—— 抢到才发,抢不到回「本次未重复发送」。 + 两个后端连同一个库时,这一句是「同键不发两封」唯一的墙 - **凭证密封**:`secret_cipher` 永不出接口,视图只回 `secretBound`;更新时 secret 缺省 = 不改; 明文只在 usecase 打开、交给 provider 那一段;所有落库的厂商原话先过 `redactSecrets` - **界面不撒谎**:探测收 / 发分栏各报;wecom 的发信一栏 `checked:false`(拿到 token 不等于能发); @@ -50,4 +69,8 @@ (回放读 `sys.mail_inbound`,永不重拉);活动侧出站经 `ToolInvoker`(`mail.send` 工具),系统侧留痕 `sys.mail_outbound`;`createFakeMailProvider` 是测试替身 -影响半径: 新模块;channel / toolbox / event 的 index / contract / port 未变 +影响半径: 新模块;channel / toolbox / event 的 index / contract / port 未变。 + 2026-08-25 终审修复轮改了 `port.ts`:`InboundStore` 新增 `getMeta`(list / getMeta 不再取 raw)、 + `OutboundStore` 新增 `claimForResend`、`MailAccountStore.savePoll` 的 `cursor` 允许显式 `null`、 + `MailProvider.pull` 回参可带 `note` —— 端口的实现方只有本模块的 adapter 与 `__tests__/fakes.ts` + 的内存件(两边语义必须一致),app 侧装配未变 diff --git a/apps/api/src/modules/mail/__tests__/account-view.test.ts b/apps/api/src/modules/mail/__tests__/account-view.test.ts index 4323bab..8dc7a63 100644 --- a/apps/api/src/modules/mail/__tests__/account-view.test.ts +++ b/apps/api/src/modules/mail/__tests__/account-view.test.ts @@ -1,6 +1,7 @@ /** 账号记录 ↔ 入参 / 视图:视图不带密文;kind / settings 变了游标清空;secret 缺省保留 */ import { describe, expect, it } from 'vitest'; import { mailAccountInputSchema } from '@contracts'; +import type { MailAccountInput } from '@contracts'; import type { MailAccountRecord } from '../internal/domain/records.ts'; import { imapCursor } from '../internal/domain/cursor.ts'; import { cursorReset, mergeAccount, secretOf, viewOf } from '../internal/domain/account-view.ts'; @@ -26,6 +27,42 @@ describe('mergeAccount / secretOf', () => { const probed: MailAccountRecord = { ...existing, address: 'app@corp.wecom.work' }; expect(mergeAccount(wecom, { id: 'w', existing: probed, secretCipher: null, now }).address).toBe('app@corp.wecom.work'); }); + it('**只有收侧的键动了才重锚**:改 smtpPort 游标留着,改 imapHost / folder 才清', () => { + /* + * 从前比对全部 settings 键:人把 465 改成 587(纯发信的事)也会清游标 → IMAP 重锚在当下 → + * 上次游标到重锚之间到达的信**永远不拉**(设计即「不回填」)。改发信的设置不该丢来信。 + */ + const existing: MailAccountRecord = { + ...mergeAccount(input, { id: 'acc-1', existing: undefined, secretCipher: 's', now }), + cursor: imapCursor('7', 9), + }; + const changed = (over: Record): MailAccountInput => + mailAccountInputSchema.parse({ ...input, settings: { ...input.settings, ...over } }); + // 发侧三键:一个都不重锚 + expect(cursorReset(existing, changed({ smtpPort: 587 }))).toBe(false); + expect(cursorReset(existing, changed({ smtpHost: 'smtp.other.com' }))).toBe(false); + expect(cursorReset(existing, changed({ smtpTls: false }))).toBe(false); + // 收侧:换信箱 / 换文件夹 / 换登录名 / 换端口都是另一个游标 + expect(cursorReset(existing, changed({ imapHost: 'imap.other.com' }))).toBe(true); + expect(cursorReset(existing, changed({ folder: 'Archive' }))).toBe(true); + expect(cursorReset(existing, changed({ imapPort: 143 }))).toBe(true); + expect(cursorReset(existing, changed({ imapTls: false }))).toBe(true); + expect(cursorReset(existing, changed({ login: 'other@we.com' }))).toBe(true); + // 只改 smtpPort 时游标真的留在记录里(不只是判据为假) + expect(mergeAccount(changed({ smtpPort: 587 }), { id: 'acc-1', existing, secretCipher: 's', now }).cursor).toEqual(imapCursor('7', 9)); + }); + + it('wecom 形按 corpId / apiBase 判重锚', () => { + const wecom = mailAccountInputSchema.parse({ kind: 'wecom', name: '应用', settings: { corpId: 'ww1' } }); + const existing: MailAccountRecord = { + ...mergeAccount(wecom, { id: 'w', existing: undefined, secretCipher: null, now }), + cursor: null, + }; + expect(cursorReset(existing, wecom)).toBe(false); + expect(cursorReset(existing, mailAccountInputSchema.parse({ ...wecom, settings: { corpId: 'ww2' } }))).toBe(true); + expect(cursorReset(existing, mailAccountInputSchema.parse({ ...wecom, settings: { corpId: 'ww1', apiBase: 'https://qyapi.example.com' } }))).toBe(true); + }); + it('更新:settings 没变游标保留;变了(或换 kind)游标与上次错误清空', () => { const existing: MailAccountRecord = { ...mergeAccount(input, { id: 'acc-1', existing: undefined, secretCipher: 's', now }), diff --git a/apps/api/src/modules/mail/__tests__/cursor.test.ts b/apps/api/src/modules/mail/__tests__/cursor.test.ts index 2e0a80a..1e224b8 100644 --- a/apps/api/src/modules/mail/__tests__/cursor.test.ts +++ b/apps/api/src/modules/mail/__tests__/cursor.test.ts @@ -53,9 +53,6 @@ describe('摘要与解析', () => { it('人读摘要;没拉过是 null', () => { expect(cursorSummary(null)).toBeNull(); expect(cursorSummary(imapCursor('7', 100))).toBe('UID 100 · validity 7'); - // brief 原字面量 1756116000 对应 2025-08-25T10:00:00.000Z(与期望的 2026 年份相差整 365 天, - // 与本文件其余用例统一锚定的 2026-08-25T10:00:00Z 不符)——已核实为 brief 的数据笔误, - // 修的是输入字面量而非断言期望值,见 task-4-report.md。 expect(cursorSummary(wecomCursor(1787652000))).toBe('至 2026-08-25T10:00:00.000Z'); }); it('从 jsonb 解回来:形状不对当没有游标(重锚),不抛', () => { @@ -65,4 +62,22 @@ describe('摘要与解析', () => { expect(parseCursor(null)).toBeNull(); expect(parseCursor('garbage')).toBeNull(); }); + it('**取值范围也算形状**:一行脏游标不许把账号页炸成 500', () => { + /* + * `cursorSummary` 跑在 `viewOf` 里:`new Date(1e13 * 1000).toISOString()` 抛 RangeError, + * 一行脏游标 → `listAccounts` 整页 500。不合格的当没有游标(下一轮重锚),不抛。 + */ + for (const bad of [1e13, -1, Number.NaN, Number.POSITIVE_INFINITY, 1.5, Number.MAX_SAFE_INTEGER + 2]) { + expect(parseCursor({ kind: 'wecom', lastTime: bad })).toBeNull(); + expect(parseCursor({ kind: 'imap', uidValidity: '7', lastUid: bad })).toBeNull(); + } + // 1e13 是安全整数,乘 1000 却出了 Date 的界 —— 「安全整数」这一句拦不住它,上界必须自己写 + expect(Number.isSafeInteger(1e13)).toBe(true); + // UID 是 32 位无符号:恰好在界内的收,越界的当没有 + expect(parseCursor({ kind: 'imap', uidValidity: '7', lastUid: 0xffff_ffff })).toEqual(imapCursor('7', 0xffff_ffff)); + expect(parseCursor({ kind: 'imap', uidValidity: '7', lastUid: 0x1_0000_0000 })).toBeNull(); + expect(parseCursor({ kind: 'wecom', lastTime: 0 })).toEqual(wecomCursor(0)); + expect(() => cursorSummary(parseCursor({ kind: 'wecom', lastTime: 1e13 }))).not.toThrow(); + expect(cursorSummary(parseCursor({ kind: 'wecom', lastTime: 1e13 }))).toBeNull(); + }); }); diff --git a/apps/api/src/modules/mail/__tests__/due.test.ts b/apps/api/src/modules/mail/__tests__/due.test.ts index 03140ad..f942d1a 100644 --- a/apps/api/src/modules/mail/__tests__/due.test.ts +++ b/apps/api/src/modules/mail/__tests__/due.test.ts @@ -22,7 +22,10 @@ describe('isDue', () => { describe('leaseMsFor / isPolling', () => { it('租约 = max(5 分钟, 3 × 间隔)', () => { - expect(leaseMsFor(30)).toBe(MIN_LEASE_MS); + // 下限那一档写死 300_000,不写 MIN_LEASE_MS:拿被测的常量当期望值是自证 —— + // 哪天有人把 MIN_LEASE_MS 改成 5 秒,这条断言照样绿 + expect(leaseMsFor(30)).toBe(300_000); + expect(MIN_LEASE_MS).toBe(300_000); expect(leaseMsFor(600)).toBe(1_800_000); }); it('polling 只在租约未过期时为真', () => { diff --git a/apps/api/src/modules/mail/__tests__/fakes.ts b/apps/api/src/modules/mail/__tests__/fakes.ts index b2bd600..0a3c4fc 100644 --- a/apps/api/src/modules/mail/__tests__/fakes.ts +++ b/apps/api/src/modules/mail/__tests__/fakes.ts @@ -96,7 +96,27 @@ export function memoryAccounts(clock: Clock): MailAccountStore & { rows: Map; deliveries: Map } { const rows = new Map(); const deliveries = new Map(); @@ -158,6 +187,7 @@ export function memoryInbound(): InboundStore & { rows: Map query.accountId === undefined || r.accountId === query.accountId) // 与真 store 同序:两级键 desc(receivedAt), desc(createdAt) —— 并列时后创建的排前。 // 少了第二级,并列会按 Map 插入序解(= createdAt 升序),与真身**方向相反**, @@ -166,6 +196,10 @@ export function memoryInbound(): InboundStore & { rows: Map ({ record, deliveries: of(record.id) })); }, + async getMeta(mailId) { + const record = rows.get(mailId); + return record === undefined ? undefined : { record: withoutRaw(record), deliveries: of(mailId) }; + }, async get(mailId) { const record = rows.get(mailId); return record === undefined ? undefined : { record, deliveries: of(mailId) }; @@ -189,6 +223,16 @@ export function memoryOutbound(): OutboundStore & { rows: Map r.idempotencyKey === key); }, + async claimForResend(id, opts) { + // 与真 store 同一句条件:failed 可抢;sending 只有停够了才算「上次崩在半路」。 + // 抢到即置 sending 并把 createdAt 推到 now —— 那一行当场离开可抢的集合,第二个人抢不到。 + const r = rows.get(id); + if (r === undefined) return false; + const claimable = r.state === 'failed' || (r.state === 'sending' && r.createdAt.getTime() <= opts.staleBefore.getTime()); + if (!claimable) return false; + rows.set(id, { ...r, state: 'sending', createdAt: opts.now, error: null }); + return true; + }, async setState(id, state, patch) { const r = rows.get(id); if (r === undefined) return; @@ -241,6 +285,8 @@ export type FakeProvider = MailProvider & { failPullWith(error: Error | undefined, opts?: { readonly afterCalls?: number }): void; failSendWith(error: Error | undefined): void; probeWith(report: MailProbeReport | undefined): void; + /** 这一轮带回一句「不算失败但人得知道」的话(真身:UIDVALIDITY 变化已重锚) */ + noteWith(note: string | undefined): void; }; /** 可脚本化的假 provider:排队的信按 limit 交出去,IMAP 形逐封带 cursorAfter */ @@ -252,6 +298,7 @@ export function fakeProvider(kind: MailAccountKind = 'imap-smtp'): FakeProvider let pullFailAfter = 0; let sendError: Error | undefined; let probeReport: MailProbeReport | undefined; + let note: string | undefined; let uidSeq = 0; return { kind, @@ -270,6 +317,9 @@ export function fakeProvider(kind: MailAccountKind = 'imap-smtp'): FakeProvider probeWith(report) { probeReport = report; }, + noteWith(value) { + note = value; + }, async probe(accountArg) { return ( probeReport ?? { @@ -293,7 +343,7 @@ export function fakeProvider(kind: MailAccountKind = 'imap-smtp'): FakeProvider }); const last = batch.at(-1)?.cursorAfter; const next = kind === 'imap-smtp' ? (last ?? cursor) : wecomCursor(Math.floor(opts.now.getTime() / 1000)); - return { messages: batch, cursor: next, more: pending.length > 0 }; + return { messages: batch, cursor: next, more: pending.length > 0, ...(note === undefined ? {} : { note }) }; }, async send(_accountArg, message) { if (sendError !== undefined) throw sendError; diff --git a/apps/api/src/modules/mail/__tests__/inbound-record.test.ts b/apps/api/src/modules/mail/__tests__/inbound-record.test.ts index 3c347e8..7e30281 100644 --- a/apps/api/src/modules/mail/__tests__/inbound-record.test.ts +++ b/apps/api/src/modules/mail/__tests__/inbound-record.test.ts @@ -10,6 +10,7 @@ import { planAttachments, rawOf, stripHtml, + stripNul, summaryOf, textPreviewOf, threadIdOf, @@ -31,10 +32,18 @@ const parsed: ParsedMail = { const now = new Date('2026-08-25T10:00:00Z'); describe('线程与正文', () => { - it('线程 = In-Reply-To;没有就取 References 末项;都没有就是 undefined', () => { - expect(threadIdOf(parsed)).toBe(''); - expect(threadIdOf({ references: ['', ''] })).toBe(''); + it('**线程 = 这一串的根**(References[0]);只给 In-Reply-To 时才退回它;都没有就是 undefined', () => { + /* + * 2026-08-25 终审改口径:原先取的是最近一跳(inReplyTo ?? references.at(-1)), + * 与 ADR-013 决策 3 和 app 侧接线正相反 —— 跑一个来回归属①级就漂掉。 + * 断言随语义更新(测试没删,期望值改对):同一封信 parsed 现在给的是根 。 + */ + expect(threadIdOf(parsed)).toBe(''); + expect(threadIdOf({ references: ['', ''] })).toBe(''); + expect(threadIdOf({ references: [], inReplyTo: '' })).toBe(''); expect(threadIdOf({ references: [] })).toBeUndefined(); + // 入站详情那一形用 null 表示「没有」—— 同一个函数要接得住(app 的 mail-wiring 传的就是它) + expect(threadIdOf({ references: [], inReplyTo: null })).toBeUndefined(); }); it('缺 Message-ID 用 providerRef@域顶上,不抛', () => { expect(fallbackMessageId('uid:7:12', 'sales@we.com')).toBe(''); @@ -75,6 +84,74 @@ describe('附件与原件', () => { }); }); +describe('NUL(U+0000)一律剥掉', () => { + /* + * PG 的 text / jsonb 列拒收 NUL,而 MIME 解出二进制垃圾时它真会出现。 + * 不剥的话这一封每 30 秒炸一次、游标停在它前面,后面所有新信永远进不来。 + */ + const NUL = '\u0000'; + it('纯函数:有就剥,没有就原样(同一个引用)', () => { + expect(stripNul(`a${NUL}b`)).toBe('ab'); + const clean = '干净的一句'; + expect(stripNul(clean)).toBe(clean); + expect(stripNul(NUL)).toBe(''); + }); + + it('落 text / jsonb 的字段全剥:主题 / 正文 / html / 地址 / 头部串 / 附件名', () => { + const dirty = buildInboundRecord({ + id: 'in-nul', + accountId: 'acc-1', + address: 'sales@we.com', + providerRef: `uid:7:${NUL}14`, + parsed: { + messageId: ``, + inReplyTo: ``, + references: [``], + from: { address: `buyer${NUL}@acme.com`, name: `张${NUL}三` }, + to: [{ address: `sales${NUL}@we.com`, name: '' }], + cc: [{ address: `cc${NUL}@we.com`, name: '' }], + subject: `询价${NUL}RFQ-1`, + text: `正文${NUL}一`, + html: `

${NUL}二

`, + attachments: [], + }, + eml: new Uint8Array([0, 1, 0]), + receivedAt: now, + attachments: [{ index: 0, name: `bom${NUL}.xlsx`, contentType: `application/${NUL}zip`, size: 3, ref: null, note: `注${NUL}` }], + now, + }); + expect(JSON.stringify(dirty)).not.toContain('\\u0000'); + expect(dirty.subject).toBe('询价RFQ-1'); + expect(dirty.text).toBe('正文一'); + expect(dirty.html).toBe('

二

'); + expect(dirty.messageId).toBe(''); + expect(dirty.providerRef).toBe('uid:7:14'); + expect(dirty.inReplyTo).toBe(''); + expect(dirty.references).toEqual(['']); + expect(dirty.from).toEqual({ address: 'buyer@acme.com', name: '张三' }); + expect(dirty.to[0]?.address).toBe('sales@we.com'); + expect(dirty.cc[0]?.address).toBe('cc@we.com'); + expect(dirty.attachments[0]).toMatchObject({ name: 'bom.xlsx', contentType: 'application/zip', note: '注' }); + // 原件是 bytea:一个字节都不动 + expect(dirty.raw).toEqual(new Uint8Array([0, 1, 0])); + }); + + it('缺 Message-ID 时顶上的那一串也剥过(它进唯一键)', () => { + const record = buildInboundRecord({ + id: 'in-nul-2', + accountId: 'acc-1', + address: 'sales@we.com', + providerRef: `uid:7:${NUL}15`, + parsed: { references: [], to: [], cc: [], subject: '', attachments: [] }, + eml: new Uint8Array(), + receivedAt: now, + attachments: [], + now, + }); + expect(record.messageId).toBe(''); + }); +}); + describe('buildInboundRecord / 视图', () => { const record = buildInboundRecord({ id: 'in-1', diff --git a/apps/api/src/modules/mail/__tests__/mail-service-guards.test.ts b/apps/api/src/modules/mail/__tests__/mail-service-guards.test.ts index 8f970c9..60a8e32 100644 --- a/apps/api/src/modules/mail/__tests__/mail-service-guards.test.ts +++ b/apps/api/src/modules/mail/__tests__/mail-service-guards.test.ts @@ -101,6 +101,42 @@ describe('幂等键', () => { expect(result.outboundId).toBe('out-9'); }); + it('**并发复用要抢**:同键两路同时重发,真信只出一封,另一路如实回没发出去', async () => { + /* + * 本仓是两个后端连同一个库:两台同时重放同一次放行,都会读到同一行 `failed`、都判「可以复用」。 + * 没有那一句条件更新的话,两台都发 —— **两封真邮件**,直接破掉契约白纸黑字的「同键不发两封」。 + */ + const { service, accounts, outbound, imap } = harness(); + await accounts.save(account()); + await outbound.insert(outboundRow({ state: 'failed', error: '上一次没发出去' })); + + const [a, b] = await Promise.all([ + service.send({ ...sendInput, idempotencyKey: 'k-1' }), + service.send({ ...sendInput, idempotencyKey: 'k-1' }), + ]); + expect(imap.sent).toHaveLength(1); + const states = [a.state, b.state].sort(); + expect(states).toEqual(['failed', 'sent']); + const loser = a.state === 'failed' ? a : b; + expect(loser.error).toContain('本次未重复发送'); + expect(loser.outboundId).toBe('out-9'); + // 抢输的那一路一个字都没往外发,也没新开一行 + expect(outbound.rows.size).toBe(1); + expect(outbound.rows.get('out-9')?.state).toBe('sent'); + }); + + it('**超时 sending 的复用同样要抢**:抢不到就走「未重复发送」那一支', async () => { + const { service, accounts, outbound, imap, clock } = harness(); + await accounts.save(account()); + await outbound.insert(outboundRow({ createdAt: new Date(clock.now().getTime() - 6 * 60_000) })); + const [a, b] = await Promise.all([ + service.send({ ...sendInput, idempotencyKey: 'k-1' }), + service.send({ ...sendInput, idempotencyKey: 'k-1' }), + ]); + expect(imap.sent).toHaveLength(1); + expect([a.state, b.state].sort()).toEqual(['failed', 'sent']); + }); + it('失败重发成功之后,那一行不留上一次的错误话', async () => { const { service, accounts, outbound, imap } = harness(); await accounts.save(account()); diff --git a/apps/api/src/modules/mail/__tests__/mail-service-poll.test.ts b/apps/api/src/modules/mail/__tests__/mail-service-poll.test.ts index 8395a13..15e013b 100644 --- a/apps/api/src/modules/mail/__tests__/mail-service-poll.test.ts +++ b/apps/api/src/modules/mail/__tests__/mail-service-poll.test.ts @@ -5,9 +5,11 @@ import { describe, expect, it } from 'vitest'; import type { MailInboundDetail } from '@contracts'; import { MailAccountDisabledError, MailPollBusyError, MailSubscriberTakenError, MailSubscriberUnknownError } from '../contract.ts'; +import { REANCHOR_NOTE } from '../internal/adapter/imap-smtp-provider.ts'; import { createMailService } from '../internal/usecase/mail-service.ts'; import { createPostalMimeParser } from '../internal/adapter/postal-mime-parser.ts'; import { imapCursor } from '../internal/domain/cursor.ts'; +import type { InboundStore, MimeParser } from '../port.ts'; import { account, eml, @@ -22,10 +24,10 @@ import { seqIds, } from './fakes.ts'; -function harness(over: { perTick?: number } = {}) { +function harness(over: { perTick?: number; mime?: MimeParser; inbound?: InboundStore } = {}) { const clock = fixedClock(); const accounts = memoryAccounts(clock); - const inbound = memoryInbound(); + const inbound = over.inbound ?? memoryInbound(); const attachments = memoryAttachments(); const imap = fakeProvider('imap-smtp'); const wecom = fakeProvider('wecom'); @@ -36,7 +38,7 @@ function harness(over: { perTick?: number } = {}) { outbound: memoryOutbound(), attachments, providers: { 'imap-smtp': imap, wecom }, - mime: createPostalMimeParser(), + mime: over.mime ?? createPostalMimeParser(), secrets: plainSecrets, clock, ids: seqIds('in'), @@ -46,6 +48,17 @@ function harness(over: { perTick?: number } = {}) { return { service, accounts, inbound, attachments, imap, wecom, clock, changes }; } +/** 解析到点名那封就炸 —— 「毒信」的最小可复现形(真实触发器是 MIME 解出二进制垃圾) */ +function poisonMime(poison: string): MimeParser { + const real = createPostalMimeParser(); + return { + async parse(emlBytes) { + if (new TextDecoder().decode(emlBytes).includes(poison)) throw new Error('MIME 解析炸了:invalid byte sequence'); + return real.parse(emlBytes); + }, + }; +} + describe('拉取', () => { it('无游标先锚定;拉到的信入库、附件入 upload、游标逐封推进、lastPolledAt 与 lastInboundAt 记下', async () => { const h = harness(); @@ -145,7 +158,10 @@ describe('拉取', () => { expect(h.imap.pullCalls).toHaveLength(0); await h.accounts.save(account({ enabled: false })); await expect(h.service.pollNow('acc-1')).rejects.toBeInstanceOf(MailAccountDisabledError); - await h.accounts.save(account({ pollingUntil: new Date(h.clock.now().getTime() + 60_000) })); + // 租约是**运行列**:`save()` 只写配置列(两台后端连同一个库时整行 upsert 会盖掉别人的租约), + // 所以这里按真路径占它 —— claimPolling 就是另一台机器那一下 + await h.accounts.save(account()); + expect(await h.accounts.claimPolling('acc-1', 60_000)).toBe(true); await expect(h.service.pollNow('acc-1')).rejects.toBeInstanceOf(MailPollBusyError); }); @@ -162,6 +178,112 @@ describe('拉取', () => { }); }); +describe('毒信不许卡死整条路(降级存档)', () => { + it('第一封解析必炸:降级行落库、游标越过它、第二封照常入库、error 点了名', async () => { + const h = harness({ mime: poisonMime('毒') }); + await h.accounts.save(account()); + h.imap.queue( + rawMessage('uid:v1:1', eml({ messageId: '', subject: '毒信一封' })), + rawMessage('uid:v1:2', eml({ messageId: '', subject: '好信' })), + ); + const hits: string[] = []; + h.service.subscribe('channel', async (m) => void hits.push(m.subject)); + + const outcome = await h.service.pollNow('acc-1'); + // 两封都落了库(降级的那封也算 stored),但只有解得开的那封投给了订阅方 + expect(outcome).toMatchObject({ pulled: 2, stored: 2, dispatched: 1 }); + expect(hits).toEqual(['好信']); + + const list = await h.service.listInbound({}); + const degraded = list.find((m) => m.messageId.includes('uid:v1:1')); + // 主题从原始字节里扒出来 —— 人在列表上认得出是哪一封卡住了 + expect(degraded?.subject).toBe('毒信一封'); + expect(degraded?.textPreview).toContain('降级存档'); + expect(list.map((m) => m.subject).sort()).toEqual(['好信', '毒信一封']); + + // **游标越过它**:下一轮不会再从这一封重来(否则后面所有新信永远进不来) + const saved = h.accounts.rows.get('acc-1'); + expect(saved?.cursor).toEqual(imapCursor('v1', 2)); + // 顺利收尾也要说实话:第几封、是谁 + expect(saved?.lastPollError).toContain('第 1 封'); + expect(saved?.lastPollError).toContain('uid:v1:1'); + expect(saved?.lastPollError).toContain('已降级存档'); + expect(outcome.error).toBe(saved?.lastPollError); + + // 降级行原件仍在:人点得开、看得到原文 + expect((await h.service.raw('in-1')).bytes.byteLength).toBeGreaterThan(0); + // 它没有任何投递痕 —— 正文是失败原因不是客户的话,不自动投;人可按「重新投递」 + expect((await h.service.getInbound('in-1')).deliveries).toEqual([]); + }); + + it('骨架也落不下:停在这一封,error 点名是谁卡着(不静默跳过,那是丢信)', async () => { + const inbound = memoryInbound(); + const real = inbound.insert.bind(inbound); + let refuse = true; + // 入库这一步整个不行(磁盘满 / 约束冲突 / 列型不合)—— 连最小骨架都写不下去 + inbound.insert = async (record) => (refuse ? Promise.reject(new Error('入库失败:disk full')) : real(record)); + const h = harness({ inbound }); + await h.accounts.save(account({ cursor: imapCursor('v1', 0) })); + h.imap.queue(rawMessage('uid:v1:7', eml({ messageId: '' }))); + + const outcome = await h.service.pollNow('acc-1'); + expect(outcome.error).toContain('uid:v1:7'); + expect(outcome.error).toContain('降级骨架也落不下'); + const saved = h.accounts.rows.get('acc-1'); + expect(saved?.lastPollError).toBe(outcome.error); + // 停在这一封**之前**:数据一封都没丢,人知道是谁卡着,修好了下一轮接着拉 + expect(saved?.cursor).toEqual(imapCursor('v1', 0)); + + refuse = false; + h.imap.queue(rawMessage('uid:v1:7', eml({ messageId: '' }))); + const recovered = await h.service.pollNow('acc-1'); + expect(recovered.stored).toBe(1); + expect(recovered.error).toBeUndefined(); + }); +}); + +describe('provider 说的话要落到界面上', () => { + it('UIDVALIDITY 变化重锚:这一轮不算失败,但 last_poll_error 里有那句说明', async () => { + const h = harness(); + await h.accounts.save(account()); + h.imap.noteWith(REANCHOR_NOTE); + h.imap.queue(rawMessage('uid:v2:1', eml({ messageId: '' }))); + const outcome = await h.service.pollNow('acc-1'); + expect(outcome.stored).toBe(1); // 拉到的照常入库 + expect(outcome.error).toBe(REANCHOR_NOTE); + expect(h.accounts.rows.get('acc-1')?.lastPollError).toContain('UIDVALIDITY'); + expect(h.accounts.rows.get('acc-1')?.lastPollError).toContain('期间来信未拉取'); + // 说明只属于报了它的那一轮:下一轮没话说就清干净,不留一条过期的红字 + h.imap.noteWith(undefined); + expect((await h.service.pollNow('acc-1')).error).toBeUndefined(); + expect(h.accounts.rows.get('acc-1')?.lastPollError).toBeNull(); + }); +}); + +describe('列表与分发不搬原件', () => { + it('list / getMeta 不带 raw 字节,而 rawAvailable 仍然正确;下载原件那条路照旧拿得到', async () => { + /* + * 一封原件上限 16MB、列表上限 200 行:每次 SSE 失效重取都可能搬几百 MB 只为算 200 字预览。 + * 这条断言在 store 那一层 —— 服务出参本来就不带 raw,只有真去数 store 交出来的东西才验得出。 + */ + const h = harness(); + await h.accounts.save(account()); + h.imap.queue(rawMessage('uid:v1:1', eml({ messageId: '' }))); + await h.service.pollNow('acc-1'); + + const listed = await h.inbound.list({ limit: 10 }); + expect(listed[0]?.record.raw).toBeNull(); + expect(listed[0]?.record.rawAvailable).toBe(true); + const meta = await h.inbound.getMeta('in-1'); + expect(meta?.record.raw).toBeNull(); + expect(meta?.record.rawAvailable).toBe(true); + // 详情面的 rawAvailable 读的是投影,不是「字节在不在手上」 + expect((await h.service.getInbound('in-1')).rawAvailable).toBe(true); + // 下载原件仍拿得到字节 + expect((await h.service.raw('in-1')).bytes.byteLength).toBeGreaterThan(0); + }); +}); + describe('分发', () => { it('订阅方收到详情与 ctx(账号 / 通道);skipped 与 failed 各自留痕,互不影响;同名注册被拒;注销后不再收', async () => { const h = harness(); diff --git a/apps/api/src/modules/mail/__tests__/mail-store.it.test.ts b/apps/api/src/modules/mail/__tests__/mail-store.it.test.ts index cbb1a71..4cc9a07 100644 --- a/apps/api/src/modules/mail/__tests__/mail-store.it.test.ts +++ b/apps/api/src/modules/mail/__tests__/mail-store.it.test.ts @@ -2,9 +2,10 @@ * mail 存储集成测试 —— 真 PG:两把唯一键、默认账号唯一、到期挑选、**跨进程租约两连接对抢**、 * 投递 upsert、出站幂等键、bytea 原件按元素拷回来(pg 的 Buffer 挂在共享内存池上)。 * - * 这组用例**共享状态、按顺序跑**:用例 2(游标)依赖用例 1 落下的 acc-a, - * 用例 4/5(listDue / SQL-domain 同源)依赖用例 3 存的那一批账号, - * 用例 7(投递)依赖用例 6(入站)写下的 in-1 —— 单独用 `-t` 挑一条跑会失败,这是设计如此,不是缺陷。 + * 这组用例**共享状态、按顺序跑**:游标那条依赖第一条落下的 acc-a,listDue / SQL-domain 同源 + * 依赖前面存的那一批账号,投递与「不搬原件」两条依赖入站那条写下的 in-1 —— + * 单独用 `-t` 挑一条跑会失败,这是设计如此,不是缺陷。 + * 同理:新增用例要**摆在对列表顺序有精确断言的那一条之后**(否则新行会挤进它的期望里)。 */ import { readFileSync } from 'node:fs'; import { afterAll, beforeAll, describe, expect, it } from 'vitest'; @@ -68,7 +69,7 @@ function inbound(over: Partial): InboundRecord { return { id: 'in-1', accountId: 'acc-a', messageId: '', providerRef: 'uid:7:1', inReplyTo: null, references: [], from: { address: 'b@acme.com', name: '' }, to: [{ address: 'a@we.com', name: '' }], cc: [], subject: 's', text: 't', html: null, - sentAt: null, receivedAt: now, attachments: [], raw: new Uint8Array([1, 2, 3]), rawNote: '', size: 3, createdAt: now, + sentAt: null, receivedAt: now, attachments: [], raw: new Uint8Array([1, 2, 3]), rawAvailable: true, rawNote: '', size: 3, createdAt: now, ...over, }; } @@ -130,8 +131,16 @@ describe.skipIf(database === undefined)('mail store(真 PG)', () => { it('listDue:从没拉过 / 到点的 / 租约过期的算;停用 / 没到点 / 租约在手的不算', async () => { const store = createMailAccountStore(db); - await store.save(account({ id: 'acc-a', name: '甲', lastPolledAt: new Date(now.getTime() - 60_000) })); - await store.save(account({ id: 'acc-b', name: '乙', address: 'b@we.com', lastPolledAt: new Date(now.getTime() - 5_000) })); + /* + * **运行列走 savePoll,不走 save** —— `save()` 只写配置列(见 accountConfigValues): + * 运营面 PUT 的同时另一台正拉着,整行 upsert 会把租约与游标一并盖回去。 + * acc-a / acc-b 在前面的用例里已经存在,所以它们的 lastPolledAt 必须显式 savePoll; + * acc-c / acc-d / acc-e 是新行,运行列跟着 insert 落下即可。 + */ + await store.save(account({ id: 'acc-a', name: '甲' })); + await store.savePoll('acc-a', { polledAt: new Date(now.getTime() - 60_000) }); + await store.save(account({ id: 'acc-b', name: '乙', address: 'b@we.com' })); + await store.savePoll('acc-b', { polledAt: new Date(now.getTime() - 5_000) }); await store.save(account({ id: 'acc-c', name: '丙', address: 'c@we.com', enabled: false })); await store.save(account({ id: 'acc-d', name: '丁', address: 'd@we.com', pollingUntil: new Date(now.getTime() + 60_000) })); // 恰好到点的那一行:钉住区间比较符是 `<=` 而不是 `<`(SQL 与 domain 都该判它到期) @@ -139,6 +148,36 @@ describe.skipIf(database === undefined)('mail store(真 PG)', () => { expect((await store.listDue(now)).map((a) => a.id).sort()).toEqual(['acc-a', 'acc-e']); }); + it('**save 只写配置列**:占着租约的账号被 save 一次,租约与游标一个字不动;要清游标显式单写那一列', async () => { + /* + * 08-24 `loop_state` 整份覆盖那一课的同款指纹:`mergeAccount` 里的运行列是**读那一刻的快照**, + * 运营面 PUT 的同时另一台正拉着 → PUT 把 polling_until 写回旧值 → 第三台抢到租约 → 两台同时拉同一信箱。 + */ + const store = createMailAccountStore(db); + await store.save(account({ id: 'acc-save', name: '存', address: 's@we.com' })); + await store.savePoll('acc-save', { cursor: imapCursor('7', 42), polledAt: now, error: '上一轮连不上', lastInboundAt: now }); + expect(await store.claimPolling('acc-save', 60_000)).toBe(true); + const held = await store.get('acc-save'); + + // 运营面 PUT:只改了显示名,而这一刻另一台正持着租约、游标刚推到 42 + await store.save(account({ id: 'acc-save', name: '存', address: 's@we.com', displayName: '改过', cursor: null, pollingUntil: null })); + const after = await store.get('acc-save'); + expect(after?.displayName).toBe('改过'); + expect(after?.cursor).toEqual(imapCursor('7', 42)); + expect(after?.pollingUntil?.toISOString()).toBe(held?.pollingUntil?.toISOString()); + expect(after?.lastPolledAt?.toISOString()).toBe(now.toISOString()); + expect(after?.lastPollError).toBe('上一轮连不上'); + expect(after?.lastInboundAt?.toISOString()).toBe(now.toISOString()); + // 收侧配置真变了那条路:用例显式单写 cursor 这一列(cursorReset 为真时) + await store.savePoll('acc-save', { cursor: null, error: null }); + const reset = await store.get('acc-save'); + expect(reset?.cursor).toBeNull(); + expect(reset?.lastPollError).toBeNull(); + // 清游标不碰租约 + expect(reset?.pollingUntil?.toISOString()).toBe(held?.pollingUntil?.toISOString()); + await store.releasePolling('acc-save'); + }); + it('**SQL 与 domain 的到期判据同源**:listDue 的返回集合 == 用 isDue 过滤 list() 的结果', async () => { // 同一条「到期」判据两处各存一份实现(SQL 里的 listDue / domain 的 isDue)—— // 本测钉住二者不悄悄分叉:哪天 SQL 把 `<=` 写成 `<`、或漏掉租约那一段,这里立刻炸。 @@ -181,6 +220,7 @@ describe.skipIf(database === undefined)('mail store(真 PG)', () => { expect(got?.record.raw).toEqual(new Uint8Array([1, 2, 3])); expect(got?.record.raw).toBeInstanceOf(Uint8Array); expect(Buffer.isBuffer(got?.record.raw)).toBe(false); + expect(got?.record.rawAvailable).toBe(true); }); it('投递:同一 (mail, subscriber) upsert;list 按收信时间倒序并带投递', async () => { @@ -195,6 +235,28 @@ describe.skipIf(database === undefined)('mail store(真 PG)', () => { expect((await store.list({ limit: 1 })).map((x) => x.record.id)).toEqual(['in-2']); }); + it('**列表与元信息读不搬原件**:raw 不取回来,而 rawAvailable 由 `raw is not null` 投影出来', async () => { + /* + * 一封原件上限 16MB、列表默认 50 行 / 上限 200:每次 SSE 失效重取都可能把几百 MB 搬进内存, + * 只为算一段 200 字预览。判「有没有原件」不必先把它扛回来。 + */ + const store = createInboundStore(db); + await store.insert(inbound({ id: 'in-noraw', messageId: '', providerRef: 'uid:7:9', raw: null, rawAvailable: false, rawNote: '原件超限未存' })); + const listed = await store.list({ accountId: 'acc-a', limit: 10 }); + for (const row of listed) expect(row.record.raw).toBeNull(); + expect(listed.find((r) => r.record.id === 'in-1')?.record.rawAvailable).toBe(true); + expect(listed.find((r) => r.record.id === 'in-noraw')?.record.rawAvailable).toBe(false); + + const meta = await store.getMeta('in-1'); + expect(meta?.record.raw).toBeNull(); + expect(meta?.record.rawAvailable).toBe(true); + expect(meta?.record.subject).toBe('s'); // 元信息一样不少 + expect(await store.getMeta('in-none')).toBeUndefined(); + // 下载原件那条路照旧拿得到字节 + expect((await store.get('in-1'))?.record.raw).toEqual(new Uint8Array([1, 2, 3])); + expect((await store.get('in-noraw'))?.record.rawAvailable).toBe(false); + }); + it('出站:insert / byIdempotencyKey / setState / list', async () => { const store = createOutboundStore(db); await store.insert(outbound({ id: 'out-1', idempotencyKey: 'k-1' })); @@ -209,4 +271,32 @@ describe.skipIf(database === undefined)('mail store(真 PG)', () => { // 同一幂等键第二次 insert 撞唯一索引 —— 用例层先查后写,这里只证明索引在 await expect(store.insert(outbound({ id: 'out-3', idempotencyKey: 'k-1' }))).rejects.toThrow(); }); + + it('**复用旧行要抢**:两条真连接同时抢同一行,只有一个抢得到;sent 的行谁也抢不到', async () => { + /* + * 两个后端连同一个库时,两台会同时读到同一行 `failed` / 超时 `sending` 并都判「可以复用」—— + * 没有这一句条件更新,两台都发,就是两封真邮件(契约白纸黑字的「同键不发两封」当场破掉)。 + */ + const a = createOutboundStore(db); + const b = createOutboundStore(dbB); + const staleBefore = new Date(now.getTime() - 5 * 60_000); + await a.insert(outbound({ id: 'out-race', messageId: '', state: 'failed', error: '上一次没发出去' })); + const wins = await Promise.all([ + a.claimForResend('out-race', { staleBefore, now }), + b.claimForResend('out-race', { staleBefore, now }), + ]); + expect(wins.filter(Boolean)).toHaveLength(1); + const row = (await a.list({ accountId: 'acc-a', limit: 50 })).find((o) => o.id === 'out-race'); + // 抢到即置 sending 并把 created_at 推到 now —— 那一行当场离开可抢的集合 + expect(row?.state).toBe('sending'); + expect(row?.error).toBeNull(); + expect(await b.claimForResend('out-race', { staleBefore, now })).toBe(false); + + // 停够了才算「上次崩在半路」:把 created_at 推到 staleBefore 之前,又抢得到 + expect(await b.claimForResend('out-race', { staleBefore: new Date(now.getTime() + 60_000), now })).toBe(true); + // 已经发出去的那一行谁也抢不到 + await a.setState('out-race', 'sent', { sentAt: now }); + expect(await a.claimForResend('out-race', { staleBefore: new Date(now.getTime() + 60_000), now })).toBe(false); + expect(await a.claimForResend('out-none', { staleBefore, now })).toBe(false); + }); }); diff --git a/apps/api/src/modules/mail/__tests__/wecom-provider.test.ts b/apps/api/src/modules/mail/__tests__/wecom-provider.test.ts index d3d94f1..24b0119 100644 --- a/apps/api/src/modules/mail/__tests__/wecom-provider.test.ts +++ b/apps/api/src/modules/mail/__tests__/wecom-provider.test.ts @@ -1,6 +1,6 @@ /** 企业微信邮件 API 适配器 —— 对假 fetch:token 缓存 / 过期刷新 / -1 重试 / 翻页 / 已知 mail_id 不再 read_mail / EML 编码兜底 / compose_send 载荷 */ import { describe, expect, it } from 'vitest'; -import { createWecomProvider, WecomApiError } from '../internal/adapter/wecom-provider.ts'; +import { createWecomProvider, tokenKeyOf, WecomApiError } from '../internal/adapter/wecom-provider.ts'; import type { FetchLike } from '../internal/adapter/wecom-provider.ts'; import { wecomCursor } from '../internal/domain/cursor.ts'; @@ -45,6 +45,61 @@ describe('token', () => { expect(calls[0]?.url).toContain('corpsecret=cs-1'); }); + it('**缓存按凭证指纹,不按 account.id**:两家公司的草稿探测都用 id:draft,各取各的 token', async () => { + /* + * 未存账号的探测在用例层一律是 `id:'draft'`(mail-service.ts 的 `stored?.id ?? 'draft'`)。 + * 按 id 缓存时,探完 A 公司再探 B 公司会命中 A 的 token —— 回的是 **A 的应用邮箱地址**, + * 而屏上写着 B 已探到;凭证填错的 B 也会显示成功。 + */ + const { fetch, calls } = scripted({ + '/cgi-bin/gettoken': (_b, nth) => ({ errcode: 0, access_token: `tok-${nth}`, expires_in: 7200 }), + 'get_email_alias': () => ({ errcode: 0, email: 'app@corp.wecom.work' }), + }); + const provider = createWecomProvider({ fetch, now: () => 0 }); + const corpA = { ...account, id: 'draft', settings: { corpId: 'wwAAA' }, secret: { corpSecret: 'cs-a' } }; + const corpB = { ...account, id: 'draft', settings: { corpId: 'wwBBB' }, secret: { corpSecret: 'cs-b' } }; + await provider.probe(corpA); + await provider.probe(corpB); + const tokenCalls = calls.filter((c) => c.url.includes('gettoken')); + expect(tokenCalls).toHaveLength(2); + expect(tokenCalls[0]?.url).toContain('corpid=wwAAA'); + expect(tokenCalls[1]?.url).toContain('corpid=wwBBB'); + // 第二家拿的是自己那份 token,不是第一家的 + expect(calls.filter((c) => c.url.includes('get_email_alias')).map((c) => c.url)).toEqual([ + expect.stringContaining('access_token=tok-1'), + expect.stringContaining('access_token=tok-2'), + ]); + // 再探第一家:凭证没变,还是它那一槽,不重取 + await provider.probe(corpA); + expect(calls.filter((c) => c.url.includes('gettoken'))).toHaveLength(2); + }); + + it('**同一账号换了 secret 就换槽**:不再拿旧 token 拉旧 corp 的信箱', async () => { + const { fetch, calls } = scripted({ + '/cgi-bin/gettoken': (_b, nth) => ({ errcode: 0, access_token: `tok-${nth}`, expires_in: 7200 }), + 'get_email_alias': () => ({ errcode: 0, email: 'app@corp.wecom.work' }), + }); + const provider = createWecomProvider({ fetch, now: () => 0 }); + await provider.probe(account); + await provider.probe(account); + expect(calls.filter((c) => c.url.includes('gettoken'))).toHaveLength(1); + // PUT 换了应用 secret,id 一个字没变 —— 旧 token 立刻作废 + await provider.probe({ ...account, secret: { corpSecret: 'cs-2' } }); + const tokenCalls = calls.filter((c) => c.url.includes('gettoken')); + expect(tokenCalls).toHaveLength(2); + expect(tokenCalls[1]?.url).toContain('corpsecret=cs-2'); + }); + + it('指纹里不含明文 secret', () => { + const key = tokenKeyOf('ww1', 'cs-1'); + expect(key).not.toContain('cs-1'); + expect(key).toMatch(/^[0-9a-f]{64}$/); + expect(tokenKeyOf('ww1', 'cs-1')).toBe(tokenKeyOf('ww1', 'cs-1')); + expect(tokenKeyOf('ww1', 'cs-1')).not.toBe(tokenKeyOf('ww2', 'cs-1')); + // 裸拼会让这两组撞在同一个指纹上 + expect(tokenKeyOf('ww1', '2cs')).not.toBe(tokenKeyOf('ww12', 'cs')); + }); + it('-1 系统繁忙重试 ≤3 次;别的错误码抛 WecomApiError 带 api 名', async () => { let busy = 2; const { fetch } = scripted({ diff --git a/apps/api/src/modules/mail/internal/adapter/imap-smtp-provider.ts b/apps/api/src/modules/mail/internal/adapter/imap-smtp-provider.ts index 37610c4..5159780 100644 --- a/apps/api/src/modules/mail/internal/adapter/imap-smtp-provider.ts +++ b/apps/api/src/modules/mail/internal/adapter/imap-smtp-provider.ts @@ -19,6 +19,9 @@ import type { MailProvider, OutboundMessage, ProviderAccount, RawMessage } from const CONNECT_TIMEOUT_MS = 30_000; const SOCKET_TIMEOUT_MS = 60_000; +/** UIDVALIDITY 变化时这一轮带回的说明 —— 落进 `last_poll_error`(测试对着这个常量断言) */ +export const REANCHOR_NOTE = '信箱 UIDVALIDITY 变化(重建 / 迁移),已重锚在当下;期间来信未拉取'; + export function imapSettingsOf(account: ProviderAccount): { settings: ImapSmtpSettings; login: string; password: string } { const settings = imapSmtpSettingsSchema.parse(account.settings); const password = account.secret['password']; @@ -151,7 +154,8 @@ export function createImapSmtpProvider(): MailProvider { const box = mailboxOf(client); if (backfillFrom === null) return imapCursor(box.uidValidity, box.uidNext - 1); const uids = await client.search({ since: backfillFrom }, { uid: true }); - const first = uids === false || uids.length === 0 ? box.uidNext : Math.min(...uids); + // `Math.min(...uids)` 在十万级 UID 的信箱上会**栈溢出**(展开成十万个实参);reduce 不铺参数 + const first = uids === false || uids.length === 0 ? box.uidNext : uids.reduce((a, b) => (b < a ? b : a), uids[0] ?? box.uidNext); return imapCursor(box.uidValidity, first - 1); } finally { lock.release(); @@ -165,8 +169,12 @@ export function createImapSmtpProvider(): MailProvider { try { const box = mailboxOf(client); if (cursor.kind !== 'imap' || cursor.uidValidity !== box.uidValidity) { - // UIDVALIDITY 变了(信箱重建 / 迁移):旧 UID 全作废,重锚在当下、不回填;万一重复由唯一键挡 - return { messages: [], cursor: imapCursor(box.uidValidity, box.uidNext - 1), more: false }; + /* + * UIDVALIDITY 变了(信箱重建 / 迁移):旧 UID 全作废,重锚在当下、不回填;万一重复由唯一键挡。 + * **重锚要出声**:上一次游标到此刻之间的来信从此拉不到了,而账号页若一片绿, + * 人只会以为「这几天没来信」。这句话经用例落进 `last_poll_error`(设计稿 §5.5)。 + */ + return { messages: [], cursor: imapCursor(box.uidValidity, box.uidNext - 1), more: false, note: REANCHOR_NOTE }; } const found = await client.search({ uid: `${cursor.lastUid + 1}:*` }, { uid: true }); const fresh = newUids(found === false ? [] : found, cursor.lastUid); diff --git a/apps/api/src/modules/mail/internal/adapter/mail-store.ts b/apps/api/src/modules/mail/internal/adapter/mail-store.ts index 124e138..d4a434d 100644 --- a/apps/api/src/modules/mail/internal/adapter/mail-store.ts +++ b/apps/api/src/modules/mail/internal/adapter/mail-store.ts @@ -51,9 +51,14 @@ function toAccount(row: AccountRow): MailAccountRecord { }; } -function accountValues(record: MailAccountRecord): typeof mailAccountTable.$inferInsert { +/** + * **配置列** —— 人在运营面填的那些,`save()` 写的就是这一组。 + * 运行列(cursor / last_polled_at / last_poll_error / last_inbound_at / polling_until)不在里面: + * `mergeAccount` 里的运行列是**读那一刻的快照**,整行 upsert 会把另一台正持着的租约与刚推进的游标 + * 一并盖回去 —— 第三台随即抢到租约,两台同时拉同一个信箱(08-24 `loop_state` 整份覆盖那一课的同款指纹)。 + */ +function accountConfigValues(record: MailAccountRecord) { return { - id: record.id, name: record.name, kind: record.kind, address: record.address, @@ -65,13 +70,21 @@ function accountValues(record: MailAccountRecord): typeof mailAccountTable.$infe backfillFrom: record.backfillFrom, settings: record.settings, secretCipher: record.secretCipher, + updatedAt: record.updatedAt, + }; +} + +/** 新行才带得上运行列(那时它们本就是空的);已存在的行只更新配置列 */ +function accountValues(record: MailAccountRecord): typeof mailAccountTable.$inferInsert { + return { + id: record.id, + ...accountConfigValues(record), cursor: record.cursor, lastPolledAt: record.lastPolledAt, lastPollError: record.lastPollError, lastInboundAt: record.lastInboundAt, pollingUntil: record.pollingUntil, createdAt: record.createdAt, - updatedAt: record.updatedAt, }; } @@ -90,7 +103,6 @@ export function createMailAccountStore(db: Db): MailAccountStore { return rows[0] === undefined ? undefined : toAccount(rows[0]); }, async save(record) { - const values = accountValues(record); await db.transaction(async (tx) => { // 默认账号只留一个:先把别的放下,库上的部分唯一索引兜底 if (record.isDefault) { @@ -99,7 +111,11 @@ export function createMailAccountStore(db: Db): MailAccountStore { .set({ isDefault: false }) .where(and(eq(mailAccountTable.isDefault, true), sql`${mailAccountTable.id} <> ${record.id}`)); } - await tx.insert(mailAccountTable).values(values).onConflictDoUpdate({ target: mailAccountTable.id, set: values }); + await tx + .insert(mailAccountTable) + .values(accountValues(record)) + // **只写配置列**:运行列归 savePoll / claimPolling / releasePolling(见 accountConfigValues) + .onConflictDoUpdate({ target: mailAccountTable.id, set: accountConfigValues(record) }); }); }, async remove(id) { @@ -133,6 +149,7 @@ export function createMailAccountStore(db: Db): MailAccountStore { }, async savePoll(id, patch) { const set: Partial = {}; + // `cursor: null` 是显式清游标(改了收侧配置要重锚);`undefined` 才是「这一次不动它」 if (patch.cursor !== undefined) set.cursor = patch.cursor; if (patch.polledAt !== undefined) set.lastPolledAt = patch.polledAt; if (patch.error !== undefined) set.lastPollError = patch.error; @@ -149,6 +166,64 @@ export function createMailAccountStore(db: Db): MailAccountStore { type InboundRow = typeof mailInboundTable.$inferSelect; type DeliveryRow = typeof mailInboundDeliveryTable.$inferSelect; +/** + * **只要元信息的那几列** —— 唯独不选 `raw`。 + * + * 一封原件上限 16MB,列表默认 50 行、上限 200:每次 SSE 失效重取都可能把几百 MB 搬进内存, + * 只为算一段 200 字的预览。`rawAvailable` 让库自己投影(`raw is not null`), + * 不靠把字节取回来再判 —— 判「有没有」不必先把它扛回来。 + */ +const inboundMetaColumns = { + id: mailInboundTable.id, + accountId: mailInboundTable.accountId, + messageId: mailInboundTable.messageId, + providerRef: mailInboundTable.providerRef, + inReplyTo: mailInboundTable.inReplyTo, + referenceIds: mailInboundTable.referenceIds, + fromAddr: mailInboundTable.fromAddr, + fromName: mailInboundTable.fromName, + toList: mailInboundTable.toList, + ccList: mailInboundTable.ccList, + subject: mailInboundTable.subject, + textBody: mailInboundTable.textBody, + htmlBody: mailInboundTable.htmlBody, + sentAt: mailInboundTable.sentAt, + receivedAt: mailInboundTable.receivedAt, + attachments: mailInboundTable.attachments, + rawAvailable: sql`${mailInboundTable.raw} is not null`, + rawNote: mailInboundTable.rawNote, + size: mailInboundTable.size, + createdAt: mailInboundTable.createdAt, +}; + +type InboundMetaRow = { [K in keyof typeof inboundMetaColumns]: K extends 'rawAvailable' ? boolean : InboundRow[K & keyof InboundRow] }; + +function toInboundMeta(row: InboundMetaRow): InboundRecord { + return { + id: row.id, + accountId: row.accountId, + messageId: row.messageId, + providerRef: row.providerRef, + inReplyTo: row.inReplyTo, + references: row.referenceIds, + from: { address: row.fromAddr, name: row.fromName }, + to: row.toList, + cc: row.ccList, + subject: row.subject, + text: row.textBody, + html: row.htmlBody, + sentAt: row.sentAt, + receivedAt: row.receivedAt, + attachments: row.attachments, + // 没取字节:`raw` 恒 null,「有没有原件」问 rawAvailable + raw: null, + rawAvailable: row.rawAvailable, + rawNote: row.rawNote, + size: row.size, + createdAt: row.createdAt, + }; +} + function toInbound(row: InboundRow): InboundRecord { return { id: row.id, @@ -171,6 +246,7 @@ function toInbound(row: InboundRow): InboundRecord { * 三份资料下载下来都是 65536 字节的前科在案(2026-08-23)。按元素拷。 */ raw: row.raw === null ? null : Uint8Array.from(row.raw), + rawAvailable: row.raw !== null, rawNote: row.rawNote, size: row.size, createdAt: row.createdAt, @@ -245,13 +321,20 @@ export function createInboundStore(db: Db): InboundStore { }, async list(query) { const rows = await db - .select() + .select(inboundMetaColumns) .from(mailInboundTable) .where(query.accountId === undefined ? undefined : eq(mailInboundTable.accountId, query.accountId)) .orderBy(desc(mailInboundTable.receivedAt), desc(mailInboundTable.createdAt)) .limit(query.limit); const deliveries = await deliveriesOf(rows.map((r) => r.id)); - return rows.map((row) => ({ record: toInbound(row), deliveries: deliveries.get(row.id) ?? [] })); + return rows.map((row) => ({ record: toInboundMeta(row), deliveries: deliveries.get(row.id) ?? [] })); + }, + async getMeta(mailId) { + const rows = await db.select(inboundMetaColumns).from(mailInboundTable).where(eq(mailInboundTable.id, mailId)).limit(1); + const row = rows[0]; + if (row === undefined) return undefined; + const deliveries = await deliveriesOf([mailId]); + return { record: toInboundMeta(row), deliveries: deliveries.get(mailId) ?? [] }; }, async get(mailId) { const rows = await db.select().from(mailInboundTable).where(eq(mailInboundTable.id, mailId)).limit(1); @@ -337,6 +420,22 @@ export function createOutboundStore(db: Db): OutboundStore { const rows = await db.select().from(mailOutboundTable).where(eq(mailOutboundTable.idempotencyKey, key)).limit(1); return rows[0] === undefined ? undefined : toOutbound(rows[0]); }, + async claimForResend(id, opts) { + /* + * **抢到才发**(设计稿 §5.4 / 契约「同键不发两封」)。条件更新一句话说完三件事: + * ①`failed` 的行可以抢;②`sending` 的行只有停够了 `staleBefore` 才算「上次崩在半路」; + * ③抢到就置 `sending` 并把 `created_at` 推到 now —— 那一行**当场离开可抢的集合**, + * 第二个人的同一句更新影响 0 行,回 false 走「本次未重复发送」那一支。 + * 两个后端连同一个库时,这一句是「两台同时重放同一次放行」与「两封真邮件」之间唯一的墙。 + */ + const result = await db.execute(sql` + update sys.mail_outbound + set state = 'sending', created_at = ${opts.now}, error = null + where id = ${id} + and (state = 'failed' or (state = 'sending' and created_at <= ${opts.staleBefore})) + returning id`); + return rowsOf(result).length > 0; + }, async setState(id, state, patch) { await db .update(mailOutboundTable) diff --git a/apps/api/src/modules/mail/internal/adapter/wecom-provider.ts b/apps/api/src/modules/mail/internal/adapter/wecom-provider.ts index abd6577..304764e 100644 --- a/apps/api/src/modules/mail/internal/adapter/wecom-provider.ts +++ b/apps/api/src/modules/mail/internal/adapter/wecom-provider.ts @@ -5,8 +5,10 @@ * exmail/app/get_mail_list(begin_time, end_time, cursor, limit ≤ 1000) → mail_id(只有应用邮箱自己的收件箱) * exmail/app/read_mail(mail_id) → mail_data(原始 EML;**文档未说编码** —— 像 EML 按原文,否则当 base64) * exmail/app/compose_send → 发件人恒为应用邮箱;**不回 Message-ID**(线程级归属对它发出的信不成立) - * token 缓存在实例闭包里按账号存(不是模块级状态);提前 5 分钟刷新;40014 / 42001 刷一次重试;-1 重试 ≤3 次。 + * token 缓存在实例闭包里按**凭证指纹**存(不是模块级状态,也不是账号 id —— 见 `tokenKeyOf`); + * 提前 5 分钟刷新;40014 / 42001 刷一次重试;-1 重试 ≤3 次。 */ +import { createHash } from 'node:crypto'; import { wecomSettingsSchema } from '@contracts'; import type { MailProbeReport } from '@contracts'; import { wecomCursor, wecomWindow } from '../domain/cursor.ts'; @@ -36,6 +38,27 @@ const TOKEN_EARLY_MS = 5 * 60_000; const BUSY_RETRIES = 3; const REQUEST_TIMEOUT_MS = 30_000; const LIST_PAGE = 100; +/** + * 一轮最多翻几页 —— 服务端若一直回 `has_more:1` 且 next_cursor 不动,这个循环就永远转下去 + * (只有单请求 30s 超时护着,而每一页都是一次成功返回,超时永远不触发)。 + * 到顶就当本轮吃不完:`more = true`、游标照旧不推进,下一轮接着翻,一封都不丢。 + */ +const MAX_LIST_PAGES = 50; + +/** + * token 缓存的键 = **凭证本身的指纹**,不是 `account.id`。 + * + * 按 id 存有两条真实的串号路径: + * ① 草稿探测共用 `id:'draft'`(用例层未存账号时就是这个字面量)—— 探完 A 公司再探 B 公司, + * B 会命中 A 的 token,`get_email_alias` 回的是 **A 的应用邮箱地址**,凭证填错的 B 也显示成功; + * ② 已存账号换了 corpId / corpSecret 而 id 不变 —— 下一轮轮询继续拿旧 token 拉**旧 corp 的收件箱**, + * 存进新配置的账号、分发给它绑的通道,窗口 ≤ 7200s。 + * 凭证一变自然换槽,两条路都封死。**secret 不进 Map 的键**:哈希掉,内存里也不留明文。 + */ +export function tokenKeyOf(corpId: string, corpSecret: string): string { + // JSON 分隔而不是裸拼:拼法有歧义(corpId 末尾多一个字符 = secret 首字少一个字符,指纹却相同) + return createHash('sha256').update(JSON.stringify([corpId, corpSecret])).digest('hex'); +} /** 文档只说「邮件 eml 内容」:头部行的形(`Name: value`)就按原文;否则当 base64 */ export function emlBytesOf(mailData: string): Uint8Array { @@ -61,11 +84,13 @@ export function createWecomProvider(deps: { fetch?: FetchLike; now?: () => numbe } async function tokenOf(account: ProviderAccount, force = false): Promise { - const cached = tokens.get(account.id); - if (!force && cached !== undefined && cached.expiresAt - TOKEN_EARLY_MS > nowMs()) return cached.token; + // 先取凭证再查缓存 —— 键就是凭证的指纹,取不到凭证也就无所谓缓存(见 tokenKeyOf) const settings = wecomSettingsSchema.parse(account.settings); const secret = account.secret['corpSecret']; if (secret === undefined || secret === '') throw new Error('缺应用 secret'); + const key = tokenKeyOf(settings.corpId, secret); + const cached = tokens.get(key); + if (!force && cached !== undefined && cached.expiresAt - TOKEN_EARLY_MS > nowMs()) return cached.token; const body = await call( 'gettoken', `${settings.apiBase}/cgi-bin/gettoken?corpid=${encodeURIComponent(settings.corpId)}&corpsecret=${encodeURIComponent(secret)}`, @@ -74,7 +99,7 @@ export function createWecomProvider(deps: { fetch?: FetchLike; now?: () => numbe const token = typeof body['access_token'] === 'string' ? body['access_token'] : ''; if (token === '') throw new Error('企业微信 gettoken 没有返回 access_token'); const expiresIn = typeof body['expires_in'] === 'number' ? body['expires_in'] : 7200; - tokens.set(account.id, { token, expiresAt: nowMs() + expiresIn * 1000 }); + tokens.set(key, { token, expiresAt: nowMs() + expiresIn * 1000 }); return token; } @@ -131,7 +156,9 @@ export function createWecomProvider(deps: { fetch?: FetchLike; now?: () => numbe const fresh: string[] = []; let pageCursor: string | undefined; let exhausted = false; - while (fresh.length < opts.limit) { + let pages = 0; + while (fresh.length < opts.limit && pages < MAX_LIST_PAGES) { + pages += 1; const page = await withToken(account, 'get_mail_list', '/cgi-bin/exmail/app/get_mail_list', { begin_time: beginTime, end_time: endTime, diff --git a/apps/api/src/modules/mail/internal/domain/account-view.ts b/apps/api/src/modules/mail/internal/domain/account-view.ts index a70b60f..fd99f22 100644 --- a/apps/api/src/modules/mail/internal/domain/account-view.ts +++ b/apps/api/src/modules/mail/internal/domain/account-view.ts @@ -1,5 +1,5 @@ /** 账号记录 ↔ 入参 / 视图(设计稿 §5.3);视图经 schema 过一遍,密文进不来 */ -import type { MailAccountInput, MailAccountView } from '@contracts'; +import type { MailAccountInput, MailAccountKind, MailAccountView } from '@contracts'; import { mailAccountViewSchema } from '@contracts'; import { cursorSummary } from './cursor.ts'; import { isPolling } from './due.ts'; @@ -36,15 +36,27 @@ export function secretOf(input: MailAccountInput): string | undefined { return input.secret === undefined ? undefined : JSON.stringify(input.secret); } -function sameSettings(a: Record, b: Record): boolean { - const keys = new Set([...Object.keys(a), ...Object.keys(b)]); - return [...keys].every((k) => JSON.stringify(a[k] ?? null) === JSON.stringify(b[k] ?? null)); +/** + * **只有收侧的键动了才谈得上重锚** —— 游标说的是「这个信箱 / 这个文件夹我读到哪儿了」。 + * + * 从前比的是全部 settings 键,连 `smtpHost` / `smtpPort` / `smtpTls` 都算: + * 人把 465 改成 587(纯发信的事)→ 游标清空 → IMAP 重锚 `lastUid = uidNext − 1` → + * 上次游标到重锚之间到达的信**永远不拉**(设计即「不回填」)。**改发信的设置不该丢来信**。 + * `login` 两侧都用,算收侧(换了登录名就是换了一个信箱在读)。 + */ +const RECEIVE_KEYS: Readonly> = { + 'imap-smtp': ['imapHost', 'imapPort', 'imapTls', 'login', 'folder'], + wecom: ['corpId', 'apiBase'], +}; + +function sameReceiveSettings(kind: MailAccountKind, a: Record, b: Record): boolean { + return RECEIVE_KEYS[kind].every((k) => JSON.stringify(a[k] ?? null) === JSON.stringify(b[k] ?? null)); } -/** kind 或 settings 变了,旧游标指的就不是同一个信箱 / 同一个文件夹 —— 清空重锚 */ +/** kind 或**收侧** settings 变了,旧游标指的就不是同一个信箱 / 同一个文件夹 —— 清空重锚 */ export function cursorReset(existing: MailAccountRecord | undefined, input: MailAccountInput): boolean { if (existing === undefined) return false; - return existing.kind !== input.kind || !sameSettings(existing.settings, input.settings); + return existing.kind !== input.kind || !sameReceiveSettings(input.kind, existing.settings, input.settings); } export function mergeAccount( diff --git a/apps/api/src/modules/mail/internal/domain/cursor.ts b/apps/api/src/modules/mail/internal/domain/cursor.ts index 1fd7cc6..0e2348a 100644 --- a/apps/api/src/modules/mail/internal/domain/cursor.ts +++ b/apps/api/src/modules/mail/internal/domain/cursor.ts @@ -53,13 +53,30 @@ export function cursorSummary(cursor: MailCursor | null): string | null { return `至 ${new Date(cursor.lastTime * 1000).toISOString()}`; } -/** 从 jsonb 解回来 —— 形状不对当没有游标(下一轮重锚),不抛 */ +/** IMAP 的 UID 是 32 位无符号(RFC 3501)—— 比它大的一定不是我们写下的那个数 */ +const MAX_UID = 0xffff_ffff; +/** + * wecom 游标是 unix 秒,`cursorSummary` 要拿 `new Date(lastTime * 1000)` 渲。 + * JS 的 Date 上限是 ±8.64e15 毫秒 —— 换算成秒就是这个数,超了 `toISOString()` 直接抛 RangeError。 + * **安全整数还不够**:1e13 是安全整数,乘 1000 就出了 Date 的界。 + */ +const MAX_UNIX_SECONDS = 8_640_000_000_000; + +/** + * 从 jsonb 解回来 —— 形状不对当没有游标(下一轮重锚),不抛。 + * + * **取值范围也算形状**:`{kind:'wecom', lastTime: 1e13}` 类型上是个 number,而 + * `new Date(lastTime * 1000).toISOString()` 会抛 `RangeError` —— 而 `cursorSummary` 跑在 `viewOf` 里, + * 一行脏游标能让 `listAccounts` **整页 500**。外部够不着这一列(只有我们自己写游标), + * 但炸起来是整页,而拦住它只要一句:非负安全整数才算数。 + */ export function parseCursor(raw: unknown): MailCursor | null { if (typeof raw !== 'object' || raw === null) return null; const c = raw as Record; - if (c['kind'] === 'imap' && typeof c['uidValidity'] === 'string' && typeof c['lastUid'] === 'number') { + const within = (v: unknown, max: number): v is number => typeof v === 'number' && Number.isSafeInteger(v) && v >= 0 && v <= max; + if (c['kind'] === 'imap' && typeof c['uidValidity'] === 'string' && within(c['lastUid'], MAX_UID)) { return imapCursor(c['uidValidity'], c['lastUid']); } - if (c['kind'] === 'wecom' && typeof c['lastTime'] === 'number') return wecomCursor(c['lastTime']); + if (c['kind'] === 'wecom' && within(c['lastTime'], MAX_UNIX_SECONDS)) return wecomCursor(c['lastTime']); return null; } diff --git a/apps/api/src/modules/mail/internal/domain/inbound-record.ts b/apps/api/src/modules/mail/internal/domain/inbound-record.ts index dc1b9d8..1694d21 100644 --- a/apps/api/src/modules/mail/internal/domain/inbound-record.ts +++ b/apps/api/src/modules/mail/internal/domain/inbound-record.ts @@ -2,7 +2,7 @@ * 解析结果 → 入站记录 / 视图(设计稿 §5.3)。零 IO:附件的存放由 usecase 按 `planAttachments` 去做, * 存完把 ref 交回来再 `buildInboundRecord`。 */ -import type { MailDeliveryView, MailInboundDetail, MailInboundSummary } from '@contracts'; +import type { MailAddress, MailDeliveryView, MailInboundDetail, MailInboundSummary } from '@contracts'; import type { DeliveryRecord, InboundAttachmentRecord, InboundRecord, ParsedAttachment, ParsedMail } from './records.ts'; /** 与 sys.upload 的上限同一个数(超了当场留注,不静默截断) */ @@ -15,15 +15,44 @@ function domainOf(address: string): string { return at < 0 || at === address.length - 1 ? 'local' : address.slice(at + 1); } +/** + * **剥掉 NUL(U+0000)** —— PG 的 `text` / `jsonb` 列一律拒收它(`invalid byte sequence` / `unsupported Unicode escape`), + * 而 MIME 解出二进制垃圾时它真会出现(声明成 text/plain 的附件、编码标错的正文)。 + * + * 落在这一层是因为:整封信卡住的病根在**一个字符**,而 domain 是纯函数、好测; + * 不剥的话这一封每 30 秒炸一次,游标停在它前面,**后面所有新信永远进不来**。 + * 只剥这一个码点,别的字节原样留着 —— 原件 `raw` 是 bytea,一个字节都不动。 + */ +export function stripNul(text: string): string { + return text.includes('\u0000') ? text.replaceAll('\u0000', '') : text; +} + +function stripNulAddress(address: MailAddress): MailAddress { + return { address: stripNul(address.address), name: stripNul(address.name) }; +} + /** 缺 Message-ID 的信(有的自动回复就没有)用 providerRef 顶上 —— 唯一键要的是一个稳定的串 */ export function fallbackMessageId(providerRef: string, address: string): string { return `<${providerRef}@${domainOf(address)}>`; } -/** 线程 = In-Reply-To(父)⊃ References 末项;归属解析①级拿它对出站的 Message-ID */ -export function threadIdOf(parsed: Pick): string | undefined { - if (parsed.inReplyTo !== undefined && parsed.inReplyTo !== '') return parsed.inReplyTo; - return parsed.references.at(-1); +/** + * 线程 = **这一串的根**(`References[0]`),不是最近一跳(ADR-013 决策 3)。 + * + * 我方出站登记的 threadId 恒是首封(`channelTransport` 回 `input.threadId ?? messageId`, + * channel 据此 `recordOutbound`),而归属解析①级是精确等值比对。取最近一跳的话跑一个来回就漂: + * 我方 M1 → 客户回 M1(命中)→ 我方再回(实际 M2、仍记在 M1 名下)→ 客户回 M2 → + * 取到 M2 而库里从没记过 M2 ⇒ ①级落空,退到③地址 + 时间窗。 + * `references[0]` 才稳定:客户端续链时首项恒是首封;客户端只给 In-Reply-To 时退回它。 + * + * 入参放宽到 `string | null`:入站详情(`MailInboundDetail`)与解析结果(`ParsedMail`) + * 对「没有」的写法一个是 null 一个是 undefined,同一套口径要能同时接住两边 —— + * 这个函数就是那**唯一一处口径**(app 的 mail-wiring 调它,不再内联一份)。 + */ +export function threadIdOf(parsed: { readonly inReplyTo?: string | null | undefined; readonly references: readonly string[] }): string | undefined { + const root = parsed.references[0]; + if (root !== undefined && root !== '') return root; + return parsed.inReplyTo === null || parsed.inReplyTo === '' ? undefined : parsed.inReplyTo; } const ENTITIES: Readonly> = { amp: '&', lt: '<', gt: '>', quot: '"', apos: "'", nbsp: ' ' }; @@ -78,6 +107,22 @@ export function planAttachments(attachments: readonly ParsedAttachment[], cap: n }); } +/** + * 解析炸了之后从**原始字节**里扒一行头 —— 「主题能取多少取多少」那一手(降级存档用)。 + * + * 不做 MIME 解码(encoded-word 原样留着):这一封本来就是解不动的那一封, + * 这里要的只是让人在列表上认出「是哪一封卡住了」,不是还原它。 + * 头名只认 ASCII 字母与连字符 —— 别的一律当没有,不把外来串塞进正则。 + */ +export function peekHeader(eml: Uint8Array, name: string): string { + if (!/^[A-Za-z][A-Za-z-]*$/.test(name)) return ''; + const head = new TextDecoder('utf-8', { fatal: false }).decode(eml.slice(0, 16 * 1024)); + const blank = head.search(/\r?\n\r?\n/); + const headers = blank < 0 ? head : head.slice(0, blank); + const hit = new RegExp(`^${name}:[ \\t]*(.*)$`, 'im').exec(headers); + return stripNul((hit?.[1] ?? '').trim()); +} + export function rawOf(eml: Uint8Array, cap: number = RAW_CAP): { raw: Uint8Array | null; note: string } { if (eml.byteLength <= cap) return { raw: eml, note: '' }; return { raw: null, note: `原件 ${(eml.byteLength / 1024 / 1024).toFixed(1)}MB,超过 ${Math.round(cap / 1024 / 1024)}MB 上限,未存` }; @@ -98,23 +143,25 @@ export function buildInboundRecord(input: { const { parsed } = input; const messageId = parsed.messageId !== undefined && parsed.messageId !== '' ? parsed.messageId : fallbackMessageId(input.providerRef, input.address); const { raw, note } = rawOf(input.eml, input.rawCap ?? RAW_CAP); + // 凡是要落 text / jsonb 的字段一律先剥 NUL(见 stripNul);raw 是 bytea,原样存 return { id: input.id, accountId: input.accountId, - messageId, - providerRef: input.providerRef, - inReplyTo: parsed.inReplyTo ?? null, - references: [...parsed.references], - from: parsed.from ?? { address: '', name: '' }, - to: [...parsed.to], - cc: [...parsed.cc], - subject: parsed.subject, - text: parsed.text ?? null, - html: parsed.html ?? null, + messageId: stripNul(messageId), + providerRef: stripNul(input.providerRef), + inReplyTo: parsed.inReplyTo === undefined ? null : stripNul(parsed.inReplyTo), + references: parsed.references.map(stripNul), + from: stripNulAddress(parsed.from ?? { address: '', name: '' }), + to: parsed.to.map(stripNulAddress), + cc: parsed.cc.map(stripNulAddress), + subject: stripNul(parsed.subject), + text: parsed.text === undefined ? null : stripNul(parsed.text), + html: parsed.html === undefined ? null : stripNul(parsed.html), sentAt: parsed.date ?? null, receivedAt: input.receivedAt, - attachments: [...input.attachments], + attachments: input.attachments.map((a) => ({ ...a, name: stripNul(a.name), contentType: stripNul(a.contentType), note: stripNul(a.note) })), raw, + rawAvailable: raw !== null, rawNote: note, size: input.eml.byteLength, createdAt: input.now, @@ -157,6 +204,7 @@ export function detailOf(record: InboundRecord, deliveries: readonly DeliveryRec text: record.text, html: record.html, size: record.size, - rawAvailable: record.raw !== null, + // 问投影出来的那一格,不问字节在不在手上 —— 列表 / 元信息读压根不取字节 + rawAvailable: record.rawAvailable, }; } diff --git a/apps/api/src/modules/mail/internal/domain/records.ts b/apps/api/src/modules/mail/internal/domain/records.ts index 208bde8..c523504 100644 --- a/apps/api/src/modules/mail/internal/domain/records.ts +++ b/apps/api/src/modules/mail/internal/domain/records.ts @@ -51,7 +51,14 @@ export type InboundRecord = { readonly sentAt: Date | null; readonly receivedAt: Date; readonly attachments: readonly InboundAttachmentRecord[]; + /** + * 原件字节 —— **只有「下载原件」那一条读取回它**(`InboundStore.get`); + * 列表与只要元信息的读一律不取(一封上限 16MB),那时这里恒是 null。 + * 「有没有原件」问 `rawAvailable`,别拿 `raw !== null` 去问 —— 那两件事只在全列读时才等价。 + */ readonly raw: Uint8Array | null; + /** 库里那一行有没有原件(`raw is not null` 投影出来的,不靠把字节搬回来再判) */ + readonly rawAvailable: boolean; readonly rawNote: string; readonly size: number; readonly createdAt: Date; diff --git a/apps/api/src/modules/mail/internal/usecase/mail-service.ts b/apps/api/src/modules/mail/internal/usecase/mail-service.ts index 90db722..27c0400 100644 --- a/apps/api/src/modules/mail/internal/usecase/mail-service.ts +++ b/apps/api/src/modules/mail/internal/usecase/mail-service.ts @@ -7,7 +7,7 @@ */ import type { MailAccountKind, MailPollOutcome, MailProbeReport, MailSendResult } from '@contracts'; import type { InboundHandler, MailService } from '../../contract.ts'; -import { mergeAccount, secretOf, viewOf } from '../domain/account-view.ts'; +import { cursorReset, mergeAccount, secretOf, viewOf } from '../domain/account-view.ts'; import { canRedeliver, stateOfOutcome } from '../domain/delivery.ts'; import { leaseMsFor } from '../domain/due.ts'; import { @@ -22,7 +22,7 @@ import { MailSubscriberTakenError, MailSubscriberUnknownError, } from '../domain/errors.ts'; -import { ATTACHMENT_CAP, buildInboundRecord, deliveryViewOf, detailOf, planAttachments, RAW_CAP, summaryOf } from '../domain/inbound-record.ts'; +import { ATTACHMENT_CAP, buildInboundRecord, deliveryViewOf, detailOf, peekHeader, planAttachments, RAW_CAP, summaryOf } from '../domain/inbound-record.ts'; import type { DeliveryRecord, InboundAttachmentRecord, InboundRecord, MailAccountRecord, OutboundRecord, ParsedAttachment } from '../domain/records.ts'; import { errorText, redactSecrets } from '../domain/redact.ts'; import { messageIdFor, pickAccount, validateSend } from '../domain/send-input.ts'; @@ -173,7 +173,8 @@ export function createMailService(deps: MailServiceDeps): MailService { only?: string, opts: { readonly reclaimStale?: boolean } = {}, ): Promise { - const existing = (await deps.inbound.get(record.id))?.deliveries ?? []; + // 只要那几条投递痕,不必把 16MB 的原件搬过来 + const existing = (await deps.inbound.getMeta(record.id))?.deliveries ?? []; const out: DeliveryRecord[] = []; for (const [name, handler] of subscribers) { if (only !== undefined && name !== only) continue; @@ -242,10 +243,56 @@ export function createMailService(deps: MailServiceDeps): MailService { return inserted ? built : undefined; } + /** + * 单封解析失败的**退路**:先试着落一行最小骨架(照迁移 0033 的 NOT NULL 填),落成了就推进游标。 + * + * 没有这条路时,一封「毒信」把整个账号永久卡死:`storeOne` 抛错 → 外层 catch 接住 → + * 游标停在这一封**之前** → 下一 tick 从同一个 UID 重拉、同样抛,**后面所有新信永远进不来**, + * 而运营面只有一行 `last_poll_error`、没有任何钮能跳过它。 + * 骨架行不是「跳过」(那是丢信):原件 `raw` 原样存着,主题能扒多少扒多少, + * 正文写明这是降级行与失败原因,人在列表上看得见、点得开、可按「重新投递」。 + */ + async function storeDegraded( + record: MailAccountRecord, + message: RawMessage, + cause: unknown, + ): Promise<{ record: InboundRecord; inserted: boolean } | undefined> { + try { + const subject = peekHeader(message.eml, 'Subject'); + const built = buildInboundRecord({ + id: deps.ids.next(), + accountId: record.id, + address: record.address, + providerRef: message.providerRef, + parsed: { + references: [], + to: [], + cc: [], + subject: subject === '' ? '(主题解析失败)' : subject, + text: `【降级存档】这一封的 MIME 解析或入库失败,原文见「下载原件」。失败原因:${errorText(cause)}`, + attachments: [], + }, + eml: message.eml, + receivedAt: message.receivedAt, + attachments: [], + now: deps.clock.now(), + rawCap: limits.rawBytes, + }); + // 唯一键撞上(这一封先前已经降级存过)也算「落下了」:那一行就在库里,照常越过去,只是不重复计数 + const { inserted } = await deps.inbound.insert(built); + return { record: built, inserted }; + } catch { + // 骨架都落不下 —— 交给调用方停在这一封并点名是谁 + return undefined; + } + } + async function pollAccount(record: MailAccountRecord): Promise { const claimed = await deps.accounts.claimPolling(record.id, leaseMsFor(record.pollIntervalSec)); if (!claimed) throw new MailPollBusyError(record.id); const tally: PollTally = { accountId: record.id, pulled: 0, stored: 0, dispatched: 0, more: false }; + /** 这一轮的「有话说」:降级存档了哪几封、provider 报的重锚说明 —— 顺利收尾也要落进 last_poll_error */ + const notes: string[] = []; let secret: Record = {}; try { secret = openSecret(record); @@ -259,6 +306,7 @@ export function createMailService(deps: MailServiceDeps): MailService { } let budget = limits.perTick; let more = true; + let seq = 0; while (more && budget > 0 && deps.clock.now().getTime() - started.getTime() < limits.tickBudgetMs) { const batch = await provider.pull(account, cursor, { limit: budget, @@ -266,10 +314,30 @@ export function createMailService(deps: MailServiceDeps): MailService { backfillFrom: record.backfillFrom, known: (refs) => deps.inbound.knownProviderRefs(record.id, refs), }); + // provider 有话说(如 UIDVALIDITY 变化已重锚)—— 这一轮不算失败,但界面上必须看得见 + if (batch.note !== undefined && batch.note !== '') notes.push(batch.note); tally.pulled += batch.messages.length; budget -= batch.messages.length; for (const message of batch.messages) { - const stored = await storeOne(record, message); + seq += 1; + let stored: InboundRecord | undefined; + let degraded = false; + try { + stored = await storeOne(record, message); + } catch (error) { + /* + * **一封解析不动的信不许卡死整条路**:先落降级骨架,落成了就照常推进游标。 + * 骨架也落不下才停在这一封 —— 那时 `last_poll_error` 必须点名是哪一封(UID / providerRef), + * 否则人只看得见一句「解析失败」,不知道是谁卡着、更不知道能不能跳过。 + */ + const fallback = await storeDegraded(record, message, error); + if (fallback === undefined) { + throw new Error(`第 ${seq} 封(${message.providerRef})存不下,降级骨架也落不下,这一轮停在它这里:${errorText(error)}`); + } + degraded = true; + stored = fallback.inserted ? fallback.record : undefined; + notes.push(`第 ${seq} 封(${message.providerRef})解析失败,已降级存档:${errorText(error)}`); + } // 逐封推进:中途栽了,下一轮从这一封之后接着 if (message.cursorAfter !== undefined) { cursor = message.cursorAfter; @@ -278,8 +346,14 @@ export function createMailService(deps: MailServiceDeps): MailService { if (stored !== undefined) { tally.stored += 1; await deps.accounts.savePoll(record.id, { lastInboundAt: stored.receivedAt }); - await dispatch(stored, { accountId: record.id, channelId: record.channelId }); - tally.dispatched += 1; + /* + * 降级行**不自动分发**:它的正文是失败原因,不是客户说的话 —— 投给通道等于让下游 + * 按一段垃圾正文起单。它照旧躺在入站列表里(带原件),人看过之后可以按「重新投递」。 + */ + if (!degraded) { + await dispatch(stored, { accountId: record.id, channelId: record.channelId }); + tally.dispatched += 1; + } } } cursor = batch.cursor; @@ -288,7 +362,10 @@ export function createMailService(deps: MailServiceDeps): MailService { if (batch.messages.length === 0) break; } tally.more = more; - await deps.accounts.savePoll(record.id, { polledAt: deps.clock.now(), error: null }); + // 这一轮没崩,但有话要说的照说(降级存档 / 重锚)—— 一片绿不是实情 + const note = notes.length === 0 ? null : notes.join(';'); + if (note !== null) tally.error = note; + await deps.accounts.savePoll(record.id, { polledAt: deps.clock.now(), error: note }); } catch (error) { const text = redactWith(secret)(errorText(error)); tally.error = text; @@ -335,9 +412,14 @@ export function createMailService(deps: MailServiceDeps): MailService { const secretCipher = secretJson !== undefined ? deps.secrets.seal(secretJson) : input.kind === existing.kind ? existing.secretCipher : null; const now = deps.clock.now(); - // kind / settings 变了游标清空重锚 —— 判定在 mergeAccount 里(cursorReset) + // kind / **收侧** settings 变了游标清空重锚 —— 判定在 mergeAccount 里(cursorReset) const record = mergeAccount(input, { id: accountId, existing, secretCipher, now }); await deps.accounts.save(record); + /* + * `save()` 只写配置列(不然会盖掉另一台正持着的租约与刚推进的游标),所以清游标要**显式单写**这一列。 + * 上一轮的错误话跟着清:它说的是旧配置下的事,留着会让人对着新配置读旧毛病。 + */ + if (cursorReset(existing, input)) await deps.accounts.savePoll(accountId, { cursor: null, error: null }); return viewOf(record, now); }, @@ -403,13 +485,14 @@ export function createMailService(deps: MailServiceDeps): MailService { }, async getInbound(mailId) { - const hit = await deps.inbound.get(mailId); + const hit = await deps.inbound.getMeta(mailId); if (hit === undefined) throw new MailInboundNotFoundError(mailId); return detailOf(hit.record, hit.deliveries); }, async attachment(mailId, index) { - const hit = await deps.inbound.get(mailId); + // 附件在 sys.upload 里,这里只要那一行的 meta —— 原件字节不必跟着出来 + const hit = await deps.inbound.getMeta(mailId); if (hit === undefined) throw new MailInboundNotFoundError(mailId); const meta = hit.record.attachments.find((a) => a.index === index); if (meta === undefined || meta.ref === null) throw new MailAttachmentNotFoundError(mailId, index); @@ -419,6 +502,7 @@ export function createMailService(deps: MailServiceDeps): MailService { }, async raw(mailId) { + // 全仓唯一要 raw 字节的一条路 const hit = await deps.inbound.get(mailId); if (hit === undefined) throw new MailInboundNotFoundError(mailId); if (hit.record.raw === null) throw new MailAttachmentNotFoundError(mailId, -1); @@ -426,7 +510,7 @@ export function createMailService(deps: MailServiceDeps): MailService { }, async redeliver(mailId, subscriber) { - const hit = await deps.inbound.get(mailId); + const hit = await deps.inbound.getMeta(mailId); if (hit === undefined) throw new MailInboundNotFoundError(mailId); if (subscriber !== undefined && !subscribers.has(subscriber)) throw new MailSubscriberUnknownError(subscriber); const owner = await deps.accounts.get(hit.record.accountId); @@ -479,6 +563,23 @@ export function createMailService(deps: MailServiceDeps): MailService { const now = deps.clock.now(); const id = prior?.id ?? deps.ids.next(); const messageId = prior?.messageId ?? messageIdFor(id, account.address); + if (prior !== undefined) { + /* + * **复用旧行之前先抢**:上面那一段只判断了「这一行看着可以复用」,而两个后端连同一个库时 + * 两台会同时判出同一个结论 —— 都发,就是两封真邮件。抢是一条条件更新(见 `claimForResend`), + * 抢不到说明另一路已经接管,这一次如实回「没发出去」。 + */ + const claimed = await deps.outbound.claimForResend(prior.id, { staleBefore: new Date(now.getTime() - SEND_STALE_MS), now }); + if (!claimed) { + return { + outboundId: prior.id, + accountId: prior.accountId, + messageId: prior.messageId, + state: 'failed', + error: '同一幂等键的上一封正被另一路重发,本次未重复发送', + }; + } + } if (prior === undefined) { const record: OutboundRecord = { id, diff --git a/apps/api/src/modules/mail/port.ts b/apps/api/src/modules/mail/port.ts index 5ab007d..5104dfe 100644 --- a/apps/api/src/modules/mail/port.ts +++ b/apps/api/src/modules/mail/port.ts @@ -23,7 +23,10 @@ export type MailAccountStore = { list(): Promise; get(id: string): Promise; findByName(name: string): Promise; - /** upsert;`isDefault:true` 时先把别的置 false(同一事务) */ + /** + * upsert **只写配置列**;`isDefault:true` 时先把别的置 false(同一事务)。 + * 运行列(cursor / lastPolledAt / lastPollError / lastInboundAt / pollingUntil)一概不动 —— 见 `savePoll`。 + */ save(record: MailAccountRecord): Promise; remove(id: string): Promise; /** enabled 且(从没拉过 或 last_polled_at + interval <= now)且租约空 / 过期 */ @@ -31,11 +34,18 @@ export type MailAccountStore = { /** 一条 SQL 判 + 占(设计稿 §9.1);抢到 true */ claimPolling(id: string, leaseMs: number): Promise; releasePolling(id: string): Promise; - /** 拉取过程的进度:游标 / 时刻 / 错误 / 最近来信 / 探到的地址;行不存在 = no-op */ + /** + * 拉取过程的进度:游标 / 时刻 / 错误 / 最近来信 / 探到的地址;行不存在 = no-op。 + * + * **运行列只走这条路**(与 `claimPolling` / `releasePolling` 一起)——`save()` 一个字都不碰它们: + * 运营面 PUT 的同时另一台正拉着,整行 upsert 会把租约与游标一并覆盖回读那一刻的快照 + * (`sys.loop_state` 整份覆盖那一课的同款指纹)。 + * `cursor: null` 是**显式清游标**(改了收侧配置要重锚时用)。 + */ savePoll( id: string, patch: { - cursor?: MailCursor; + cursor?: MailCursor | null; polledAt?: Date; error?: string | null; lastInboundAt?: Date; @@ -48,7 +58,14 @@ export type InboundStore = { /** (account_id, message_id) / (account_id, provider_ref) 冲突 → inserted:false,不抛 */ insert(record: InboundRecord): Promise<{ inserted: boolean }>; knownProviderRefs(accountId: string, refs: readonly string[]): Promise>; + /** + * 列表 —— **不取 `raw` 原件字节**(一封上限 16MB,50 行就是几百 MB,只为算 200 字预览); + * `rawAvailable` 由 `raw is not null` 投影出来,不靠把字节搬回来再判。 + */ list(query: { accountId?: string | undefined; limit: number }): Promise>; + /** 只要元信息的读(分发 / 详情 / 附件 / 重投都走它)—— 同样不取 `raw` 字节 */ + getMeta(mailId: string): Promise<{ record: InboundRecord; deliveries: DeliveryRecord[] } | undefined>; + /** 全列读 —— 只有「下载原件」那一条路要它 */ get(mailId: string): Promise<{ record: InboundRecord; deliveries: DeliveryRecord[] } | undefined>; /** upsert on (mail_id, subscriber) */ setDelivery(delivery: DeliveryRecord): Promise; @@ -57,6 +74,15 @@ export type InboundStore = { export type OutboundStore = { insert(record: OutboundRecord): Promise; byIdempotencyKey(key: string): Promise; + /** + * 复用旧行重发之前**先抢**(一条条件更新,抢到才回 true): + * `failed` 的行、或 `created_at <= staleBefore` 的 `sending` 行才抢得到;抢到即置 `sending` 并把 + * `created_at` 推到 `now` —— 这一手让那一行**离开可抢的集合**,第二个人当场抢不到。 + * + * 本仓是两个后端连同一个库:没有这一抢,两台同时重放同一次放行都会判「超时可复用」, + * 发出**两封真邮件**,直接破掉契约白纸黑字的「同键不发两封」。 + */ + claimForResend(id: string, opts: { readonly staleBefore: Date; readonly now: Date }): Promise; setState(id: string, state: 'sent' | 'failed', patch: { error?: string; providerRef?: string; sentAt?: Date }): Promise; list(query: { accountId?: string | undefined; limit: number }): Promise; }; @@ -91,6 +117,11 @@ export type MailProvider = { probe(account: ProviderAccount): Promise; /** 无游标时锚定初始游标(不回填;设了 backfillFrom 才从那天起) */ anchor(account: ProviderAccount, backfillFrom: Date | null, now: Date): Promise; + /** + * 拉一批。回参的 `note` 是这一轮**不算失败、但人必须知道**的事 + * (当前唯一一种:IMAP 的 UIDVALIDITY 变了、已重锚在当下、期间来信不回填)—— + * 用例把它落进 `last_poll_error`。丢了一段信而界面一片绿,那是界面撒谎。 + */ pull( account: ProviderAccount, cursor: MailCursor, @@ -101,7 +132,7 @@ export type MailProvider = { /** 库里已有的 provider_ref —— wecom 用它在 read_mail 之前把重叠窗里的旧信滤掉;IMAP 靠 UID 游标不用 */ readonly known: (refs: readonly string[]) => Promise>; }, - ): Promise<{ messages: RawMessage[]; cursor: MailCursor; more: boolean }>; + ): Promise<{ messages: RawMessage[]; cursor: MailCursor; more: boolean; note?: string | undefined }>; send(account: ProviderAccount, message: OutboundMessage): Promise<{ providerRef?: string | undefined }>; }; diff --git a/apps/api/src/modules/toolbox/__tests__/tool-contract.test.ts b/apps/api/src/modules/toolbox/__tests__/tool-contract.test.ts index 7d6b995..ed6feaf 100644 --- a/apps/api/src/modules/toolbox/__tests__/tool-contract.test.ts +++ b/apps/api/src/modules/toolbox/__tests__/tool-contract.test.ts @@ -138,20 +138,3 @@ describe('staticAudit —— 注册两审之静态审', () => { expect(auditPassed(findings)).toBe(true); }); }); - -describe('mail.send 2.0.0 的演进', () => { - it('保留 v1 的 body / queued、不新增必填 —— 演进检查零违规', () => { - const v1 = { - name: 'mail.send', displayName: '发送邮件', version: '1.0.0', description: '发出邮件(报价单 / 询价函 / 催办)', - input: { type: 'object', required: ['to', 'subject'], properties: { to: { type: 'string' }, subject: { type: 'string' }, body: { type: 'string' } } }, - output: { type: 'object', properties: { queued: { type: 'boolean' } } }, - effect: 'write' as const, errors: [], idempotent: true, timeoutMs: 10_000, retries: 2, connectionKeys: ['endpoint'], - }; - const v2 = { - ...v1, version: '2.0.0', connectionKeys: ['account'], - input: { type: 'object', required: ['to', 'subject'], properties: { ...v1.input.properties, to: { type: 'array', items: { type: 'string' } }, html: { type: 'string' }, dryRun: { type: 'boolean' } } }, - output: { type: 'object', properties: { ...v1.output.properties, outboundId: { type: 'string' }, state: { type: 'string' } } }, - }; - expect(checkEvolution(v1, v2)).toEqual([]); - }); -}); diff --git a/docs/API文档-邮件收发.md b/docs/API文档-邮件收发.md index 35638d5..4bac558 100644 --- a/docs/API文档-邮件收发.md +++ b/docs/API文档-邮件收发.md @@ -201,6 +201,14 @@ body `{ "subscriber": "channel" }`(可省 = 全部订阅方;body 也可整个不 放行落地会带 `idempotencyKey`(同一次放行重放不发两封),重试分三种局面:上一封已 `sent` → 直接短路回那次的结果,不再发; 上一封 `failed`,或停在 `sending` **超过 5 分钟**(算作上次进程崩在半路)→ 复用那一行重发;停在 `sending` **不足 5 分钟** (算作正在飞)→ **不发**,如实回 `state:'failed'` 加一句「同一幂等键的上一封仍在发送中,本次未重复发送」,过一会再放行会重试。 + 复用那一行之前还要**抢**(一条条件更新):两个后端连同一个库时两台会同时判出「可以复用」,抢不到的那一路回 + 「同一幂等键的上一封正被另一路重发,本次未重复发送」,一封真信只出一次。 +- **发出去的信不登记 `outbound_ref`**(实情,2026-08-25 终审核实):`event.recordOutbound` 的唯一调用方是 + `channel.sendMail`,而 `channel.sendMail` 在 src 里**零调用方** —— 放行落地走的是 `deps.tools.execute` → `mail.send` 工具这条路, + 它不经 channel,也就不写 `sys.outbound_ref`。所以:**活动经 `mail.send` 发出的信,回信归属只靠②字段提取与触发器** + (以及③地址 + 时间窗);①级(`outbound_ref` 对表)目前只对 channel 路成立,而 channel 路暂无调用方。 + 下面 §5 里 `sys.mail_outbound.message_id`「线程级归属的对表来源」那句话,对生产唯一的发信路径**尚不成立**。 + 根治要 toolbox 把 `taskId` 透进 `ToolHandler` 签名(跨模块契约变更,单独一件事),已记 `docs/debt.md`。 ## 5. Schema 附录(与 `packages/contracts/src/mail.ts` 同名) diff --git a/docs/adr/ADR-013-mail-gateway.md b/docs/adr/ADR-013-mail-gateway.md index 56f1211..7be2928 100644 --- a/docs/adr/ADR-013-mail-gateway.md +++ b/docs/adr/ADR-013-mail-gateway.md @@ -32,8 +32,10 @@ (`channelTransport.send` 回 `input.threadId ?? messageId`,channel 据此 `recordOutbound`), 而归属解析①级是精确等值比对。若入站取 `In-Reply-To`(最近一跳),跑一个来回就漂: 我方 M1 → 客户回 M1(①命中)→ 我方再回(实际 M2、仍记在 M1 名下)→ 客户回 M2 → - 库里从没记过 M2 ⇒ ①级落空,退到③地址+时间窗。故 `mail-wiring.ts` 的 `toInboundMail` 取 - `references[0] ?? inReplyTo`;客户端只给 `In-Reply-To` 时退回它,全新来信两者皆无则不带线程。 + 库里从没记过 M2 ⇒ ①级落空,退到③地址+时间窗。故取 `references[0] ?? inReplyTo`; + 客户端只给 `In-Reply-To` 时退回它,全新来信两者皆无则不带线程。 + **这条口径全仓只有一处**:mail 模块 domain 的 `threadIdOf`,`mail-wiring.ts` 的 `toInboundMail` 调它 + ——2026-08-25 终审逮到过两套相反的实现(导出的那个取最近一跳、app 里内联的取根),已并成一份。 4. **分发自动只投一次**:每个订阅方一条留痕 `dispatching → done | skipped | failed(原话脱敏)`,不自动重试 —— channel → event 按 `cause.ref` 不幂等,重投会起两单;「重新投递」是人按的钮,那一刻起是至少一次。 订阅是代码级装配(不落库、不回放老信)。 @@ -51,6 +53,16 @@ 执行,幂等键随调用一起带来);`connectionKeys: ['account']` 说「从哪个信箱发」。 **幂等保护只在这一条路上**:`channel.sendMail`(`channelTransport.send`)与运营面 `POST …/send-test` 都不传 `idempotencyKey`,不受这三支逻辑保护 —— 各自的调用方式本就不重放。 + **复用旧行之前还要抢**(2026-08-25 终审补):判「可以复用」与「真去发」之间隔着 await, + 两个后端连同一个库时两台会同时判出同一个结论 —— 都发就是两封真邮件。故 `OutboundStore.claimForResend` + 是一条条件更新(`failed`、或 `created_at <= now − 5min` 的 `sending` 才抢得到;抢到即置 `sending` + 并把 `created_at` 推到 now,那一行当场离开可抢的集合),**抢到才发**,抢不到回 + `{state:'failed', error:'同一幂等键的上一封正被另一路重发,本次未重复发送'}`。 + **`outbound_ref` 的实情**:`event.recordOutbound` 的唯一调用方是 `channel.sendMail`,而 + `channel.sendMail` 在 src 里**零调用方** —— 放行落地走 `deps.tools.execute` → `mail.send` 工具,不经 channel。 + 于是**活动经 `mail.send` 发出的信不登记 `outbound_ref`**:回信归属只靠②字段提取与触发器(以及③地址 + 时间窗), + ①级(`outbound_ref` 对表)目前只对 channel 路成立,而 channel 路暂无调用方。 + 根治要 toolbox 把 `taskId` 透进 `ToolHandler` 签名(跨模块契约变更,单独一件事),已记 debt。 7. **凭证**:`secret_cipher` 密封(AES-256-GCM,`JCAI_SECRET_KEY`);视图只回 `secretBound`;更新时 secret 缺省 = 不改;明文只在用例打开、交给 provider 那一段;所有落库 / 上接口的厂商原话先脱敏(明文 / base64 / URL 编码三形)。 8. **界面不撒谎**:探测收 / 发分栏各报;wecom 的发信一栏 `checked:false`(拿到 token 不等于能发, @@ -71,10 +83,14 @@ - 收信最快延迟 = `pollIntervalSec`(下限 10 秒);没有 IDLE 推送(记 debt)。 - wecom 发出的信拿不到 Message-ID,线程级归属(①级)对它不成立;③级地址窗仍可用。 + 还有一半要说清:wecom 的 `compose_send` 载荷里**没有 `inReplyTo` / `references` 的位置** + (适配器发出去的就是 to/cc/bcc/subject/content/attachment_list)—— 所以 **wecom 账号发出的信在对方的邮件客户端里 + 不与来信串成一个线程**;imap-smtp 那条路走 nodemailer,两个头都带得上,线程正常。 - 附件走 `sys.upload` 的 8MB 上限,更大的只留注;原件 .eml 超 16MB 不存。 - `mail.send` 工具契约升 2.0.0:`to` 由 string 变 array(演进检查不管类型;处理器仍容忍逗号串), 唯一消费方 quote-* 活动已卸载,无在途调用;`createSimulatedMailTransport` 与 `sys.outbound_mail` 暂留待退役。 - 前端接入前:账号只能经接口配;配好账号 + 通道 + 触发器,来信才会起单 —— `channel` 配置仍是只读种子(无 upsert 路由),真要按业务收信得先把通道种上(记 debt)。 -- 部署:迁移 0033 先于启动;`reset-runs` 清三张留痕表、留账号。 +- 部署:迁移 0033 先于启动;`reset-runs` 清三张留痕表、**留账号但清它的游标与拉取状态** + (留着游标而信没了,wecom 那十分钟重叠窗会把最近的信当新信再投一次)。 - 欠账见 `docs/debt.md`「邮件收发」段。 diff --git a/docs/debt.md b/docs/debt.md index 3a6144c..d20efcd 100644 --- a/docs/debt.md +++ b/docs/debt.md @@ -1131,9 +1131,12 @@ pnpm 硬链到全局 store,真实占盘远小于这个数,但 `du` 看着吓人 `{ ...(state==='sent'?{error:null}:{}), ...(patch.error===undefined?{}:{error:patch.error}) }` 里 `patch.error` 排在后面, `setState(id,'sent',{error:'x'})` 在类型上合法且会让它赢,静默破掉「sent 的行没有错误话」这条 store 侧不变量。 当下没有调用点传得到这个组合,只是没人踩雷;把 `patch.error` 那条挪到 `state==='sent'` 判断前面即可让它成为真不变量。 -⑨ **`parseCursor` 只查类型不查取值范围**(`cursor.ts`):`{kind:'wecom', lastTime: 1e13}` 这类脏游标能让 `cursorSummary` - 的 `new Date(lastTime*1000).toISOString()` 抛 `RangeError`,而它渲的正是账号视图 `status.cursor` 那一格 —— - 一行脏游标能让 `listAccounts` 整个 500。补一句 `Number.isSafeInteger` 校验即可封死。 +⑨ ~~**`parseCursor` 只查类型不查取值范围**(`cursor.ts`)~~(2026-08-25 终审修复轮销账): + `{kind:'wecom', lastTime: 1e13}` 这类脏游标能让 `cursorSummary` 的 `new Date(lastTime*1000).toISOString()` + 抛 `RangeError`,而它渲的正是账号视图 `status.cursor` 那一格 —— 一行脏游标能让 `listAccounts` 整个 500。 + 已修:两形的数值都要**非负安全整数且在界内**(imap 的 `lastUid` ≤ 0xFFFFFFFF,RFC 3501 的 32 位; + wecom 的 `lastTime` ≤ 8.64e12 秒 = JS Date 的上限)—— 只写「安全整数」还不够,`1e13` 正是安全整数, + 乘 1000 才出界。不合格的当没有游标(照原注释「形状不对当没有游标,不抛」)。`cursor.test.ts` 钉住。 ⑩ **Message-ID 撞车(providerRef 是新的)留孤儿附件**:`storeOne` 先按 `provider_ref` 查重、再解析存附件、最后才 `insert`; 若真正撞车的是 `(account_id, message_id)` 那把键(provider_ref 没撞),附件已经落进 `sys.upload` 而 `insert` 报 `inserted:false`, 没人再引用这批附件。根治要给 `InboundStore` 加一个按 `messageId` 查重的口,插入前先挡。 @@ -1148,3 +1151,20 @@ pnpm 硬链到全局 store,真实占盘远小于这个数,但 `du` 看着吓人 与已经修过的「缺 `uploadRef` 当场拒」是同一族问题的更窄一种,还没堵上。 ⑮ **`JCAI_MAIL_POLL_TICK_MS` 给垃圾值会得 `NaN`**:`Number(env[...] ?? 5000)` 不校验,`setInterval(fn, NaN)` 按最小间隔跑; 与既有的 `dispatchIntervalMs` 是同一款约定(未做输入校验),但这一条每拍要真打一次库,代价更大。 + +### 2026-08-25 终审修复轮之后新增 / 变更的几笔 + +⑯ **工具路发出的信不登记 `outbound_ref`**:`event.recordOutbound` 的唯一调用方是 `channel.sendMail`, + 而 `channel.sendMail` 在 src 里**零调用方** —— 放行落地走 `deps.tools.execute` → `mail.send` 工具,不经 channel。 + 于是生产唯一的发信路径**不写** `sys.outbound_ref`:回信归属只剩②字段提取与③地址窗,①级(线程号对表)落空。 + 根治要 toolbox 把 `taskId` 透进 `ToolHandler` 签名(处理器现在只拿得到 input / connection / idempotencyKey), + 再由 `mailSendTool` 在发成功后调 `event.recordOutbound` —— 那是跨模块契约变更,单独一件事。 + 本轮只把文档改老实(ADR-013 决策 6、`docs/API文档-邮件收发.md` §4)。 +⑰ **`imapNeedsReanchor` 导出而无人调用**:UIDVALIDITY 变没变的判断在 `imap-smtp-provider.pull` 里内联了一份 + (`cursor.kind !== 'imap' || cursor.uidValidity !== box.uidValidity`)。两处同义、只有一处在跑 —— + 与本轮修掉的 `threadIdOf` 是同一族(导出的那个没人用,内联的那个才是实情),下一轮并成一处。 +⑱ **降级存档行不自动分发,而「重新投递」会把降级正文投给通道**:解析不动的那一封落的是骨架行 + (正文是失败原因,不是客户说的话),自动路径跳过它是对的;但人按「重新投递」时基座不拦 —— + 下游会按一段「【降级存档】…」的正文起单。要么在重投前提醒,要么给这类行一个标(库里现在只能靠正文认)。 +⑲ **`SEND_STALE_MS` 与 `claimForResend` 的 5 分钟是两处各写一份**:用例算 `staleBefore = now − SEND_STALE_MS` 传进去, + store 只按传进来的时刻判 —— 目前只有一个调用方所以对得上,但这个数没有一处共同的定义点。 diff --git a/docs/superpowers/specs/2026-08-25-mail-design.md b/docs/superpowers/specs/2026-08-25-mail-design.md index cd74042..ad348df 100644 --- a/docs/superpowers/specs/2026-08-25-mail-design.md +++ b/docs/superpowers/specs/2026-08-25-mail-design.md @@ -1,7 +1,7 @@ # 邮件收发模块(`mail`)—— 后端设计 - 日期:2026-08-25 -- 状态:设计已通过(用户 2026-08-25 裁决),待实施(实施计划另出) +- 状态:设计已通过(用户 2026-08-25 裁决),**已实施** —— 落地以 `docs/adr/ADR-013-mail-gateway.md` 与代码为准 - 范围:**只做后端**(apps/api + packages/contracts + db + 文档);web 一个文件不动, 前端按本轮产出的《前端对接说明-邮件收发》另做 - 依据(官方文档,核实日期 2026-08-25): @@ -17,6 +17,14 @@ - 全局错误码 / 访问频率限制: / - 老版企业邮箱开放接口(`api.exmail.qq.com`): +> **这份稿子里有三处已被 ADR-013 取代,别照它接线**(2026-08-25 终审补记): +> ① **§7.2 的读路由 `access`** —— 稿子把 7 条 GET 标成 `design|ops`,落地时 15 条**一律 `ops`**(ADR-013 决策 10: +> 邮箱那一屏没有设计端消费方,入站正文是客户来往内容); +> ② **§5.3 的 `threadId`** —— 稿子取最近一跳,落地取**这一串的根** `references[0] ?? inReplyTo`(ADR-013 决策 3); +> ③ **§7.3 的「`body` 改名 `text`」与 `connectionKeys: []`** —— 都没有那么做:契约演进只增不破坏, +> 2.0.0 **保留** v1 的 `body`(纯文本正文,与 `html` 二选一或并存),连接键是 **`['account']`** +> (「从哪个信箱发」:`'default'` 或某个账号 id),不是空数组(ADR-013 决策 6)。 + ## 1. 背景与目标 现状:`channel` 模块(L2)已有整条「入站归一 → 事件(归属解析 / 触发器起单 / 任务 `listen` 监听 /