Arcrun/cypher-executor/src/scheduled.ts

/**
 * scheduled() handler — 對應 wrangler.toml [triggers].crons 觸發。
 *
 * 流程：
 * 1. 單次 get cron index（cron-idx:_all，集中存所有 cron workflow 的 cron_expr）
 * 2. 在記憶體比對每筆 cron_expr 跟 event.scheduledTime（UTC 分鐘精度）
 * 3. 匹配才去讀完整 workflow record（{apiKey}:wf:{name}）
 * 4. 匹配 → executeWebhookGraph 跑（waitUntil 背景，不擋）
 * 5. 每天固定一分鐘（UTC 02:30）順便叫 KBDB 清一批過期執行紀錄（P7 保留期，見下方 §5）
 *
 * 8.P0 止血（SDD §8.2）：原本每分鐘 WEBHOOKS.list('cron-idx:') = 1440 list/日 爆 KV 上限，
 * 改成單一固定 key 只 get 一次 → list 歸零。
 *
 * SDD: arcrun.md 三-A P1 #3 / kbdb-base §8.2
 */

import type { ExecutionContext, ScheduledController } from '@cloudflare/workers-types';
import type { Bindings } from './types';
import { cronMatch } from './lib/cron-match';
import { readCronIndex, parseCronEntryKey } from './lib/cron-index';
import { executeWebhookGraph } from './actions/webhook-handlers';
import { kbdbBase } from './routes/kbdb-proxy';

type StoredWorkflowRecord = {
  graph: Record<string, unknown>;
  cron_expr?: string;
  // 其他欄位（id, name, created_at 等）忽略
};

export async function handleScheduled(
  controller: ScheduledController,
  env: Bindings,
  ctx: ExecutionContext,
): Promise<void> {
  const now = new Date(controller.scheduledTime);
  console.log('[scheduled] tick', now.toISOString(), 'controller.cron=', controller.cron);

  // 8.P0：單次 get 集中索引（取代每分鐘 list），主 workflow record 仍在 {apiKey}:wf:{name}
  const index = await readCronIndex(env.WEBHOOKS);
  const entries = Object.entries(index);

  let triggered = 0;
  for (const [entryKey, cronExpr] of entries) {
    const parsed = parseCronEntryKey(entryKey);
    if (!parsed) continue;
    const { apiKey, name } = parsed;

    if (!cronExpr) continue;
    if (!cronMatch(cronExpr, now)) continue;

    // 匹配才去讀完整 workflow record
    const wfKey = `${apiKey}:wf:${name}`;
    const wfRaw = await env.WEBHOOKS.get(wfKey, 'text');
    if (!wfRaw) {
      console.warn('[scheduled] cron-idx 對應 workflow 不存在', wfKey);
      continue;
    }
    let record: StoredWorkflowRecord;
    try { record = JSON.parse(wfRaw) as StoredWorkflowRecord; } catch { continue; }
    triggered++;

    console.log('[scheduled] trigger', name, 'apiKey=', apiKey.slice(0, 12) + '...', 'cron=', cronExpr);
    // 把 apiKey 也放進 triggerContext，讓 workflow 內節點能用 {{api_key}}（跟 webhook trigger 慣例一致）
    const triggerContext = {
      api_key: apiKey,
      _triggered_by: 'cron' as const,
      _scheduled_at: now.toISOString(),
    };
    ctx.waitUntil(
      executeWebhookGraph(env, record.graph, triggerContext, name, apiKey)
        .then(
          (r) => console.log('[scheduled] done', name, r.success, r.duration_ms + 'ms'),
          (e) => console.error('[scheduled] fail', name, e),
        ),
    );
  }
  console.log(`[scheduled] scanned ${entries.length} cron-idx entries, ${triggered} triggered`);

  // §5 P7 保留期清理（2026-08-09）：不新增排程基礎設施（wrangler.toml [triggers] 是受保護
  // 檔案，AI 不可編輯——見 InkStoneCo 頂層 pending-changes.md P9 段 L1 權限閘），改「搭便車」：
  // 這支 handler 本來就每分鐘醒一次（給上面的 cron workflow 用），挑固定一分鐘（UTC 02:30，
  // 避開整點/半點常見的 cron 表達式擁擠時段）順手打一次 fire-and-forget 給 KBDB 的
  // POST /execution-log/cleanup。頻率仍是「一天一次」，不是輪詢外部系統要狀態，是既有 tick
  // 順手打理自己的表。呼叫失敗不影響上面的 cron workflow 觸發（各自 try/catch，互不拖累）。
  if (now.getUTCHours() === 2 && now.getUTCMinutes() === 30) {
    const { base, headers } = kbdbBase(env);
    ctx.waitUntil(
      fetch(`${base}/execution-log/cleanup`, { method: 'POST', headers })
        .then(async (r) => {
          const body = await r.json().catch(() => null);
          console.log('[scheduled] execution-log cleanup', r.status, JSON.stringify(body));
        })
        .catch((e) => console.error('[scheduled] execution-log cleanup failed', e)),
    );
  }
}