JX-BPOS/apps/api/src/modules/trigger/internal/usecase/trigger-service.ts

/**
 * 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' };
      /*
       * 主题优先取登记人填的 `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;
    },
  };
}