Arcrun/cypher-executor/src/lib/component-loader.ts

/**
 * arcrun component loader
 *
 * 解析優先序：
 *
 * 0. trigger_workflow 內建 orchestration 零件（in-process call，繞 CF self-fetch 死鎖）
 * 1. 內建零件（BUILTIN_COMPONENTS）— 純 JS，最快
 * 2. 外部 URL（https://...）— 直接 fetch，n8n/MCP/任何 HTTP 服務
 * 3. cmp_xxxxxxxx hash → 查 WEBHOOKS KV idx → canonical_id → 邏輯 Worker
 * 4. rec_xxxxxxxx hash → 查 RECIPES KV idx → recipe 執行
 * 5. 邏輯零件 canonical_id → Service Binding（同帳號不走公網）
 * 5.5. Auth recipe（平台預建）→ Auth Recipe Runner
 * 6. KV recipe canonical_id → 從 RECIPES KV 讀取 recipe → fetch 外部 API
 * 7. WASM HTTP runner（auth primitive / API 零件 → 獨立 Worker URL）
 * 8. 找不到 → 報錯
 */

import { BUILTIN_COMPONENTS } from './constants';
import { isComponentHash, isRecipeHash } from './hash';
import { resolveRecipe, resolveAuthRecipe } from '../routes/recipes';
import type { AuthRecipeDefinition } from '../routes/recipes';
import type { Bindings, ComponentRunner, ServiceBinding } from '../types';
import { renderBodyTemplate, applyResponseMap } from './recipe-payload';

/**
 * WASM HTTP runner：canonical_id → 對應獨立 Worker URL。
 *
 * 所有 WASM 零件（auth primitive / API 零件 / 未來用戶自製）都是獨立部署的 Worker，
 * 以 `{canonical-id-kebab}.arcrun.dev` 為 URL 慣例。cypher-executor 不做 WASM
 * instantiate,只做 HTTP fetch。這層是 API 零件(及 auth primitive)的唯一入口。
 *
 * R2 動態注入 WASM 路徑作廢（CF workerd 不支援以 R2 物件臨時 instantiate）。
 */
// TODO（架構債，2026-05-07）：白名單寫死違反 arcrun 「新零件無需改 cypher-executor」承諾
// 應改為從 component-registry KV 動態查（registry 已有 backfill index，知道所有 canonical_id）
// SDD 待開：cypher-executor-dynamic-component-discovery
const WASM_HTTP_RUNNER_IDS: ReadonlySet<string> = new Set([
  // 通用 HTTP 零件
  'http_request',
  // 通用 code 零件（sandbox inline JS，Arcrun#10 / 07-thin-shell §3.5 code-node）：獨立 Worker，
  // URL 走 wasmWorkerUrl 通用推導（arcrun-code.{WORKER_SUBDOMAIN}.workers.dev，
  // self-hosted 由 WORKER_SUBDOMAIN var 注入自己的 subdomain，無寫死官方域名）。
  // 漏這行 = workflow 寫 `component: code` 落到 step 8 直接「找不到零件」（#29 發現）。
  'code',
  // gmail / telegram / line_notify / google_sheets 已降級為 recipe（2026-05-29 Phase 2）：
  //   recipe:gmail_send / telegram_send / line_notify_send / google_sheets_read|append
  //   走 step 6 KV recipe 解析，不再是零件。零件目錄已刪。
  'cron',
  // Auth primitives
  'auth_static_key',
  'auth_service_account',
  'auth_oauth2',
  'auth_mtls',
]);

/**
 * canonical_id → component worker URL（走 workers.dev 子域，避開同 zone 自循環死鎖）
 *
 * 為何不用 *.arcrun.dev：cypher-executor 本身綁 cypher.arcrun.dev/*，
 * fetch 同 zone *.arcrun.dev 會撞 CF 的 zone 自循環防護回 522。
 * 詳見 arcrun.md P0 #9（2026-05-13）。
 *
 * subdomain 來自 wrangler.toml [vars] WORKER_SUBDOMAIN（預設 uncle6-me，self-hosted fork 改自己的）。
 */
