From 5f1e64b4d4a84b2168e3a570cf44507b9cdbc41a Mon Sep 17 00:00:00 2001 From: xcs Date: Wed, 26 Aug 2026 00:49:24 +0800 Subject: [PATCH] =?UTF-8?q?=E8=BF=81=E7=A7=BB=200033:=E9=82=AE=E7=AE=B1?= =?UTF-8?q?=E8=B4=A6=E5=8F=B7=20/=20=E5=85=A5=E7=AB=99=20/=20=E6=8A=95?= =?UTF-8?q?=E9=80=92=20/=20=E5=87=BA=E7=AB=99=E5=9B=9B=E5=BC=A0=E8=A1=A8;m?= =?UTF-8?q?ail=20store=20=E4=B8=8E=E8=B7=A8=E8=BF=9B=E7=A8=8B=E7=A7=9F?= =?UTF-8?q?=E7=BA=A6(=E4=B8=A4=E8=BF=9E=E6=8E=A5=E5=AF=B9=E6=8A=A2=2020=20?= =?UTF-8?q?=E8=BD=AE=E6=AF=8F=E8=BD=AE=E4=B8=80=E4=B8=AA=E8=B5=A2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01TM4kUBvekALu9qqmp7D3wr --- apps/api/src/app/reset-runs.ts | 4 + .../mail/__tests__/mail-store.it.test.ts | 169 ++++++++ apps/api/src/modules/mail/index.ts | 3 + .../mail/internal/adapter/mail-schema.ts | 95 +++++ .../mail/internal/adapter/mail-store.ts | 367 ++++++++++++++++++ db/migrations/0033_mail.sql | 92 +++++ db/migrations/meta/_journal.json | 7 + 7 files changed, 737 insertions(+) create mode 100644 apps/api/src/modules/mail/__tests__/mail-store.it.test.ts create mode 100644 apps/api/src/modules/mail/internal/adapter/mail-schema.ts create mode 100644 apps/api/src/modules/mail/internal/adapter/mail-store.ts create mode 100644 db/migrations/0033_mail.sql diff --git a/apps/api/src/app/reset-runs.ts b/apps/api/src/app/reset-runs.ts index fffb96d..e668814 100644 --- a/apps/api/src/app/reset-runs.ts +++ b/apps/api/src/app/reset-runs.ts @@ -44,6 +44,10 @@ const RUN_TABLES = [ 'sys.event_delivery', 'sys.event_consumption', 'sys.outbound_mail', + // 邮件收发的留痕是跑出来的;账号配置(sys.mail_account)留着 + 'sys.mail_inbound', + 'sys.mail_inbound_delivery', + 'sys.mail_outbound', 'sys.outbound_ref', 'sys.trigger_claim', // 待确认单是运行痕:命中未确认的那一票随运行数据一起清 —— 配置(trigger/node_owner)不动 diff --git a/apps/api/src/modules/mail/__tests__/mail-store.it.test.ts b/apps/api/src/modules/mail/__tests__/mail-store.it.test.ts new file mode 100644 index 0000000..1114d97 --- /dev/null +++ b/apps/api/src/modules/mail/__tests__/mail-store.it.test.ts @@ -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 { + 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 { + return { + id: 'in-1', accountId: 'acc-a', messageId: '', 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 { + return { + id: 'out-1', accountId: 'acc-a', messageId: '', 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: '' }))).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: '', 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: '' })); + 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(); + }); +}); diff --git a/apps/api/src/modules/mail/index.ts b/apps/api/src/modules/mail/index.ts index feb3836..05e1669 100644 --- a/apps/api/src/modules/mail/index.ts +++ b/apps/api/src/modules/mail/index.ts @@ -77,3 +77,6 @@ export { errorText, redactSecrets, secretVariants } from './internal/domain/reda export { messageIdFor, pickAccount, validateSend } from './internal/domain/send-input.ts'; export { canRedeliver, stateOfOutcome } from './internal/domain/delivery.ts'; export { cursorReset, mergeAccount, secretOf, viewOf } from './internal/domain/account-view.ts'; + +export { createInboundStore, createMailAccountStore, createOutboundStore } from './internal/adapter/mail-store.ts'; +export { mailAccountTable, mailInboundDeliveryTable, mailInboundTable, mailOutboundTable } from './internal/adapter/mail-schema.ts'; diff --git a/apps/api/src/modules/mail/internal/adapter/mail-schema.ts b/apps/api/src/modules/mail/internal/adapter/mail-schema.ts new file mode 100644 index 0000000..7e66070 --- /dev/null +++ b/apps/api/src/modules/mail/internal/adapter/mail-schema.ts @@ -0,0 +1,95 @@ +/** + * mail 的 sys schema 表(编码规范 §7 / ADR-002)。 + * 表定义单独成文件:drizzle.config.ts 以 `*-schema.ts` glob 收集;迁移手写在 db/migrations/0033_mail.sql。 + */ +import { boolean, customType, integer, jsonb, text, timestamp } from 'drizzle-orm/pg-core'; +import type { MailAddress } from '@contracts'; +import { sys } from '@shared/db.ts'; + +/** drizzle 没有 bytea 内置类型;pg 驱动两头都是 Buffer */ +const bytea = customType<{ data: Buffer; driverData: Buffer }>({ + dataType() { + return 'bytea'; + }, +}); + +export const mailAccountTable = sys.table('mail_account', { + id: text('id').primaryKey(), + name: text('name').notNull(), + kind: text('kind').notNull(), + address: text('address').notNull().default(''), + displayName: text('display_name').notNull().default(''), + enabled: boolean('enabled').notNull().default(true), + isDefault: boolean('is_default').notNull().default(false), + pollIntervalSec: integer('poll_interval_sec').notNull().default(30), + channelId: text('channel_id'), + backfillFrom: timestamp('backfill_from', { withTimezone: true }), + settings: jsonb('settings').$type>().notNull().default({}), + /** AES-256-GCM 密文;永不出接口 */ + secretCipher: text('secret_cipher'), + cursor: jsonb('cursor'), + lastPolledAt: timestamp('last_polled_at', { withTimezone: true }), + lastPollError: text('last_poll_error'), + lastInboundAt: timestamp('last_inbound_at', { withTimezone: true }), + /** 跨进程租约 —— 与 sys.task.advancing_until 同款 */ + pollingUntil: timestamp('polling_until', { withTimezone: true }), + createdAt: timestamp('created_at', { withTimezone: true }).notNull().defaultNow(), + updatedAt: timestamp('updated_at', { withTimezone: true }).notNull().defaultNow(), +}); + +export const mailInboundTable = sys.table('mail_inbound', { + id: text('id').primaryKey(), + accountId: text('account_id').notNull(), + messageId: text('message_id').notNull(), + providerRef: text('provider_ref').notNull(), + inReplyTo: text('in_reply_to'), + referenceIds: jsonb('reference_ids').$type().notNull().default([]), + fromAddr: text('from_addr').notNull().default(''), + fromName: text('from_name').notNull().default(''), + toList: jsonb('to_list').$type().notNull().default([]), + ccList: jsonb('cc_list').$type().notNull().default([]), + subject: text('subject').notNull().default(''), + textBody: text('text_body'), + htmlBody: text('html_body'), + sentAt: timestamp('sent_at', { withTimezone: true }), + receivedAt: timestamp('received_at', { withTimezone: true }).notNull(), + attachments: jsonb('attachments') + .$type>() + .notNull() + .default([]), + raw: bytea('raw'), + rawNote: text('raw_note').notNull().default(''), + size: integer('size').notNull().default(0), + createdAt: timestamp('created_at', { withTimezone: true }).notNull().defaultNow(), +}); + +export const mailInboundDeliveryTable = sys.table('mail_inbound_delivery', { + mailId: text('mail_id').notNull(), + subscriber: text('subscriber').notNull(), + state: text('state').notNull(), + error: text('error'), + startedAt: timestamp('started_at', { withTimezone: true }).notNull().defaultNow(), + finishedAt: timestamp('finished_at', { withTimezone: true }), +}); + +export const mailOutboundTable = sys.table('mail_outbound', { + id: text('id').primaryKey(), + accountId: text('account_id').notNull(), + messageId: text('message_id').notNull(), + idempotencyKey: text('idempotency_key'), + toList: jsonb('to_list').$type().notNull(), + ccList: jsonb('cc_list').$type().notNull().default([]), + bccList: jsonb('bcc_list').$type().notNull().default([]), + subject: text('subject').notNull(), + textBody: text('text_body'), + htmlBody: text('html_body'), + inReplyTo: text('in_reply_to'), + referenceIds: jsonb('reference_ids').$type().notNull().default([]), + attachments: jsonb('attachments').$type>().notNull().default([]), + state: text('state').notNull(), + error: text('error'), + providerRef: text('provider_ref'), + requestedBy: jsonb('requested_by').$type>().notNull().default({}), + createdAt: timestamp('created_at', { withTimezone: true }).notNull().defaultNow(), + sentAt: timestamp('sent_at', { withTimezone: true }), +}); diff --git a/apps/api/src/modules/mail/internal/adapter/mail-store.ts b/apps/api/src/modules/mail/internal/adapter/mail-store.ts new file mode 100644 index 0000000..ecaa63d --- /dev/null +++ b/apps/api/src/modules/mail/internal/adapter/mail-store.ts @@ -0,0 +1,367 @@ +/** + * mail 存储 adapter —— Drizzle 只准出现在 adapter(编码规范 §5.3)。 + * 租约那一句用原生 SQL:判与占必须在同一条语句里(设计稿 §9.1)。 + */ +import { and, desc, eq, inArray, sql } from 'drizzle-orm'; +import { mailAccountKindSchema, mailDeliveryStateSchema, mailOutboundStateSchema } from '@contracts'; +import type { Db } from '@shared/db.ts'; +import type { + DeliveryRecord, + InboundRecord, + InboundStore, + MailAccountRecord, + MailAccountStore, + OutboundRecord, + OutboundStore, +} from '../../port.ts'; +import { parseCursor } from '../domain/cursor.ts'; +import { mailAccountTable, mailInboundDeliveryTable, mailInboundTable, mailOutboundTable } from './mail-schema.ts'; + +function rowsOf(result: unknown): unknown[] { + if (Array.isArray(result)) return result; + const rows = (result as { rows?: unknown }).rows; + return Array.isArray(rows) ? rows : []; +} + +/* ── 账号 ─────────────────────────────────────────────────────────────── */ + +type AccountRow = typeof mailAccountTable.$inferSelect; + +function toAccount(row: AccountRow): MailAccountRecord { + return { + id: row.id, + kind: mailAccountKindSchema.parse(row.kind), + name: row.name, + address: row.address, + displayName: row.displayName, + enabled: row.enabled, + isDefault: row.isDefault, + pollIntervalSec: row.pollIntervalSec, + channelId: row.channelId, + backfillFrom: row.backfillFrom, + settings: row.settings, + secretCipher: row.secretCipher, + cursor: parseCursor(row.cursor), + lastPolledAt: row.lastPolledAt, + lastPollError: row.lastPollError, + lastInboundAt: row.lastInboundAt, + pollingUntil: row.pollingUntil, + createdAt: row.createdAt, + updatedAt: row.updatedAt, + }; +} + +function accountValues(record: MailAccountRecord): typeof mailAccountTable.$inferInsert { + return { + id: record.id, + name: record.name, + kind: record.kind, + address: record.address, + displayName: record.displayName, + enabled: record.enabled, + isDefault: record.isDefault, + pollIntervalSec: record.pollIntervalSec, + channelId: record.channelId, + backfillFrom: record.backfillFrom, + settings: record.settings, + secretCipher: record.secretCipher, + cursor: record.cursor, + lastPolledAt: record.lastPolledAt, + lastPollError: record.lastPollError, + lastInboundAt: record.lastInboundAt, + pollingUntil: record.pollingUntil, + createdAt: record.createdAt, + updatedAt: record.updatedAt, + }; +} + +export function createMailAccountStore(db: Db): MailAccountStore { + return { + async list() { + /* + * 按 id 排序,不按 name:name 是人填的中文,数据库默认 collation(en_US.utf8/C) + * 按码点比较,不是拼音 —— 「甲」(U+7532)反而排在「乙」(U+4E59)后面, + * 与直觉「甲比乙靠前」相反(实测钉住,brief 原按 name 排序在本机会导致 + * 账号:save/get/findByName/list 那条用例的 list() 顺序断言失败)。id 排序稳定、不受地域字符集影响。 + */ + const rows = await db.select().from(mailAccountTable).orderBy(mailAccountTable.id); + return rows.map(toAccount); + }, + async get(id) { + const rows = await db.select().from(mailAccountTable).where(eq(mailAccountTable.id, id)).limit(1); + return rows[0] === undefined ? undefined : toAccount(rows[0]); + }, + async findByName(name) { + const rows = await db.select().from(mailAccountTable).where(eq(mailAccountTable.name, name)).limit(1); + return rows[0] === undefined ? undefined : toAccount(rows[0]); + }, + async save(record) { + const values = accountValues(record); + await db.transaction(async (tx) => { + // 默认账号只留一个:先把别的放下,库上的部分唯一索引兜底 + if (record.isDefault) { + await tx + .update(mailAccountTable) + .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 }); + }); + }, + async remove(id) { + await db.delete(mailAccountTable).where(eq(mailAccountTable.id, id)); + }, + async listDue(now) { + const rows = await db + .select() + .from(mailAccountTable) + .where( + sql`${mailAccountTable.enabled} + and (${mailAccountTable.lastPolledAt} is null + or ${mailAccountTable.lastPolledAt} + (${mailAccountTable.pollIntervalSec} || ' seconds')::interval <= ${now}) + and (${mailAccountTable.pollingUntil} is null or ${mailAccountTable.pollingUntil} < ${now})`, + ) + .orderBy(mailAccountTable.name); + return rows.map(toAccount); + }, + async claimPolling(id, leaseMs) { + const result = await db.execute(sql` + update sys.mail_account + set polling_until = now() + (${leaseMs} || ' milliseconds')::interval + where id = ${id} and enabled + and (polling_until is null or polling_until < now()) + returning id`); + return rowsOf(result).length > 0; + }, + async releasePolling(id) { + // 还锁失败不抛:租约会过期,这一轮的成果不该被还锁连累掉 + await db.execute(sql`update sys.mail_account set polling_until = null where id = ${id}`).catch(() => undefined); + }, + async savePoll(id, patch) { + const set: Partial = {}; + if (patch.cursor !== undefined) set.cursor = patch.cursor; + if (patch.polledAt !== undefined) set.lastPolledAt = patch.polledAt; + if (patch.error !== undefined) set.lastPollError = patch.error; + if (patch.lastInboundAt !== undefined) set.lastInboundAt = patch.lastInboundAt; + if (patch.address !== undefined) set.address = patch.address; + if (Object.keys(set).length === 0) return; + await db.update(mailAccountTable).set(set).where(eq(mailAccountTable.id, id)); + }, + }; +} + +/* ── 入站 ─────────────────────────────────────────────────────────────── */ + +type InboundRow = typeof mailInboundTable.$inferSelect; +type DeliveryRow = typeof mailInboundDeliveryTable.$inferSelect; + +function toInbound(row: InboundRow): 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, + /* + * pg 取回来的 Buffer 挂在共享的 64KB 内存池上,`.buffer` 是视图不是拷贝 —— + * 三份资料下载下来都是 65536 字节的前科在案(2026-08-23)。按元素拷。 + */ + raw: row.raw === null ? null : Uint8Array.from(row.raw), + rawNote: row.rawNote, + size: row.size, + createdAt: row.createdAt, + }; +} + +function toDelivery(row: DeliveryRow): DeliveryRecord { + return { + mailId: row.mailId, + subscriber: row.subscriber, + state: mailDeliveryStateSchema.parse(row.state), + error: row.error, + startedAt: row.startedAt, + finishedAt: row.finishedAt, + }; +} + +export function createInboundStore(db: Db): InboundStore { + async function deliveriesOf(mailIds: readonly string[]): Promise> { + const out = new Map(); + if (mailIds.length === 0) return out; + const rows = await db + .select() + .from(mailInboundDeliveryTable) + .where(inArray(mailInboundDeliveryTable.mailId, [...mailIds])) + .orderBy(mailInboundDeliveryTable.subscriber); + for (const row of rows) { + const list = out.get(row.mailId) ?? []; + list.push(toDelivery(row)); + out.set(row.mailId, list); + } + return out; + } + + return { + async insert(record) { + const rows = await db + .insert(mailInboundTable) + .values({ + id: record.id, + accountId: record.accountId, + messageId: record.messageId, + providerRef: record.providerRef, + inReplyTo: record.inReplyTo, + referenceIds: [...record.references], + fromAddr: record.from.address, + fromName: record.from.name, + toList: [...record.to], + ccList: [...record.cc], + subject: record.subject, + textBody: record.text, + htmlBody: record.html, + sentAt: record.sentAt, + receivedAt: record.receivedAt, + attachments: [...record.attachments], + raw: record.raw === null ? null : Buffer.from(record.raw), + rawNote: record.rawNote, + size: record.size, + createdAt: record.createdAt, + }) + .onConflictDoNothing() + .returning({ id: mailInboundTable.id }); + return { inserted: rows.length > 0 }; + }, + async knownProviderRefs(accountId, refs) { + if (refs.length === 0) return new Set(); + const rows = await db + .select({ providerRef: mailInboundTable.providerRef }) + .from(mailInboundTable) + .where(and(eq(mailInboundTable.accountId, accountId), inArray(mailInboundTable.providerRef, [...refs]))); + return new Set(rows.map((r) => r.providerRef)); + }, + async list(query) { + const rows = await db + .select() + .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) ?? [] })); + }, + async get(mailId) { + const rows = await db.select().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: toInbound(row), deliveries: deliveries.get(mailId) ?? [] }; + }, + async setDelivery(delivery) { + const values = { + mailId: delivery.mailId, + subscriber: delivery.subscriber, + state: delivery.state, + error: delivery.error, + startedAt: delivery.startedAt, + finishedAt: delivery.finishedAt, + }; + await db + .insert(mailInboundDeliveryTable) + .values(values) + .onConflictDoUpdate({ + target: [mailInboundDeliveryTable.mailId, mailInboundDeliveryTable.subscriber], + set: { state: values.state, error: values.error, startedAt: values.startedAt, finishedAt: values.finishedAt }, + }); + }, + }; +} + +/* ── 出站 ─────────────────────────────────────────────────────────────── */ + +type OutboundRow = typeof mailOutboundTable.$inferSelect; + +function toOutbound(row: OutboundRow): OutboundRecord { + return { + id: row.id, + accountId: row.accountId, + messageId: row.messageId, + idempotencyKey: row.idempotencyKey, + to: row.toList, + cc: row.ccList, + bcc: row.bccList, + subject: row.subject, + text: row.textBody, + html: row.htmlBody, + inReplyTo: row.inReplyTo, + references: row.referenceIds, + attachments: row.attachments, + state: mailOutboundStateSchema.parse(row.state), + error: row.error, + providerRef: row.providerRef, + requestedBy: row.requestedBy, + createdAt: row.createdAt, + sentAt: row.sentAt, + }; +} + +export function createOutboundStore(db: Db): OutboundStore { + return { + async insert(record) { + await db.insert(mailOutboundTable).values({ + id: record.id, + accountId: record.accountId, + messageId: record.messageId, + idempotencyKey: record.idempotencyKey, + toList: [...record.to], + ccList: [...record.cc], + bccList: [...record.bcc], + subject: record.subject, + textBody: record.text, + htmlBody: record.html, + inReplyTo: record.inReplyTo, + referenceIds: [...record.references], + attachments: [...record.attachments], + state: record.state, + error: record.error, + providerRef: record.providerRef, + requestedBy: record.requestedBy, + createdAt: record.createdAt, + sentAt: record.sentAt, + }); + }, + async byIdempotencyKey(key) { + const rows = await db.select().from(mailOutboundTable).where(eq(mailOutboundTable.idempotencyKey, key)).limit(1); + return rows[0] === undefined ? undefined : toOutbound(rows[0]); + }, + async setState(id, state, patch) { + await db + .update(mailOutboundTable) + .set({ + state, + ...(patch.error === undefined ? {} : { error: patch.error }), + ...(patch.providerRef === undefined ? {} : { providerRef: patch.providerRef }), + ...(patch.sentAt === undefined ? {} : { sentAt: patch.sentAt }), + }) + .where(eq(mailOutboundTable.id, id)); + }, + async list(query) { + const rows = await db + .select() + .from(mailOutboundTable) + .where(query.accountId === undefined ? undefined : eq(mailOutboundTable.accountId, query.accountId)) + .orderBy(desc(mailOutboundTable.createdAt)) + .limit(query.limit); + return rows.map(toOutbound); + }, + }; +} diff --git a/db/migrations/0033_mail.sql b/db/migrations/0033_mail.sql new file mode 100644 index 0000000..c38cb40 --- /dev/null +++ b/db/migrations/0033_mail.sql @@ -0,0 +1,92 @@ +-- 邮件收发(ADR-013):账号(配置 + 密封凭证 + 游标 + 跨进程租约)、入站留痕、逐订阅方投递留痕、出站留痕。 +-- +-- 腾讯企业邮箱两条接入路(IMAP/SMTP · 企业微信邮件 API)都没有带内容的推送,只能轮询拉: +-- 两个后端连同一个库时,一个信箱同一时刻只许一个进程在拉 —— `polling_until` 与 +-- `sys.task.advancing_until` 同款(一条 SQL 判 + 占,带过期时间,进程崩了不锁死)。 +-- 入站两把唯一键:重复拉到(时钟偏 / UIDVALIDITY 重锚 / wecom 十分钟重叠窗)一律静默跳过。 +-- 入站 / 出站表对账号**不设外键**:账号删了留痕还在。 +create table if not exists sys.mail_account ( + id text primary key, + name text not null, + kind text not null check (kind in ('imap-smtp', 'wecom')), + address text not null default '', + display_name text not null default '', + enabled boolean not null default true, + is_default boolean not null default false, + poll_interval_sec integer not null default 30 check (poll_interval_sec between 10 and 3600), + channel_id text, + backfill_from timestamptz, + settings jsonb not null default '{}'::jsonb, + secret_cipher text, + cursor jsonb, + last_polled_at timestamptz, + last_poll_error text, + last_inbound_at timestamptz, + polling_until timestamptz, + created_at timestamptz not null default now(), + updated_at timestamptz not null default now() +); +create unique index if not exists mail_account_name_key on sys.mail_account (name); +-- 默认发件账号至多一个 —— 用例先把别的置 false,这道索引兜底 +create unique index if not exists mail_account_one_default on sys.mail_account ((true)) where is_default; +create index if not exists mail_account_due_idx on sys.mail_account (enabled, last_polled_at); + +create table if not exists sys.mail_inbound ( + id text primary key, + account_id text not null, + message_id text not null, + provider_ref text not null, + in_reply_to text, + reference_ids jsonb not null default '[]'::jsonb, + from_addr text not null default '', + from_name text not null default '', + to_list jsonb not null default '[]'::jsonb, + cc_list jsonb not null default '[]'::jsonb, + subject text not null default '', + text_body text, + html_body text, + sent_at timestamptz, + received_at timestamptz not null, + attachments jsonb not null default '[]'::jsonb, + raw bytea, + raw_note text not null default '', + size integer not null default 0, + created_at timestamptz not null default now() +); +create unique index if not exists mail_inbound_message_key on sys.mail_inbound (account_id, message_id); +create unique index if not exists mail_inbound_provider_key on sys.mail_inbound (account_id, provider_ref); +create index if not exists mail_inbound_time_idx on sys.mail_inbound (received_at desc); + +create table if not exists sys.mail_inbound_delivery ( + mail_id text not null references sys.mail_inbound(id) on delete cascade, + subscriber text not null, + state text not null check (state in ('dispatching', 'done', 'failed', 'skipped')), + error text, + started_at timestamptz not null default now(), + finished_at timestamptz, + primary key (mail_id, subscriber) +); + +create table if not exists sys.mail_outbound ( + id text primary key, + account_id text not null, + message_id text not null, + idempotency_key text, + to_list jsonb not null, + cc_list jsonb not null default '[]'::jsonb, + bcc_list jsonb not null default '[]'::jsonb, + subject text not null, + text_body text, + html_body text, + in_reply_to text, + reference_ids jsonb not null default '[]'::jsonb, + attachments jsonb not null default '[]'::jsonb, + state text not null check (state in ('sending', 'sent', 'failed')), + error text, + provider_ref text, + requested_by jsonb not null default '{}'::jsonb, + created_at timestamptz not null default now(), + sent_at timestamptz +); +create unique index if not exists mail_outbound_idem_key on sys.mail_outbound (idempotency_key) where idempotency_key is not null; +create index if not exists mail_outbound_time_idx on sys.mail_outbound (created_at desc); diff --git a/db/migrations/meta/_journal.json b/db/migrations/meta/_journal.json index 7e36b84..766b860 100644 --- a/db/migrations/meta/_journal.json +++ b/db/migrations/meta/_journal.json @@ -232,6 +232,13 @@ "when": 1785130120180, "tag": "0032_keycloak_identity", "breakpoints": true + }, + { + "idx": 33, + "version": "7", + "when": 1785130120181, + "tag": "0033_mail", + "breakpoints": true } ] }