durable-trigger

@asm/durable-trigger 位于 Routine 前面，统一事件格式、去重记录和路由。它提供事件入口适配器，宿主提供 HTTP 服务、认证、轮询和关闭流程。

连接日历入口

以下是宿主集成片段。先按 Routine 文档 提供 routineStore、routineLock、openHarness 和已注册的 task：

import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import { withTenantContext } from "@asm/durable-routine";
import {
  CronCalendarTriggerSource,
  SqliteTriggerStore,
  TriggerEngine,
  TriggerRoutineRunner,
} from "@asm/durable-trigger";

const context = withTenantContext(
  { tenantId: "demo-tenant", scopeId: "demo-store" },
  BACKGROUND_CONTEXT,
);
const store = new SqliteTriggerStore("./triggers.sqlite");
const engine = new TriggerEngine({
  store,
  runner: new TriggerRoutineRunner({
    store: routineStore,
    lock: routineLock,
    openHarness,
  }),
});
engine.registerRoute({
  triggerId: "inventory.refresh",
  routineId: "inventory.refresh",
  profile: "inventory",
  task,
  buildInput: (envelope) => envelope.payload,
});
const calendar = new CronCalendarTriggerSource({ store });
engine.registerSource(calendar);
await engine.start(context);
try {
  await calendar.configure({
    tenantId: "demo-tenant",
    scopeId: "demo-store",
    triggerId: "inventory.refresh",
    kind: "cron",
    cron: "0 7 * * *",
    timezone: "Asia/Tokyo",
  }, Date.now(), context);
  // 宿主定时调用 tick；此处只示范一次，不等待下一个执行时刻。
  console.log(await calendar.tick("demo-tenant", Date.now(), context));
} finally {
  await engine.stop(context);
  store.close();
}

start() 只连接入口，不启动定时器。configure() 用于创建或修改订阅，会重置下次执行的游标；不要在每轮 tick() 前重新配置，否则会不断推迟到未来。

使用 TriggerRoutineRunner 连接 Routine。普通 RoutineRunner 按 Routine 身份和时间计算运行键；同一毫秒的两个事件可能碰撞。适配器把事件去重键纳入底层运行身份，对外保留 runKey = dedupeKey。

事件身份和恢复

自定义来源使用 createTriggerEnvelope() 构造事件，再调用 engine.ingest(envelope, context)。必填字段是 tenantId、scopeId、source: { kind, provider }、triggerId、eventId、dedupeKey、occurredAt、scheduledAt 和 JSON payload。函数生成规范 JSON 的 payloadHash，调用方不提供该摘要。

同一租户的 dedupeKey 必须稳定。相同键换成不同范围、来源或载荷会抛出 PAYLOAD_CONFLICT；投递重试保留最初的名义时间。

返回状态	宿主的处理
triggered	已取得 Routine 终态；检查 routineResult，它也可能表示业务失败
filtered	路由过滤器跳过该事件，可以确认消费
duplicate	已有终态记录，可以确认消费
in_progress	仍在执行，保留重试或安排持久恢复

engine.recoverIncomplete(tenantId, context) 重投该租户的 processing、failed 记录，使用保存的事件和范围。宿主需提供有权限执行租户级恢复的 Context，并安排调用时机；该方法不会启动后台 worker。

Webhook 与持久化

WebhookTriggerSource 的 endpoints 配置 provider、triggerId、signatureScheme 和 resolveSecrets。支持 standard-webhooks、stripe、hmac-sha256。宿主保留原始请求字节，并调用 handle({ tenantId, scopeId, provider, headers, rawBody }, context)；身份必须由认证中间件确定。

签名验证不代替租户授权。密钥由 resolveSecrets(tenantId, scopeId, context) 提供；不应把密钥写进文档或模型参数。自定义签名若不覆盖时间戳或事件 ID，宿主还需保证这些字段可信。

InMemoryTriggerStore 只适合进程内示例；SQLite 使用 SqliteTriggerStore；PostgreSQL 使用 await PostgresTriggerStore.open(pool, context)，自动初始化表和强制租户 RLS。数据库池仍由宿主关闭。

关闭服务时，先停止宿主的 HTTP 和轮询入口并等待在途调用，再 engine.stop(context)，最后关闭存储。stop() 不会替宿主等待已经启动的 tick。

源码依据：公共类型、引擎、日历入口、Routine 适配。