export function wasmWorkerUrl(canonicalId: string, subdomain: string): string {
  const kebab = canonicalId.replace(/_/g, '-');
  // 平台慣例：component worker 名稱 = `arcrun-{kebab}`（見 rule 03 / rule 05），
  // 例如 canonical_id=http_request → worker 名 arcrun-http-request → URL arcrun-http-request.{subdomain}.workers.dev
  return `https://arcrun-${kebab}.${subdomain}.workers.dev`;
}

/** 邏輯零件 canonical_id → Service Binding key */
const LOGIC_BINDING_MAP: Record<string, keyof Bindings> = {
  if_control:           'SVC_IF_CONTROL',
  switch:               'SVC_SWITCH',
  foreach_control:      'SVC_FOREACH_CONTROL',
  filter:               'SVC_FILTER',
  merge:                'SVC_MERGE',
  try_catch:            'SVC_TRY_CATCH',
  wait:                 'SVC_WAIT',
  set:                  'SVC_SET',
  array_ops:            'SVC_ARRAY_OPS',
  string_ops:           'SVC_STRING_OPS',
  number_ops:           'SVC_NUMBER_OPS',
  date_ops:             'SVC_DATE_OPS',
  validate_json:        'SVC_VALIDATE_JSON',
  // ai_transform_compile / ai_transform_run 已刪除（2026-05-29）：
  // Arcrun 是 AI 呼叫的工具，工作流不該內嵌 AI 節點回頭呼叫 AI（n8n 才需要，因它沒大腦）。
};

/**
 * 「查得到 vs 真的有」的單一真相源（Arcrun#88，2026-08-11）。
 *
 * 病因：`/cypher/search`（`search-nodes.ts`）只查 component registry（`SUBMISSIONS_KV`，
 * 經 `submitComponent`/`index-only` 才會有記錄）；而本檔 0/1/5/7 四步驟能直接解析、
 * **完全不查 registry** 的一整類零件（trigger_workflow、BUILTIN_COMPONENTS、
 * LOGIC_BINDING_MAP、WASM_HTTP_RUNNER_IDS）從未被 submit 過（也不需要——它們是
 * cypher-executor 自帶的，不是投稿存量）。實測 leo21c 實例：`/components/catalog`
 * 404（registry 是舊版沒這端點／索引空），search 因此對 `if_control`／`http_request`
 * 誠實地回「兩庫都查過沒有」——但這兩顆其實跑得動（leo 08-11 探測工作流已證）。
 *
 * 修法：把「執行期真的解析得動」的這份清單匯出給 search-nodes.ts，在查 registry
 * **之前**先比對——讓「查得到」不受 registry 是否可達／是否已 backfill 影響。
 *
 * 刻意不做的事：不去掃 `registry/components/*` 目錄當清單來源——那是零件原始碼
 * 存放處，含已標記待刪的死碼（`km_writer`／`kbdb_upsert_block`，見
 * `system-dev/docs/3-specs/arcrun-usable/cleanup-dead-code.md`）；07-30 曾把這類死碼
 * 誤灌進 registry（leo 點名的錯）。這裡改用**執行期真正拿去 resolve 的白名單本身**
 * （本檔 1/5/7 步驟既有的三份清單）——精確等於「解析得動」，不會多一顆、不會少一顆。
 */
export const RUNTIME_NATIVE_COMPONENT_IDS: ReadonlySet<string> = new Set([
  'trigger_workflow',
  ...BUILTIN_COMPONENTS.keys(),
  ...Object.keys(LOGIC_BINDING_MAP),
  ...WASM_HTTP_RUNNER_IDS,
]);

