Executor

class Executor extends AsyncEventEmitter {
  // Methods
  execute(workflow: Workflow, options?: ExecutionOptions): Promise<ExecutionResult>;
  cancel(executionId: string, reason?: CancellationReason): void;
  cancelAll(reason?: CancellationReason): void;
  isExecutionActive(executionId: string): boolean;
  
  // Event Methods (inherited from AsyncEventEmitter)
  on(event: string, listener: Function): this;
  once(event: string, listener: Function): this;
  off(event: string, listener: Function): this;
  emit(event: string, ...args: any[]): Promise<boolean>;
}

execute(workflow: Workflow, options?: ExecutionOptions): Promise<ExecutionResult>;

import { Executor } from '@crystalflow/core';

const executor = new Executor();

const result = await executor.execute(workflow, {
  timeout: 30000,
  variables: { apiKey: 'xxx' }
});

console.log(result.status); // 'success'
console.log(result.duration); // 1523 (ms)

cancel(executionId: string, reason?: CancellationReason): void;

let executionId: string;

executor.on('beforeExecute', (context) => {
  executionId = context.executionId;
});

// Cancel after 5 seconds
setTimeout(() => {
  executor.cancel(executionId, CancellationReason.UserCancelled);
}, 5000);

cancelAll(reason?: CancellationReason): void;

executor.cancelAll(CancellationReason.UserCancelled);

isExecutionActive(executionId: string): boolean;

if (executor.isExecutionActive(executionId)) {
  console.log('Execution is running');
}

executor.on('beforeExecute', (context: ExecutionContext) => {
  console.log(`Starting execution: ${context.executionId}`);
});

executor.on('afterExecute', (result: ExecutionResult) => {
  console.log(`Execution ${result.status}: ${result.duration}ms`);
});

executor.on('onNodeStart', (nodeId: string, node: Node) => {
  console.log(`Node ${nodeId} starting...`);
});

executor.on('onNodeComplete', (nodeId: string, result: NodeExecutionResult) => {
  console.log(`Node ${nodeId} completed:`, result.outputs);
});

executor.on('onNodeError', (nodeId: string, error: Error) => {
  console.error(`Node ${nodeId} failed:`, error);
});

executor.on('onError', (error: Error) => {
  console.error('Workflow failed:', error);
});

executor.on('onCancellation', (
  executionId: string,
  reason: CancellationReason,
  cancelledAt: Date
) => {
  console.log(`Execution ${executionId} cancelled: ${reason}`);
});

interface ExecutionOptions {
  timeout?: number;
  variables?: Record<string, any>;
  abortSignal?: AbortSignal;
}

interface ExecutionResult {
  id: string;
  workflowId: string;
  status: 'success' | 'failed' | 'cancelled';
  startTime: Date;
  endTime?: Date;
  duration?: number;
  nodeResults: Map<string, NodeExecutionResult>;
  error?: Error;
  cancellationReason?: CancellationReason;
  cancelledAt?: Date;
}

interface NodeExecutionResult {
  nodeId: string;
  state: NodeState;
  outputs: Record<string, any>;
  error?: Error;
  startTime: Date;
  endTime?: Date;
  duration?: number;
}

interface ExecutionContext {
  executionId: string;
  workflowId: string;
  startTime: Date;
  variables: Record<string, any>;
  nodeResults: Map<string, NodeExecutionResult>;
}

enum CancellationReason {
  Timeout = 'timeout',
  UserCancelled = 'user-cancelled',
  ExternalSignal = 'external-signal',
  ResourceLimit = 'resource-limit'
}

import { Executor } from '@crystalflow/core';

const executor = new Executor();

const result = await executor.execute(workflow);
console.log(result.status); // 'success'

const executor = new Executor();

executor.on('beforeExecute', (context) => {
  console.log(`Starting: ${context.executionId}`);
});

executor.on('onNodeStart', (nodeId, node) => {
  console.log(`Executing ${node.type} (${nodeId})`);
});

executor.on('onNodeComplete', (nodeId, result) => {
  console.log(`Completed ${nodeId}:`, result.outputs);
});

executor.on('afterExecute', (result) => {
  console.log(`Finished in ${result.duration}ms`);
});

await executor.execute(workflow);

executor.on('onNodeComplete', async (nodeId, result) => {
  // Executor waits for async handlers
  await saveToDatabase(result);
  await sendWebhook(result);
  console.log('Persisted node result');
});

await executor.execute(workflow);

try {
  const result = await executor.execute(workflow, {
    timeout: 10000 // 10 seconds
  });
} catch (error) {
  if (error instanceof TimeoutError) {
    console.error('Execution timed out');
  }
}

let executionId: string;

executor.on('beforeExecute', (context) => {
  executionId = context.executionId;
});

executor.on('onCancellation', (id, reason) => {
  console.log(`Cancelled: ${reason}`);
});

// Start execution
const promise = executor.execute(workflow);

// Cancel after 5 seconds
setTimeout(() => {
  executor.cancel(executionId);
}, 5000);

try {
  await promise;
} catch (error) {
  if (error instanceof CancellationError) {
    console.log('Execution was cancelled');
  }
}

const controller = new AbortController();

// Start execution
const promise = executor.execute(workflow, {
  abortSignal: controller.signal
});

// Cancel from outside
setTimeout(() => controller.abort(), 5000);

try {
  await promise;
} catch (error) {
  console.error('Cancelled:', error);
}

const totalNodes = workflow.nodes.size;
let completedNodes = 0;

executor.on('onNodeStart', (nodeId) => {
  console.log(`Progress: ${completedNodes}/${totalNodes}`);
});

executor.on('onNodeComplete', (nodeId) => {
  completedNodes++;
  const progress = (completedNodes / totalNodes) * 100;
  console.log(`Progress: ${progress.toFixed(0)}%`);
});

await executor.execute(workflow);

executor.on('onNodeError', async (nodeId, error) => {
  console.error(`Node ${nodeId} failed:`, error);
  
  // Log to external service
  await logError({
    nodeId,
    error: error.message,
    timestamp: new Date()
  });
});

executor.on('onError', async (error) => {
  // Workflow failed - send notification
  await sendAlertEmail({
    subject: 'Workflow Failed',
    body: error.message
  });
});

try {
  await executor.execute(workflow);
} catch (error) {
  // Handle gracefully
  console.log('Execution failed, but errors were logged');
}