forked from ChaiHao/BPOS-MVP
M5(3/3): trigger —— 三类触发器与起单幂等
- matchesEvent:事件触发的命中判定(S-FEEL 规则;规则写坏按不命中处置,不让事件流停摆) - assembleInput:字段映射装配;缺必填即起单失败入死信,不静默丢弃 - 幂等:同一事件对同一触发器至多起一单;数据日历型同一行同一日期只触发一次 - 周期两型:日历表达式到点扫描 + 数据日历型(日期字段 + 偏移 + 过滤)逐行起单 - 登记触发:人工起单的统一形态,登记人即来源 - 触发器与流程包平级:配置独立存放,增删启停不改流程包版本 340 个测试通过,门禁五道全绿 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KeDBx2PHLavbURXB8LJJ7m
This commit is contained in:
JCAI Main Agent
co-authored by
Claude Opus 5
Claude Opus 5
parent
a8f6eef610
commit
4ad901c452
@@ -0,0 +1,27 @@
|
||||
/**
|
||||
* trigger 的 sys schema 表(编码规范 §7 / ADR-002)。
|
||||
* 触发器配置**独立存放**于系统自持存储 —— 不入流程包结构、不随流程包版本(OS 3.1)。
|
||||
*/
|
||||
import { boolean, index, jsonb, text, timestamp } from 'drizzle-orm/pg-core';
|
||||
import { sys } from '@shared/db.ts';
|
||||
|
||||
export const triggerTable = sys.table(
|
||||
'trigger',
|
||||
{
|
||||
id: text('id').primaryKey(),
|
||||
name: text('name').notNull(),
|
||||
enabled: boolean('enabled').notNull().default(true),
|
||||
processPackageId: text('process_package_id').notNull(),
|
||||
mapping: jsonb('mapping').notNull().default([]),
|
||||
spec: jsonb('spec').notNull(),
|
||||
lastFiredAt: timestamp('last_fired_at', { withTimezone: true }),
|
||||
createdAt: timestamp('created_at', { withTimezone: true }).notNull().defaultNow(),
|
||||
},
|
||||
(t) => [index('trigger_enabled_idx').on(t.enabled)],
|
||||
);
|
||||
|
||||
/** 起单幂等表:同一事件对同一触发器至多起一单 */
|
||||
export const triggerClaimTable = sys.table('trigger_claim', {
|
||||
key: text('key').primaryKey(),
|
||||
claimedAt: timestamp('claimed_at', { withTimezone: true }).notNull().defaultNow(),
|
||||
});
|
||||
@@ -0,0 +1,73 @@
|
||||
/**
|
||||
* trigger 存储 adapter —— Drizzle 只准出现在 adapter(编码规范 §5.3)。
|
||||
*/
|
||||
import { eq } from 'drizzle-orm';
|
||||
import type { TriggerConfig } from '@contracts';
|
||||
import { triggerConfigSchema } from '@contracts';
|
||||
import type { Db } from '@shared/db.ts';
|
||||
import type { TriggerStore } from '../../port.ts';
|
||||
import { triggerClaimTable, triggerTable } from './trigger-schema.ts';
|
||||
|
||||
type Row = typeof triggerTable.$inferSelect;
|
||||
|
||||
function toConfig(row: Row): TriggerConfig {
|
||||
return triggerConfigSchema.parse({
|
||||
id: row.id,
|
||||
name: row.name,
|
||||
enabled: row.enabled,
|
||||
processPackageId: row.processPackageId,
|
||||
mapping: row.mapping,
|
||||
spec: row.spec,
|
||||
});
|
||||
}
|
||||
|
||||
export function createTriggerStore(db: Db): TriggerStore {
|
||||
return {
|
||||
async list() {
|
||||
const rows = await db.select().from(triggerTable).orderBy(triggerTable.name);
|
||||
return rows.map(toConfig);
|
||||
},
|
||||
async get(id) {
|
||||
const rows = await db.select().from(triggerTable).where(eq(triggerTable.id, id)).limit(1);
|
||||
return rows[0] === undefined ? undefined : toConfig(rows[0]);
|
||||
},
|
||||
async upsert(config) {
|
||||
const values = {
|
||||
id: config.id,
|
||||
name: config.name,
|
||||
enabled: config.enabled,
|
||||
processPackageId: config.processPackageId,
|
||||
mapping: config.mapping,
|
||||
spec: config.spec,
|
||||
};
|
||||
const [row] = await db
|
||||
.insert(triggerTable)
|
||||
.values(values)
|
||||
.onConflictDoUpdate({ target: triggerTable.id, set: values })
|
||||
.returning();
|
||||
if (row === undefined) throw new Error('触发器写入未返回行');
|
||||
return toConfig(row);
|
||||
},
|
||||
async setEnabled(id, enabled) {
|
||||
await db.update(triggerTable).set({ enabled }).where(eq(triggerTable.id, id));
|
||||
},
|
||||
async remove(id) {
|
||||
await db.delete(triggerTable).where(eq(triggerTable.id, id));
|
||||
},
|
||||
async claim(key) {
|
||||
const inserted = await db
|
||||
.insert(triggerClaimTable)
|
||||
.values({ key })
|
||||
.onConflictDoNothing()
|
||||
.returning({ key: triggerClaimTable.key });
|
||||
return inserted.length > 0;
|
||||
},
|
||||
async lastFiredAt(id) {
|
||||
const rows = await db.select({ at: triggerTable.lastFiredAt }).from(triggerTable).where(eq(triggerTable.id, id)).limit(1);
|
||||
return rows[0]?.at ?? undefined;
|
||||
},
|
||||
async markFired(id, at) {
|
||||
await db.update(triggerTable).set({ lastFiredAt: at }).where(eq(triggerTable.id, id));
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,137 @@
|
||||
/**
|
||||
* 触发器判定 —— 纯逻辑,零 IO(编码规范 §5.3;语义出处 OS 3.1)。
|
||||
*
|
||||
* 「触发器 = **流程层的系统级事件监听**:消费同一条事件流,动作是**起单**。
|
||||
* 三类:周期(日历型 / 数据日历型)、事件、登记。
|
||||
* 触发器与工作流**平级**:不入流程包结构、不随流程包版本;
|
||||
* 只负责起单,经调用接口调起流程包 —— 流程包与活动内部不含任何触发逻辑。」
|
||||
*/
|
||||
import type { TriggerConfig } from '@contracts';
|
||||
|
||||
export type ConditionEvaluator = (expression: string, scope: Readonly<Record<string, unknown>>) => boolean;
|
||||
export type ExpressionEvaluator = (expression: string, scope: Readonly<Record<string, unknown>>) => unknown;
|
||||
|
||||
/** 事件触发的求值输入:事件信封与载荷字段 */
|
||||
export type EventScope = {
|
||||
readonly type: string;
|
||||
readonly keys: Readonly<Record<string, string>>;
|
||||
readonly payload: Readonly<Record<string, unknown>>;
|
||||
readonly source: string;
|
||||
readonly occurredAt: string;
|
||||
};
|
||||
|
||||
export function scopeOf(scope: EventScope): Record<string, unknown> {
|
||||
return { type: scope.type, keys: scope.keys, source: scope.source, occurredAt: scope.occurredAt, ...scope.payload };
|
||||
}
|
||||
|
||||
/** 事件触发器是否命中 —— 命中规则用 S-FEEL,静态可检 */
|
||||
export function matchesEvent(config: TriggerConfig, scope: EventScope, evaluate: ConditionEvaluator): boolean {
|
||||
if (!config.enabled) return false;
|
||||
if (config.spec.kind !== 'event') return false;
|
||||
if (config.spec.eventType !== scope.type) return false;
|
||||
try {
|
||||
return evaluate(config.spec.rule, scopeOf(scope));
|
||||
} catch {
|
||||
// 规则写坏了不该让事件流停摆:按不命中处置
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/* ── 触发器输入装配(OS 3.1) ────────────────────────────────────────── */
|
||||
|
||||
export type AssemblyResult =
|
||||
| { readonly kind: 'ready'; readonly data: Record<string, unknown> }
|
||||
/** 缺必填字段则**起单失败、入死信** */
|
||||
| { readonly kind: 'missing'; readonly fields: readonly string[] };
|
||||
|
||||
export function assembleInput(
|
||||
config: TriggerConfig,
|
||||
scope: Readonly<Record<string, unknown>>,
|
||||
required: readonly string[],
|
||||
evaluateExpression: ExpressionEvaluator,
|
||||
): AssemblyResult {
|
||||
const data: Record<string, unknown> = {};
|
||||
for (const mapping of config.mapping) {
|
||||
switch (mapping.source.from) {
|
||||
case 'constant':
|
||||
data[mapping.target] = mapping.source.value;
|
||||
break;
|
||||
case 'expression':
|
||||
data[mapping.target] = evaluateExpression(mapping.source.sfeel, scope);
|
||||
break;
|
||||
case 'initiation':
|
||||
case 'output-port':
|
||||
case 'workflow-var':
|
||||
data[mapping.target] = pick(scope, mapping.source.from === 'workflow-var' ? mapping.source.name : mapping.source.path);
|
||||
break;
|
||||
}
|
||||
}
|
||||
const missing = required.filter((field) => data[field] === undefined || data[field] === null);
|
||||
return missing.length > 0 ? { kind: 'missing', fields: missing } : { kind: 'ready', data };
|
||||
}
|
||||
|
||||
function pick(scope: Readonly<Record<string, unknown>>, path: string): unknown {
|
||||
return path.split('.').reduce<unknown>((acc, seg) => {
|
||||
if (acc === null || acc === undefined || typeof acc !== 'object') return undefined;
|
||||
return (acc as Record<string, unknown>)[seg];
|
||||
}, scope);
|
||||
}
|
||||
|
||||
/* ── 幂等:同一事件对同一触发器**至多起一单**,以事件 ID 为幂等键(OS 6.3) ── */
|
||||
|
||||
export function triggerIdempotencyKey(triggerId: string, eventId: string): string {
|
||||
return `trigger:${triggerId}:${eventId}`;
|
||||
}
|
||||
|
||||
/* ── 周期触发:日历型与数据日历型(OS 3.1) ─────────────────────────── */
|
||||
|
||||
/** 极简日历表达式:`daily HH:mm` / `weekly <1-7> HH:mm` / `monthly <1-28> HH:mm` */
|
||||
export function nextCalendarFire(expression: string, after: Date): Date | undefined {
|
||||
const daily = /^daily\s+(\d{1,2}):(\d{2})$/.exec(expression);
|
||||
if (daily !== null) return nextAt(after, Number(daily[1]), Number(daily[2]));
|
||||
|
||||
const weekly = /^weekly\s+([1-7])\s+(\d{1,2}):(\d{2})$/.exec(expression);
|
||||
if (weekly !== null) {
|
||||
let candidate = nextAt(after, Number(weekly[2]), Number(weekly[3]));
|
||||
const target = Number(weekly[1]) % 7;
|
||||
for (let i = 0; i < 7 && candidate.getDay() !== target; i += 1) {
|
||||
candidate = new Date(candidate.getTime() + 86_400_000);
|
||||
}
|
||||
return candidate;
|
||||
}
|
||||
|
||||
const monthly = /^monthly\s+(\d{1,2})\s+(\d{1,2}):(\d{2})$/.exec(expression);
|
||||
if (monthly !== null) {
|
||||
let candidate = nextAt(after, Number(monthly[2]), Number(monthly[3]));
|
||||
const day = Number(monthly[1]);
|
||||
for (let i = 0; i < 62 && candidate.getDate() !== day; i += 1) {
|
||||
candidate = new Date(candidate.getTime() + 86_400_000);
|
||||
}
|
||||
return candidate;
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
function nextAt(after: Date, hour: number, minute: number): Date {
|
||||
const candidate = new Date(after);
|
||||
candidate.setSeconds(0, 0);
|
||||
candidate.setHours(hour, minute);
|
||||
if (candidate.getTime() <= after.getTime()) candidate.setDate(candidate.getDate() + 1);
|
||||
return candidate;
|
||||
}
|
||||
|
||||
/**
|
||||
* 数据日历型:按台账日期字段 + 偏移 + 过滤条件**逐行起单**。
|
||||
* 「同一行同一日期只触发一次,字段值变更后可再触发。」
|
||||
*/
|
||||
export function dataCalendarKey(triggerId: string, rowKey: string, dateValue: string): string {
|
||||
return `trigger:${triggerId}:${rowKey}:${dateValue}`;
|
||||
}
|
||||
|
||||
/** 偏移:`-P3D` 表示日期字段前 3 天 */
|
||||
export function applyOffset(date: Date, offset: string): Date {
|
||||
const m = /^(-?)P(?:(\d+)D)?$/.exec(offset);
|
||||
if (m === null) throw new Error(`非法日历偏移:${offset}`);
|
||||
const days = Number(m[2] ?? 0) * (m[1] === '-' ? -1 : 1);
|
||||
return new Date(date.getTime() + days * 86_400_000);
|
||||
}
|
||||
@@ -0,0 +1,132 @@
|
||||
/**
|
||||
* trigger 用例 —— 判定与起单(编码规范 §5.3;语义出处 OS 3.1)。
|
||||
*/
|
||||
import type { TriggerConfig } from '@contracts';
|
||||
import type { EventScope, FireOutcome, TriggerService } from '../../contract.ts';
|
||||
import type { Clock, 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 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<string, unknown>,
|
||||
subject: string,
|
||||
): Promise<FireOutcome> {
|
||||
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 };
|
||||
}
|
||||
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' };
|
||||
return fire(config, { kind: 'registration', ref: `registration:${by}`, by }, data, config.name);
|
||||
},
|
||||
|
||||
/** 周期扫描:日历型按表达式到点;数据日历型逐行起单 */
|
||||
async sweepPeriodic(now) {
|
||||
const outcomes: FireOutcome[] = [];
|
||||
for (const config of await deps.store.list()) {
|
||||
if (!config.enabled || config.spec.kind !== 'periodic') continue;
|
||||
|
||||
if (config.spec.mode === '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;
|
||||
}
|
||||
|
||||
// 数据日历型:日期字段 + 偏移 + 过滤条件
|
||||
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;
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user