mail 用例测试:拉取 / 去重 / 逐封游标 / 失败停位 / 分发三态 / 重新投递 / 附件原件

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TM4kUBvekALu9qqmp7D3wr

apps/api/src/modules/mail/__tests__/mail-service-poll.test.ts

@@ -0,0 +1,203 @@
+/**
+ * 拉取与分发:先持久后分发、两把唯一键去重、逐封推进游标、中途失败停在最后一封成功处、
+ * 订阅方 skipped / failed 互不影响、自动只投一次、重新投递、租约被占 409、凭证空记错误。
+ */
+import { describe, expect, it } from 'vitest';
+import type { MailInboundDetail } from '@contracts';
+import { MailAccountDisabledError, MailPollBusyError, MailSubscriberTakenError, MailSubscriberUnknownError } from '../contract.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 {
+  account,
+  eml,
+  fakeProvider,
+  fixedClock,
+  memoryAccounts,
+  memoryAttachments,
+  memoryInbound,
+  memoryOutbound,
+  plainSecrets,
+  rawMessage,
+  seqIds,
+} from './fakes.ts';
+
+function harness(over: { perTick?: number } = {}) {
+  const clock = fixedClock();
+  const accounts = memoryAccounts(clock);
+  const inbound = memoryInbound();
+  const attachments = memoryAttachments();
+  const imap = fakeProvider('imap-smtp');
+  const wecom = fakeProvider('wecom');
+  const changes: string[] = [];
+  const service = createMailService({
+    accounts,
+    inbound,
+    outbound: memoryOutbound(),
+    attachments,
+    providers: { 'imap-smtp': imap, wecom },
+    mime: createPostalMimeParser(),
+    secrets: plainSecrets,
+    clock,
+    ids: seqIds('in'),
+    notify: { changed: (r) => void changes.push(r) },
+    limits: { perTick: over.perTick ?? 50 },
+  });
+  return { service, accounts, inbound, attachments, imap, wecom, clock, changes };
+}
+
+describe('拉取', () => {
+  it('无游标先锚定;拉到的信入库、附件入 upload、游标逐封推进、lastPolledAt 与 lastInboundAt 记下', async () => {
+    const h = harness();
+    await h.accounts.save(account());
+    h.imap.queue(
+      rawMessage('uid:v1:1', eml({ messageId: '<m1@acme.com>', subject: '一', attachment: { name: 'bom.xlsx', contentType: 'application/vnd.ms-excel', content: 'PK' } })),
+      rawMessage('uid:v1:2', eml({ messageId: '<m2@acme.com>', subject: '二' })),
+    );
+    const outcome = await h.service.pollNow('acc-1');
+    expect(outcome).toEqual({ accountId: 'acc-1', pulled: 2, stored: 2, dispatched: 2, more: false });
+    const saved = h.accounts.rows.get('acc-1');
+    expect(saved?.cursor).toEqual(imapCursor('v1', 2));
+    expect(saved?.lastPolledAt).toEqual(h.clock.now());
+    expect(saved?.lastInboundAt).not.toBeNull();
+    expect(saved?.lastPollError).toBeNull();
+    expect(saved?.pollingUntil).toBeNull();
+    const list = await h.service.listInbound({});
+    // 两封 receivedAt 相同(夹具默认值),按插入顺序稳定排 —— 只断集合,不赌顺序
+    expect(list.map((m) => m.subject).sort()).toEqual(['一', '二']);
+    const one = list.find((m) => m.subject === '一');
+    expect(one?.attachments[0]?.ref).toBe('up-1');
+    expect(h.attachments.files.get('up-1')?.note).toBe('邮件附件:一');
+    expect(h.changes).toContain('inbound');
+  });
+
+  it('重复拉到(同 provider_ref 或同 Message-ID)不入库不分发、附件不重存', async () => {
+    const h = harness();
+    await h.accounts.save(account());
+    const hits: string[] = [];
+    h.service.subscribe('spy', async (m) => void hits.push(m.messageId));
+    h.imap.queue(rawMessage('uid:v1:1', eml({ messageId: '<m1@acme.com>', attachment: { name: 'a.txt', contentType: 'text/plain', content: 'x' } })));
+    await h.service.pollNow('acc-1');
+    h.imap.queue(
+      rawMessage('uid:v1:1', eml({ messageId: '<m1@acme.com>' })),
+      rawMessage('uid:v1:9', eml({ messageId: '<m1@acme.com>' })),
+    );
+    const again = await h.service.pollNow('acc-1');
+    expect(again.pulled).toBe(2);
+    expect(again.stored).toBe(0);
+    expect(hits).toEqual(['<m1@acme.com>']);
+    expect(h.attachments.files.size).toBe(1);
+  });
+
+  it('每轮封顶 perTick:余下的报 more=true,下一轮接着;wecom 形整批推进游标', async () => {
+    const h = harness({ perTick: 2 });
+    await h.accounts.save(account({ id: 'w', kind: 'wecom', name: '应用', address: 'app@corp.wecom.work', secretCipher: `sealed:${JSON.stringify({ corpSecret: 'cs' })}`, settings: { corpId: 'ww1' } }));
+    h.wecom.queue(rawMessage('mid-1', eml({ messageId: '<1@a>' })), rawMessage('mid-2', eml({ messageId: '<2@a>' })), rawMessage('mid-3', eml({ messageId: '<3@a>' })));
+    const first = await h.service.pollNow('w');
+    expect(first).toMatchObject({ pulled: 2, stored: 2, more: true });
+    expect(h.accounts.rows.get('w')?.cursor?.kind).toBe('wecom');
+    const second = await h.service.pollNow('w');
+    expect(second).toMatchObject({ pulled: 1, stored: 1, more: false });
+  });
+
+  it('拉取抛错:记 last_poll_error(脱敏)、游标停在最后一封成功处、租约还了、下一轮照常', async () => {
+    const h = harness();
+    await h.accounts.save(account({ cursor: imapCursor('v1', 5) }));
+    h.imap.failPullWith(new Error('LOGIN pw-1234 rejected'));
+    const outcome = await h.service.pollNow('acc-1');
+    expect(outcome.error).toBe('LOGIN *** rejected');
+    const saved = h.accounts.rows.get('acc-1');
+    expect(saved?.lastPollError).toBe('LOGIN *** rejected');
+    expect(saved?.cursor).toEqual(imapCursor('v1', 5));
+    expect(saved?.pollingUntil).toBeNull();
+    h.imap.failPullWith(undefined);
+    expect((await h.service.pollNow('acc-1')).error).toBeUndefined();
+    expect(h.accounts.rows.get('acc-1')?.lastPollError).toBeNull();
+  });
+
+  it('凭证空:不连 provider,记「未配置凭证」;停用 → 412;租约被占 → 409', async () => {
+    const h = harness();
+    await h.accounts.save(account({ secretCipher: null }));
+    const outcome = await h.service.pollNow('acc-1');
+    expect(outcome.error).toContain('未配置凭证');
+    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) }));
+    await expect(h.service.pollNow('acc-1')).rejects.toBeInstanceOf(MailPollBusyError);
+  });
+
+  it('pollDue:到期的账号各拉一轮,一个出错不影响别的', async () => {
+    const h = harness();
+    await h.accounts.save(account({ id: 'a', name: 'a', address: 'a@we.com' }));
+    await h.accounts.save(account({ id: 'b', name: 'b', address: 'b@we.com', secretCipher: null }));
+    await h.accounts.save(account({ id: 'c', name: 'c', address: 'c@we.com', enabled: false }));
+    h.imap.queue(rawMessage('uid:v1:1', eml({ messageId: '<x@a>' })));
+    const outcomes = await h.service.pollDue(h.clock.now());
+    expect(outcomes.map((o) => o.accountId).sort()).toEqual(['a', 'b']);
+    expect(outcomes.find((o) => o.accountId === 'a')?.stored).toBe(1);
+    expect(outcomes.find((o) => o.accountId === 'b')?.error).toContain('未配置凭证');
+  });
+});
+
+describe('分发', () => {
+  it('订阅方收到详情与 ctx(账号 / 通道);skipped 与 failed 各自留痕,互不影响;同名注册被拒;注销后不再收', async () => {
+    const h = harness();
+    await h.accounts.save(account({ channelId: 'ch-9' }));
+    const seen: Array<{ mail: MailInboundDetail; channelId: string | undefined }> = [];
+    h.service.subscribe('ok', async (mail, ctx) => void seen.push({ mail, channelId: ctx.channelId }));
+    h.service.subscribe('skip', async () => ({ skipped: '与我无关' }));
+    const off = h.service.subscribe('boom', async () => {
+      throw new Error('通道实例已停用:ch-9');
+    });
+    expect(() => h.service.subscribe('ok', async () => undefined)).toThrow(MailSubscriberTakenError);
+    h.imap.queue(rawMessage('uid:v1:1', eml({ messageId: '<m1@a>', subject: '主题' })));
+    await h.service.pollNow('acc-1');
+    expect(seen[0]?.mail.subject).toBe('主题');
+    expect(seen[0]?.channelId).toBe('ch-9');
+    const detail = await h.service.getInbound('in-1');
+    expect(detail.deliveries.map((d) => [d.subscriber, d.state, d.error])).toEqual([
+      ['boom', 'failed', '通道实例已停用:ch-9'],
+      ['ok', 'done', null],
+      ['skip', 'skipped', '与我无关'],
+    ]);
+    off();
+    h.imap.queue(rawMessage('uid:v1:2', eml({ messageId: '<m2@a>' })));
+    await h.service.pollNow('acc-1');
+    expect((await h.service.getInbound('in-2')).deliveries.map((d) => d.subscriber)).toEqual(['ok', 'skip']);
+  });
+
+  it('重新投递:指定订阅方或全部;正在 dispatching 的跳过;不认识的订阅方 404', async () => {
+    const h = harness();
+    await h.accounts.save(account());
+    let fail = true;
+    h.service.subscribe('flaky', async () => {
+      if (fail) throw new Error('第一次不行');
+    });
+    h.imap.queue(rawMessage('uid:v1:1', eml({ messageId: '<m1@a>' })));
+    await h.service.pollNow('acc-1');
+    expect((await h.service.getInbound('in-1')).deliveries[0]?.state).toBe('failed');
+    fail = false;
+    const redone = await h.service.redeliver('in-1', 'flaky');
+    expect(redone).toEqual([expect.objectContaining({ subscriber: 'flaky', state: 'done' })]);
+    await expect(h.service.redeliver('in-1', 'nobody')).rejects.toBeInstanceOf(MailSubscriberUnknownError);
+    await h.inbound.setDelivery({ mailId: 'in-1', subscriber: 'flaky', state: 'dispatching', error: null, startedAt: h.clock.now(), finishedAt: null });
+    expect((await h.service.redeliver('in-1'))[0]?.state).toBe('dispatching');
+  });
+});
+
+describe('附件与原件', () => {
+  it('attachment 取回字节;超限没存的 404;raw 取回原件', async () => {
+    const h = harness();
+    await h.accounts.save(account());
+    h.imap.queue(rawMessage('uid:v1:1', eml({ messageId: '<m1@a>', attachment: { name: 'a.txt', contentType: 'text/plain', content: 'hello' } })));
+    await h.service.pollNow('acc-1');
+    const file = await h.service.attachment('in-1', 0);
+    expect(new TextDecoder().decode(file.bytes)).toBe('hello');
+    expect(file.name).toBe('a.txt');
+    await expect(h.service.attachment('in-1', 5)).rejects.toThrow('附件没有存下来');
+    const raw = await h.service.raw('in-1');
+    expect(raw.name).toBe('m1_a.eml');
+    expect(new TextDecoder().decode(raw.bytes)).toContain('Subject:');
+  });
+});