export function createComponentLoader(env: Bindings) {
  return async (componentId: string): Promise<ComponentRunner> => {

    // 0. 平台內建 orchestration 零件（需要 env / 跨 workflow 能力）
    //    這類零件「是 orchestrator 的職責」（不是業務邏輯），故不違反「業務邏輯走 WASM」規則。
    //    目前只有 trigger_workflow：用 in-process call 觸發另一個 named workflow，
    //    繞掉 CF 同 zone self-fetch 死鎖（避免 cypher-executor 自打 http_request → 1042）。
    if (componentId === 'trigger_workflow') {
      return makeTriggerWorkflowRunner(env);
    }

    // 1. 內建零件（純 JS，最優先）
    const builtin = BUILTIN_COMPONENTS.get(componentId);
    if (builtin) return builtin;

    // 2. 外部 URL
    if (componentId.startsWith('http://') || componentId.startsWith('https://')) {
      return makeHttpRunner(componentId);
    }

    // 3. cmp_hash → 查 WEBHOOKS KV idx → canonical_id → 邏輯 Worker
    if (isComponentHash(componentId)) {
      const canonicalId = await env.WEBHOOKS.get(`idx:${componentId}`);
      if (canonicalId) {
        const runner = makeLogicRunner(canonicalId, env);
        if (runner) return runner;
      }
      throw new Error(`找不到零件 hash "${componentId}"，請確認已透過 acr push 上傳`);
    }

    // 4. rec_hash → 查 RECIPES KV idx → recipe 執行
    if (isRecipeHash(componentId)) {
      const recipe = await resolveRecipe(componentId, env.RECIPES);
      if (recipe) return pickRecipeRunner(recipe, env);
      throw new Error(`找不到 recipe hash "${componentId}"，請確認已透過 acr push 上傳`);
    }

    // 5. 邏輯零件 canonical_id → Service Binding
    const logicRunner = makeLogicRunner(componentId, env);
    if (logicRunner) return logicRunner;

    // 5.5 Auth recipe（平台預建，auth_recipe:{service} in RECIPES KV）
    const authRecipe = await resolveAuthRecipe(componentId, env.RECIPES);
    if (authRecipe) return makeAuthRecipeRunner(authRecipe);

    // 6. KV recipe（動態，用戶 push 的）
    const kvRecipe = await resolveRecipe(componentId, env.RECIPES);
    if (kvRecipe) return pickRecipeRunner(kvRecipe, env);

    // 7. WASM HTTP runner:auth primitive / API 零件 → 獨立 Worker URL
    //    白名單見 WASM_HTTP_RUNNER_IDS（http_request、5 個待降級 API 零件、4 個 auth primitive）。
    //    對應 Worker 部署於 arcrun-{canonical-id-kebab}.{WORKER_SUBDOMAIN}.workers.dev
    //    （見 P0 #9 / rule 03）。
    if (WASM_HTTP_RUNNER_IDS.has(componentId)) {
      return makeHttpRunner(wasmWorkerUrl(componentId, env.WORKER_SUBDOMAIN));
    }

    // 8. 找不到
    throw new Error(
      `找不到零件 "${componentId}"。\n` +
      `邏輯零件：${Object.keys(LOGIC_BINDING_MAP).join(', ')}\n` +
      `或傳入外部 URL（https://...）、recipe hash（rec_xxxxxxxx）、零件 hash（cmp_xxxxxxxx）`
    );
  };
}

// ── 執行器工廠 ────────────────────────────────────────────────────────────────

/**
 * trigger_workflow 內建 orchestration 零件
 *
 * 用途：在 workflow A 內 in-process 觸發 workflow B，繞 CF 同 zone self-fetch 死鎖。
 *
 * 動機：mira_feed_watcher 之前用 http_request 自打 cypher.arcrun.dev → CF 1042。
 *      就算改打 arcrun-cypher-executor.{subdomain}.workers.dev，Worker → 自身 URL 仍
 *      被 CF 「self subrequest」防護擋（即使 hostname 不同）。
 *      改用 in-process call executeWebhookGraph 徹底繞掉外部 HTTP。
 *
 * 不違反「業務邏輯走 WASM」鐵律：trigger_workflow 是 orchestrator 自己的 routing 能力
 * （像 CALLS_SUBFLOW），不是業務邏輯（不解密 / 不簽 JWT / 不打外部 API）。
 *
 * Input ctx：
 *   - workflow_name: string   (必填，目標 workflow 名稱)
 *   - api_key:       string   (必填，KV 查 key prefix)
 *   - input:         object   (可選，傳給子 workflow 當 triggerContext)
 *   - wait:          boolean  (預設 true，await 完成；false = fire-and-forget 用 waitUntil)
 *
 * 動態 import webhook-handlers 避循環依賴（webhook-handlers → component-loader → 自己）。
 */
