From 0c6e8e2c1bac5be1dc8b90cdc229d2152bd3bd29 Mon Sep 17 00:00:00 2001 From: xcs Date: Wed, 26 Aug 2026 02:05:50 +0800 Subject: [PATCH] =?UTF-8?q?app=20=E8=A3=85=E9=85=8D=20mail:channel=20?= =?UTF-8?q?=E4=BD=9C=E9=A6=96=E4=B8=AA=E8=AE=A2=E9=98=85=E6=96=B9=E3=80=81?= =?UTF-8?q?=E5=87=BA=E7=AB=99=E7=BB=8F=20mail.send=E3=80=81mail.send=20?= =?UTF-8?q?=E5=B7=A5=E5=85=B7=E7=9C=9F=E5=8F=91=E4=BF=A1(dryRun=20?= =?UTF-8?q?=E5=8F=AA=E6=A0=A1=E9=AA=8C)=E3=80=81=E8=BD=AE=E8=AF=A2?= =?UTF-8?q?=E5=B7=A5=E4=BD=9C=E8=80=85;=E4=B8=8A=E4=BC=A0=E4=BB=B6?= =?UTF-8?q?=E5=AD=98=E5=8F=96=E6=8A=BD=E5=87=BA=E5=85=B1=E7=94=A8?= 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 --- .../api/src/app/__tests__/mail-poller.test.ts | 47 ++++++ .../api/src/app/__tests__/mail-wiring.test.ts | 111 +++++++++++++ apps/api/src/app/config.ts | 3 + apps/api/src/app/container.ts | 56 +++++-- apps/api/src/app/mail-wiring.ts | 150 ++++++++++++++++++ apps/api/src/app/main.ts | 5 + apps/api/src/app/routes/governance-ops.ts | 32 +--- apps/api/src/app/tools/first-party.ts | 10 -- apps/api/src/app/upload-store.ts | 42 +++++ apps/api/src/app/workers/mail-poller.ts | 41 +++++ 10 files changed, 449 insertions(+), 48 deletions(-) create mode 100644 apps/api/src/app/__tests__/mail-poller.test.ts create mode 100644 apps/api/src/app/__tests__/mail-wiring.test.ts create mode 100644 apps/api/src/app/mail-wiring.ts create mode 100644 apps/api/src/app/upload-store.ts create mode 100644 apps/api/src/app/workers/mail-poller.ts diff --git a/apps/api/src/app/__tests__/mail-poller.test.ts b/apps/api/src/app/__tests__/mail-poller.test.ts new file mode 100644 index 0000000..1e9be32 --- /dev/null +++ b/apps/api/src/app/__tests__/mail-poller.test.ts @@ -0,0 +1,47 @@ +/** 轮询工作者:每 tick 调 pollDue;上一 tick 没跑完就跳过这一拍;出错的账号写 stderr;stop 后不再跑 */ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { MailPollOutcome } from '@contracts'; +import { pollMailOnce, startMailPoller } from '../workers/mail-poller.ts'; + +function fakeMail(outcomes: MailPollOutcome[], holdMs = 0) { + const calls: Date[] = []; + return { + calls, + mail: { + pollDue: async (now: Date) => { + calls.push(now); + if (holdMs > 0) await new Promise((r) => setTimeout(r, holdMs)); + return outcomes; + }, + }, + }; +} + +describe('pollMailOnce', () => { + it('回用例的结果;出错的账号写一行 stderr', async () => { + const write = vi.spyOn(process.stderr, 'write').mockImplementation(() => true); + const { mail } = fakeMail([{ accountId: 'a', pulled: 1, stored: 1, dispatched: 1, more: false }, { accountId: 'b', pulled: 0, stored: 0, dispatched: 0, more: false, error: '连不上' }]); + const out = await pollMailOnce({ container: { mail: mail as never } }); + expect(out).toHaveLength(2); + expect(write).toHaveBeenCalledWith(expect.stringContaining('[mail-poller] b:连不上')); + write.mockRestore(); + }); +}); + +describe('startMailPoller', () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it('按 tick 调;在跑时跳过;stop 后停', async () => { + const { mail, calls } = fakeMail([], 1000); + const poller = startMailPoller({ container: { mail: mail as never } }, 100); + await vi.advanceTimersByTimeAsync(350); + expect(calls).toHaveLength(1); // 第一拍 1000ms 没跑完,后面两拍跳过 + await vi.advanceTimersByTimeAsync(1000); + expect(calls.length).toBeGreaterThanOrEqual(2); + poller.stop(); + const before = calls.length; + await vi.advanceTimersByTimeAsync(2000); + expect(calls).toHaveLength(before); + }); +}); diff --git a/apps/api/src/app/__tests__/mail-wiring.test.ts b/apps/api/src/app/__tests__/mail-wiring.test.ts new file mode 100644 index 0000000..1444d3f --- /dev/null +++ b/apps/api/src/app/__tests__/mail-wiring.test.ts @@ -0,0 +1,111 @@ +/** mail ↔ channel / toolbox 的接线:三段纯映射各自成立;dryRun 不发;失败翻成工具错误 */ +import { describe, expect, it } from 'vitest'; +import type { MailInboundDetail, MailSendInput, MailSendResult } from '@contracts'; +import type { InboundMail } from '@contracts'; +import { channelSubscriber, channelTransport, mailSendTool, toInboundMail } from '../mail-wiring.ts'; + +const detail: MailInboundDetail = { + id: 'in-1', + accountId: 'acc-1', + messageId: '', + from: { address: 'buyer@acme.com', name: '张三' }, + to: [{ address: 'sales@we.com', name: '' }], + cc: [{ address: 'ops@we.com', name: 'Ops' }], + subject: '询价 RFQ-1', + sentAt: '2026-08-25T09:00:00.000Z', + receivedAt: '2026-08-25T09:30:00.000Z', + textPreview: '你好', + hasHtml: true, + attachments: [ + { index: 0, name: 'bom.xlsx', contentType: 'application/vnd.ms-excel', size: 10, ref: 'up-1', note: '' }, + { index: 1, name: 'big.zip', contentType: 'application/zip', size: 9_000_000, ref: null, note: '超限' }, + ], + deliveries: [], + inReplyTo: null, + references: ['', ''], + text: null, + html: '

你好

', + size: 100, + rawAvailable: true, +}; + +function fakeSend(result: Partial = {}): { send: (input: MailSendInput) => Promise; calls: MailSendInput[] } { + const calls: MailSendInput[] = []; + return { + calls, + send: async (input) => { + calls.push(input); + return { outboundId: 'out-1', accountId: 'acc-1', messageId: '', state: 'sent', ...result }; + }, + }; +} + +describe('toInboundMail', () => { + it('线程取 References 末项(没 In-Reply-To);只带存下来的附件;纯 html 剥成正文', () => { + const mail: InboundMail = toInboundMail(detail); + expect(mail.threadId).toBe(''); + expect(mail.from).toBe('buyer@acme.com'); + expect(mail.to).toEqual(['sales@we.com']); + expect(mail.cc).toEqual(['ops@we.com']); + expect(mail.body).toBe('你好'); + expect(mail.attachments).toEqual([{ name: 'bom.xlsx', contentType: 'application/vnd.ms-excel', size: 10, ref: 'up-1' }]); + expect(mail.receivedAt).toBe('2026-08-25T09:30:00.000Z'); + }); +}); + +describe('channelSubscriber', () => { + it('账号没绑通道 → skipped;绑了就 ingestMail(通道 id + 归一后的信)', async () => { + const calls: Array<[string, InboundMail]> = []; + const handler = channelSubscriber({ + ingestMail: async (id, mail) => { + calls.push([id, mail]); + return undefined; + }, + }); + expect(await handler(detail, { accountId: 'acc-1', channelId: undefined })).toEqual({ skipped: '账号未绑定通道' }); + expect(await handler(detail, { accountId: 'acc-1', channelId: 'ch-1' })).toBeUndefined(); + expect(calls[0]?.[0]).toBe('ch-1'); + expect(calls[0]?.[1].messageId).toBe(''); + }); +}); + +describe('channelTransport', () => { + it('按 from 找账号、threadId 进 inReplyTo / references、附件 ref 变 uploadRef;回 messageId 与线程', async () => { + const mail = fakeSend(); + const sent = await channelTransport(mail).send({ from: 'sales@we.com', to: ['b@acme.com'], cc: [], subject: '报价', body: '正文', threadId: '', attachments: [{ name: 'q.pdf', ref: 'up-9' }] }); + expect(sent).toEqual({ messageId: '', threadId: '' }); + expect(mail.calls[0]).toMatchObject({ from: 'sales@we.com', inReplyTo: '', references: [''], text: '正文', attachments: [{ name: 'q.pdf', uploadRef: 'up-9' }], requestedBy: { kind: 'channel' } }); + const fresh = await channelTransport(fakeSend()).send({ from: 'sales@we.com', to: ['b@acme.com'], cc: [], subject: 's', body: 'b', attachments: [] }); + expect(fresh.threadId).toBe(''); + }); + it('厂商失败抛出,让 channel.sendMail 的调用方看到', async () => { + await expect(channelTransport(fakeSend({ state: 'failed', error: '535 auth failed' })).send({ from: 'sales@we.com', to: ['b@acme.com'], cc: [], subject: 's', body: 'b', attachments: [] })).rejects.toThrow('535 auth failed'); + }); +}); + +describe('mailSendTool', () => { + const base = { connection: { account: 'default' }, operation: undefined, idempotencyKey: 'k-1' }; + it('数组 / 逗号串都认作收件人;body → text;幂等键与连接配置的账号带过去;出参保留 queued', async () => { + const mail = fakeSend(); + const result = await mailSendTool(mail)({ ...base, connection: { account: 'acc-2' }, input: { to: 'a@x.com; b@x.com', subject: 's', body: '正文', inReplyTo: '', attachments: [{ uploadRef: 'up-1', name: 'q.pdf' }, { name: 'no-ref' }] } }); + expect(result).toEqual({ ok: true, output: { queued: true, outboundId: 'out-1', messageId: '', state: 'sent' } }); + expect(mail.calls[0]).toMatchObject({ accountId: 'acc-2', to: ['a@x.com', 'b@x.com'], text: '正文', inReplyTo: '', references: [''], idempotencyKey: 'k-1', attachments: [{ uploadRef: 'up-1', name: 'q.pdf' }], requestedBy: { kind: 'tool', tool: 'mail.send' } }); + await mailSendTool(mail)({ ...base, input: { to: ['a@x.com'], subject: 's', body: 'b' } }); + expect(mail.calls[1]?.accountId).toBeUndefined(); + }); + it('dryRun 只校验不发', async () => { + const mail = fakeSend(); + expect(await mailSendTool(mail)({ ...base, input: { to: ['x@y.com'], subject: '询价', body: 'b', dryRun: true } })).toEqual({ ok: true, output: { queued: false, dryRun: true } }); + expect((await mailSendTool(mail)({ ...base, input: { to: [], subject: '', dryRun: true } })).ok).toBe(false); + expect(mail.calls).toHaveLength(0); + }); + it('厂商失败 → send-failed 可重试;没账号 → no-account 不重试;入参问题 → invalid', async () => { + const failed = await mailSendTool(fakeSend({ state: 'failed', error: '535' }))({ ...base, input: { to: ['a@x.com'], subject: 's', body: 'b' } }); + expect(failed).toEqual({ ok: false, error: { code: 'send-failed', retryable: true, message: '535' } }); + const { MailNoDefaultAccountError, MailSendInvalidError } = await import('@modules/mail'); + const noAccount = await mailSendTool({ send: async () => { throw new MailNoDefaultAccountError(); } })({ ...base, input: { to: ['a@x.com'], subject: 's', body: 'b' } }); + expect(noAccount.error?.code).toBe('no-account'); + const invalid = await mailSendTool({ send: async () => { throw new MailSendInvalidError(['正文 text / html 至少一个']); } })({ ...base, input: { to: ['a@x.com'], subject: 's' } }); + expect(invalid.error?.code).toBe('invalid'); + }); +}); diff --git a/apps/api/src/app/config.ts b/apps/api/src/app/config.ts index 5eff790..84a7297 100644 --- a/apps/api/src/app/config.ts +++ b/apps/api/src/app/config.ts @@ -23,6 +23,8 @@ export type AppConfig = { readonly attachmentRoot: string; /** 派发器轮询间隔(毫秒) */ readonly dispatchIntervalMs: number; + /** 邮件轮询工作者的 tick(毫秒);每个账号自己的间隔另算 */ + readonly mailPollTickMs: number; /** 三样齐了才有;缺一样即 undefined —— main.ts 据此拒绝启动,集成测试可不配 */ readonly keycloak?: KeycloakConfig; /** 测试账号(Keycloak 用户名):首次建档 / 播种时落 TEST 岗 + 独立工作组 */ @@ -65,6 +67,7 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env): AppConfig { databaseUrl: env['DATABASE_URL'] ?? DEFAULT_DATABASE_URL, attachmentRoot: env['JCAI_ATTACHMENT_ROOT'] ?? join(process.cwd(), '.data', 'attachments'), dispatchIntervalMs: Number(env['JCAI_DISPATCH_INTERVAL_MS'] ?? 2000), + mailPollTickMs: Number(env['JCAI_MAIL_POLL_TICK_MS'] ?? 5000), ...(keycloak === undefined ? {} : { keycloak }), testAccounts: (env['JCAI_TEST_ACCOUNTS'] ?? '') .split(',') diff --git a/apps/api/src/app/container.ts b/apps/api/src/app/container.ts index f7c415d..83ff4eb 100644 --- a/apps/api/src/app/container.ts +++ b/apps/api/src/app/container.ts @@ -74,12 +74,19 @@ import type { TraceService } from '@modules/trace'; import { wireEngines } from './wiring/engines.ts'; import { wireDeskElements } from './wiring/desk-elements.ts'; import type { ChannelService } from '@modules/channel'; +import { createChannelRepository, createChannelService, createFileAttachmentArchive } from '@modules/channel'; +import type { MailService } from '@modules/mail'; import { - createChannelRepository, - createChannelService, - createFileAttachmentArchive, - createSimulatedMailTransport, -} from '@modules/channel'; + createImapSmtpProvider, + createInboundStore, + createMailAccountStore, + createMailService, + createOutboundStore, + createPostalMimeParser, + createWecomProvider, +} from '@modules/mail'; +import { channelSubscriber, channelTransport, mailSendTool } from './mail-wiring.ts'; +import { readUpload, storeUpload } from './upload-store.ts'; import type { AppConfig } from './config.ts'; import { createIdGenerator, systemClock } from './runtime.ts'; @@ -107,6 +114,8 @@ export type Container = { readonly toolInvoker: ToolInvoker; readonly message: MessageService; readonly channel: ChannelService; + /** 邮件收发(ADR-013):账号 / 轮询 / 订阅 / 发信 */ + readonly mail: MailService; readonly humanTool: HumanToolService; readonly interaction: InteractionService; readonly registry: ActivityRegistryService; @@ -172,9 +181,6 @@ const sfeelEvaluator = (expression: string, scope: Readonly(FIRST_PARTY_TOOLS); - /** 集成测试的注入缝:验签与名册来源都是外界,测试给替身,不联网 */ export type ContainerOverrides = { readonly tokenVerifier?: TokenVerifier; @@ -369,9 +375,36 @@ export async function createContainer(config: AppConfig, overrides: ContainerOve ids, }); + /* + * 邮件收发(ADR-013)—— L1,只要库 / 密封 / 时钟;先于工具运行时装配,`mail.send` 的处理器要闭包持它。 + * 变更经 ChangeNotifier 接到广播:轮询器不走 /api/gov/* 中间件,新来信 / 拉取出错得自己说一声。 + * 附件落 sys.upload —— 与上传路由同一段存取(8MB 上限、文本抽取、下载路由一并复用)。 + */ + const mail = createMailService({ + accounts: createMailAccountStore(db), + inbound: createInboundStore(db), + outbound: createOutboundStore(db), + attachments: { + put: async (file) => ({ + ref: (await storeUpload(db, { name: file.name, contentType: file.contentType, bytes: file.bytes, uploadedBy: null, note: file.note })).ref, + }), + get: (ref) => readUpload(db, ref), + }, + providers: { 'imap-smtp': createImapSmtpProvider(), wecom: createWecomProvider() }, + mime: createPostalMimeParser(), + secrets, + clock: systemClock, + ids, + notify: { changed: (reason) => broadcaster.publish({ kind: 'system', id: 'mail', reason }) }, + }); + const toolRegistry = createToolRegistry(db); + /** 第一方工具实现 —— `mail.send` 真发信,处理器在这儿闭包持 mail(放行时经 ToolInvoker 执行) */ // 导入的自定义工具走通用 HTTP 实现:契约申报 endpoint,运行时只把那一个连接目标交出去 - const toolRuntime = createToolRuntime(firstPartyToolHandlers, httpToolHandler); + const toolRuntime = createToolRuntime( + new Map([...FIRST_PARTY_TOOLS, ['mail.send', mailSendTool(mail)]]), + httpToolHandler, + ); const schemas = createSchemaValidator(); const toolbox = createToolboxService({ registry: toolRegistry, @@ -399,11 +432,13 @@ export async function createContainer(config: AppConfig, overrides: ContainerOve recordOutbound: (ref) => event.recordOutbound(ref), }, ledger: { systemWrite: (input) => ledger.systemWrite(input) }, - mail: createSimulatedMailTransport(db, ids), + mail: channelTransport(mail), archive: createFileAttachmentArchive(config.attachmentRoot), clock: systemClock, ids, }); + // channel 是第一个订阅方:来信归一进事件流,活动经触发器 / 监听收到;账号没绑通道的信留在运营面 + mail.subscribe('channel', channelSubscriber(channel)); /* ── L3/L4:交互、人类工具、活动库、放行门 ─────────────────────── */ @@ -775,6 +810,7 @@ export async function createContainer(config: AppConfig, overrides: ContainerOve toolInvoker, message, channel, + mail, humanTool, interaction, registry, diff --git a/apps/api/src/app/mail-wiring.ts b/apps/api/src/app/mail-wiring.ts new file mode 100644 index 0000000..c135221 --- /dev/null +++ b/apps/api/src/app/mail-wiring.ts @@ -0,0 +1,150 @@ +/** + * mail 与 channel / toolbox 的接线(app 层;设计稿 §8;ADR-013): + * ① channel 订阅方:入站 → channel.ingestMail(账号绑的通道)—— 活动经现有事件链收到信 + * ② channel 的 MailTransport:出站 → mail.send(按 from 找账号,对不上不回落默认) + * ③ `mail.send` 工具处理器:活动侧出站,放行时经 ToolInvoker 执行;dryRun 只校验不发(注册试运行会真调处理器) + * 三段都是纯映射 + 一次调用,单测直接对函数。 + */ +import type { InboundMail, MailSendInput } from '@contracts'; +import type { MailTransport } from '@modules/channel'; +import type { InboundHandler, MailInboundDetail, MailService } from '@modules/mail'; +import { + bodyTextOf, + MailAccountDisabledError, + MailAccountNotFoundError, + MailNoDefaultAccountError, + MailSecretMissingError, + MailSendInvalidError, +} from '@modules/mail'; +import type { ToolHandler } from '@modules/toolbox'; + +/** mail 的入站详情 → channel 的 InboundMail:只带存下来的附件;线程 = In-Reply-To ⊃ References 末项 */ +export function toInboundMail(mail: MailInboundDetail): InboundMail { + const threadId = mail.inReplyTo ?? mail.references.at(-1); + return { + messageId: mail.messageId, + ...(threadId === undefined ? {} : { threadId }), + from: mail.from.address, + to: mail.to.map((a) => a.address), + cc: mail.cc.map((a) => a.address), + subject: mail.subject, + body: bodyTextOf({ ...(mail.text === null ? {} : { text: mail.text }), ...(mail.html === null ? {} : { html: mail.html }) }), + receivedAt: mail.receivedAt, + attachments: mail.attachments.flatMap((a) => + a.ref === null ? [] : [{ name: a.name, contentType: a.contentType, size: a.size, ref: a.ref }], + ), + }; +} + +/** 只要 `ingestMail` 这一个口,出参不看(InboundOutcome 由 channel 自己留痕) */ +export function channelSubscriber(channel: { ingestMail(channelId: string, mail: InboundMail): Promise }): InboundHandler { + return async (mail, ctx) => { + if (ctx.channelId === undefined) return { skipped: '账号未绑定通道' }; + await channel.ingestMail(ctx.channelId, toInboundMail(mail)); + // 显式回 undefined —— 投递成功就是「没话说」;`noImplicitReturns` 不许一条路返值一条路掉出去 + return undefined; + }; +} + +/** channel 的出站面:厂商失败抛出 —— channel.sendMail 的调用方(放行落地)要看到失败 */ +export function channelTransport(mail: Pick): MailTransport { + return { + async send(input) { + const result = await mail.send({ + from: input.from, + to: [...input.to], + cc: [...input.cc], + bcc: [], + subject: input.subject, + text: input.body, + ...(input.threadId === undefined ? {} : { inReplyTo: input.threadId }), + references: input.threadId === undefined ? [] : [input.threadId], + attachments: input.attachments.map((a) => ({ name: a.name, contentType: 'application/octet-stream', uploadRef: a.ref })), + requestedBy: { kind: 'channel' }, + }); + if (result.state === 'failed') throw new Error(`发信失败:${result.error ?? '未知'}`); + return { messageId: result.messageId, threadId: input.threadId ?? result.messageId }; + }, + }; +} + +function recipientsOf(value: unknown): string[] { + if (Array.isArray(value)) return value.filter((v): v is string => typeof v === 'string').map((v) => v.trim()).filter((v) => v !== ''); + if (typeof value === 'string') return value.split(/[,;]/).map((s) => s.trim()).filter((s) => s !== ''); + return []; +} + +function attachmentsOf(value: unknown): MailSendInput['attachments'] { + if (!Array.isArray(value)) return []; + return value.flatMap((v) => { + if (typeof v !== 'object' || v === null) return []; + const a = v as Record; + const uploadRef = typeof a['uploadRef'] === 'string' && a['uploadRef'] !== '' ? a['uploadRef'] : undefined; + if (uploadRef === undefined) return []; + const name = typeof a['name'] === 'string' && a['name'] !== '' ? a['name'] : uploadRef; + return [{ name, contentType: 'application/octet-stream', uploadRef }]; + }); +} + +export function mailSendTool(mail: Pick): ToolHandler { + return async ({ input, connection, idempotencyKey }) => { + const to = recipientsOf(input['to']); + const cc = recipientsOf(input['cc']); + const bcc = recipientsOf(input['bcc']); + const subject = typeof input['subject'] === 'string' ? input['subject'] : ''; + const body = typeof input['body'] === 'string' && input['body'] !== '' ? input['body'] : undefined; + const html = typeof input['html'] === 'string' && input['html'] !== '' ? input['html'] : undefined; + const inReplyTo = typeof input['inReplyTo'] === 'string' && input['inReplyTo'] !== '' ? input['inReplyTo'] : undefined; + // 账号:入参指定 > 连接配置('default' = 默认账号) + const configured = connection['account']; + const accountId = + typeof input['accountId'] === 'string' && input['accountId'] !== '' + ? input['accountId'] + : configured !== undefined && configured !== '' && configured !== 'default' + ? configured + : undefined; + + if (input['dryRun'] === true) { + const problems = [ + ...(to.length === 0 ? ['收件人至少一个'] : []), + ...(subject === '' ? ['主题不能为空'] : []), + ...(body === undefined && html === undefined ? ['正文 body / html 至少一个'] : []), + ]; + return problems.length === 0 + ? { ok: true, output: { queued: false, dryRun: true } } + : { ok: false, error: { code: 'invalid', retryable: false, message: problems.join('; ') } }; + } + + try { + const result = await mail.send({ + ...(accountId === undefined ? {} : { accountId }), + to, + cc, + bcc, + subject, + ...(body === undefined ? {} : { text: body }), + ...(html === undefined ? {} : { html }), + ...(inReplyTo === undefined ? {} : { inReplyTo }), + references: inReplyTo === undefined ? [] : [inReplyTo], + attachments: attachmentsOf(input['attachments']), + ...(idempotencyKey === undefined ? {} : { idempotencyKey }), + requestedBy: { kind: 'tool', tool: 'mail.send' }, + }); + if (result.state === 'failed') { + return { ok: false, error: { code: 'send-failed', retryable: true, message: result.error ?? '发信失败' } }; + } + return { ok: true, output: { queued: true, outboundId: result.outboundId, messageId: result.messageId, state: result.state } }; + } catch (error) { + if (error instanceof MailSendInvalidError) return { ok: false, error: { code: 'invalid', retryable: false, message: error.message } }; + if ( + error instanceof MailNoDefaultAccountError || + error instanceof MailAccountNotFoundError || + error instanceof MailAccountDisabledError || + error instanceof MailSecretMissingError + ) { + return { ok: false, error: { code: 'no-account', retryable: false, message: error.message } }; + } + throw error; + } + }; +} diff --git a/apps/api/src/app/main.ts b/apps/api/src/app/main.ts index 27590a2..41ebbf6 100644 --- a/apps/api/src/app/main.ts +++ b/apps/api/src/app/main.ts @@ -10,6 +10,7 @@ import { createServer } from './server.ts'; import { createConsumers } from './workers/consumers.ts'; import { dispatchOnce, startDispatcher } from './workers/dispatcher.ts'; import { advanceOnce, startTaskAdvancer } from './workers/task-advancer.ts'; +import { startMailPoller } from './workers/mail-poller.ts'; import { registerPulse } from './workers/pulse.ts'; export async function boot(): Promise { @@ -38,6 +39,9 @@ export async function boot(): Promise { // 过程在后端:运行中的任务由推进器一拍一拍推,不依赖浏览器在线 const advancer = startTaskAdvancer({ container }, config.dispatchIntervalMs); + // 邮件轮询:腾讯没有带内容的推送,信箱只能拉;跨进程租约让两台后端不重复拉同一个信箱 + const mailPoller = startMailPoller({ container }, config.mailPollTickMs); + /* * 人一答完就踢一脚 —— **顺序有意义**:先把事件投递下去(任务才醒),再推它一拍。 * 反过来登记的话,推进那一轮跑在投递之前,任务还没醒,白跑一轮, @@ -54,6 +58,7 @@ export async function boot(): Promise { const shutdown = async (): Promise => { dispatcher.stop(); advancer.stop(); + mailPoller.stop(); await container.close(); process.exit(0); }; diff --git a/apps/api/src/app/routes/governance-ops.ts b/apps/api/src/app/routes/governance-ops.ts index 0e73a47..7b4f0d3 100644 --- a/apps/api/src/app/routes/governance-ops.ts +++ b/apps/api/src/app/routes/governance-ops.ts @@ -5,7 +5,7 @@ * 「治理动作本身留痕入事件流 —— 治理也被治理。」 */ import { sql } from 'drizzle-orm'; -import { extractText } from '../upload-extract.ts'; +import { readUpload, storeUpload, UPLOAD_CAP } from '../upload-store.ts'; import type { Hono } from 'hono'; import type { ApiError } from '@contracts'; import { @@ -721,9 +721,6 @@ export function registerGovernanceOpsRoutes(app: Hono, container: Contai /* ── 上传件:附件要能真正进活动 ─────────────────────────────────── */ - /** 上限 8MB —— 超了当场拒,不静默截断(截断过的文件比没有更害人) */ - const UPLOAD_CAP = 8 * 1024 * 1024; - app.post(uploadFileRoute.path, async (c) => { const parsed = uploadFileRoute.body.safeParse(await c.req.json()); if (!parsed.success) { @@ -744,37 +741,16 @@ export function registerGovernanceOpsRoutes(app: Hono, container: Contai return c.json(body, 400); } // 读文本就在入口这一步做完:活动那一侧读得懂的只有文字,读不出就如实记 note - const got = extractText(name, contentType, bytes); - const me = staffOf(c); - const ref = crypto.randomUUID(); - await container.db.execute(sql` - insert into sys.upload (id, name, content_type, size, bytes, text, note, uploaded_by) - values (${ref}, ${name}, ${contentType}, ${bytes.length}, ${Buffer.from(bytes)}, - ${got.text ?? null}, ${got.note}, ${me?.id ?? null})`); - return c.json({ - ref, - name, - contentType, - size: bytes.length, - ...(got.text === undefined ? {} : { text: got.text }), - note: got.note, - }); + return c.json(await storeUpload(container.db, { name, contentType, bytes, uploadedBy: staffOf(c)?.id ?? null })); }); app.get(downloadFileRoute.path, async (c) => { - const rows = await container.db.execute<{ name: string; content_type: string; bytes: Buffer }>( - sql`select name, content_type, bytes from sys.upload where id = ${c.req.param('id')}`, - ); - const hit = rowsOf<{ name: string; content_type: string; bytes: Buffer }>(rows)[0]; + const hit = await readUpload(container.db, c.req.param('id')); if (hit === undefined) { const body: ApiError = { code: 'not-found', message: '这个文件不在了' }; return c.json(body, 404); } - return c.json({ - name: hit.name, - contentType: hit.content_type, - dataBase64: Buffer.from(hit.bytes).toString('base64'), - }); + return c.json({ name: hit.name, contentType: hit.contentType, dataBase64: Buffer.from(hit.bytes).toString('base64') }); }); /* ── 试跑日志:历次试跑的存档,按实例管理 ────────────────────────── */ diff --git a/apps/api/src/app/tools/first-party.ts b/apps/api/src/app/tools/first-party.ts index e9f7c92..367da49 100644 --- a/apps/api/src/app/tools/first-party.ts +++ b/apps/api/src/app/tools/first-party.ts @@ -286,15 +286,6 @@ const rankKeyPersons: ToolHandler = async ({ input }) => { return { ok: true, output: { ranked, counts } }; }; -/** 出站发送 —— 写类:过程中登记为动作意向,放行后由系统执行落地(OS 7.3) */ -const mailSend: ToolHandler = async ({ input, connection }) => { - // 运行时只拿得到申报过的连接目标;真发信由 channel 的出站面承担 - return { - ok: true, - output: { queued: true, to: String(input['to'] ?? ''), via: connection['endpoint'] ?? 'simulated' }, - }; -}; - /* ── 活动槽位的占位产出(编排层跑通期专用) ───────────────────────────── * * 六个活动都只有契约面、没有实现。而出口载荷只从**落地事实**推导, @@ -349,7 +340,6 @@ export const FIRST_PARTY_TOOLS: ReadonlyArray = ['quote-pricing.floor-gate', floorGate], ['quote-archive.profile', profile], ['quote-archive.landing-rate', landingRate], - ['mail.send', mailSend], ['slot.placeholder', slotPlaceholder], ['crm-relationship-coverage.recompute-score', recomputeScore], ['crm-relationship-coverage.rank-key-persons', rankKeyPersons], diff --git a/apps/api/src/app/upload-store.ts b/apps/api/src/app/upload-store.ts new file mode 100644 index 0000000..97ab9aa --- /dev/null +++ b/apps/api/src/app/upload-store.ts @@ -0,0 +1,42 @@ +/** + * sys.upload 的存取 —— 上传路由与 mail 的附件注入件同用这一段(设计稿 §8)。 + * 表是 0024 手写的 SQL,不在 Drizzle 里;照旧走 sql``。 + */ +import { sql } from 'drizzle-orm'; +import type { Db } from '@shared/db.ts'; +import { extractText } from './upload-extract.ts'; + +/** 上限 8MB —— 超了当场拒,不静默截断(截断过的文件比没有更害人) */ +export const UPLOAD_CAP = 8 * 1024 * 1024; + +function rowsOf(result: unknown): T[] { + const bag = result as { rows?: T[] } | T[]; + return Array.isArray(bag) ? bag : (bag.rows ?? []); +} + +export type StoredUpload = { ref: string; name: string; contentType: string; size: number; text?: string; note: string }; + +/** 存字节 + 读出来的文本;`note` 给了就拼在抽取说明前面(邮件附件写「邮件附件:主题」) */ +export async function storeUpload( + db: Db, + file: { name: string; contentType: string; bytes: Uint8Array; uploadedBy: string | null; note?: string | undefined }, +): Promise { + const got = extractText(file.name, file.contentType, file.bytes); + const ref = crypto.randomUUID(); + const note = file.note === undefined || file.note === '' ? got.note : `${file.note};${got.note}`; + await db.execute(sql` + insert into sys.upload (id, name, content_type, size, bytes, text, note, uploaded_by) + values (${ref}, ${file.name}, ${file.contentType}, ${file.bytes.length}, ${Buffer.from(file.bytes)}, + ${got.text ?? null}, ${note}, ${file.uploadedBy})`); + return { ref, name: file.name, contentType: file.contentType, size: file.bytes.length, ...(got.text === undefined ? {} : { text: got.text }), note }; +} + +export async function readUpload(db: Db, ref: string): Promise<{ name: string; contentType: string; bytes: Uint8Array } | undefined> { + const rows = await db.execute<{ name: string; content_type: string; bytes: Buffer }>( + sql`select name, content_type, bytes from sys.upload where id = ${ref}`, + ); + const hit = rowsOf<{ name: string; content_type: string; bytes: Buffer }>(rows)[0]; + if (hit === undefined) return undefined; + // pg 的 Buffer 挂在共享内存池上,按元素拷(2026-08-23 的前科) + return { name: hit.name, contentType: hit.content_type, bytes: Uint8Array.from(hit.bytes) }; +} diff --git a/apps/api/src/app/workers/mail-poller.ts b/apps/api/src/app/workers/mail-poller.ts new file mode 100644 index 0000000..6454369 --- /dev/null +++ b/apps/api/src/app/workers/mail-poller.ts @@ -0,0 +1,41 @@ +/** + * 邮件轮询工作者 —— 每 tick 挑到期账号各拉一轮(设计稿 §4);跨进程租约在 store 里, + * 两个后端连同一个库时一个信箱同一时刻只有一个在拉。 + * 与 startDispatcher / startTaskAdvancer 同形;上一 tick 没跑完(账号多、信多)就跳过这一拍,不叠着跑。 + */ +import type { MailPollOutcome } from '@modules/mail'; +import type { Container } from '../container.ts'; + +export type MailPollerDeps = { readonly container: Pick; readonly now?: () => Date }; + +export async function pollMailOnce(deps: MailPollerDeps): Promise { + const outcomes = await deps.container.mail.pollDue((deps.now ?? (() => new Date()))()); + for (const outcome of outcomes) { + if (outcome.error !== undefined) process.stderr.write(`[mail-poller] ${outcome.accountId}:${outcome.error}\n`); + } + return outcomes; +} + +export function startMailPoller(deps: MailPollerDeps, tickMs: number): { stop(): void } { + let stopped = false; + let running = false; + const tick = async (): Promise => { + if (stopped || running) return; + running = true; + try { + await pollMailOnce(deps); + } catch (error) { + process.stderr.write(`[mail-poller] ${error instanceof Error ? error.message : String(error)}\n`); + } finally { + running = false; + } + }; + const timer = setInterval(() => void tick(), tickMs); + timer.unref(); + return { + stop() { + stopped = true; + clearInterval(timer); + }, + }; +}