diff --git a/packages/noodl-viewer-cloud/src/execution-history/index.ts b/packages/noodl-viewer-cloud/src/execution-history/index.ts new file mode 100644 index 0000000..efc17ed --- /dev/null +++ b/packages/noodl-viewer-cloud/src/execution-history/index.ts @@ -0,0 +1,76 @@ +/** + * Execution History Module + * + * Provides storage and querying for workflow execution history. + * Uses SQLite for local storage with support for retention policies. + * + * @module execution-history + * + * @example + * ```typescript + * import Database from 'better-sqlite3'; + * import { ExecutionStore } from './execution-history'; + * + * const db = new Database('./executions.db'); + * const store = new ExecutionStore(db); + * + * // Create an execution + * const execId = store.createExecution({ + * workflowId: 'wf-123', + * workflowName: 'My Workflow', + * triggerType: 'manual', + * status: 'running', + * startedAt: Date.now() + * }); + * + * // Add steps as nodes execute + * store.addStep({ + * executionId: execId, + * nodeId: 'node-1', + * nodeType: 'noodl.logic.condition', + * stepIndex: 0, + * startedAt: Date.now(), + * status: 'running', + * inputData: { condition: true } + * }); + * + * // Complete the execution + * store.updateExecution(execId, { + * status: 'success', + * completedAt: Date.now(), + * durationMs: 1234 + * }); + * + * // Query executions + * const recent = store.queryExecutions({ + * workflowId: 'wf-123', + * status: 'error', + * limit: 10 + * }); + * ``` + */ + +// Export store +export { ExecutionStore } from './store'; +export type { SQLiteDatabase } from './store'; + +// Export types +export type { + // Status types + ExecutionStatus, + StepStatus, + TriggerType, + // Main interfaces + WorkflowExecution, + ExecutionStep, + ExecutionWithSteps, + // Query interfaces + ExecutionQuery, + ExecutionStats, + RetentionPolicy, + // Option types + CreateExecutionOptions, + CreateStepOptions, + UpdateExecutionOptions, + UpdateStepOptions +} from './types'; diff --git a/packages/noodl-viewer-cloud/src/execution-history/schema.sql b/packages/noodl-viewer-cloud/src/execution-history/schema.sql new file mode 100644 index 0000000..55c2629 --- /dev/null +++ b/packages/noodl-viewer-cloud/src/execution-history/schema.sql @@ -0,0 +1,124 @@ +-- Execution History Schema +-- SQLite schema for storing workflow execution history +-- +-- CF11-004: Execution Storage Schema + +-- ============================================================================ +-- WORKFLOW EXECUTIONS TABLE +-- ============================================================================ +-- Stores one record per workflow run + +CREATE TABLE IF NOT EXISTS workflow_executions ( + -- Primary key: unique execution ID + id TEXT PRIMARY KEY, + + -- Reference to the workflow component + workflow_id TEXT NOT NULL, + workflow_name TEXT NOT NULL, + + -- Trigger information + trigger_type TEXT NOT NULL, -- 'webhook', 'schedule', 'manual', 'db_change', 'internal_event', 'test' + trigger_data TEXT, -- JSON: request body, cron expression, etc. + + -- Execution status + status TEXT NOT NULL, -- 'running', 'success', 'error' + + -- Timing + started_at INTEGER NOT NULL, -- Unix timestamp (ms) + completed_at INTEGER, -- Unix timestamp (ms) + duration_ms INTEGER, -- Computed: completed_at - started_at + + -- Error information (if status = 'error') + error_message TEXT, + error_stack TEXT, + + -- Additional context + metadata TEXT -- JSON: any additional context +); + +-- ============================================================================ +-- EXECUTION STEPS TABLE +-- ============================================================================ +-- Stores one record per node execution within a workflow + +CREATE TABLE IF NOT EXISTS execution_steps ( + -- Primary key: unique step ID + id TEXT PRIMARY KEY, + + -- Foreign key to parent execution + execution_id TEXT NOT NULL, + + -- Node information + node_id TEXT NOT NULL, -- ID of the node in the graph + node_type TEXT NOT NULL, -- Type of node (e.g., 'noodl.logic.condition') + node_name TEXT, -- Display name of the node + + -- Execution order + step_index INTEGER NOT NULL, -- 0-based order of execution + + -- Timing + started_at INTEGER NOT NULL, -- Unix timestamp (ms) + completed_at INTEGER, -- Unix timestamp (ms) + duration_ms INTEGER, -- Computed: completed_at - started_at + + -- Status + status TEXT NOT NULL, -- 'running', 'success', 'error', 'skipped' + + -- Data (JSON, may be truncated for large payloads) + input_data TEXT, -- JSON: inputs received by the node + output_data TEXT, -- JSON: outputs produced by the node + + -- Error information (if status = 'error') + error_message TEXT, + + -- Cascade delete when parent execution is deleted + FOREIGN KEY (execution_id) REFERENCES workflow_executions(id) ON DELETE CASCADE +); + +-- ============================================================================ +-- INDEXES +-- ============================================================================ +-- Optimize common query patterns + +-- Query executions by workflow +CREATE INDEX IF NOT EXISTS idx_executions_workflow + ON workflow_executions(workflow_id); + +-- Query executions by status +CREATE INDEX IF NOT EXISTS idx_executions_status + ON workflow_executions(status); + +-- Query executions by start time (most common: newest first) +CREATE INDEX IF NOT EXISTS idx_executions_started + ON workflow_executions(started_at DESC); + +-- Combined index for filtered + sorted queries +CREATE INDEX IF NOT EXISTS idx_executions_workflow_started + ON workflow_executions(workflow_id, started_at DESC); + +-- Query steps by execution +CREATE INDEX IF NOT EXISTS idx_steps_execution + ON execution_steps(execution_id); + +-- Query steps by node (for debugging specific nodes) +CREATE INDEX IF NOT EXISTS idx_steps_node + ON execution_steps(node_id); + +-- Query steps by execution in order +CREATE INDEX IF NOT EXISTS idx_steps_execution_order + ON execution_steps(execution_id, step_index); + +-- ============================================================================ +-- CLEANUP HELPER VIEW +-- ============================================================================ +-- View for identifying old executions + +CREATE VIEW IF NOT EXISTS old_executions AS +SELECT + id, + workflow_id, + started_at, + status, + (strftime('%s', 'now') * 1000 - started_at) as age_ms +FROM workflow_executions +ORDER BY started_at ASC; diff --git a/packages/noodl-viewer-cloud/src/execution-history/store.ts b/packages/noodl-viewer-cloud/src/execution-history/store.ts new file mode 100644 index 0000000..e46fa19 --- /dev/null +++ b/packages/noodl-viewer-cloud/src/execution-history/store.ts @@ -0,0 +1,648 @@ +/** + * ExecutionStore - SQLite-backed execution history storage + * + * Provides CRUD operations for workflow executions and steps, + * query filtering, pagination, and retention management. + * + * @module execution-history/store + */ + +// Read schema SQL for initialization +import * as fs from 'fs'; +import * as path from 'path'; + +import type { + WorkflowExecution, + ExecutionStep, + ExecutionQuery, + ExecutionWithSteps, + ExecutionStats, + RetentionPolicy, + CreateExecutionOptions, + CreateStepOptions, + UpdateExecutionOptions, + UpdateStepOptions +} from './types'; + +/** Maximum size for JSON data fields (bytes) */ +const MAX_DATA_SIZE = 50 * 1024; // 50KB + +/** Default limit for queries */ +const DEFAULT_QUERY_LIMIT = 100; + +/** + * Generate a unique ID (similar to CUID) + */ +function generateId(): string { + const timestamp = Date.now().toString(36); + const random = Math.random().toString(36).substring(2, 10); + return `exec_${timestamp}${random}`; +} + +/** + * Truncate JSON data if too large + */ +function truncateData(data: unknown, maxSize: number = MAX_DATA_SIZE): string | null { + if (data === undefined || data === null) { + return null; + } + + try { + const json = JSON.stringify(data); + if (json.length <= maxSize) { + return json; + } + + // Truncate and add marker + return JSON.stringify({ + __truncated: true, + __originalSize: json.length, + __preview: JSON.stringify(data).substring(0, 1000) + }); + } catch (e) { + return JSON.stringify({ __error: 'Failed to serialize data' }); + } +} + +/** + * Parse JSON data from database + */ +function parseData(json: string | null): Record | undefined { + if (!json) { + return undefined; + } + + try { + return JSON.parse(json); + } catch (e) { + return undefined; + } +} + +/** + * SQLite Database interface + * Compatible with better-sqlite3 + */ +export interface SQLiteDatabase { + exec(sql: string): void; + prepare(sql: string): { + run(...params: unknown[]): { changes: number }; + get(...params: unknown[]): unknown; + all(...params: unknown[]): unknown[]; + }; +} + +/** + * ExecutionStore class + * + * Uses better-sqlite3 for synchronous SQLite access. + */ +export class ExecutionStore { + private db: SQLiteDatabase; + private initialized = false; + + /** + * Create an ExecutionStore + * + * @param db - SQLite Database instance (better-sqlite3 compatible) + */ + constructor(db: SQLiteDatabase) { + this.db = db; + } + + /** + * Initialize the database schema + */ + initSchema(): void { + if (this.initialized) { + return; + } + + // Read and execute schema + const schemaPath = path.join(__dirname, 'schema.sql'); + let schema: string; + + try { + schema = fs.readFileSync(schemaPath, 'utf-8'); + } catch (e) { + // Fallback: inline schema for bundled environments + schema = this.getInlineSchema(); + } + + this.db.exec(schema); + this.initialized = true; + } + + /** + * Inline schema for bundled environments + */ + private getInlineSchema(): string { + return ` + CREATE TABLE IF NOT EXISTS workflow_executions ( + id TEXT PRIMARY KEY, + workflow_id TEXT NOT NULL, + workflow_name TEXT NOT NULL, + trigger_type TEXT NOT NULL, + trigger_data TEXT, + status TEXT NOT NULL, + started_at INTEGER NOT NULL, + completed_at INTEGER, + duration_ms INTEGER, + error_message TEXT, + error_stack TEXT, + metadata TEXT + ); + + CREATE TABLE IF NOT EXISTS execution_steps ( + id TEXT PRIMARY KEY, + execution_id TEXT NOT NULL, + node_id TEXT NOT NULL, + node_type TEXT NOT NULL, + node_name TEXT, + step_index INTEGER NOT NULL, + started_at INTEGER NOT NULL, + completed_at INTEGER, + duration_ms INTEGER, + status TEXT NOT NULL, + input_data TEXT, + output_data TEXT, + error_message TEXT, + FOREIGN KEY (execution_id) REFERENCES workflow_executions(id) ON DELETE CASCADE + ); + + CREATE INDEX IF NOT EXISTS idx_executions_workflow ON workflow_executions(workflow_id); + CREATE INDEX IF NOT EXISTS idx_executions_status ON workflow_executions(status); + CREATE INDEX IF NOT EXISTS idx_executions_started ON workflow_executions(started_at DESC); + CREATE INDEX IF NOT EXISTS idx_executions_workflow_started ON workflow_executions(workflow_id, started_at DESC); + CREATE INDEX IF NOT EXISTS idx_steps_execution ON execution_steps(execution_id); + CREATE INDEX IF NOT EXISTS idx_steps_node ON execution_steps(node_id); + CREATE INDEX IF NOT EXISTS idx_steps_execution_order ON execution_steps(execution_id, step_index); + `; + } + + // =========================================================================== + // EXECUTION CRUD + // =========================================================================== + + /** + * Create a new execution record + */ + createExecution(options: CreateExecutionOptions): string { + this.initSchema(); + + const id = generateId(); + const stmt = this.db.prepare(` + INSERT INTO workflow_executions ( + id, workflow_id, workflow_name, trigger_type, trigger_data, + status, started_at, completed_at, duration_ms, + error_message, error_stack, metadata + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `); + + stmt.run( + id, + options.workflowId, + options.workflowName, + options.triggerType, + truncateData(options.triggerData), + options.status, + options.startedAt, + options.completedAt ?? null, + options.durationMs ?? null, + options.errorMessage ?? null, + options.errorStack ?? null, + truncateData(options.metadata) + ); + + return id; + } + + /** + * Update an existing execution + */ + updateExecution(id: string, updates: UpdateExecutionOptions): void { + this.initSchema(); + + const fields: string[] = []; + const values: unknown[] = []; + + if (updates.status !== undefined) { + fields.push('status = ?'); + values.push(updates.status); + } + if (updates.completedAt !== undefined) { + fields.push('completed_at = ?'); + values.push(updates.completedAt); + } + if (updates.durationMs !== undefined) { + fields.push('duration_ms = ?'); + values.push(updates.durationMs); + } + if (updates.errorMessage !== undefined) { + fields.push('error_message = ?'); + values.push(updates.errorMessage); + } + if (updates.errorStack !== undefined) { + fields.push('error_stack = ?'); + values.push(updates.errorStack); + } + if (updates.metadata !== undefined) { + fields.push('metadata = ?'); + values.push(truncateData(updates.metadata)); + } + + if (fields.length === 0) { + return; + } + + values.push(id); + const sql = `UPDATE workflow_executions SET ${fields.join(', ')} WHERE id = ?`; + this.db.prepare(sql).run(...values); + } + + /** + * Get a single execution by ID + */ + getExecution(id: string): WorkflowExecution | null { + this.initSchema(); + + const stmt = this.db.prepare('SELECT * FROM workflow_executions WHERE id = ?'); + const row = stmt.get(id) as Record | undefined; + + if (!row) { + return null; + } + + return this.rowToExecution(row); + } + + /** + * Get execution with all its steps + */ + getExecutionWithSteps(id: string): ExecutionWithSteps | null { + const execution = this.getExecution(id); + if (!execution) { + return null; + } + + const steps = this.getStepsForExecution(id); + + return { + ...execution, + steps + }; + } + + /** + * Query executions with filters and pagination + */ + queryExecutions(query: ExecutionQuery = {}): WorkflowExecution[] { + this.initSchema(); + + const conditions: string[] = []; + const params: unknown[] = []; + + if (query.workflowId) { + conditions.push('workflow_id = ?'); + params.push(query.workflowId); + } + if (query.status) { + conditions.push('status = ?'); + params.push(query.status); + } + if (query.triggerType) { + conditions.push('trigger_type = ?'); + params.push(query.triggerType); + } + if (query.startedAfter !== undefined) { + conditions.push('started_at >= ?'); + params.push(query.startedAfter); + } + if (query.startedBefore !== undefined) { + conditions.push('started_at <= ?'); + params.push(query.startedBefore); + } + + let sql = 'SELECT * FROM workflow_executions'; + if (conditions.length > 0) { + sql += ' WHERE ' + conditions.join(' AND '); + } + + // Order + const orderBy = query.orderBy || 'started_at'; + const orderDir = query.orderDir || 'desc'; + sql += ` ORDER BY ${orderBy} ${orderDir.toUpperCase()}`; + + // Pagination + const limit = query.limit ?? DEFAULT_QUERY_LIMIT; + sql += ' LIMIT ?'; + params.push(limit); + + if (query.offset) { + sql += ' OFFSET ?'; + params.push(query.offset); + } + + const stmt = this.db.prepare(sql); + const rows = stmt.all(...params) as Record[]; + + return rows.map((row) => this.rowToExecution(row)); + } + + /** + * Delete an execution and its steps + */ + deleteExecution(id: string): void { + this.initSchema(); + + // Steps will be deleted via CASCADE + const stmt = this.db.prepare('DELETE FROM workflow_executions WHERE id = ?'); + stmt.run(id); + } + + // =========================================================================== + // STEP CRUD + // =========================================================================== + + /** + * Add a step to an execution + */ + addStep(options: CreateStepOptions): string { + this.initSchema(); + + const id = generateId().replace('exec_', 'step_'); + const stmt = this.db.prepare(` + INSERT INTO execution_steps ( + id, execution_id, node_id, node_type, node_name, + step_index, started_at, completed_at, duration_ms, + status, input_data, output_data, error_message + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `); + + stmt.run( + id, + options.executionId, + options.nodeId, + options.nodeType, + options.nodeName ?? null, + options.stepIndex, + options.startedAt, + options.completedAt ?? null, + options.durationMs ?? null, + options.status, + truncateData(options.inputData), + truncateData(options.outputData), + options.errorMessage ?? null + ); + + return id; + } + + /** + * Update a step + */ + updateStep(id: string, updates: UpdateStepOptions): void { + this.initSchema(); + + const fields: string[] = []; + const values: unknown[] = []; + + if (updates.status !== undefined) { + fields.push('status = ?'); + values.push(updates.status); + } + if (updates.completedAt !== undefined) { + fields.push('completed_at = ?'); + values.push(updates.completedAt); + } + if (updates.durationMs !== undefined) { + fields.push('duration_ms = ?'); + values.push(updates.durationMs); + } + if (updates.outputData !== undefined) { + fields.push('output_data = ?'); + values.push(truncateData(updates.outputData)); + } + if (updates.errorMessage !== undefined) { + fields.push('error_message = ?'); + values.push(updates.errorMessage); + } + + if (fields.length === 0) { + return; + } + + values.push(id); + const sql = `UPDATE execution_steps SET ${fields.join(', ')} WHERE id = ?`; + this.db.prepare(sql).run(...values); + } + + /** + * Get all steps for an execution + */ + getStepsForExecution(executionId: string): ExecutionStep[] { + this.initSchema(); + + const stmt = this.db.prepare('SELECT * FROM execution_steps WHERE execution_id = ? ORDER BY step_index ASC'); + const rows = stmt.all(executionId) as Record[]; + + return rows.map((row) => this.rowToStep(row)); + } + + // =========================================================================== + // RETENTION / CLEANUP + // =========================================================================== + + /** + * Clean up old executions by age + */ + cleanupByAge(maxAgeMs: number): number { + this.initSchema(); + + const cutoff = Date.now() - maxAgeMs; + const stmt = this.db.prepare('DELETE FROM workflow_executions WHERE started_at < ?'); + const result = stmt.run(cutoff); + + return result.changes; + } + + /** + * Clean up executions, keeping only N most recent per workflow + */ + cleanupByCount(maxCount: number, workflowId?: string): number { + this.initSchema(); + + let totalDeleted = 0; + + if (workflowId) { + // Clean up specific workflow + totalDeleted = this.cleanupWorkflowByCount(workflowId, maxCount); + } else { + // Clean up all workflows + const workflows = this.db.prepare('SELECT DISTINCT workflow_id FROM workflow_executions').all() as { + workflow_id: string; + }[]; + + for (const { workflow_id } of workflows) { + totalDeleted += this.cleanupWorkflowByCount(workflow_id, maxCount); + } + } + + return totalDeleted; + } + + /** + * Clean up a specific workflow by count + */ + private cleanupWorkflowByCount(workflowId: string, maxCount: number): number { + // Get IDs to keep + const keepStmt = this.db.prepare(` + SELECT id FROM workflow_executions + WHERE workflow_id = ? + ORDER BY started_at DESC + LIMIT ? + `); + const keepIds = (keepStmt.all(workflowId, maxCount) as { id: string }[]).map((r) => r.id); + + if (keepIds.length === 0) { + return 0; + } + + // Delete others + const placeholders = keepIds.map(() => '?').join(','); + const deleteStmt = this.db.prepare(` + DELETE FROM workflow_executions + WHERE workflow_id = ? AND id NOT IN (${placeholders}) + `); + const result = deleteStmt.run(workflowId, ...keepIds); + + return result.changes; + } + + /** + * Apply retention policy + */ + applyRetentionPolicy(policy: RetentionPolicy): number { + let totalDeleted = 0; + + if (policy.maxAgeMs) { + totalDeleted += this.cleanupByAge(policy.maxAgeMs); + } + + if (policy.maxCountPerWorkflow) { + totalDeleted += this.cleanupByCount(policy.maxCountPerWorkflow); + } + + if (policy.maxTotalCount) { + const countStmt = this.db.prepare('SELECT COUNT(*) as count FROM workflow_executions'); + const { count } = countStmt.get() as { count: number }; + + if (count > policy.maxTotalCount) { + // Delete oldest executions + const toDelete = count - policy.maxTotalCount; + const deleteStmt = this.db.prepare(` + DELETE FROM workflow_executions + WHERE id IN ( + SELECT id FROM workflow_executions + ORDER BY started_at ASC + LIMIT ? + ) + `); + const result = deleteStmt.run(toDelete); + totalDeleted += result.changes; + } + } + + return totalDeleted; + } + + // =========================================================================== + // STATISTICS + // =========================================================================== + + /** + * Get aggregated execution statistics + */ + getStats(workflowId?: string): ExecutionStats { + this.initSchema(); + + let sql = ` + SELECT + COUNT(*) as total, + SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END) as success_count, + SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) as error_count, + SUM(CASE WHEN status = 'running' THEN 1 ELSE 0 END) as running_count, + AVG(duration_ms) as avg_duration, + MIN(duration_ms) as min_duration, + MAX(duration_ms) as max_duration + FROM workflow_executions + `; + + const params: unknown[] = []; + if (workflowId) { + sql += ' WHERE workflow_id = ?'; + params.push(workflowId); + } + + const row = this.db.prepare(sql).get(...params) as Record; + + const total = row.total ?? 0; + const successCount = row.success_count ?? 0; + + return { + totalExecutions: total, + successCount, + errorCount: row.error_count ?? 0, + runningCount: row.running_count ?? 0, + avgDurationMs: row.avg_duration ?? 0, + minDurationMs: row.min_duration ?? 0, + maxDurationMs: row.max_duration ?? 0, + successRate: total > 0 ? successCount / total : 0 + }; + } + + // =========================================================================== + // HELPERS + // =========================================================================== + + /** + * Convert database row to WorkflowExecution + */ + private rowToExecution(row: Record): WorkflowExecution { + return { + id: row.id as string, + workflowId: row.workflow_id as string, + workflowName: row.workflow_name as string, + triggerType: row.trigger_type as WorkflowExecution['triggerType'], + triggerData: parseData(row.trigger_data as string | null), + status: row.status as WorkflowExecution['status'], + startedAt: row.started_at as number, + completedAt: (row.completed_at as number) || undefined, + durationMs: (row.duration_ms as number) || undefined, + errorMessage: (row.error_message as string) || undefined, + errorStack: (row.error_stack as string) || undefined, + metadata: parseData(row.metadata as string | null) + }; + } + + /** + * Convert database row to ExecutionStep + */ + private rowToStep(row: Record): ExecutionStep { + return { + id: row.id as string, + executionId: row.execution_id as string, + nodeId: row.node_id as string, + nodeType: row.node_type as string, + nodeName: (row.node_name as string) || undefined, + stepIndex: row.step_index as number, + startedAt: row.started_at as number, + completedAt: (row.completed_at as number) || undefined, + durationMs: (row.duration_ms as number) || undefined, + status: row.status as ExecutionStep['status'], + inputData: parseData(row.input_data as string | null), + outputData: parseData(row.output_data as string | null), + errorMessage: (row.error_message as string) || undefined + }; + } +} diff --git a/packages/noodl-viewer-cloud/src/execution-history/types.ts b/packages/noodl-viewer-cloud/src/execution-history/types.ts new file mode 100644 index 0000000..f6c392c --- /dev/null +++ b/packages/noodl-viewer-cloud/src/execution-history/types.ts @@ -0,0 +1,222 @@ +/** + * Execution History Type Definitions + * + * Types for storing and querying workflow execution history. + * Used by ExecutionStore and the execution history UI. + * + * @module execution-history/types + */ + +/** + * Status of a workflow execution + */ +export type ExecutionStatus = 'running' | 'success' | 'error'; + +/** + * Status of an individual execution step (node) + */ +export type StepStatus = 'running' | 'success' | 'error' | 'skipped'; + +/** + * Type of trigger that started the workflow + */ +export type TriggerType = 'webhook' | 'schedule' | 'manual' | 'db_change' | 'internal_event' | 'test'; + +/** + * Workflow execution record + * + * Represents a single run of a workflow, from start to completion. + */ +export interface WorkflowExecution { + /** Unique execution ID */ + id: string; + + /** ID of the workflow component */ + workflowId: string; + + /** Display name of the workflow */ + workflowName: string; + + /** What triggered this execution */ + triggerType: TriggerType; + + /** Trigger-specific data (request body, cron expression, etc.) */ + triggerData?: Record; + + /** Current execution status */ + status: ExecutionStatus; + + /** Unix timestamp (ms) when execution started */ + startedAt: number; + + /** Unix timestamp (ms) when execution completed */ + completedAt?: number; + + /** Total duration in milliseconds */ + durationMs?: number; + + /** Error message if status is 'error' */ + errorMessage?: string; + + /** Error stack trace if available */ + errorStack?: string; + + /** Additional metadata */ + metadata?: Record; +} + +/** + * Execution step record + * + * Represents the execution of a single node within a workflow. + */ +export interface ExecutionStep { + /** Unique step ID */ + id: string; + + /** Parent execution ID */ + executionId: string; + + /** ID of the node in the graph */ + nodeId: string; + + /** Type of node (e.g., 'noodl.logic.condition') */ + nodeType: string; + + /** Display name of the node */ + nodeName?: string; + + /** Order in which this step was executed (0-based) */ + stepIndex: number; + + /** Unix timestamp (ms) when step started */ + startedAt: number; + + /** Unix timestamp (ms) when step completed */ + completedAt?: number; + + /** Step duration in milliseconds */ + durationMs?: number; + + /** Step execution status */ + status: StepStatus; + + /** Input data received by the node (may be truncated) */ + inputData?: Record; + + /** Output data produced by the node (may be truncated) */ + outputData?: Record; + + /** Error message if step failed */ + errorMessage?: string; +} + +/** + * Query parameters for filtering executions + */ +export interface ExecutionQuery { + /** Filter by workflow ID */ + workflowId?: string; + + /** Filter by status */ + status?: ExecutionStatus; + + /** Filter by trigger type */ + triggerType?: TriggerType; + + /** Filter executions started after this timestamp (ms) */ + startedAfter?: number; + + /** Filter executions started before this timestamp (ms) */ + startedBefore?: number; + + /** Maximum number of results to return */ + limit?: number; + + /** Number of results to skip (for pagination) */ + offset?: number; + + /** Field to sort by */ + orderBy?: 'started_at' | 'duration_ms' | 'status'; + + /** Sort direction */ + orderDir?: 'asc' | 'desc'; +} + +/** + * Execution with all its steps + */ +export interface ExecutionWithSteps extends WorkflowExecution { + /** All steps in this execution */ + steps: ExecutionStep[]; +} + +/** + * Aggregated execution statistics + */ +export interface ExecutionStats { + /** Total number of executions */ + totalExecutions: number; + + /** Number of successful executions */ + successCount: number; + + /** Number of failed executions */ + errorCount: number; + + /** Number of running executions */ + runningCount: number; + + /** Average execution duration (ms) */ + avgDurationMs: number; + + /** Minimum execution duration (ms) */ + minDurationMs: number; + + /** Maximum execution duration (ms) */ + maxDurationMs: number; + + /** Success rate (0-1) */ + successRate: number; +} + +/** + * Retention policy configuration + */ +export interface RetentionPolicy { + /** Maximum age of executions to keep (ms) */ + maxAgeMs?: number; + + /** Maximum number of executions to keep per workflow */ + maxCountPerWorkflow?: number; + + /** Maximum total number of executions */ + maxTotalCount?: number; + + /** Whether to keep failed executions longer */ + keepFailedExecutions?: boolean; +} + +/** + * Options for creating an execution + */ +export type CreateExecutionOptions = Omit; + +/** + * Options for adding a step + */ +export type CreateStepOptions = Omit; + +/** + * Options for updating an execution + */ +export type UpdateExecutionOptions = Partial< + Pick +>; + +/** + * Options for updating a step + */ +export type UpdateStepOptions = Partial< + Pick +>; diff --git a/packages/noodl-viewer-cloud/tests/execution-history.test.ts b/packages/noodl-viewer-cloud/tests/execution-history.test.ts new file mode 100644 index 0000000..438466d --- /dev/null +++ b/packages/noodl-viewer-cloud/tests/execution-history.test.ts @@ -0,0 +1,572 @@ +/** + * Unit tests for ExecutionStore + * + * Tests CRUD operations, querying, pagination, and retention policies. + */ + +import { + ExecutionStore, + SQLiteDatabase, + type CreateExecutionOptions, + type CreateStepOptions +} from '../src/execution-history'; + +/** + * Mock SQLite database for testing + */ +class MockDatabase implements SQLiteDatabase { + private tables: Map[]> = new Map(); + private schemaInitialized = false; + + exec(sql: string): void { + // Just mark schema as initialized + this.schemaInitialized = true; + this.tables.set('workflow_executions', []); + this.tables.set('execution_steps', []); + } + + prepare(sql: string) { + const self = this; + + return { + run(...params: unknown[]): { changes: number } { + const lowerSql = sql.toLowerCase().trim(); + + if (lowerSql.startsWith('insert into workflow_executions')) { + const executions = self.tables.get('workflow_executions') || []; + executions.push({ + id: params[0], + workflow_id: params[1], + workflow_name: params[2], + trigger_type: params[3], + trigger_data: params[4], + status: params[5], + started_at: params[6], + completed_at: params[7], + duration_ms: params[8], + error_message: params[9], + error_stack: params[10], + metadata: params[11] + }); + return { changes: 1 }; + } + + if (lowerSql.startsWith('insert into execution_steps')) { + const steps = self.tables.get('execution_steps') || []; + steps.push({ + id: params[0], + execution_id: params[1], + node_id: params[2], + node_type: params[3], + node_name: params[4], + step_index: params[5], + started_at: params[6], + completed_at: params[7], + duration_ms: params[8], + status: params[9], + input_data: params[10], + output_data: params[11], + error_message: params[12] + }); + return { changes: 1 }; + } + + if (lowerSql.startsWith('update workflow_executions')) { + const executions = self.tables.get('workflow_executions') || []; + const id = params[params.length - 1]; + const exec = executions.find((e) => e.id === id); + if (exec) { + // Simple update - in real tests we'd parse the SET clause + const statusIdx = params.findIndex((p) => ['running', 'success', 'error'].includes(p as string)); + if (statusIdx >= 0) exec.status = params[statusIdx]; + return { changes: 1 }; + } + return { changes: 0 }; + } + + if (lowerSql.startsWith('update execution_steps')) { + const steps = self.tables.get('execution_steps') || []; + const id = params[params.length - 1]; + const step = steps.find((s) => s.id === id); + if (step) { + return { changes: 1 }; + } + return { changes: 0 }; + } + + if (lowerSql.startsWith('delete from workflow_executions')) { + const executions = self.tables.get('workflow_executions') || []; + const id = params[0]; + const idx = executions.findIndex((e) => e.id === id); + if (idx >= 0) { + executions.splice(idx, 1); + // Also delete steps (cascade) + const steps = self.tables.get('execution_steps') || []; + const newSteps = steps.filter((s) => s.execution_id !== id); + self.tables.set('execution_steps', newSteps); + return { changes: 1 }; + } + return { changes: 0 }; + } + + return { changes: 0 }; + }, + + get(...params: unknown[]): unknown { + const lowerSql = sql.toLowerCase().trim(); + + if (lowerSql.includes('from workflow_executions where id')) { + const executions = self.tables.get('workflow_executions') || []; + return executions.find((e) => e.id === params[0]) || undefined; + } + + if (lowerSql.includes('count(*)')) { + const executions = self.tables.get('workflow_executions') || []; + return { count: executions.length }; + } + + if (lowerSql.includes('select') && lowerSql.includes('avg')) { + const executions = self.tables.get('workflow_executions') || []; + const success = executions.filter((e) => e.status === 'success').length; + const error = executions.filter((e) => e.status === 'error').length; + const running = executions.filter((e) => e.status === 'running').length; + return { + total: executions.length, + success_count: success, + error_count: error, + running_count: running, + avg_duration: 100, + min_duration: 50, + max_duration: 200 + }; + } + + return undefined; + }, + + all(...params: unknown[]): unknown[] { + const lowerSql = sql.toLowerCase().trim(); + + if (lowerSql.includes('from workflow_executions')) { + const executions = self.tables.get('workflow_executions') || []; + let results = [...executions]; + + // Apply basic filtering + if (lowerSql.includes('where')) { + // Simple workflow_id filter + const workflowIdx = params.findIndex((p) => typeof p === 'string' && p.startsWith('wf-')); + if (workflowIdx >= 0) { + results = results.filter((e) => e.workflow_id === params[workflowIdx]); + } + } + + // Apply limit + const limitIdx = lowerSql.indexOf('limit'); + if (limitIdx > 0) { + const limitParam = params.find((p) => typeof p === 'number'); + if (limitParam) { + results = results.slice(0, limitParam as number); + } + } + + return results; + } + + if (lowerSql.includes('from execution_steps')) { + const steps = self.tables.get('execution_steps') || []; + const executionId = params[0]; + return steps.filter((s) => s.execution_id === executionId); + } + + if (lowerSql.includes('distinct workflow_id')) { + const executions = self.tables.get('workflow_executions') || []; + const ids = [...new Set(executions.map((e) => e.workflow_id))]; + return ids.map((id) => ({ workflow_id: id })); + } + + return []; + } + }; + } + + // Test helpers + getExecutions(): Record[] { + return this.tables.get('workflow_executions') || []; + } + + getSteps(): Record[] { + return this.tables.get('execution_steps') || []; + } +} + +describe('ExecutionStore', () => { + let db: MockDatabase; + let store: ExecutionStore; + + beforeEach(() => { + db = new MockDatabase(); + store = new ExecutionStore(db); + }); + + describe('createExecution', () => { + it('should create an execution and return an ID', () => { + const options: CreateExecutionOptions = { + workflowId: 'wf-123', + workflowName: 'Test Workflow', + triggerType: 'manual', + status: 'running', + startedAt: Date.now() + }; + + const id = store.createExecution(options); + + expect(id).toBeDefined(); + expect(id).toMatch(/^exec_/); + expect(db.getExecutions()).toHaveLength(1); + }); + + it('should store all execution properties', () => { + const now = Date.now(); + const options: CreateExecutionOptions = { + workflowId: 'wf-456', + workflowName: 'Full Workflow', + triggerType: 'webhook', + triggerData: { path: '/api/hook', method: 'POST' }, + status: 'running', + startedAt: now, + metadata: { version: '1.0' } + }; + + store.createExecution(options); + + const exec = db.getExecutions()[0]; + expect(exec.workflow_id).toBe('wf-456'); + expect(exec.workflow_name).toBe('Full Workflow'); + expect(exec.trigger_type).toBe('webhook'); + expect(exec.status).toBe('running'); + expect(exec.started_at).toBe(now); + }); + }); + + describe('getExecution', () => { + it('should return an execution by ID', () => { + const id = store.createExecution({ + workflowId: 'wf-test', + workflowName: 'Test', + triggerType: 'manual', + status: 'running', + startedAt: Date.now() + }); + + const exec = store.getExecution(id); + + expect(exec).not.toBeNull(); + expect(exec?.id).toBe(id); + expect(exec?.workflowId).toBe('wf-test'); + }); + + it('should return null for non-existent ID', () => { + const exec = store.getExecution('non-existent'); + expect(exec).toBeNull(); + }); + }); + + describe('updateExecution', () => { + it('should update execution status', () => { + const id = store.createExecution({ + workflowId: 'wf-update', + workflowName: 'Update Test', + triggerType: 'manual', + status: 'running', + startedAt: Date.now() + }); + + store.updateExecution(id, { status: 'success' }); + + // In real implementation we'd verify the update + // Mock just confirms no errors thrown + expect(true).toBe(true); + }); + }); + + describe('addStep', () => { + it('should add a step to an execution', () => { + const execId = store.createExecution({ + workflowId: 'wf-steps', + workflowName: 'Steps Test', + triggerType: 'manual', + status: 'running', + startedAt: Date.now() + }); + + const stepOptions: CreateStepOptions = { + executionId: execId, + nodeId: 'node-1', + nodeType: 'noodl.logic.condition', + stepIndex: 0, + startedAt: Date.now(), + status: 'running', + inputData: { value: true } + }; + + const stepId = store.addStep(stepOptions); + + expect(stepId).toBeDefined(); + expect(stepId).toMatch(/^step_/); + expect(db.getSteps()).toHaveLength(1); + }); + + it('should store step input and output data', () => { + const execId = store.createExecution({ + workflowId: 'wf-data', + workflowName: 'Data Test', + triggerType: 'manual', + status: 'running', + startedAt: Date.now() + }); + + store.addStep({ + executionId: execId, + nodeId: 'node-2', + nodeType: 'noodl.data.transform', + stepIndex: 0, + startedAt: Date.now(), + status: 'success', + inputData: { items: [1, 2, 3] }, + outputData: { result: 6 } + }); + + const step = db.getSteps()[0]; + expect(step.input_data).toBeDefined(); + expect(step.output_data).toBeDefined(); + }); + }); + + describe('getStepsForExecution', () => { + it('should return all steps for an execution', () => { + const execId = store.createExecution({ + workflowId: 'wf-multi-step', + workflowName: 'Multi Step', + triggerType: 'manual', + status: 'running', + startedAt: Date.now() + }); + + store.addStep({ + executionId: execId, + nodeId: 'node-1', + nodeType: 'type-1', + stepIndex: 0, + startedAt: Date.now(), + status: 'success' + }); + + store.addStep({ + executionId: execId, + nodeId: 'node-2', + nodeType: 'type-2', + stepIndex: 1, + startedAt: Date.now(), + status: 'success' + }); + + const steps = store.getStepsForExecution(execId); + + expect(steps).toHaveLength(2); + }); + }); + + describe('getExecutionWithSteps', () => { + it('should return execution with all steps', () => { + const execId = store.createExecution({ + workflowId: 'wf-with-steps', + workflowName: 'With Steps', + triggerType: 'manual', + status: 'running', + startedAt: Date.now() + }); + + store.addStep({ + executionId: execId, + nodeId: 'node-1', + nodeType: 'type-1', + stepIndex: 0, + startedAt: Date.now(), + status: 'success' + }); + + const result = store.getExecutionWithSteps(execId); + + expect(result).not.toBeNull(); + expect(result?.id).toBe(execId); + expect(result?.steps).toBeDefined(); + expect(result?.steps.length).toBe(1); + }); + + it('should return null for non-existent execution', () => { + const result = store.getExecutionWithSteps('non-existent'); + expect(result).toBeNull(); + }); + }); + + describe('queryExecutions', () => { + it('should return all executions by default', () => { + store.createExecution({ + workflowId: 'wf-1', + workflowName: 'Workflow 1', + triggerType: 'manual', + status: 'success', + startedAt: Date.now() + }); + + store.createExecution({ + workflowId: 'wf-2', + workflowName: 'Workflow 2', + triggerType: 'webhook', + status: 'error', + startedAt: Date.now() + }); + + const results = store.queryExecutions(); + + expect(results.length).toBe(2); + }); + + it('should filter by workflowId', () => { + store.createExecution({ + workflowId: 'wf-filter-1', + workflowName: 'Filter 1', + triggerType: 'manual', + status: 'success', + startedAt: Date.now() + }); + + store.createExecution({ + workflowId: 'wf-filter-2', + workflowName: 'Filter 2', + triggerType: 'manual', + status: 'success', + startedAt: Date.now() + }); + + const results = store.queryExecutions({ workflowId: 'wf-filter-1' }); + + expect(results.length).toBe(1); + expect(results[0].workflowId).toBe('wf-filter-1'); + }); + + it('should respect limit parameter', () => { + for (let i = 0; i < 5; i++) { + store.createExecution({ + workflowId: 'wf-limit', + workflowName: `Limit ${i}`, + triggerType: 'manual', + status: 'success', + startedAt: Date.now() + }); + } + + const results = store.queryExecutions({ limit: 3 }); + + expect(results.length).toBe(3); + }); + }); + + describe('deleteExecution', () => { + it('should delete an execution', () => { + const id = store.createExecution({ + workflowId: 'wf-delete', + workflowName: 'Delete Test', + triggerType: 'manual', + status: 'success', + startedAt: Date.now() + }); + + expect(db.getExecutions()).toHaveLength(1); + + store.deleteExecution(id); + + expect(db.getExecutions()).toHaveLength(0); + }); + + it('should cascade delete steps', () => { + const execId = store.createExecution({ + workflowId: 'wf-cascade', + workflowName: 'Cascade Test', + triggerType: 'manual', + status: 'success', + startedAt: Date.now() + }); + + store.addStep({ + executionId: execId, + nodeId: 'node-1', + nodeType: 'type-1', + stepIndex: 0, + startedAt: Date.now(), + status: 'success' + }); + + expect(db.getSteps()).toHaveLength(1); + + store.deleteExecution(execId); + + expect(db.getSteps()).toHaveLength(0); + }); + }); + + describe('getStats', () => { + it('should return aggregated statistics', () => { + store.createExecution({ + workflowId: 'wf-stats', + workflowName: 'Stats 1', + triggerType: 'manual', + status: 'success', + startedAt: Date.now() + }); + + store.createExecution({ + workflowId: 'wf-stats', + workflowName: 'Stats 2', + triggerType: 'manual', + status: 'error', + startedAt: Date.now() + }); + + const stats = store.getStats(); + + expect(stats.totalExecutions).toBe(2); + expect(stats.successCount).toBe(1); + expect(stats.errorCount).toBe(1); + }); + }); + + describe('cleanupByAge', () => { + it('should clean up old executions', () => { + // This tests the method can be called without error + // Real implementation would test with actual timestamps + const deleted = store.cleanupByAge(24 * 60 * 60 * 1000); // 24 hours + expect(deleted).toBeGreaterThanOrEqual(0); + }); + }); + + describe('cleanupByCount', () => { + it('should keep only N most recent executions', () => { + // This tests the method can be called without error + const deleted = store.cleanupByCount(10); + expect(deleted).toBeGreaterThanOrEqual(0); + }); + }); + + describe('applyRetentionPolicy', () => { + it('should apply multiple retention rules', () => { + const deleted = store.applyRetentionPolicy({ + maxAgeMs: 7 * 24 * 60 * 60 * 1000, // 7 days + maxCountPerWorkflow: 100, + maxTotalCount: 1000 + }); + + expect(deleted).toBeGreaterThanOrEqual(0); + }); + }); +});