function makeTriggerWorkflowRunner(env: Bindings): ComponentRunner {
  return async (ctx: unknown) => {
    const c = (ctx && typeof ctx === 'object') ? ctx as Record<string, unknown> : {};
    const workflowName = String(c.workflow_name ?? '');
    const apiKey = String(c.api_key ?? '');
    const input = (c.input && typeof c.input === 'object')
      ? c.input as Record<string, unknown>
      : {};
    const wait = c.wait !== false; // 預設 true

    if (!workflowName) return { success: false, error: 'trigger_workflow 缺 workflow_name' };
    if (!apiKey) return { success: false, error: 'trigger_workflow 缺 api_key' };

    // 從 WEBHOOKS KV 撈目標 workflow 的 graph
    const wfKey = `${apiKey}:wf:${workflowName}`;
    const wfRaw = await env.WEBHOOKS.get(wfKey, 'text');
    if (!wfRaw) return { success: false, error: `找不到 workflow "${workflowName}" (key=${wfKey})` };

    let record: { graph?: Record<string, unknown> };
    try { record = JSON.parse(wfRaw); }
    catch { return { success: false, error: `workflow "${workflowName}" KV 內容非 JSON` }; }
    if (!record.graph) return { success: false, error: `workflow "${workflowName}" 缺 graph 欄位` };

    // 動態 import 避循環依賴
    const { executeWebhookGraph } = await import('../actions/webhook-handlers');

    const triggerContext = { ...input, _triggered_by: 'trigger_workflow' };

    if (wait) {
      const r = await executeWebhookGraph(env, record.graph, triggerContext, workflowName, apiKey);
      // paused 是預期狀態（claude_api 等待外部 callback resume），不算失敗
      // executeWebhookGraph 內部把 ExecutionError + "paused at node X" 包成 success:false + 含 error 字串
      //
      // 2026-05-16 rename per LI roadmap (block e924c231) 自評建議：
      //   舊 `paused_awaiting_resume` 容易被誤讀成「掛起出問題」
      //   新 `running_async` 強調「已接受，繼續在背景跑」— 行為一致，命名更清楚
      const isPaused = !r.success && typeof r.error === 'string' && /workflow paused/i.test(r.error);
      return {
        success: r.success || isPaused,
        triggered_workflow: workflowName,
        status: r.success ? 'completed' : (isPaused ? 'running_async' : 'failed'),
        sub_result: r,
      };
    } else {
      // fire-and-forget — 不 await，但因為沒拿到 ctx.waitUntil，這裡 promise 可能被 cancel
      // 目前不啟用，留 wait=true 為預設。未來想要 fire-and-forget 需 plumb ExecutionContext
      void executeWebhookGraph(env, record.graph, triggerContext, workflowName, apiKey)
        .catch((e) => console.error('[trigger_workflow] fire-and-forget fail', workflowName, e));
      return { success: true, triggered_workflow: workflowName, mode: 'fire_and_forget' };
    }
  };
}

function makeHttpRunner(url: string): ComponentRunner {
  return async (ctx: unknown) => {
    const res = await fetch(url, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(ctx),
    });
    if (!res.ok) {
      const text = await res.text();
      return { success: false, status: res.status, error: text.slice(0, 200) };
    }
    try { return await res.json(); }
    catch { return { success: true, data: await res.text() }; }
  };
}

function makeLogicRunner(canonicalId: string, env: Bindings): ComponentRunner | null {
  const bindingKey = LOGIC_BINDING_MAP[canonicalId];
  if (!bindingKey) return null;

  const svc = env[bindingKey] as ServiceBinding | undefined;
  if (svc) {
    return async (ctx: unknown) => {
      const res = await svc.fetch(new Request('https://component/', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify(ctx),
      }));
      if (!res.ok) {
        const text = await res.text();
        return { success: false, error: `${canonicalId} 回傳 ${res.status}: ${text.slice(0, 200)}` };
      }
      try { return await res.json(); }
      catch { return { success: false, error: `${canonicalId} 回傳非 JSON` }; }
    };
  }

  // Service Binding 未配置時 fallback 到公網（自製零件 or 開發環境）
  // 走 workers.dev 子域避開同 zone 死鎖（P0 #9）
  return makeHttpRunner(wasmWorkerUrl(canonicalId, env.WORKER_SUBDOMAIN));
}

