迁移 0033:邮箱账号 / 入站 / 投递 / 出站四张表;mail store 与跨进程租约(两连接对抢 20 轮每轮一个赢)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TM4kUBvekALu9qqmp7D3wr
2026-08-26 00:49:24 +08:00
parent 8d7e56aa92
commit 5f1e64b4d4
7 changed files with 737 additions and 0 deletions
@@ -0,0 +1,169 @@
+/**
+ * mail 存储集成测试 —— 真 PG:两把唯一键、默认账号唯一、到期挑选、**跨进程租约两连接对抢**、
+ * 投递 upsert、出站幂等键、bytea 原件按元素拷回来(pg 的 Buffer 挂在共享内存池上)。
+ */
+import { readFileSync } from 'node:fs';
+import { afterAll, beforeAll, describe, expect, it } from 'vitest';
+import { Pool } from 'pg';
+import { drizzle } from 'drizzle-orm/node-postgres';
+import { startItDatabase } from '@testing';
+import type { ItDatabase } from '@testing';
+import type { Db } from '@shared/db.ts';
+import type { InboundRecord, MailAccountRecord, OutboundRecord } from '../port.ts';
+import { imapCursor } from '../internal/domain/cursor.ts';
+import { isDue } from '../internal/domain/due.ts';
+import { createInboundStore, createMailAccountStore, createOutboundStore } from '../internal/adapter/mail-store.ts';
+
+const database: ItDatabase | undefined = await startItDatabase();
+let poolA: Pool | undefined;
+let poolB: Pool | undefined;
+let db: Db;
+let dbB: Db;
+
+const MIGRATION = new URL('../../../../../../db/migrations/0033_mail.sql', import.meta.url);
+const now = new Date('2026-08-25T10:00:00Z');
+
+function account(over: Partial<MailAccountRecord>): MailAccountRecord {
+  return {
+    id: 'acc-a', kind: 'imap-smtp', name: '甲', address: 'a@we.com', displayName: '', enabled: true, isDefault: false,
+    pollIntervalSec: 30, channelId: null, backfillFrom: null, settings: { folder: 'INBOX' }, secretCipher: 'v1.x.y.z',
+    cursor: null, lastPolledAt: null, lastPollError: null, lastInboundAt: null, pollingUntil: null, createdAt: now, updatedAt: now,
+    ...over,
+  };
+}
+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,
+    ...over,
+  };
+}
+function outbound(over: Partial<OutboundRecord>): OutboundRecord {
+  return {
+    id: 'out-1', accountId: 'acc-a', messageId: '<o1@we.com>', idempotencyKey: null, to: ['b@acme.com'], cc: [], bcc: [],
+    subject: 's', text: 't', html: null, inReplyTo: null, references: [], attachments: [], state: 'sending', error: null,
+    providerRef: null, requestedBy: { kind: 'ops' }, createdAt: now, sentAt: null,
+    ...over,
+  };
+}
+
+beforeAll(async () => {
+  if (database === undefined) return;
+  poolA = new Pool({ connectionString: database.url, max: 2 });
+  poolB = new Pool({ connectionString: database.url, max: 2 });
+  await poolA.query('drop table if exists sys.mail_inbound_delivery, sys.mail_inbound, sys.mail_outbound, sys.mail_account cascade');
+  await poolA.query(readFileSync(MIGRATION, 'utf8'));
+  db = drizzle(poolA);
+  dbB = drizzle(poolB);
+});
+
+afterAll(async () => {
+  await poolA?.end();
+  await poolB?.end();
+  await database?.stop();
+});
+
+describe.skipIf(database === undefined)('mail store(真 PG)', () => {
+  it('账号:save / get / findByName / list;isDefault 只留一个', async () => {
+    const store = createMailAccountStore(db);
+    await store.save(account({ id: 'acc-a', name: '甲', isDefault: true }));
+    await store.save(account({ id: 'acc-b', name: '乙', address: 'b@we.com', isDefault: true }));
+    expect((await store.get('acc-a'))?.isDefault).toBe(false);
+    expect((await store.get('acc-b'))?.isDefault).toBe(true);
+    expect((await store.findByName('甲'))?.id).toBe('acc-a');
+    expect((await store.list()).map((a) => a.id)).toEqual(['acc-a', 'acc-b']);
+  });
+
+  it('游标 jsonb 存取回来还是同一形;savePoll 只改给了的列', async () => {
+    const store = createMailAccountStore(db);
+    await store.savePoll('acc-a', { cursor: imapCursor('7', 12), polledAt: now, error: null });
+    const got = await store.get('acc-a');
+    expect(got?.cursor).toEqual(imapCursor('7', 12));
+    expect(got?.lastPolledAt?.toISOString()).toBe(now.toISOString());
+    await store.savePoll('acc-a', { error: '连不上' });
+    expect((await store.get('acc-a'))?.cursor).toEqual(imapCursor('7', 12));
+    expect((await store.get('acc-a'))?.lastPollError).toBe('连不上');
+    await expect(store.savePoll('no-such', { error: 'x' })).resolves.toBeUndefined();
+  });
+
+  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) }));
+    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) }));
+    expect((await store.listDue(now)).map((a) => a.id).sort()).toEqual(['acc-a']);
+  });
+
+  it('**SQL 与 domain 的到期判据同源**:listDue 的返回集合 == 用 isDue 过滤 list() 的结果', async () => {
+    // 同一条「到期」判据两处各存一份实现(SQL 里的 listDue / domain 的 isDue)——
+    // 本测钉住二者不悄悄分叉:哪天 SQL 把 `<=` 写成 `<`、或漏掉租约那一段,这里立刻炸。
+    const store = createMailAccountStore(db);
+    const fromSql = (await store.listDue(now)).map((a) => a.id).sort();
+    const fromDomain = (await store.list())
+      .filter((a) => isDue(a, now))
+      .map((a) => a.id)
+      .sort();
+    expect(fromSql).toEqual(fromDomain);
+  });
+
+  it('**跨进程租约**:两条真连接对抢 20 轮,每轮恰好一个赢;还了再抢得到;过期了别人接得上', async () => {
+    const a = createMailAccountStore(db);
+    const b = createMailAccountStore(dbB);
+    await a.save(account({ id: 'acc-lease', name: '租', address: 'l@we.com' }));
+    for (let round = 0; round < 20; round += 1) {
+      const wins = await Promise.all([a.claimPolling('acc-lease', 60_000), b.claimPolling('acc-lease', 60_000)]);
+      expect(wins.filter(Boolean)).toHaveLength(1);
+      await a.releasePolling('acc-lease');
+    }
+    expect(await a.claimPolling('acc-lease', 60_000)).toBe(true);
+    expect(await b.claimPolling('acc-lease', 60_000)).toBe(false);
+    await poolA!.query(`update sys.mail_account set polling_until = now() - interval '1 second' where id = 'acc-lease'`);
+    expect(await b.claimPolling('acc-lease', 60_000)).toBe(true);
+    await b.releasePolling('acc-lease');
+    // 停用的账号抢不到
+    await a.save(account({ id: 'acc-lease', name: '租', address: 'l@we.com', enabled: false }));
+    expect(await a.claimPolling('acc-lease', 60_000)).toBe(false);
+  });
+
+  it('入站:两把唯一键都回 inserted:false;knownProviderRefs 只回库里有的;raw 拷回来是独立字节', async () => {
+    const store = createInboundStore(db);
+    expect(await store.insert(inbound({}))).toEqual({ inserted: true });
+    expect(await store.insert(inbound({ id: 'in-dup-msg', providerRef: 'uid:7:2' }))).toEqual({ inserted: false });
+    expect(await store.insert(inbound({ id: 'in-dup-ref', messageId: '<other@x>' }))).toEqual({ inserted: false });
+    expect(await store.knownProviderRefs('acc-a', ['uid:7:1', 'uid:7:99'])).toEqual(new Set(['uid:7:1']));
+    expect(await store.knownProviderRefs('acc-a', [])).toEqual(new Set());
+    const got = await store.get('in-1');
+    expect(got?.record.raw).toEqual(new Uint8Array([1, 2, 3]));
+    expect(got?.record.raw).toBeInstanceOf(Uint8Array);
+    expect(Buffer.isBuffer(got?.record.raw)).toBe(false);
+  });
+
+  it('投递:同一 (mail, subscriber) upsert;list 按收信时间倒序并带投递', async () => {
+    const store = createInboundStore(db);
+    await store.insert(inbound({ id: 'in-2', messageId: '<m2@x>', providerRef: 'uid:7:2', receivedAt: new Date(now.getTime() + 1000) }));
+    await store.setDelivery({ mailId: 'in-2', subscriber: 'channel', state: 'dispatching', error: null, startedAt: now, finishedAt: null });
+    await store.setDelivery({ mailId: 'in-2', subscriber: 'channel', state: 'failed', error: '通道停用', startedAt: now, finishedAt: now });
+    const list = await store.list({ accountId: 'acc-a', limit: 10 });
+    expect(list.map((x) => x.record.id)).toEqual(['in-2', 'in-1']);
+    expect(list[0]?.deliveries).toEqual([{ mailId: 'in-2', subscriber: 'channel', state: 'failed', error: '通道停用', startedAt: now, finishedAt: now }]);
+    expect(await store.list({ accountId: 'acc-zzz', limit: 10 })).toEqual([]);
+    expect((await store.list({ limit: 1 })).map((x) => x.record.id)).toEqual(['in-2']);
+  });
+
+  it('出站:insert / byIdempotencyKey / setState / list', async () => {
+    const store = createOutboundStore(db);
+    await store.insert(outbound({ id: 'out-1', idempotencyKey: 'k-1' }));
+    await store.insert(outbound({ id: 'out-2', messageId: '<o2@we.com>' }));
+    expect((await store.byIdempotencyKey('k-1'))?.id).toBe('out-1');
+    expect(await store.byIdempotencyKey('k-none')).toBeUndefined();
+    await store.setState('out-1', 'sent', { providerRef: 'smtp:250 ok', sentAt: now });
+    await store.setState('out-2', 'failed', { error: '535 auth failed' });
+    const list = await store.list({ accountId: 'acc-a', limit: 10 });
+    expect(list.find((o) => o.id === 'out-1')).toMatchObject({ state: 'sent', providerRef: 'smtp:250 ok' });
+    expect(list.find((o) => o.id === 'out-2')).toMatchObject({ state: 'failed', error: '535 auth failed' });
+    // 同一幂等键第二次 insert 撞唯一索引 —— 用例层先查后写,这里只证明索引在
+    await expect(store.insert(outbound({ id: 'out-3', idempotencyKey: 'k-1' }))).rejects.toThrow();
+  });
+});