/** * trigger 用例 —— 判定与起单(编码规范 §5.3;语义出处 OS 3.1)。 */ import type { TriggerConfig } from '@contracts'; import type { EventScope, FireOutcome, TriggerService } from '../../contract.ts'; import type { Clock, ConfirmDesk, DeadLetterSink, IdGenerator, LedgerScanner, ProcessGateway, Sfeel, TriggerStore } from '../../port.ts'; import { applyOffset, assembleInput, dataCalendarKey, matchesEvent, nextCalendarFire, scopeOf, triggerIdempotencyKey, } from '../domain/matching.ts'; export type TriggerServiceDeps = { readonly store: TriggerStore; readonly process: ProcessGateway; readonly sfeel: Sfeel; readonly deadLetters: DeadLetterSink; readonly confirmDesk: ConfirmDesk; readonly ledger: LedgerScanner; readonly clock: Clock; readonly ids: IdGenerator; }; export function createTriggerService(deps: TriggerServiceDeps): TriggerService { /** 装配 → 起单;缺必填即入死信,不静默丢弃 */ async function fire( config: TriggerConfig, source: { kind: 'periodic' | 'event' | 'registration'; ref: string; by?: string }, scope: Record, subject: string, ): Promise { const required = await deps.process.requiredInputs(config.processPackageId); const assembled = assembleInput(config, scope, required, deps.sfeel.expression); if (assembled.kind === 'missing') { await deps.deadLetters.report({ triggerId: config.id, reason: `起单失败:缺必填字段 ${assembled.fields.join('、')}`, detail: { scope }, }); return { triggerId: config.id, skipped: 'input-missing', missing: assembled.fields }; } /* * 触发确认链路(OS 用户口述五步的第 4/5 步,2026-07-30): * 勾了「需人工确认」,命中**不等于起单** —— 装配好的入参存成待确认单, * 带来件内容的系统消息送到指定人工作台;他点「确认起单」流程才起。 * 登记触发不走这条:登记那一下本身就是人在确认。 */ if (config.confirm.required && source.kind !== 'registration') { const staffId = config.confirm.staffId; if (staffId === undefined) { // 勾了要确认却没指定谁 —— 这是配置错误,如实入死信,不静默丢、也不擅自直起 await deps.deadLetters.report({ triggerId: config.id, reason: '勾了「需人工确认」但没有指定确认人 —— 去运营中心 · 工作流补上', detail: { subject }, }); return { triggerId: config.id, skipped: 'no-confirmer' }; } const pendingId = deps.ids.next(); // 来件预览只收字符串字段 —— 消息上带内容,确认的人不用回头翻邮箱 const preview = Object.fromEntries( (['from', 'subject', 'body'] as const).flatMap((k) => { const v = scope[k]; return typeof v === 'string' ? [[k, v.slice(0, 2000)]] : []; }), ); await deps.store.createPending({ id: pendingId, triggerId: config.id, packageId: config.processPackageId, staffId, subject, data: assembled.data, source, preview, createdAt: deps.clock.now().toISOString(), }); await deps.confirmDesk.request({ pendingId, staffId, triggerName: config.name, subject, preview }); return { triggerId: config.id, pendingId }; } const workflow = await deps.process.start({ packageId: config.processPackageId, subject, source, data: assembled.data, }); await deps.store.markFired(config.id, deps.clock.now()); return { triggerId: config.id, workflowId: workflow.id }; } return { async list() { return deps.store.list(); }, async upsert(config) { return deps.store.upsert(config); }, async setEnabled(id, enabled) { await deps.store.setEnabled(id, enabled); }, async remove(id) { await deps.store.remove(id); }, /** 供 event 的 TriggerMatcher 端口:同一事件流,不另设通路 */ async match(scope: EventScope) { const all = await deps.store.list(); return all.filter((c) => matchesEvent(c, scope, deps.sfeel.condition)).map((c) => c.id); }, async fireByEvent(triggerId, eventId, scope) { const config = await deps.store.get(triggerId); if (config === undefined || !config.enabled) return { triggerId, skipped: 'disabled' }; // 同一事件对同一触发器至多起一单 if (!(await deps.store.claim(triggerIdempotencyKey(triggerId, eventId)))) { return { triggerId, skipped: 'duplicate' }; } return fire(config, { kind: 'event', ref: eventId }, scopeOf(scope), config.name); }, /** 登记触发:登记人即来源 */ async fireByRegistration(triggerId, by, data) { const config = await deps.store.get(triggerId); if (config === undefined || !config.enabled) return { triggerId, skipped: 'disabled' }; /* * 主题优先取登记人填的 `subject` —— 拿触发器名当主题的话, * 同一个登记入口起的所有单主题全一样,清单上分不出哪条是哪条。 */ const filled = data['subject']; const subject = typeof filled === 'string' && filled.trim() !== '' ? filled.trim() : config.name; return fire(config, { kind: 'registration', ref: `registration:${by}`, by }, data, subject); }, /** 周期扫描:日历型按表达式到点;数据日历型逐行起单 */ async sweepPeriodic(now) { const outcomes: FireOutcome[] = []; for (const config of await deps.store.list()) { if (!config.enabled) continue; if (config.spec.kind === 'periodic-calendar') { const last = (await deps.store.lastFiredAt(config.id)) ?? new Date(now.getTime() - 86_400_000); const due = nextCalendarFire(config.spec.calendar, last); if (due === undefined || due.getTime() > now.getTime()) continue; if (!(await deps.store.claim(`${config.id}:${due.toISOString()}`))) continue; outcomes.push(await fire(config, { kind: 'periodic', ref: due.toISOString() }, { firedAt: due.toISOString() }, config.name)); continue; } if (config.spec.kind !== 'periodic-data') continue; // 数据日历型:日期字段 + 偏移 + 过滤条件 const spec = config.spec; const rows = await deps.ledger.scan({ domain: spec.domain, object: spec.object, dateField: spec.dateField, ...(spec.filter === undefined ? {} : { filter: spec.filter }), }); for (const row of rows) { const due = applyOffset(new Date(row.date), spec.offset); if (due.getTime() > now.getTime()) continue; const rowKey = Object.entries(row.key) .sort(([a], [b]) => (a < b ? -1 : 1)) .map(([k, v]) => `${k}=${String(v)}`) .join('|'); // 同一行同一日期只触发一次;字段值变更后可再触发 if (!(await deps.store.claim(dataCalendarKey(config.id, rowKey, row.date)))) continue; outcomes.push( await fire(config, { kind: 'periodic', ref: `${rowKey}@${row.date}` }, { ...row.row, ...row.key }, config.name), ); } } return outcomes; }, async setConfirm(id, confirm) { await deps.store.setConfirm(id, confirm); }, async confirmPending(pendingId, by) { const pending = await deps.store.getPending(pendingId); if (pending === undefined) throw new Error(`无此待确认单:${pendingId}`); if (pending.consumedAt !== undefined) { if (pending.workflowId === undefined) throw new Error('这一票正在被确认 —— 稍等两秒再看'); return { workflowId: pending.workflowId, alreadyStarted: true }; } /* * 先认领后起单:consumed_at 空着才认得到,两人同时点确认只有一个赢 —— * 绝不起两条流。起单失败就把认领退回去,这一票还能再确认。 */ const won = await deps.store.claimPending(pendingId, deps.clock.now()); if (!won) { const again = await deps.store.getPending(pendingId); if (again?.workflowId === undefined) throw new Error('这一票正在被确认 —— 稍等两秒再看'); return { workflowId: again.workflowId, alreadyStarted: true }; } try { const workflow = await deps.process.start({ packageId: pending.packageId, subject: pending.subject, // 确认人记进来源:这条流是谁点头放行的,留痕上要看得到 source: { ...pending.source, by }, data: pending.data, }); await deps.store.setPendingWorkflow(pendingId, workflow.id); await deps.store.markFired(pending.triggerId, deps.clock.now()); return { workflowId: workflow.id, alreadyStarted: false }; } catch (error) { await deps.store.releasePending(pendingId); throw error; } }, async pendingByIds(ids) { return deps.store.pendingByIds(ids); }, }; }