邮件收发终审修复轮:token 按凭证指纹、毒信有退路、只改发信设置不丢信、重发先抢、列表不搬原件、save 不盖运行列、重锚出声、脏游标不炸页、线程口径归一

全分支终审(跨任务视角)逮出的十条 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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TM4kUBvekALu9qqmp7D3wr
This commit is contained in:
xcs
2026-08-26 03:47:48 +08:00
co-authored by Claude Fable 5
parent 18ca6dacfe
commit 816389a185
30 changed files with 1105 additions and 114 deletions
@@ -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);
});
@@ -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: '<q1@we.com>' };
expect(toInboundMail(onlyHop).threadId).toBe(threadIdOf(onlyHop));
expect(toInboundMail(onlyHop).threadId).toBe('<q1@we.com>');
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');
});
});
+5 -7
View File
@@ -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 }),
+12 -1
View File
@@ -5,7 +5,7 @@
*
* **清的是「跑出来的」**:任务、工作流、卡、任务记忆、事件、监听、台账行、活动记忆…
* **留的是「配置好的」**:员工与岗位授权、会话、工作组、流程包、活动库、工具、
* 触发器、通道、模型实例、纪要。
* 触发器、通道、模型实例、纪要、**邮箱账号**(它身上的游标与拉取状态是运行痕,跟着清)。
*
* 直接走 SQL 而不绕模块:模块出口里没有、也不该有「删掉所有运行痕迹」这种动作 ——
* 留痕只增不改是系统的规矩,清库是**开发期的外科手术**,不是系统的一项能力。
@@ -89,6 +89,17 @@ export async function resetRuns(): Promise<void> {
);
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'`,
);
+1 -1
View File
@@ -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: '发送邮件',
+24 -1
View File
@@ -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 侧装配未变
@@ -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<string, unknown>): 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 }),
@@ -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();
});
});
@@ -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 只在租约未过期时为真', () => {
+52 -2
View File
@@ -96,7 +96,27 @@ export function memoryAccounts(clock: Clock): MailAccountStore & { rows: Map<str
if (record.isDefault) {
for (const [k, v] of rows) if (k !== record.id && v.isDefault) rows.set(k, { ...v, isDefault: false });
}
rows.set(record.id, record);
const existing = rows.get(record.id);
/*
* 与真 store 同一条不变量:**upsert 只写配置列**。运行列(游标 / 上次拉取 / 上次错误 /
* 最近来信 / 租约)一概保留库里当下那一份 —— 整行覆盖会把另一台正持着的租约与刚推进的游标
* 盖回读那一刻的快照。替身与真身在这一条上分叉,测试就会连错的点(本项目已踩过两次)。
* 要在测试里摆运行列:用 `savePoll` / `claimPolling`,或直接动 `rows`。
*/
rows.set(
record.id,
existing === undefined
? record
: {
...record,
cursor: existing.cursor,
lastPolledAt: existing.lastPolledAt,
lastPollError: existing.lastPollError,
lastInboundAt: existing.lastInboundAt,
pollingUntil: existing.pollingUntil,
createdAt: existing.createdAt,
},
);
},
async remove(id) {
rows.delete(id);
@@ -133,6 +153,15 @@ export function memoryAccounts(clock: Clock): MailAccountStore & { rows: Map<str
};
}
/**
* 与真 store 同款:列表与只要元信息的读**不带 `raw` 原件字节**(只有 `get()` 带)。
* 「有没有原件」一律问 `rawAvailable` —— 替身若照旧把字节递出来,
* 「列表不搬原件」这条就没有任何测试拦得住了。
*/
function withoutRaw(record: InboundRecord): InboundRecord {
return { ...record, raw: null };
}
export function memoryInbound(): InboundStore & { rows: Map<string, InboundRecord>; deliveries: Map<string, DeliveryRecord> } {
const rows = new Map<string, InboundRecord>();
const deliveries = new Map<string, DeliveryRecord>();
@@ -158,6 +187,7 @@ export function memoryInbound(): InboundStore & { rows: Map<string, InboundRecor
},
async list(query) {
return [...rows.values()]
.map(withoutRaw)
.filter((r) => query.accountId === undefined || r.accountId === query.accountId)
// 与真 store 同序:两级键 desc(receivedAt), desc(createdAt) —— 并列时后创建的排前。
// 少了第二级,并列会按 Map 插入序解(= createdAt 升序),与真身**方向相反**,
@@ -166,6 +196,10 @@ export function memoryInbound(): InboundStore & { rows: Map<string, InboundRecor
.slice(0, query.limit)
.map((record) => ({ 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<string, OutboundRe
async byIdempotencyKey(key) {
return [...rows.values()].find((r) => 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;
@@ -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('<q1@we.com>');
expect(threadIdOf({ references: ['<a>', '<b>'] })).toBe('<b>');
it('**线程 = 这一串的根**(References[0]);只给 In-Reply-To 时才退回它;都没有就是 undefined', () => {
/*
* 2026-08-25 终审改口径:原先取的是最近一跳(inReplyTo ?? references.at(-1)),
* 与 ADR-013 决策 3 和 app 侧接线正相反 —— 跑一个来回归属①级就漂掉。
* 断言随语义更新(测试没删,期望值改对):同一封信 parsed 现在给的是根 <root@we.com>。
*/
expect(threadIdOf(parsed)).toBe('<root@we.com>');
expect(threadIdOf({ references: ['<a>', '<b>'] })).toBe('<a>');
expect(threadIdOf({ references: [], inReplyTo: '<only-hop>' })).toBe('<only-hop>');
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('<uid:7:12@we.com>');
@@ -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: `<m${NUL}1@acme.com>`,
inReplyTo: `<q${NUL}1@we.com>`,
references: [`<root${NUL}@we.com>`],
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: `<p>${NUL}二</p>`,
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('<p>二</p>');
expect(dirty.messageId).toBe('<m1@acme.com>');
expect(dirty.providerRef).toBe('uid:7:14');
expect(dirty.inReplyTo).toBe('<q1@we.com>');
expect(dirty.references).toEqual(['<root@we.com>']);
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('<uid:7:15@we.com>');
});
});
describe('buildInboundRecord / 视图', () => {
const record = buildInboundRecord({
id: 'in-1',
@@ -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());
@@ -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: '<bad@a>', subject: '毒信一封' })),
rawMessage('uid:v1:2', eml({ messageId: '<ok@a>', 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: '<x@a>' })));
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: '<x@a>' })));
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: '<m1@a>' })));
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: '<m1@a>' })));
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();
@@ -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>): InboundRecord {
return {
id: 'in-1', accountId: 'acc-a', messageId: '<m1@x>', 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: '<m-noraw@x>', 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: '<race@we.com>', 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);
});
});
@@ -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({
@@ -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);
@@ -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<typeof mailAccountTable.$inferInsert> = {};
// `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<boolean>`${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)
@@ -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<string> {
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,
@@ -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<string, unknown>, b: Record<string, unknown>): 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<Record<MailAccountKind, readonly string[]>> = {
'imap-smtp': ['imapHost', 'imapPort', 'imapTls', 'login', 'folder'],
wecom: ['corpId', 'apiBase'],
};
function sameReceiveSettings(kind: MailAccountKind, a: Record<string, unknown>, b: Record<string, unknown>): 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(
@@ -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<string, unknown>;
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;
}
@@ -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<ParsedMail, 'inReplyTo' | 'references'>): 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<Record<string, string>> = { 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,
};
}
@@ -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;
@@ -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<DeliveryRecord[]> {
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<MailPollOutcome> {
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<string, string> = {};
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,
+35 -4
View File
@@ -23,7 +23,10 @@ export type MailAccountStore = {
list(): Promise<MailAccountRecord[]>;
get(id: string): Promise<MailAccountRecord | undefined>;
findByName(name: string): Promise<MailAccountRecord | undefined>;
/** upsert;`isDefault:true` 时先把别的置 false(同一事务) */
/**
* upsert **只写配置列**;`isDefault:true` 时先把别的置 false(同一事务)。
* 运行列(cursor / lastPolledAt / lastPollError / lastInboundAt / pollingUntil)一概不动 —— 见 `savePoll`。
*/
save(record: MailAccountRecord): Promise<void>;
remove(id: string): Promise<void>;
/** enabled 且(从没拉过 或 last_polled_at + interval <= now)且租约空 / 过期 */
@@ -31,11 +34,18 @@ export type MailAccountStore = {
/** 一条 SQL 判 + 占(设计稿 §9.1);抢到 true */
claimPolling(id: string, leaseMs: number): Promise<boolean>;
releasePolling(id: string): Promise<void>;
/** 拉取过程的进度:游标 / 时刻 / 错误 / 最近来信 / 探到的地址;行不存在 = 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<Set<string>>;
/**
* 列表 —— **不取 `raw` 原件字节**(一封上限 16MB,50 行就是几百 MB,只为算 200 字预览);
* `rawAvailable` 由 `raw is not null` 投影出来,不靠把字节搬回来再判。
*/
list(query: { accountId?: string | undefined; limit: number }): Promise<Array<{ record: InboundRecord; deliveries: DeliveryRecord[] }>>;
/** 只要元信息的读(分发 / 详情 / 附件 / 重投都走它)—— 同样不取 `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<void>;
@@ -57,6 +74,15 @@ export type InboundStore = {
export type OutboundStore = {
insert(record: OutboundRecord): Promise<void>;
byIdempotencyKey(key: string): Promise<OutboundRecord | undefined>;
/**
* 复用旧行重发之前**先抢**(一条条件更新,抢到才回 true):
* `failed` 的行、或 `created_at <= staleBefore` 的 `sending` 行才抢得到;抢到即置 `sending` 并把
* `created_at` 推到 `now` —— 这一手让那一行**离开可抢的集合**,第二个人当场抢不到。
*
* 本仓是两个后端连同一个库:没有这一抢,两台同时重放同一次放行都会判「超时可复用」,
* 发出**两封真邮件**,直接破掉契约白纸黑字的「同键不发两封」。
*/
claimForResend(id: string, opts: { readonly staleBefore: Date; readonly now: Date }): Promise<boolean>;
setState(id: string, state: 'sent' | 'failed', patch: { error?: string; providerRef?: string; sentAt?: Date }): Promise<void>;
list(query: { accountId?: string | undefined; limit: number }): Promise<OutboundRecord[]>;
};
@@ -91,6 +117,11 @@ export type MailProvider = {
probe(account: ProviderAccount): Promise<MailProbeReport>;
/** 无游标时锚定初始游标(不回填;设了 backfillFrom 才从那天起) */
anchor(account: ProviderAccount, backfillFrom: Date | null, now: Date): Promise<MailCursor>;
/**
* 拉一批。回参的 `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<Set<string>>;
},
): 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 }>;
};
@@ -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([]);
});
});