/**
 * recipe → runner 的分派（3.12）：auth='binding' 走平台 binding（免金鑰），
 * 其餘一律走既有 HTTP 路徑（沒宣告 auth 的舊 recipe 完全不受影響）。
 */
function pickRecipeRunner(
  recipe: import('../routes/recipes').RecipeDefinition,
  env: Bindings,
): ComponentRunner {
  return recipe.auth === 'binding'
    ? makeBindingRecipeRunner(recipe, env)
    : makeRecipeRunner(recipe);
}

/**
 * auth='binding' 的 recipe runner（3.12 第四型認證）：不打外部 HTTP、不需要任何金鑰，
 * 直接用平台 binding（env.AI／VECTORIZE／…）⇒ leo 要的「開機就可用」。
 *
 * 為什麼要開這型：recipe 的舊抽象＝「打一個外部 HTTP API」（endpoint+method+auth_service），
 * 而 Cloudflare 的 binding 呼叫不是 HTTP ⇒ **整類能力被排除在 recipe 之外**。
 * 開這一型不是為 Workers AI 開特例，是一次打開 env.AI／VECTORIZE／BROWSER／QUEUE 整排。
 */
function makeBindingRecipeRunner(
  recipe: import('../routes/recipes').RecipeDefinition,
  env: Bindings,
): ComponentRunner {
  return async (ctx: unknown) => {
    const ctxObj = (ctx && typeof ctx === 'object') ? ctx as Record<string, unknown> : {};
    const name = recipe.binding_name ?? 'AI';
    const binding = (env as unknown as Record<string, unknown>)[name];

    if (!binding) {
      return {
        success: false,
        error:
          `recipe "${recipe.canonical_id}" 宣告 auth: binding、binding_name: "${name}"，` +
          `但這個部署沒有綁定 ${name}。請在 wrangler.toml 補上該 binding 後重新部署。`,
      };
    }

    // endpoint 在 binding 型當作「要呼叫的資源名」（例 Workers AI 的模型 id）
    const target = recipe.endpoint;
    const payload = renderBodyTemplate(recipe.body_template ?? recipe.body, ctxObj)
      ?? Object.fromEntries(Object.entries(ctxObj).filter(([k]) => !k.startsWith('_')));

    try {
      const runner = binding as { run?: (model: string, input: unknown) => Promise<unknown> };
      if (typeof runner.run !== 'function') {
        return {
          success: false,
          error: `binding "${name}" 沒有 run() 方法，目前 binding 型只支援 run(model, input) 形狀（如 env.AI）。`,
        };
      }
      const data = await runner.run(target, payload);
      if (recipe.response_map) {
        const normalized = applyResponseMap(data, recipe.response_map);
        return { success: true, data, text: normalized.text };
      }
      return { success: true, data };
    } catch (e) {
      return {
        success: false,
        error: `binding "${name}" 呼叫失敗（${target}）：${e instanceof Error ? e.message : String(e)}`,
      };
    }
  };
}

