mail adapter:企业微信邮件 API provider(token 缓存与刷新 / 翻页 / 已知 mail_id 不再读 / EML 编码兜底 / compose_send)

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

apps/api/src/modules/mail/__tests__/wecom-provider.test.ts

@@ -0,0 +1,147 @@
+/** 企业微信邮件 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 type { FetchLike } from '../internal/adapter/wecom-provider.ts';
+import { wecomCursor } from '../internal/domain/cursor.ts';
+
+type Call = { url: string; body: Record<string, unknown> };
+
+function scripted(handlers: Record<string, (body: Record<string, unknown>, nth: number) => unknown>): { fetch: FetchLike; calls: Call[] } {
+  const calls: Call[] = [];
+  const counts = new Map<string, number>();
+  const fetch: FetchLike = async (url, init) => {
+    const body = init?.body === undefined ? {} : (JSON.parse(init.body) as Record<string, unknown>);
+    calls.push({ url, body });
+    const key = Object.keys(handlers).find((k) => url.includes(k));
+    if (key === undefined) return { ok: false, status: 404, json: async () => ({}) };
+    const nth = (counts.get(key) ?? 0) + 1;
+    counts.set(key, nth);
+    return { ok: true, status: 200, json: async () => handlers[key]!(body, nth) };
+  };
+  return { fetch, calls };
+}
+
+const account = { id: 'w', kind: 'wecom' as const, address: '', displayName: '应用', settings: { corpId: 'ww1' }, secret: { corpSecret: 'cs-1' } };
+const known = async (): Promise<Set<string>> => new Set<string>();
+const now = new Date('2026-08-25T10:00:00Z');
+const EML = 'From: a@x.com\r\nTo: b@y.com\r\nSubject: hi\r\n\r\nbody\r\n';
+
+describe('token', () => {
+  it('第一次取 token 之后缓存;过期前 5 分钟重取;42001 刷一次重试', async () => {
+    let clock = 0;
+    let issued = 0;
+    const { fetch, calls } = scripted({
+      '/cgi-bin/gettoken': () => ({ errcode: 0, access_token: `tok-${(issued += 1)}`, expires_in: 7200 }),
+      'get_email_alias': (_b, nth) => (nth === 2 ? { errcode: 42001, errmsg: 'access_token expired' } : { errcode: 0, email: 'app@corp.wecom.work' }),
+    });
+    const provider = createWecomProvider({ fetch, now: () => clock });
+    await provider.probe(account);
+    await provider.probe(account);
+    expect(calls.filter((c) => c.url.includes('gettoken'))).toHaveLength(2); // 第二次 alias 回 42001 → 刷一次
+    clock = 7200_000 - 4 * 60_000; // 离过期不到 5 分钟
+    await provider.probe(account);
+    expect(calls.filter((c) => c.url.includes('gettoken'))).toHaveLength(3);
+    expect(calls[0]?.url).toContain('corpid=ww1');
+    expect(calls[0]?.url).toContain('corpsecret=cs-1');
+  });
+
+  it('-1 系统繁忙重试 ≤3 次;别的错误码抛 WecomApiError 带 api 名', async () => {
+    let busy = 2;
+    const { fetch } = scripted({
+      '/cgi-bin/gettoken': () => ({ errcode: 0, access_token: 't', expires_in: 7200 }),
+      'get_email_alias': () => (busy-- > 0 ? { errcode: -1, errmsg: 'system busy' } : { errcode: 48002, errmsg: 'api forbidden' }),
+    });
+    const report = await createWecomProvider({ fetch }).probe(account);
+    expect(report.receive.ok).toBe(false);
+    expect(report.receive.error).toContain('get_email_alias');
+    expect(report.receive.error).toContain('48002');
+    expect(report.send).toMatchObject({ checked: false });
+    await expect(
+      createWecomProvider({ fetch: scripted({ '/cgi-bin/gettoken': () => ({ errcode: 40001, errmsg: 'invalid credential' }) }).fetch }).pull(account, wecomCursor(1), { limit: 5, now, backfillFrom: null, known }),
+    ).rejects.toBeInstanceOf(WecomApiError);
+  });
+});
+
+describe('probe / anchor', () => {
+  it('探到应用邮箱地址;发信一栏 checked=false', async () => {
+    const { fetch } = scripted({
+      '/cgi-bin/gettoken': () => ({ errcode: 0, access_token: 't', expires_in: 7200 }),
+      'get_email_alias': () => ({ errcode: 0, email: 'app@corp.wecom.work', alias_list: [] }),
+    });
+    const report = await createWecomProvider({ fetch }).probe(account);
+    expect(report).toMatchObject({ kind: 'wecom', address: 'app@corp.wecom.work', receive: { ok: true }, send: { checked: false } });
+  });
+  it('锚定 = 当下(或回填起点)的秒', async () => {
+    const provider = createWecomProvider({ fetch: scripted({}).fetch });
+    expect(await provider.anchor(account, null, now)).toEqual(wecomCursor(Math.floor(now.getTime() / 1000)));
+    expect(await provider.anchor(account, new Date('2026-08-20T00:00:00Z'), now)).toEqual(wecomCursor(Math.floor(new Date('2026-08-20T00:00:00Z').getTime() / 1000)));
+  });
+});
+
+describe('pull', () => {
+  it('窗口 = lastTime−600 到当下;翻页;库里已有的 mail_id 不再 read_mail;EML 原文与 base64 都解得出;more 反映还没翻完', async () => {
+    const { fetch, calls } = scripted({
+      '/cgi-bin/gettoken': () => ({ errcode: 0, access_token: 't', expires_in: 7200 }),
+      'get_mail_list': (body) =>
+        body['cursor'] === undefined
+          ? { errcode: 0, next_cursor: 'c2', has_more: 1, mail_list: [{ mail_id: 'm1' }, { mail_id: 'm2' }] }
+          : { errcode: 0, next_cursor: '', has_more: 0, mail_list: [{ mail_id: 'm3' }] },
+      'read_mail': (body) => ({ errcode: 0, mail_data: body['mail_id'] === 'm3' ? Buffer.from(EML).toString('base64') : EML }),
+    });
+    const provider = createWecomProvider({ fetch });
+    const cursor = wecomCursor(Math.floor(now.getTime() / 1000) - 1000);
+    const result = await provider.pull(account, cursor, { limit: 10, now, backfillFrom: null, known: async (refs) => new Set(refs.filter((r) => r === 'm2')) });
+    const listCall = calls.find((c) => c.url.includes('get_mail_list'))?.body;
+    expect(listCall).toMatchObject({ begin_time: cursor.lastTime - 600, end_time: Math.floor(now.getTime() / 1000) });
+    expect(result.messages.map((m) => m.providerRef)).toEqual(['m1', 'm3']);
+    expect(new TextDecoder().decode(result.messages[1]?.eml)).toBe(EML);
+    expect(calls.filter((c) => c.url.includes('read_mail')).map((c) => c.body['mail_id'])).toEqual(['m1', 'm3']);
+    expect(result.cursor).toEqual(wecomCursor(Math.floor(now.getTime() / 1000)));
+    expect(result.more).toBe(false);
+  });
+
+  it('取满 limit 就停,more=true', async () => {
+    const { fetch } = scripted({
+      '/cgi-bin/gettoken': () => ({ errcode: 0, access_token: 't', expires_in: 7200 }),
+      'get_mail_list': () => ({ errcode: 0, next_cursor: 'c2', has_more: 1, mail_list: [{ mail_id: 'm1' }, { mail_id: 'm2' }] }),
+      'read_mail': () => ({ errcode: 0, mail_data: EML }),
+    });
+    const cursor = wecomCursor(1);
+    const result = await createWecomProvider({ fetch }).pull(account, cursor, { limit: 2, now, backfillFrom: null, known });
+    expect(result.messages).toHaveLength(2);
+    expect(result.more).toBe(true);
+    // **截断时游标原样退回**:没读到的那几封不能被跳过去(重叠窗兜不住一次积压)
+    expect(result.cursor).toEqual(cursor);
+  });
+});
+
+describe('send', () => {
+  it('compose_send 载荷:emails 三组、html 优先、附件 base64;成功不回 providerRef', async () => {
+    const { fetch, calls } = scripted({
+      '/cgi-bin/gettoken': () => ({ errcode: 0, access_token: 't', expires_in: 7200 }),
+      'compose_send': () => ({ errcode: 0, errmsg: 'ok' }),
+    });
+    const result = await createWecomProvider({ fetch }).send(account, {
+      messageId: '<x@y>',
+      from: { address: 'app@corp.wecom.work', name: '应用' },
+      to: ['a@x.com'],
+      cc: ['b@x.com'],
+      bcc: [],
+      subject: '报价',
+      text: '纯文本',
+      html: '<p>html</p>',
+      references: [],
+      attachments: [{ name: 'q.pdf', contentType: 'application/pdf', bytes: new Uint8Array([1, 2]) }],
+    });
+    expect(result).toEqual({});
+    const payload = calls.find((c) => c.url.includes('compose_send'))?.body;
+    expect(payload).toEqual({
+      to: { emails: ['a@x.com'] },
+      cc: { emails: ['b@x.com'] },
+      subject: '报价',
+      content: '<p>html</p>',
+      content_type: 'html',
+      attachment_list: [{ file_name: 'q.pdf', content: Buffer.from([1, 2]).toString('base64') }],
+    });
+  });
+});

apps/api/src/modules/mail/index.ts

@@ -86,3 +86,6 @@ export type { MailServiceDeps } from './internal/usecase/mail-service.ts';
 export { createMailService } from './internal/usecase/mail-service.ts';
 
 export { createImapSmtpProvider } from './internal/adapter/imap-smtp-provider.ts';
+
+export type { FetchLike } from './internal/adapter/wecom-provider.ts';
+export { createWecomProvider, WecomApiError } from './internal/adapter/wecom-provider.ts';

apps/api/src/modules/mail/internal/adapter/wecom-provider.ts

@@ -0,0 +1,183 @@
+/**
+ * 企业微信邮件 API 适配器(外界;设计稿 §5.5;文档核实 2026-08-25):
+ *   gettoken(corpid, corpsecret) → access_token(7200s)
+ *   exmail/app/get_email_alias → 应用邮箱地址
+ *   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 次。
+ */
+import { wecomSettingsSchema } from '@contracts';
+import type { MailProbeReport } from '@contracts';
+import { wecomCursor, wecomWindow } from '../domain/cursor.ts';
+import type { MailCursor } from '../domain/cursor.ts';
+import { errorText } from '../domain/redact.ts';
+import type { MailProvider, ProviderAccount, RawMessage } from '../../port.ts';
+
+export type FetchLike = (
+  url: string,
+  init?: { method?: string; headers?: Record<string, string>; body?: string; signal?: AbortSignal },
+) => Promise<{ ok: boolean; status: number; json(): Promise<unknown> }>;
+
+type WecomResponse = Record<string, unknown> & { errcode?: number; errmsg?: string };
+
+export class WecomApiError extends Error {
+  constructor(
+    readonly errcode: number,
+    readonly errmsg: string,
+    readonly api: string,
+  ) {
+    super(`企业微信 ${api} 失败:errcode ${errcode} ${errmsg}`);
+    this.name = 'WecomApiError';
+  }
+}
+
+const TOKEN_EARLY_MS = 5 * 60_000;
+const BUSY_RETRIES = 3;
+const REQUEST_TIMEOUT_MS = 30_000;
+const LIST_PAGE = 100;
+
+/** 文档只说「邮件 eml 内容」:头部行的形(`Name: value`)就按原文;否则当 base64 */
+export function emlBytesOf(mailData: string): Uint8Array {
+  const looksLikeEml = /^[A-Za-z][A-Za-z0-9-]*:\s/m.test(mailData.slice(0, 2048));
+  return looksLikeEml ? new TextEncoder().encode(mailData) : new Uint8Array(Buffer.from(mailData, 'base64'));
+}
+
+export function createWecomProvider(deps: { fetch?: FetchLike; now?: () => number } = {}): MailProvider {
+  const doFetch: FetchLike = deps.fetch ?? ((url, init) => fetch(url, init));
+  const nowMs = deps.now ?? (() => Date.now());
+  const tokens = new Map<string, { token: string; expiresAt: number }>();
+
+  async function call(api: string, url: string, init?: Parameters<FetchLike>[1]): Promise<WecomResponse> {
+    for (let attempt = 0; ; attempt += 1) {
+      const res = await doFetch(url, { ...init, signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS) });
+      if (!res.ok) throw new Error(`企业微信 ${api} HTTP ${res.status}`);
+      const body = (await res.json()) as WecomResponse;
+      const code = typeof body.errcode === 'number' ? body.errcode : 0;
+      if (code === 0) return body;
+      if (code === -1 && attempt < BUSY_RETRIES) continue;
+      throw new WecomApiError(code, String(body.errmsg ?? ''), api);
+    }
+  }
+
+  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;
+    const settings = wecomSettingsSchema.parse(account.settings);
+    const secret = account.secret['corpSecret'];
+    if (secret === undefined || secret === '') throw new Error('缺应用 secret');
+    const body = await call(
+      'gettoken',
+      `${settings.apiBase}/cgi-bin/gettoken?corpid=${encodeURIComponent(settings.corpId)}&corpsecret=${encodeURIComponent(secret)}`,
+      { method: 'GET' },
+    );
+    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 });
+    return token;
+  }
+
+  async function withToken(account: ProviderAccount, api: string, path: string, payload?: unknown): Promise<WecomResponse> {
+    const settings = wecomSettingsSchema.parse(account.settings);
+    const run = (token: string): Promise<WecomResponse> =>
+      call(
+        api,
+        `${settings.apiBase}${path}?access_token=${encodeURIComponent(token)}`,
+        payload === undefined
+          ? { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' }
+          : { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify(payload) },
+      );
+    try {
+      return await run(await tokenOf(account));
+    } catch (error) {
+      if (error instanceof WecomApiError && (error.errcode === 40014 || error.errcode === 42001)) {
+        return run(await tokenOf(account, true));
+      }
+      throw error;
+    }
+  }
+
+  return {
+    kind: 'wecom',
+
+    async probe(account): Promise<MailProbeReport> {
+      const started = nowMs();
+      try {
+        const alias = await withToken(account, 'get_email_alias', '/cgi-bin/exmail/app/get_email_alias');
+        const address = typeof alias['email'] === 'string' ? alias['email'] : '';
+        return {
+          kind: 'wecom',
+          address,
+          receive: { ok: true, latencyMs: nowMs() - started },
+          send: { checked: false, note: '凭证有效;发信权限只能在真发时验证(compose_send 没有只验不发的口)' },
+        };
+      } catch (error) {
+        return { kind: 'wecom', receive: { ok: false, error: errorText(error) }, send: { checked: false, note: '收信凭证未通过,发信未验' } };
+      }
+    },
+
+    async anchor(_account, backfillFrom, now): Promise<MailCursor> {
+      return wecomCursor(Math.floor((backfillFrom ?? now).getTime() / 1000));
+    },
+
+    async pull(account, cursor, opts) {
+      const { beginTime, endTime } = wecomWindow(cursor, opts.backfillFrom, opts.now);
+      /*
+       * **先滤库里已有的,再数够不够** —— 顺序反了会卡死:一个窗口里若前 limit 条恰好都已入库,
+       * 这一轮就交出 0 封,而引擎见空批即收工、游标又没动,下一轮还是这 0 封,永远吃不到后面的新信。
+       * 滤在前,每一轮都实实在在往前吃 limit 封**新**的。
+       */
+      const fresh: string[] = [];
+      let pageCursor: string | undefined;
+      let exhausted = false;
+      while (fresh.length < opts.limit) {
+        const page = await withToken(account, 'get_mail_list', '/cgi-bin/exmail/app/get_mail_list', {
+          begin_time: beginTime,
+          end_time: endTime,
+          ...(pageCursor === undefined ? {} : { cursor: pageCursor }),
+          limit: LIST_PAGE,
+        });
+        const list = Array.isArray(page['mail_list']) ? (page['mail_list'] as Array<{ mail_id?: unknown }>) : [];
+        const ids = list.flatMap((item) => (typeof item.mail_id === 'string' ? [item.mail_id] : []));
+        const known = await opts.known(ids);
+        for (const id of ids) if (!known.has(id)) fresh.push(id);
+        const hasMore = page['has_more'] === 1 || page['has_more'] === true;
+        pageCursor = typeof page['next_cursor'] === 'string' && page['next_cursor'] !== '' ? page['next_cursor'] : undefined;
+        if (!hasMore || list.length === 0 || pageCursor === undefined) {
+          exhausted = true;
+          break;
+        }
+      }
+      const wanted = fresh.slice(0, opts.limit);
+      const messages: RawMessage[] = [];
+      for (const mailId of wanted) {
+        const mail = await withToken(account, 'read_mail', '/cgi-bin/exmail/app/read_mail', { mail_id: mailId });
+        const data = mail['mail_data'];
+        if (typeof data !== 'string' || data === '') continue;
+        messages.push({ providerRef: mailId, eml: emlBytesOf(data), receivedAt: opts.now });
+      }
+      /*
+       * 游标**只在把这个窗口翻完时**才推到窗口上沿;被 limit 截断时原样退回旧游标 ——
+       * 否则没读到的那几封就被跳过去了(重叠窗只兜 10 分钟,兜不住一次积压)。
+       * 退回不会卡住:已入库的下一轮被 `known` 滤掉,每轮照旧往前吃一批。
+       */
+      return { messages, cursor: exhausted ? wecomCursor(endTime) : cursor, more: !exhausted || fresh.length > wanted.length };
+    },
+
+    async send(account, message) {
+      await withToken(account, 'compose_send', '/cgi-bin/exmail/app/compose_send', {
+        to: { emails: [...message.to] },
+        ...(message.cc.length === 0 ? {} : { cc: { emails: [...message.cc] } }),
+        ...(message.bcc.length === 0 ? {} : { bcc: { emails: [...message.bcc] } }),
+        subject: message.subject,
+        content: message.html ?? message.text ?? '',
+        content_type: message.html !== undefined ? 'html' : 'text',
+        ...(message.attachments.length === 0
+          ? {}
+          : { attachment_list: message.attachments.map((a) => ({ file_name: a.name, content: Buffer.from(a.bytes).toString('base64') })) }),
+      });
+      return {};
+    },
+  };
+}