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, 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<string, unknown>,
    subject: string,
  ): Promise<FireOutcome> {
    const required = await deps.process.requiredInputs(config.processPackageId);
    const assembled = assembleInput(config, scope, required, deps.sfeel.expression);
    /*
     * 宽进(2026-07-30,用户定):**缺必填不再入死信** —— 装到多少带多少照样起单。
     * 收敛点在活动:进活动后受理归一,信息不全由任务侧派补全单收齐。
     * 门口只负责把流起起来;把单卡死在门外,人连「差什么」都要翻死信才知道。
     */
    /*
     * 触发确认链路(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);
    },
  };
}