Archived
mail 用例:账号建改删与探测(草稿探测可借库里密文;wecom 探到地址回填);收发用例同份落地待测
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TM4kUBvekALu9qqmp7D3wr
This commit is contained in:
xcs
@@ -0,0 +1,100 @@
|
||||
/** 账号用例:建 / 改 / 删 / 探测;密文只进库不出接口;名字唯一;secret 缺省保留;kind 变了游标清空 */
|
||||
import { describe, expect, it } from 'vitest';
|
||||
import { mailAccountInputSchema } from '@contracts';
|
||||
import { MailAccountNameTakenError, MailAccountNotFoundError, MailSecretMissingError } from '../contract.ts';
|
||||
import { createMailService } from '../internal/usecase/mail-service.ts';
|
||||
import { createPostalMimeParser } from '../internal/adapter/postal-mime-parser.ts';
|
||||
import { fakeProvider, fixedClock, memoryAccounts, memoryAttachments, memoryInbound, memoryOutbound, plainSecrets, seqIds } from './fakes.ts';
|
||||
|
||||
function harness() {
|
||||
const clock = fixedClock();
|
||||
const accounts = memoryAccounts(clock);
|
||||
const imap = fakeProvider('imap-smtp');
|
||||
const wecom = fakeProvider('wecom');
|
||||
const changes: string[] = [];
|
||||
const service = createMailService({
|
||||
accounts,
|
||||
inbound: memoryInbound(),
|
||||
outbound: memoryOutbound(),
|
||||
attachments: memoryAttachments(),
|
||||
providers: { 'imap-smtp': imap, wecom },
|
||||
mime: createPostalMimeParser(),
|
||||
secrets: plainSecrets,
|
||||
clock,
|
||||
ids: seqIds('acc'),
|
||||
notify: { changed: (r) => void changes.push(r) },
|
||||
});
|
||||
return { service, accounts, imap, wecom, clock, changes };
|
||||
}
|
||||
// 入参是判别联合:`{...imapInput, address}` 若停在联合上,TS 会拿 wecom 那一形来判超额属性(它没有 address)。
|
||||
// 先窄到 imap-smtp 这一形,下面各处展开才成立 —— 窄的是夹具的静态型,值一个字节没变。
|
||||
const parsedImapInput = mailAccountInputSchema.parse({ kind: 'imap-smtp', name: '销售', address: 'sales@we.com', isDefault: true, secret: { password: 'pw-1234' } });
|
||||
if (parsedImapInput.kind !== 'imap-smtp') throw new Error('夹具:imapInput 应是 imap-smtp 那一形');
|
||||
const imapInput = parsedImapInput;
|
||||
|
||||
describe('账号', () => {
|
||||
it('建:视图不带密文、secretBound 真;库里密文是密封过的', async () => {
|
||||
const { service, accounts } = harness();
|
||||
const view = await service.createAccount(imapInput);
|
||||
expect(view.id).toBe('acc-1');
|
||||
expect(view.secretBound).toBe(true);
|
||||
expect(JSON.stringify(view)).not.toContain('pw-1234');
|
||||
expect(accounts.rows.get('acc-1')?.secretCipher).toBe(`sealed:${JSON.stringify({ password: 'pw-1234' })}`);
|
||||
expect((await service.listAccounts()).map((a) => a.name)).toEqual(['销售']);
|
||||
});
|
||||
|
||||
it('名字唯一:重名 409;改名撞别人也 409', async () => {
|
||||
const { service } = harness();
|
||||
await service.createAccount(imapInput);
|
||||
await expect(service.createAccount(imapInput)).rejects.toBeInstanceOf(MailAccountNameTakenError);
|
||||
const second = await service.createAccount({ ...imapInput, name: '采购', address: 'buy@we.com' });
|
||||
await expect(service.updateAccount(second.id, { ...imapInput, name: '销售' })).rejects.toBeInstanceOf(MailAccountNameTakenError);
|
||||
});
|
||||
|
||||
it('改:secret 缺省保留原密文;给了就换;换 kind 没给 secret 时旧密文作废', async () => {
|
||||
const { service, accounts } = harness();
|
||||
const created = await service.createAccount(imapInput);
|
||||
await service.updateAccount(created.id, { ...imapInput, secret: undefined, displayName: '销售一部' });
|
||||
expect(accounts.rows.get(created.id)?.secretCipher).toContain('pw-1234');
|
||||
await service.updateAccount(created.id, { ...imapInput, secret: { password: 'new-pw-9' } });
|
||||
expect(accounts.rows.get(created.id)?.secretCipher).toContain('new-pw-9');
|
||||
const wecomInput = mailAccountInputSchema.parse({ kind: 'wecom', name: '销售', settings: { corpId: 'ww1' } });
|
||||
const switched = await service.updateAccount(created.id, wecomInput);
|
||||
expect(switched.secretBound).toBe(false);
|
||||
expect(switched.address).toBe('');
|
||||
});
|
||||
|
||||
it('默认账号只有一个;删了就没了(404)', async () => {
|
||||
const { service } = harness();
|
||||
const a = await service.createAccount(imapInput);
|
||||
const b = await service.createAccount({ ...imapInput, name: '采购', address: 'buy@we.com', isDefault: true });
|
||||
expect((await service.getAccount(a.id)).isDefault).toBe(false);
|
||||
expect((await service.getAccount(b.id)).isDefault).toBe(true);
|
||||
await service.deleteAccount(a.id);
|
||||
await expect(service.getAccount(a.id)).rejects.toBeInstanceOf(MailAccountNotFoundError);
|
||||
await expect(service.deleteAccount(a.id)).rejects.toBeInstanceOf(MailAccountNotFoundError);
|
||||
});
|
||||
|
||||
it('探测(已存):用库里凭证,报告里的厂商原话脱敏;wecom 探到地址回填', async () => {
|
||||
const { service, wecom, imap } = harness();
|
||||
imap.probeWith({ kind: 'imap-smtp', receive: { ok: false, error: 'LOGIN sales@we.com pw-1234 failed' }, send: { checked: true, ok: true } });
|
||||
const a = await service.createAccount(imapInput);
|
||||
const report = await service.probe({ accountId: a.id });
|
||||
expect(report.receive.error).toBe('LOGIN sales@we.com *** failed');
|
||||
const w = await service.createAccount(mailAccountInputSchema.parse({ kind: 'wecom', name: '应用', settings: { corpId: 'ww1' }, secret: { corpSecret: 'cs-9999' } }));
|
||||
wecom.probeWith({ kind: 'wecom', address: 'app@corp.wecom.work', receive: { ok: true }, send: { checked: false, note: '发信权限只能在真发时验证' } });
|
||||
await service.probe({ accountId: w.id });
|
||||
expect((await service.getAccount(w.id)).address).toBe('app@corp.wecom.work');
|
||||
});
|
||||
|
||||
it('探测(草稿):不落库;没 secret 又没 accountId → 412;给了 accountId 就借库里的密文', async () => {
|
||||
const { service, accounts } = harness();
|
||||
const report = await service.probe({ input: imapInput });
|
||||
expect(report.receive.ok).toBe(true);
|
||||
expect(accounts.rows.size).toBe(0);
|
||||
await expect(service.probe({ input: { ...imapInput, secret: undefined } })).rejects.toBeInstanceOf(MailSecretMissingError);
|
||||
const a = await service.createAccount(imapInput);
|
||||
const borrowed = await service.probe({ input: { ...imapInput, secret: undefined, displayName: '改过' }, accountId: a.id });
|
||||
expect(borrowed.receive.ok).toBe(true);
|
||||
});
|
||||
});
|
||||
@@ -81,3 +81,6 @@ export { cursorReset, mergeAccount, secretOf, viewOf } from './internal/domain/a
|
||||
export { createInboundStore, createMailAccountStore, createOutboundStore } from './internal/adapter/mail-store.ts';
|
||||
export { mailAccountTable, mailInboundDeliveryTable, mailInboundTable, mailOutboundTable } from './internal/adapter/mail-schema.ts';
|
||||
export { createPostalMimeParser } from './internal/adapter/postal-mime-parser.ts';
|
||||
|
||||
export type { MailServiceDeps } from './internal/usecase/mail-service.ts';
|
||||
export { createMailService } from './internal/usecase/mail-service.ts';
|
||||
|
||||
@@ -0,0 +1,497 @@
|
||||
/**
|
||||
* mail 用例 —— 账号、收、发(编码规范 §5.3;设计稿 §5.4;ADR-013)。
|
||||
*
|
||||
* 收:抢租约 → 无游标先锚 → 循环 pull(封顶 50 封 / 20s)→ 逐封解析 / 存附件 / 入库(唯一键去重)
|
||||
* → 逐封推进游标 → 新入库的分发给订阅方 → 还租约。任何异常:记 last_poll_error(脱敏),游标停在最后一封成功处。
|
||||
* 发:选账号 → 幂等键 → 校验 → 落 sending → provider → sent / failed(原话脱敏);厂商失败不抛。
|
||||
*/
|
||||
import type { MailAccountKind, MailPollOutcome, MailProbeReport, MailSendResult } from '@contracts';
|
||||
import type { InboundHandler, MailService } from '../../contract.ts';
|
||||
import { mergeAccount, secretOf, viewOf } from '../domain/account-view.ts';
|
||||
import { canRedeliver, stateOfOutcome } from '../domain/delivery.ts';
|
||||
import { leaseMsFor } from '../domain/due.ts';
|
||||
import {
|
||||
MailAccountDisabledError,
|
||||
MailAccountNameTakenError,
|
||||
MailAccountNotFoundError,
|
||||
MailAttachmentNotFoundError,
|
||||
MailInboundNotFoundError,
|
||||
MailPollBusyError,
|
||||
MailSecretMissingError,
|
||||
MailSendInvalidError,
|
||||
MailSubscriberTakenError,
|
||||
MailSubscriberUnknownError,
|
||||
} from '../domain/errors.ts';
|
||||
import { ATTACHMENT_CAP, buildInboundRecord, deliveryViewOf, detailOf, planAttachments, RAW_CAP, summaryOf } from '../domain/inbound-record.ts';
|
||||
import type { DeliveryRecord, InboundAttachmentRecord, InboundRecord, MailAccountRecord, OutboundRecord, ParsedAttachment } from '../domain/records.ts';
|
||||
import { errorText, redactSecrets } from '../domain/redact.ts';
|
||||
import { messageIdFor, pickAccount, validateSend } from '../domain/send-input.ts';
|
||||
import type {
|
||||
AttachmentStore,
|
||||
ChangeNotifier,
|
||||
Clock,
|
||||
IdGenerator,
|
||||
InboundStore,
|
||||
MailAccountStore,
|
||||
MailProvider,
|
||||
MimeParser,
|
||||
OutboundStore,
|
||||
ProviderAccount,
|
||||
RawMessage,
|
||||
SecretBox,
|
||||
} from '../../port.ts';
|
||||
|
||||
export type MailServiceDeps = {
|
||||
readonly accounts: MailAccountStore;
|
||||
readonly inbound: InboundStore;
|
||||
readonly outbound: OutboundStore;
|
||||
readonly attachments: AttachmentStore;
|
||||
readonly providers: Readonly<Record<MailAccountKind, MailProvider>>;
|
||||
readonly mime: MimeParser;
|
||||
readonly secrets: SecretBox;
|
||||
readonly clock: Clock;
|
||||
readonly ids: IdGenerator;
|
||||
readonly notify: ChangeNotifier;
|
||||
readonly limits?:
|
||||
| {
|
||||
readonly perTick?: number | undefined;
|
||||
readonly tickBudgetMs?: number | undefined;
|
||||
readonly attachmentBytes?: number | undefined;
|
||||
readonly rawBytes?: number | undefined;
|
||||
}
|
||||
| undefined;
|
||||
};
|
||||
|
||||
const DEFAULT_LIST_LIMIT = 50;
|
||||
const MAX_LIST_LIMIT = 200;
|
||||
|
||||
type PollTally = { accountId: string; pulled: number; stored: number; dispatched: number; more: boolean; error?: string };
|
||||
|
||||
function tallyToOutcome(t: PollTally): MailPollOutcome {
|
||||
return {
|
||||
accountId: t.accountId,
|
||||
pulled: t.pulled,
|
||||
stored: t.stored,
|
||||
dispatched: t.dispatched,
|
||||
more: t.more,
|
||||
...(t.error === undefined ? {} : { error: t.error }),
|
||||
};
|
||||
}
|
||||
|
||||
function safeFileName(messageId: string): string {
|
||||
const cleaned = messageId.replace(/[^A-Za-z0-9._-]+/g, '_').replace(/^_+|_+$/g, '');
|
||||
return cleaned === '' ? 'mail' : cleaned;
|
||||
}
|
||||
|
||||
function resultOf(record: OutboundRecord): MailSendResult {
|
||||
return {
|
||||
outboundId: record.id,
|
||||
accountId: record.accountId,
|
||||
messageId: record.messageId,
|
||||
state: record.state === 'sent' ? 'sent' : 'failed',
|
||||
...(record.error === null ? {} : { error: record.error }),
|
||||
...(record.providerRef === null ? {} : { providerRef: record.providerRef }),
|
||||
};
|
||||
}
|
||||
|
||||
export function createMailService(deps: MailServiceDeps): MailService {
|
||||
const limits = {
|
||||
perTick: deps.limits?.perTick ?? 50,
|
||||
tickBudgetMs: deps.limits?.tickBudgetMs ?? 20_000,
|
||||
attachmentBytes: deps.limits?.attachmentBytes ?? ATTACHMENT_CAP,
|
||||
rawBytes: deps.limits?.rawBytes ?? RAW_CAP,
|
||||
};
|
||||
const subscribers = new Map<string, InboundHandler>();
|
||||
|
||||
async function requireAccount(accountId: string): Promise<MailAccountRecord> {
|
||||
const record = await deps.accounts.get(accountId);
|
||||
if (record === undefined) throw new MailAccountNotFoundError(accountId);
|
||||
return record;
|
||||
}
|
||||
|
||||
/** 明文只在这里打开、交给 provider 那一段;不进任何留痕 */
|
||||
function openSecret(record: Pick<MailAccountRecord, 'id' | 'secretCipher'>): Record<string, string> {
|
||||
if (record.secretCipher === null || record.secretCipher === '') throw new MailSecretMissingError(record.id);
|
||||
const parsed: unknown = JSON.parse(deps.secrets.open(record.secretCipher));
|
||||
if (typeof parsed !== 'object' || parsed === null) throw new MailSecretMissingError(record.id);
|
||||
return Object.fromEntries(Object.entries(parsed as Record<string, unknown>).map(([k, v]) => [k, String(v)]));
|
||||
}
|
||||
|
||||
function providerAccount(record: MailAccountRecord, secret: Record<string, string>): ProviderAccount {
|
||||
return { id: record.id, kind: record.kind, address: record.address, displayName: record.displayName, settings: record.settings, secret };
|
||||
}
|
||||
|
||||
function redactWith(secret: Record<string, string>): (text: string) => string {
|
||||
const values = Object.values(secret);
|
||||
return (text) => redactSecrets(text, values);
|
||||
}
|
||||
|
||||
function redactReport(report: MailProbeReport, secret: Record<string, string>): MailProbeReport {
|
||||
const redact = redactWith(secret);
|
||||
return {
|
||||
...report,
|
||||
receive: { ...report.receive, ...(report.receive.error === undefined ? {} : { error: redact(report.receive.error) }) },
|
||||
send: { ...report.send, ...(report.send.error === undefined ? {} : { error: redact(report.send.error) }) },
|
||||
};
|
||||
}
|
||||
|
||||
function limitOf(query: { limit?: number | undefined }): number {
|
||||
return Math.min(MAX_LIST_LIMIT, Math.max(1, query.limit ?? DEFAULT_LIST_LIMIT));
|
||||
}
|
||||
|
||||
/* ── 分发 ────────────────────────────────────────────────────────── */
|
||||
|
||||
async function dispatch(record: InboundRecord, ctx: { accountId: string; channelId: string | null }, only?: string): Promise<DeliveryRecord[]> {
|
||||
const existing = (await deps.inbound.get(record.id))?.deliveries ?? [];
|
||||
const out: DeliveryRecord[] = [];
|
||||
for (const [name, handler] of subscribers) {
|
||||
if (only !== undefined && name !== only) continue;
|
||||
const current = existing.find((d) => d.subscriber === name);
|
||||
// 正在投的那条不重投:进程还没回话,再投一遍就是两条线同时改一件事
|
||||
if (!canRedeliver(current?.state)) {
|
||||
if (current !== undefined) out.push(current);
|
||||
continue;
|
||||
}
|
||||
const startedAt = deps.clock.now();
|
||||
const dispatching: DeliveryRecord = { mailId: record.id, subscriber: name, state: 'dispatching', error: null, startedAt, finishedAt: null };
|
||||
await deps.inbound.setDelivery(dispatching);
|
||||
let settled: DeliveryRecord;
|
||||
try {
|
||||
const outcome = await handler(detailOf(record, existing), { accountId: ctx.accountId, channelId: ctx.channelId ?? undefined });
|
||||
const state = stateOfOutcome(outcome);
|
||||
settled = {
|
||||
...dispatching,
|
||||
state,
|
||||
// skipped 的理由也放 error 列:人在列表上看得见「为什么没投」
|
||||
error: state === 'skipped' && outcome !== undefined && typeof outcome === 'object' ? outcome.skipped : null,
|
||||
finishedAt: deps.clock.now(),
|
||||
};
|
||||
} catch (error) {
|
||||
settled = { ...dispatching, state: 'failed', error: errorText(error), finishedAt: deps.clock.now() };
|
||||
}
|
||||
await deps.inbound.setDelivery(settled);
|
||||
out.push(settled);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/* ── 拉取 ────────────────────────────────────────────────────────── */
|
||||
|
||||
async function storeOne(record: MailAccountRecord, message: RawMessage): Promise<InboundRecord | undefined> {
|
||||
// 先按 provider 给的引用查重,免得重复拉到的信把附件再存一遍
|
||||
const known = await deps.inbound.knownProviderRefs(record.id, [message.providerRef]);
|
||||
if (known.has(message.providerRef)) return undefined;
|
||||
const parsed = await deps.mime.parse(message.eml);
|
||||
const attachments: InboundAttachmentRecord[] = [];
|
||||
for (const plan of planAttachments(parsed.attachments, limits.attachmentBytes)) {
|
||||
let ref: string | null = null;
|
||||
if (plan.store) {
|
||||
const bytes = parsed.attachments[plan.index]?.bytes ?? new Uint8Array();
|
||||
ref = (await deps.attachments.put({ name: plan.name, contentType: plan.contentType, bytes, note: `邮件附件:${parsed.subject}` })).ref;
|
||||
}
|
||||
attachments.push({ index: plan.index, name: plan.name, contentType: plan.contentType, size: plan.size, ref, note: plan.note });
|
||||
}
|
||||
const built = buildInboundRecord({
|
||||
id: deps.ids.next(),
|
||||
accountId: record.id,
|
||||
address: record.address,
|
||||
providerRef: message.providerRef,
|
||||
parsed,
|
||||
eml: message.eml,
|
||||
receivedAt: message.receivedAt,
|
||||
attachments,
|
||||
now: deps.clock.now(),
|
||||
rawCap: limits.rawBytes,
|
||||
});
|
||||
const { inserted } = await deps.inbound.insert(built);
|
||||
return inserted ? built : undefined;
|
||||
}
|
||||
|
||||
async function pollAccount(record: MailAccountRecord): Promise<MailPollOutcome> {
|
||||
const claimed = await deps.accounts.claimPolling(record.id, leaseMsFor(record.pollIntervalSec));
|
||||
if (!claimed) throw new MailPollBusyError(record.id);
|
||||
const tally: PollTally = { accountId: record.id, pulled: 0, stored: 0, dispatched: 0, more: false };
|
||||
let secret: Record<string, string> = {};
|
||||
try {
|
||||
secret = openSecret(record);
|
||||
const provider = deps.providers[record.kind];
|
||||
const account = providerAccount(record, secret);
|
||||
const started = deps.clock.now();
|
||||
let cursor = record.cursor;
|
||||
if (cursor === null) {
|
||||
cursor = await provider.anchor(account, record.backfillFrom, started);
|
||||
await deps.accounts.savePoll(record.id, { cursor });
|
||||
}
|
||||
let budget = limits.perTick;
|
||||
let more = true;
|
||||
while (more && budget > 0 && deps.clock.now().getTime() - started.getTime() < limits.tickBudgetMs) {
|
||||
const batch = await provider.pull(account, cursor, {
|
||||
limit: budget,
|
||||
now: deps.clock.now(),
|
||||
backfillFrom: record.backfillFrom,
|
||||
known: (refs) => deps.inbound.knownProviderRefs(record.id, refs),
|
||||
});
|
||||
tally.pulled += batch.messages.length;
|
||||
budget -= batch.messages.length;
|
||||
for (const message of batch.messages) {
|
||||
const stored = await storeOne(record, message);
|
||||
// 逐封推进:中途栽了,下一轮从这一封之后接着
|
||||
if (message.cursorAfter !== undefined) {
|
||||
cursor = message.cursorAfter;
|
||||
await deps.accounts.savePoll(record.id, { cursor });
|
||||
}
|
||||
if (stored !== undefined) {
|
||||
tally.stored += 1;
|
||||
await deps.accounts.savePoll(record.id, { lastInboundAt: stored.receivedAt });
|
||||
await dispatch(stored, { accountId: record.id, channelId: record.channelId });
|
||||
tally.dispatched += 1;
|
||||
}
|
||||
}
|
||||
cursor = batch.cursor;
|
||||
await deps.accounts.savePoll(record.id, { cursor });
|
||||
more = batch.more;
|
||||
if (batch.messages.length === 0) break;
|
||||
}
|
||||
tally.more = more;
|
||||
await deps.accounts.savePoll(record.id, { polledAt: deps.clock.now(), error: null });
|
||||
} catch (error) {
|
||||
const text = redactWith(secret)(errorText(error));
|
||||
tally.error = text;
|
||||
await deps.accounts.savePoll(record.id, { polledAt: deps.clock.now(), error: text });
|
||||
} finally {
|
||||
await deps.accounts.releasePolling(record.id);
|
||||
}
|
||||
if (tally.stored > 0 || tally.error !== undefined) deps.notify.changed('inbound');
|
||||
return tallyToOutcome(tally);
|
||||
}
|
||||
|
||||
/* ── 服务 ────────────────────────────────────────────────────────── */
|
||||
|
||||
return {
|
||||
async listAccounts() {
|
||||
const now = deps.clock.now();
|
||||
return (await deps.accounts.list()).map((r) => viewOf(r, now));
|
||||
},
|
||||
|
||||
async getAccount(accountId) {
|
||||
return viewOf(await requireAccount(accountId), deps.clock.now());
|
||||
},
|
||||
|
||||
async createAccount(input) {
|
||||
if ((await deps.accounts.findByName(input.name)) !== undefined) throw new MailAccountNameTakenError(input.name);
|
||||
const secretJson = secretOf(input);
|
||||
const now = deps.clock.now();
|
||||
const record = mergeAccount(input, {
|
||||
id: deps.ids.next(),
|
||||
existing: undefined,
|
||||
secretCipher: secretJson === undefined ? null : deps.secrets.seal(secretJson),
|
||||
now,
|
||||
});
|
||||
await deps.accounts.save(record);
|
||||
return viewOf(record, now);
|
||||
},
|
||||
|
||||
async updateAccount(accountId, input) {
|
||||
const existing = await requireAccount(accountId);
|
||||
const clash = await deps.accounts.findByName(input.name);
|
||||
if (clash !== undefined && clash.id !== accountId) throw new MailAccountNameTakenError(input.name);
|
||||
const secretJson = secretOf(input);
|
||||
// 只写不读:secret 留空即保持原密文;换了 kind 旧密文不是这一形的,作废
|
||||
const secretCipher =
|
||||
secretJson !== undefined ? deps.secrets.seal(secretJson) : input.kind === existing.kind ? existing.secretCipher : null;
|
||||
const now = deps.clock.now();
|
||||
// kind / settings 变了游标清空重锚 —— 判定在 mergeAccount 里(cursorReset)
|
||||
const record = mergeAccount(input, { id: accountId, existing, secretCipher, now });
|
||||
await deps.accounts.save(record);
|
||||
return viewOf(record, now);
|
||||
},
|
||||
|
||||
async deleteAccount(accountId) {
|
||||
await requireAccount(accountId);
|
||||
await deps.accounts.remove(accountId);
|
||||
},
|
||||
|
||||
async probe(target) {
|
||||
if ('input' in target) {
|
||||
const stored = target.accountId === undefined ? undefined : await requireAccount(target.accountId);
|
||||
const secretJson = secretOf(target.input);
|
||||
const secret =
|
||||
secretJson !== undefined
|
||||
? (JSON.parse(secretJson) as Record<string, string>)
|
||||
: stored !== undefined && stored.kind === target.input.kind
|
||||
? openSecret(stored)
|
||||
: (() => {
|
||||
throw new MailSecretMissingError(target.accountId ?? 'draft');
|
||||
})();
|
||||
const draft = mergeAccount(target.input, { id: stored?.id ?? 'draft', existing: stored, secretCipher: null, now: deps.clock.now() });
|
||||
const report = await deps.providers[target.input.kind].probe(providerAccount(draft, secret));
|
||||
return redactReport(report, secret);
|
||||
}
|
||||
const record = await requireAccount(target.accountId);
|
||||
const secret = openSecret(record);
|
||||
const report = redactReport(await deps.providers[record.kind].probe(providerAccount(record, secret)), secret);
|
||||
// wecom 的地址由探测回填;imap-smtp 的地址是人填的,不改
|
||||
if (record.kind === 'wecom' && report.address !== undefined && report.address !== record.address) {
|
||||
await deps.accounts.savePoll(record.id, { address: report.address });
|
||||
}
|
||||
return report;
|
||||
},
|
||||
|
||||
async pollDue(now) {
|
||||
const due = await deps.accounts.listDue(now);
|
||||
// 账号间并行,单账号内串行;一个账号的异常只记在它自己头上
|
||||
const settled = await Promise.allSettled(due.map((record) => pollAccount(record)));
|
||||
return settled.map((r, i) =>
|
||||
r.status === 'fulfilled'
|
||||
? r.value
|
||||
: { accountId: due[i]?.id ?? '?', pulled: 0, stored: 0, dispatched: 0, more: false, error: errorText(r.reason) },
|
||||
);
|
||||
},
|
||||
|
||||
async pollNow(accountId) {
|
||||
const record = await requireAccount(accountId);
|
||||
if (!record.enabled) throw new MailAccountDisabledError(accountId);
|
||||
return pollAccount(record);
|
||||
},
|
||||
|
||||
subscribe(name, handler) {
|
||||
if (subscribers.has(name)) throw new MailSubscriberTakenError(name);
|
||||
subscribers.set(name, handler);
|
||||
return () => {
|
||||
subscribers.delete(name);
|
||||
};
|
||||
},
|
||||
|
||||
async listInbound(query) {
|
||||
const rows = await deps.inbound.list({ accountId: query.accountId, limit: limitOf(query) });
|
||||
return rows.map((r) => summaryOf(r.record, r.deliveries));
|
||||
},
|
||||
|
||||
async getInbound(mailId) {
|
||||
const hit = await deps.inbound.get(mailId);
|
||||
if (hit === undefined) throw new MailInboundNotFoundError(mailId);
|
||||
return detailOf(hit.record, hit.deliveries);
|
||||
},
|
||||
|
||||
async attachment(mailId, index) {
|
||||
const hit = await deps.inbound.get(mailId);
|
||||
if (hit === undefined) throw new MailInboundNotFoundError(mailId);
|
||||
const meta = hit.record.attachments.find((a) => a.index === index);
|
||||
if (meta === undefined || meta.ref === null) throw new MailAttachmentNotFoundError(mailId, index);
|
||||
const file = await deps.attachments.get(meta.ref);
|
||||
if (file === undefined) throw new MailAttachmentNotFoundError(mailId, index);
|
||||
return { name: meta.name, contentType: meta.contentType, bytes: file.bytes };
|
||||
},
|
||||
|
||||
async raw(mailId) {
|
||||
const hit = await deps.inbound.get(mailId);
|
||||
if (hit === undefined) throw new MailInboundNotFoundError(mailId);
|
||||
if (hit.record.raw === null) throw new MailAttachmentNotFoundError(mailId, -1);
|
||||
return { name: `${safeFileName(hit.record.messageId)}.eml`, bytes: hit.record.raw };
|
||||
},
|
||||
|
||||
async redeliver(mailId, subscriber) {
|
||||
const hit = await deps.inbound.get(mailId);
|
||||
if (hit === undefined) throw new MailInboundNotFoundError(mailId);
|
||||
if (subscriber !== undefined && !subscribers.has(subscriber)) throw new MailSubscriberUnknownError(subscriber);
|
||||
const owner = await deps.accounts.get(hit.record.accountId);
|
||||
const deliveries = await dispatch(hit.record, { accountId: hit.record.accountId, channelId: owner?.channelId ?? null }, subscriber);
|
||||
deps.notify.changed('inbound');
|
||||
return deliveries.map(deliveryViewOf);
|
||||
},
|
||||
|
||||
async send(input) {
|
||||
const account = pickAccount(await deps.accounts.list(), { accountId: input.accountId, from: input.from });
|
||||
// 幂等键:放行重放不发两封;上一次失败 / 中断的那一行复用,重发
|
||||
const prior = input.idempotencyKey === undefined ? undefined : await deps.outbound.byIdempotencyKey(input.idempotencyKey);
|
||||
if (prior !== undefined && prior.state === 'sent') return resultOf(prior);
|
||||
|
||||
const files: ParsedAttachment[] = [];
|
||||
for (const a of input.attachments) {
|
||||
if (a.dataBase64 !== undefined) {
|
||||
files.push({ name: a.name, contentType: a.contentType, bytes: new Uint8Array(Buffer.from(a.dataBase64, 'base64')) });
|
||||
} else if (a.uploadRef !== undefined) {
|
||||
const got = await deps.attachments.get(a.uploadRef);
|
||||
if (got === undefined) throw new MailSendInvalidError([`附件不存在:${a.uploadRef}`]);
|
||||
files.push({ name: a.name, contentType: a.contentType === 'application/octet-stream' ? got.contentType : a.contentType, bytes: got.bytes });
|
||||
} else {
|
||||
throw new MailSendInvalidError([`附件 ${a.name} 既没给字节也没给 uploadRef`]);
|
||||
}
|
||||
}
|
||||
const problems = validateSend(input, account.kind, files.map((f) => f.bytes.byteLength));
|
||||
if (problems.length > 0) throw new MailSendInvalidError(problems);
|
||||
const secret = openSecret(account);
|
||||
|
||||
const now = deps.clock.now();
|
||||
const id = prior?.id ?? deps.ids.next();
|
||||
const messageId = prior?.messageId ?? messageIdFor(id, account.address);
|
||||
if (prior === undefined) {
|
||||
const record: OutboundRecord = {
|
||||
id,
|
||||
accountId: account.id,
|
||||
messageId,
|
||||
idempotencyKey: input.idempotencyKey ?? null,
|
||||
to: [...input.to],
|
||||
cc: [...input.cc],
|
||||
bcc: [...input.bcc],
|
||||
subject: input.subject,
|
||||
text: input.text ?? null,
|
||||
html: input.html ?? null,
|
||||
inReplyTo: input.inReplyTo ?? null,
|
||||
references: [...input.references],
|
||||
attachments: files.map((f, i) => ({ name: f.name, contentType: f.contentType, size: f.bytes.byteLength, ref: input.attachments[i]?.uploadRef ?? null })),
|
||||
state: 'sending',
|
||||
error: null,
|
||||
providerRef: null,
|
||||
requestedBy: { ...input.requestedBy },
|
||||
createdAt: now,
|
||||
sentAt: null,
|
||||
};
|
||||
await deps.outbound.insert(record);
|
||||
}
|
||||
|
||||
try {
|
||||
const sent = await deps.providers[account.kind].send(providerAccount(account, secret), {
|
||||
messageId,
|
||||
from: { address: account.address, name: account.displayName },
|
||||
to: input.to,
|
||||
cc: input.cc,
|
||||
bcc: input.bcc,
|
||||
subject: input.subject,
|
||||
...(input.text === undefined ? {} : { text: input.text }),
|
||||
...(input.html === undefined ? {} : { html: input.html }),
|
||||
...(input.inReplyTo === undefined ? {} : { inReplyTo: input.inReplyTo }),
|
||||
references: input.references,
|
||||
attachments: files,
|
||||
});
|
||||
await deps.outbound.setState(id, 'sent', { sentAt: deps.clock.now(), ...(sent.providerRef === undefined ? {} : { providerRef: sent.providerRef }) });
|
||||
deps.notify.changed('outbound');
|
||||
return { outboundId: id, accountId: account.id, messageId, state: 'sent', ...(sent.providerRef === undefined ? {} : { providerRef: sent.providerRef }) };
|
||||
} catch (error) {
|
||||
const text = redactWith(secret)(errorText(error));
|
||||
await deps.outbound.setState(id, 'failed', { error: text });
|
||||
deps.notify.changed('outbound');
|
||||
return { outboundId: id, accountId: account.id, messageId, state: 'failed', error: text };
|
||||
}
|
||||
},
|
||||
|
||||
async listOutbound(query) {
|
||||
const rows = await deps.outbound.list({ accountId: query.accountId, limit: limitOf(query) });
|
||||
return rows.map((r) => ({
|
||||
id: r.id,
|
||||
accountId: r.accountId,
|
||||
messageId: r.messageId,
|
||||
to: [...r.to],
|
||||
cc: [...r.cc],
|
||||
bcc: [...r.bcc],
|
||||
subject: r.subject,
|
||||
state: r.state,
|
||||
error: r.error,
|
||||
providerRef: r.providerRef,
|
||||
requestedBy: r.requestedBy,
|
||||
createdAt: r.createdAt.toISOString(),
|
||||
sentAt: r.sentAt === null ? null : r.sentAt.toISOString(),
|
||||
}));
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user