import type { Bindings, ExecutionGraph, ExecutionContext } from '../types'; import { ExecutionError } from '../types'; import { GraphExecutor } from '../graph-executor'; import { graphSchema } from '../lib/schemas'; import { createComponentLoader } from '../lib/component-loader'; import { recordTelemetry } from '../lib/telemetry'; /** * kbdb-base §7.1+§7.5.h:一條工作流執行結束後,把這次用到的 recipe 各記一次成功/失敗到 KBDB 市場星數。 * 判定單位是「工作流執行」(n8n execution):整體成功 → 用到的每個 recipe key +1 成功;整體失敗 → 各 +1 失敗。 * **key = recipe uuid**(per-uuid,能區分同 canonical 的不同作者版本 §7.5.5;舊資料 fallback canonical_id)。 * * fire-and-forget(用 ctx.waitUntil,仿 recordTelemetry):記錄失敗不影響工作流結果。 * KBDB 端點 POST {KBDB_BASE_URL}/recipe-stats/record { canonical_id, ok, at }—— * 該欄位名為 canonical_id 但語意已是 recipe key(uuid),KBDB 端只當 stat 的主鍵字串用。 */ function recordRecipeStats( env: Bindings, recipeKeys: Set, ok: boolean, at: number, ctx?: ExecutionContext, ): void { if (recipeKeys.size === 0) return; const base = (env.KBDB_BASE_URL ?? 'https://kbdb.finally.click').replace(/\/$/, ''); const headers: Record = { 'Content-Type': 'application/json' }; if (env.KBDB_INTERNAL_TOKEN) headers['Authorization'] = `Bearer ${env.KBDB_INTERNAL_TOKEN}`; const promise = Promise.all( [...recipeKeys].map(key => fetch(`${base}/recipe-stats/record`, { method: 'POST', headers, body: JSON.stringify({ canonical_id: key, ok, at }), }).catch(() => undefined), ), ).then(() => undefined); if (ctx?.waitUntil) ctx.waitUntil(promise); else void promise; } type WebhookRecord = { graph: Record; description: string; created_at: string; }; export function generateToken(): string { const tokenBytes = crypto.getRandomValues(new Uint8Array(16)); return Array.from(tokenBytes).map(b => b.toString(16).padStart(2, '0')).join(''); } export async function validateAndParseWebhook(raw: string): Promise { try { return JSON.parse(raw) as WebhookRecord; } catch { return null; } } export async function executeWebhookGraph( env: Bindings, graph: Record, triggerContext: Record, token: string, apiKey?: string, ctx?: ExecutionContext, // 可選 — 用 waitUntil 把 telemetry 推到背景 userAgent?: string, // MCP / SDK client 帶過來 ): Promise<{ success: boolean; data?: unknown; error?: string; trace?: unknown; duration_ms: number }> { const parsed = graphSchema.safeParse(graph); if (!parsed.success) { return { success: false, error: '圖定義已失效', duration_ms: 0 }; } const loader = createComponentLoader(env); const executor = new GraphExecutor(loader, undefined, env, apiKey); const start = Date.now(); try { const result = await executor.execute( parsed.data as ExecutionGraph, { ...triggerContext, _webhook_token: token }, env.EXEC_CONTEXT, ); const duration_ms = Date.now() - start; // Implicit telemetry:成功 run(含 paused 也算「成功啟動」由 trigger_workflow 那層分類) recordTelemetry(env, apiKey, { event_type: 'run_success', workflow_name: token, duration_ms, agent_user_agent: userAgent, }, ctx); // kbdb-base §7.1:整體成功 → 用到的 recipe 各記成功一次。 recordRecipeStats(env, executor.usedRecipeKeys, true, Date.now(), ctx); return { success: true, data: result.data, duration_ms }; } catch (err) { const duration_ms = Date.now() - start; const errMsg = err instanceof Error ? err.message : String(err); const isPaused = /workflow paused/i.test(errMsg); // Implicit telemetry:paused 算 run_success;真錯才 run_fail recordTelemetry(env, apiKey, { event_type: isPaused ? 'run_success' : 'run_fail', workflow_name: token, error_code: isPaused ? 'paused_awaiting_resume' : 'execution_error', duration_ms, agent_user_agent: userAgent, }, ctx); // kbdb-base §7.1:真錯(非 paused)→ 用到的 recipe 各記失敗一次。 // paused 是「執行中暫停等 callback」非失敗,不記(resume 後成功才會在那條路徑記成功)。 if (!isPaused) { recordRecipeStats(env, executor.usedRecipeKeys, false, Date.now(), ctx); } if (err instanceof ExecutionError) { const traceFormatted = err.trace.map(s => ({ node: s.nodeId, status: s.error ? 'failed' : 'success', ...(s.error ? { error: s.error } : {}), })); return { success: false, error: errMsg, trace: traceFormatted, duration_ms, }; } return { success: false, error: errMsg, duration_ms }; } }