diff --git a/apps/api/src/modules/mail/__tests__/fakes.ts b/apps/api/src/modules/mail/__tests__/fakes.ts index 5a1063f..56d2a5f 100644 --- a/apps/api/src/modules/mail/__tests__/fakes.ts +++ b/apps/api/src/modules/mail/__tests__/fakes.ts @@ -195,6 +195,8 @@ export function memoryOutbound(): OutboundStore & { rows: Map void changes.push(r) }, + }); + return { service, accounts, inbound, outbound, imap, wecom, clock, changes }; +} + +const sendInput = mailSendInputSchema.parse({ to: ['buyer@acme.com'], subject: '报价', text: '见附', requestedBy: { kind: 'ops' } }); + +function outboundRow(over: Partial = {}): OutboundRecord { + return { + id: 'out-9', + accountId: 'acc-1', + messageId: '', + idempotencyKey: 'k-1', + to: ['buyer@acme.com'], + cc: [], + bcc: [], + subject: '报价', + text: '见附', + html: null, + inReplyTo: null, + references: [], + attachments: [], + state: 'sending', + error: null, + providerRef: null, + requestedBy: { kind: 'ops' }, + createdAt: new Date('2026-08-25T10:00:00Z'), + sentAt: null, + ...over, + }; +} + +describe('凭证解不开', () => { + it('密文解得开但不是 JSON:错误话里没有明文,只说「未配置凭证」', async () => { + const { service, accounts, imap } = harness(); + // 解得开(plainSecrets 会回出 `pw-1234`),但那不是 JSON —— JSON.parse 的 SyntaxError 把原文嵌在 message 里 + await accounts.save(account({ secretCipher: 'sealed:pw-1234' })); + const outcome = await service.pollNow('acc-1'); + expect(outcome.error).not.toContain('pw-1234'); + expect(outcome.error).toContain('未配置凭证'); + expect(accounts.rows.get('acc-1')?.lastPollError).not.toContain('pw-1234'); + expect(accounts.rows.get('acc-1')?.lastPollError).toContain('未配置凭证'); + // 凭证都没打开,一次都不该去够厂商 + expect(imap.pullCalls).toEqual([]); + }); +}); + +describe('幂等键', () => { + it('同键的上一封还在飞:不发第二封,如实回没发出去', async () => { + const { service, accounts, outbound, imap, clock } = harness(); + await accounts.save(account()); + await outbound.insert(outboundRow({ createdAt: clock.now() })); + const result = await service.send({ ...sendInput, idempotencyKey: 'k-1' }); + expect(result.state).toBe('failed'); + expect(result.error).toContain('仍在发送中'); + expect(imap.sent).toHaveLength(0); + expect(outbound.rows.size).toBe(1); + expect(outbound.rows.get('out-9')?.state).toBe('sending'); + }); + + 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 result = await service.send({ ...sendInput, idempotencyKey: 'k-1' }); + expect(result.state).toBe('sent'); + expect(imap.sent).toHaveLength(1); + expect(outbound.rows.get('out-9')?.state).toBe('sent'); + // 复用,不是新开一行 + expect(outbound.rows.size).toBe(1); + expect(result.outboundId).toBe('out-9'); + }); + + it('失败重发成功之后,那一行不留上一次的错误话', async () => { + const { service, accounts, outbound, imap } = harness(); + await accounts.save(account()); + imap.failSendWith(new Error('暂时不行')); + const first = await service.send({ ...sendInput, idempotencyKey: 'k-1' }); + expect(first.state).toBe('failed'); + expect(outbound.rows.get(first.outboundId)?.error).toContain('暂时不行'); + + imap.failSendWith(undefined); + const second = await service.send({ ...sendInput, idempotencyKey: 'k-1' }); + expect(second.state).toBe('sent'); + expect(outbound.rows.get(first.outboundId)?.error).toBeNull(); + expect((await service.listOutbound({}))[0]?.error).toBeNull(); + + // 第三次同键走 resultOf(prior):既说成功又带错误的结果会让调用方按 error 分支判反 + const third = await service.send({ ...sendInput, idempotencyKey: 'k-1' }); + expect(third.state).toBe('sent'); + expect(third.error).toBeUndefined(); + expect(imap.sent).toHaveLength(1); + }); +}); + +describe('卡死的 dispatching', () => { + const parsed: ParsedMail = { + messageId: '', + references: [], + from: { address: 'buyer@acme.com', name: '买家' }, + to: [{ address: 'sales@we.com', name: '销售' }], + cc: [], + subject: '询价', + text: '请报价', + attachments: [], + }; + + it('自动路径不进(墙),人按的重新投递能接管(路)', async () => { + const { service, accounts, inbound, clock } = harness(); + await accounts.save(account()); + await inbound.insert( + buildInboundRecord({ + id: 'in-1', + accountId: 'acc-1', + address: 'sales@we.com', + providerRef: 'uid-1', + parsed, + eml: new Uint8Array(), + receivedAt: clock.now(), + attachments: [], + now: clock.now(), + }), + ); + let calls = 0; + service.subscribe('channel', async () => { + calls += 1; + }); + + // 进程崩在写 dispatching 与写结果之间:这条留痕永远停在 dispatching + const stuck = { mailId: 'in-1', subscriber: 'channel', state: 'dispatching' as const, error: null, finishedAt: null }; + await inbound.setDelivery({ ...stuck, startedAt: new Date(clock.now().getTime() - 11 * 60_000) }); + const reclaimed = await service.redeliver('in-1'); + expect(reclaimed[0]?.state).toBe('done'); + expect(calls).toBe(1); + + // 新鲜的那条仍是墙:人按也不抢,免得两条线同时改一件事 + await inbound.setDelivery({ ...stuck, startedAt: clock.now() }); + const skipped = await service.redeliver('in-1'); + expect(skipped[0]?.state).toBe('dispatching'); + expect(calls).toBe(1); + }); +}); 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 f7a3765..124e138 100644 --- a/apps/api/src/modules/mail/internal/adapter/mail-store.ts +++ b/apps/api/src/modules/mail/internal/adapter/mail-store.ts @@ -342,6 +342,12 @@ export function createOutboundStore(db: Db): OutboundStore { .update(mailOutboundTable) .set({ state, + /* + * **发成功的那一行不留上一次的错误话**:同一行可能先 failed、再被幂等键复用重发成功。 + * 不清的话,`listOutbound` 会摆出「state: sent 而 error: 上次那句」, + * `resultOf` 也会回出一个既说成功又带错误的结果 —— 调用方按 error 分支就判反了。 + */ + ...(state === 'sent' ? { error: null } : {}), ...(patch.error === undefined ? {} : { error: patch.error }), ...(patch.providerRef === undefined ? {} : { providerRef: patch.providerRef }), ...(patch.sentAt === undefined ? {} : { sentAt: patch.sentAt }), 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 9ffe893..90db722 100644 --- a/apps/api/src/modules/mail/internal/usecase/mail-service.ts +++ b/apps/api/src/modules/mail/internal/usecase/mail-service.ts @@ -65,6 +65,20 @@ export type MailServiceDeps = { const DEFAULT_LIST_LIMIT = 50; const MAX_LIST_LIMIT = 200; +/** + * 同键的上一封停在 `sending` 多久之后,算作「上次崩在半路」而不是「正在飞」。 + * 这两种局面在库里长得一模一样,只能靠时限分开:短于此值一律不发(宁可让调用方再试一次, + * 也不能发出第二封真邮件);长于此值就当那一行没人管了,复用它重发。 + */ +const SEND_STALE_MS = 5 * 60_000; + +/** + * 一条投递停在 `dispatching` 多久之后,允许**人手**的「重新投递」把它接管过去。 + * 自动路径永远不重投(那是防双投的墙);但进程若崩在写 dispatching 与写结果之间, + * 那条留痕就永远卡着 —— 墙要配路:人按的那一下得有门进。 + */ +const DISPATCH_STALE_MS = 10 * 60_000; + type PollTally = { accountId: string; pulled: number; stored: number; dispatched: number; more: boolean; error?: string }; function tallyToOutcome(t: PollTally): MailPollOutcome { @@ -112,7 +126,19 @@ export function createMailService(deps: MailServiceDeps): MailService { /** 明文只在这里打开、交给 provider 那一段;不进任何留痕 */ function openSecret(record: Pick): Record { if (record.secretCipher === null || record.secretCipher === '') throw new MailSecretMissingError(record.id); - const parsed: unknown = JSON.parse(deps.secrets.open(record.secretCipher)); + /* + * 解密与解析都包进来:`JSON.parse` 抛的 SyntaxError **把被解析的原文嵌在 message 里** + * (V8:`Unexpected token 'p', "pw-1234" is not valid JSON`),而这一抛发生在调用方把 `secret` + * 赋值**之前** —— 脱敏那一层此刻手里是空的、一个字都洗不掉,明文会经 `savePoll({error})` + * 落进 `last_poll_error`,再由账号视图上到运营台。 + * 解不开就是解不开:一律换成「未配置凭证」,原文一个字不外传。 + */ + let parsed: unknown; + try { + parsed = JSON.parse(deps.secrets.open(record.secretCipher)); + } catch { + throw new MailSecretMissingError(record.id); + } if (typeof parsed !== 'object' || parsed === null) throw new MailSecretMissingError(record.id); return Object.fromEntries(Object.entries(parsed as Record).map(([k, v]) => [k, String(v)])); } @@ -141,14 +167,23 @@ export function createMailService(deps: MailServiceDeps): MailService { /* ── 分发 ────────────────────────────────────────────────────────── */ - async function dispatch(record: InboundRecord, ctx: { accountId: string; channelId: string | null }, only?: string): Promise { + async function dispatch( + record: InboundRecord, + ctx: { accountId: string; channelId: string | null }, + only?: string, + opts: { readonly reclaimStale?: boolean } = {}, + ): Promise { const existing = (await deps.inbound.get(record.id))?.deliveries ?? []; const out: DeliveryRecord[] = []; for (const [name, handler] of subscribers) { if (only !== undefined && name !== only) continue; const current = existing.find((d) => d.subscriber === name); // 正在投的那条不重投:进程还没回话,再投一遍就是两条线同时改一件事 - if (!canRedeliver(current?.state)) { + const stale = + opts.reclaimStale === true && + current?.state === 'dispatching' && + deps.clock.now().getTime() - current.startedAt.getTime() >= DISPATCH_STALE_MS; + if (!canRedeliver(current?.state) && !stale) { if (current !== undefined) out.push(current); continue; } @@ -395,16 +430,35 @@ export function createMailService(deps: MailServiceDeps): MailService { 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); - const deliveries = await dispatch(hit.record, { accountId: hit.record.accountId, channelId: owner?.channelId ?? null }, subscriber); + const deliveries = await dispatch(hit.record, { accountId: hit.record.accountId, channelId: owner?.channelId ?? null }, subscriber, { + reclaimStale: true, + }); deps.notify.changed('inbound'); return deliveries.map(deliveryViewOf); }, async send(input) { const account = pickAccount(await deps.accounts.list(), { accountId: input.accountId, from: input.from }); - // 幂等键:放行重放不发两封;上一次失败 / 中断的那一行复用,重发 + // 幂等键:放行重放不发两封 const prior = input.idempotencyKey === undefined ? undefined : await deps.outbound.byIdempotencyKey(input.idempotencyKey); if (prior !== undefined && prior.state === 'sent') return resultOf(prior); + /* + * **同键不发两封**。`sending` 有两种局面而库里长得一样:正在飞 / 上次崩在半路 —— 用时限分开。 + * 没超时的:不发,如实回「没发出去」并说清缘由(调用方再试一次即可;第一封落地后同键就短路成 sent 了)。 + * 超时的、以及 `failed` 的:复用那一行重发。 + */ + if (prior !== undefined && prior.state === 'sending') { + const ageMs = deps.clock.now().getTime() - prior.createdAt.getTime(); + if (ageMs < SEND_STALE_MS) { + return { + outboundId: prior.id, + accountId: prior.accountId, + messageId: prior.messageId, + state: 'failed', + error: `同一幂等键的上一封仍在发送中(${Math.round(ageMs / 1000)}s),本次未重复发送`, + }; + } + } const files: ParsedAttachment[] = []; for (const a of input.attachments) {