function makeRecipeRunner(recipe: import('../routes/recipes').RecipeDefinition): ComponentRunner {
  return async (ctx: unknown) => {
    const ctxObj = (ctx && typeof ctx === 'object') ? ctx as Record<string, unknown> : {};

    // 模板替換：{{key}} 從 ctx 取；{{auth.K}} 從 _auth_path 取
    // （_auth_path 由 auth primitive 解密後注入，供 URL path 用，如 telegram /bot{{auth.token}}/）
    const authPath = (ctxObj._auth_path as Record<string, string>) ?? {};
    const interpolate = (s: string) =>
      s.replace(/\{\{(auth\.)?(\w+)\}\}/g, (_, authPrefix, k) =>
        String(authPrefix ? (authPath[k] ?? '') : (ctxObj[k] ?? '')),
      );

    const method = (recipe.method ?? 'POST').toUpperCase();
    const authHeaders = (ctxObj._auth_headers as Record<string, string>) ?? {};
    const headers: Record<string, string> = {
      'Content-Type': 'application/json',
      ...authHeaders,
    };
    for (const [k, v] of Object.entries(recipe.headers ?? {})) {
      headers[k] = interpolate(v);
    }

    // body：優先 body_template（③ payload 層，3.12——支援巢狀/dot path/保留型別），
    // 其次既有 recipe.body（淺層 {{key}}，舊 recipe 照舊），最後才拿 ctx 當 body。
    let bodyStr: string | undefined;
    if (recipe.body_template) {
      bodyStr = JSON.stringify(renderBodyTemplate(recipe.body_template, ctxObj));
    } else if (recipe.body) {
      bodyStr = interpolate(JSON.stringify(recipe.body));
    } else if (method !== 'GET') {
      // 沒指定 body template → 用 ctx 當 body，但剔除 _ 前綴的內部欄位
      // （_path / _auth_headers / _auth_query / _auth_body 不該漏進下游請求）
      const bodyObj = Object.fromEntries(
        Object.entries(ctxObj).filter(([k]) => !k.startsWith('_')),
      );
      bodyStr = JSON.stringify(bodyObj);
    }

    const res = await fetch(interpolate(recipe.endpoint), {
      method,
      headers,
      body: bodyStr,
    });

    const data = await readBodyOnce(res);

    // ③ 回應正規化（3.12）：未設 response_map ⇒ 原樣回傳（既有 recipe 零行為變化）。
    // 設了 ⇒ 額外附 `text`（各家形狀差異收在 recipe 裡，換源不必改 workflow）。
    if (recipe.response_map) {
      const normalized = applyResponseMap(data, recipe.response_map);
      return { success: res.ok, status: res.status, data, text: normalized.text };
    }
    return { success: res.ok, status: res.status, data };
  };
}

// ── Auth Recipe Runner ────────────────────────────────────────────────────────
//
// auth-dispatcher 已先將認證資訊注入為 _auth_headers / _auth_query / _auth_body。
// 這裡只需要讀取這些欄位，合併進 fetch，再清除 _auth_* 不傳給下游。

function makeAuthRecipeRunner(recipe: AuthRecipeDefinition): ComponentRunner {
  return async (ctx: unknown) => {
    const ctxObj = (ctx && typeof ctx === 'object') ? ctx as Record<string, unknown> : {};

    const authHeaders = (ctxObj._auth_headers as Record<string, string>) ?? {};
    const authQuery   = (ctxObj._auth_query   as Record<string, string>) ?? {};

    // _path 讓呼叫者指定 endpoint 後綴（e.g. /pages, /messages），可選
    const path = typeof ctxObj._path === 'string' ? ctxObj._path : '';
    const method = ((ctxObj.method as string) ?? 'POST').toUpperCase();

    const url = new URL(recipe.base_url.replace(/\/$/, '') + path);
    for (const [k, v] of Object.entries(authQuery)) {
      url.searchParams.set(k, v);
    }

    const headers: Record<string, string> = {
      'Content-Type': 'application/json',
      ...authHeaders,
    };

    // body：剔除所有 _ 前綴的內部欄位，以及 method
    const bodyObj = Object.fromEntries(
      Object.entries(ctxObj).filter(([k]) => !k.startsWith('_') && k !== 'method'),
    );

    const res = await fetch(url.toString(), {
      method,
      headers,
      body: method !== 'GET' ? JSON.stringify(bodyObj) : undefined,
    });

    const data = await readBodyOnce(res);
    return { success: res.ok, status: res.status, data };
  };
}

// 讀 response body 一次：先取 text，再嘗試 parse JSON。
// 不可用 `res.json().catch(() => res.text())` —— res.json() 失敗時 body 已被消費，
// 第二次讀會丟 "Body has already been used"。
async function readBodyOnce(res: Response): Promise<unknown> {
  const text = await res.text();
  try { return JSON.parse(text); }
  catch { return text; }
}