408 lines
15 KiB
TypeScript
408 lines
15 KiB
TypeScript
/**
|
|
* Pipeline Engine
|
|
* Matches incoming webhooks to pipelines, executes steps sequentially,
|
|
* resolves template variables, and accumulates context between steps.
|
|
*/
|
|
|
|
import { postgresClient } from './postgres-client';
|
|
import {
|
|
WebhookPipeline,
|
|
PipelineStep,
|
|
PipelineWithSteps,
|
|
PipelineContext,
|
|
PipelineStatus,
|
|
StepExecutorResult,
|
|
TriggerCondition,
|
|
} from '../types/pipeline';
|
|
|
|
// Step executor registry — populated by individual step files
|
|
type StepExecutorFn = (
|
|
step: PipelineStep,
|
|
context: PipelineContext,
|
|
executionId: number
|
|
) => Promise<StepExecutorResult>;
|
|
|
|
const stepExecutors: Map<string, StepExecutorFn> = new Map();
|
|
|
|
export function registerStepExecutor(stepType: string, executor: StepExecutorFn): void {
|
|
stepExecutors.set(stepType, executor);
|
|
}
|
|
|
|
export class PipelineEngine {
|
|
/**
|
|
* Find and execute all matching pipelines for a trigger source + payload.
|
|
* Called from webhook routes after raw logging.
|
|
*/
|
|
async processTrigger(
|
|
triggerSource: string,
|
|
payload: Record<string, any>
|
|
): Promise<number[]> {
|
|
const pipelines = await this.findMatchingPipelines(triggerSource, payload);
|
|
|
|
if (pipelines.length === 0) {
|
|
return [];
|
|
}
|
|
|
|
console.log(`[PIPELINE] ${pipelines.length} pipeline(s) matched for ${triggerSource}`);
|
|
|
|
const executionIds: number[] = [];
|
|
for (const pipeline of pipelines) {
|
|
try {
|
|
const execId = await this.executePipeline(pipeline, triggerSource, payload);
|
|
executionIds.push(execId);
|
|
} catch (err) {
|
|
console.error(`[PIPELINE] Failed to execute pipeline "${pipeline.name}":`, err);
|
|
}
|
|
}
|
|
|
|
return executionIds;
|
|
}
|
|
|
|
/**
|
|
* Find active pipelines matching the trigger source and conditions.
|
|
*/
|
|
async findMatchingPipelines(
|
|
triggerSource: string,
|
|
payload: Record<string, any>
|
|
): Promise<PipelineWithSteps[]> {
|
|
const result = await postgresClient.query<WebhookPipeline>(
|
|
`SELECT * FROM webhook_pipelines
|
|
WHERE is_active = true AND trigger_source = $1
|
|
ORDER BY sort_order`,
|
|
[triggerSource]
|
|
);
|
|
|
|
const matched: PipelineWithSteps[] = [];
|
|
|
|
for (const pipeline of result.rows) {
|
|
const conditions: TriggerCondition[] = Array.isArray(pipeline.trigger_conditions)
|
|
? pipeline.trigger_conditions
|
|
: [];
|
|
|
|
if (this.evaluateConditions(conditions, payload)) {
|
|
const stepsResult = await postgresClient.query<PipelineStep>(
|
|
`SELECT * FROM pipeline_steps
|
|
WHERE pipeline_id = $1 AND is_active = true
|
|
ORDER BY step_order`,
|
|
[pipeline.id]
|
|
);
|
|
matched.push({ ...pipeline, steps: stepsResult.rows });
|
|
}
|
|
}
|
|
|
|
return matched;
|
|
}
|
|
|
|
/**
|
|
* Execute a single pipeline: create execution record, run steps, update status.
|
|
*/
|
|
async executePipeline(
|
|
pipeline: PipelineWithSteps,
|
|
triggerSource: string,
|
|
payload: Record<string, any>
|
|
): Promise<number> {
|
|
const execResult = await postgresClient.query<{ id: number }>(
|
|
`INSERT INTO pipeline_executions (pipeline_id, trigger_source, trigger_payload, status)
|
|
VALUES ($1, $2, $3, 'running')
|
|
RETURNING id`,
|
|
[pipeline.id, triggerSource, JSON.stringify(payload)]
|
|
);
|
|
const executionId = execResult.rows[0].id;
|
|
|
|
const context: PipelineContext = { trigger: payload };
|
|
let finalStatus: PipelineStatus = 'completed';
|
|
let errorMessage: string | null = null;
|
|
|
|
console.log(`[PIPELINE] Executing "${pipeline.name}" (exec #${executionId}), ${pipeline.steps.length} steps`);
|
|
|
|
for (const step of pipeline.steps) {
|
|
// Update current step
|
|
await postgresClient.query(
|
|
`UPDATE pipeline_executions SET current_step = $1, context = $2 WHERE id = $3`,
|
|
[step.step_order, JSON.stringify(context), executionId]
|
|
);
|
|
|
|
const stepStart = Date.now();
|
|
|
|
// Log step start
|
|
await postgresClient.query(
|
|
`INSERT INTO pipeline_execution_steps (execution_id, step_order, step_type, step_name, status, started_at, input_data)
|
|
VALUES ($1, $2, $3, $4, 'running', NOW(), $5)`,
|
|
[executionId, step.step_order, step.step_type, step.name, JSON.stringify({ config: step.config })]
|
|
);
|
|
|
|
const executor = stepExecutors.get(step.step_type);
|
|
if (!executor) {
|
|
const err = `No executor registered for step type: ${step.step_type}`;
|
|
console.error(`[PIPELINE] ${err}`);
|
|
await this.updateStepLog(executionId, step.step_order, 'failed', null, err, Date.now() - stepStart);
|
|
|
|
if (step.on_failure === 'stop') {
|
|
finalStatus = 'failed';
|
|
errorMessage = err;
|
|
break;
|
|
}
|
|
continue;
|
|
}
|
|
|
|
try {
|
|
// Resolve template variables in step config
|
|
const resolvedConfig = this.resolveTemplates(step.config, context);
|
|
const resolvedStep = { ...step, config: resolvedConfig };
|
|
|
|
const result = await executor(resolvedStep, context, executionId);
|
|
const duration = Date.now() - stepStart;
|
|
|
|
if (result.waiting) {
|
|
await this.updateStepLog(executionId, step.step_order, 'waiting', result.output, null, duration);
|
|
finalStatus = 'waiting';
|
|
break;
|
|
}
|
|
|
|
if (result.success) {
|
|
// Merge output into context
|
|
if (result.output) {
|
|
Object.assign(context, result.output);
|
|
}
|
|
await this.updateStepLog(executionId, step.step_order, 'completed', result.output, null, duration);
|
|
console.log(`[PIPELINE] Step ${step.step_order} "${step.name}" completed (${duration}ms)`);
|
|
} else {
|
|
await this.updateStepLog(executionId, step.step_order, 'failed', result.output, result.error || null, duration);
|
|
console.error(`[PIPELINE] Step ${step.step_order} "${step.name}" failed: ${result.error}`);
|
|
|
|
if (step.on_failure === 'stop') {
|
|
finalStatus = 'failed';
|
|
errorMessage = `Step ${step.step_order} "${step.name}": ${result.error}`;
|
|
break;
|
|
} else if (step.on_failure === 'skip_to' && step.skip_to_step) {
|
|
// Skip ahead — handled by finding the next step with matching order
|
|
// For simplicity, we just continue; the skip_to logic would need step reordering
|
|
continue;
|
|
}
|
|
// on_failure === 'continue' → keep going
|
|
}
|
|
} catch (err) {
|
|
const errMsg = err instanceof Error ? err.message : String(err);
|
|
const duration = Date.now() - stepStart;
|
|
await this.updateStepLog(executionId, step.step_order, 'failed', null, errMsg, duration);
|
|
console.error(`[PIPELINE] Step ${step.step_order} "${step.name}" threw: ${errMsg}`);
|
|
|
|
if (step.on_failure === 'stop') {
|
|
finalStatus = 'failed';
|
|
errorMessage = errMsg;
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
// Finalize execution
|
|
await postgresClient.query(
|
|
`UPDATE pipeline_executions
|
|
SET status = $1, context = $2, completed_at = NOW(),
|
|
duration_ms = EXTRACT(EPOCH FROM (NOW() - started_at)) * 1000,
|
|
error_message = $3
|
|
WHERE id = $4`,
|
|
[finalStatus, JSON.stringify(context), errorMessage, executionId]
|
|
);
|
|
|
|
console.log(`[PIPELINE] Execution #${executionId} finished: ${finalStatus}`);
|
|
return executionId;
|
|
}
|
|
|
|
/**
|
|
* Resume a waiting pipeline (e.g., after approval callback).
|
|
*/
|
|
async resumeExecution(executionId: number, approvalResult: Record<string, any>): Promise<void> {
|
|
const execResult = await postgresClient.query<any>(
|
|
`SELECT pe.*, wp.name as pipeline_name FROM pipeline_executions pe
|
|
JOIN webhook_pipelines wp ON wp.id = pe.pipeline_id
|
|
WHERE pe.id = $1 AND pe.status = 'waiting'`,
|
|
[executionId]
|
|
);
|
|
|
|
if (execResult.rows.length === 0) {
|
|
throw new Error(`Execution #${executionId} not found or not in waiting state`);
|
|
}
|
|
|
|
const execution = execResult.rows[0];
|
|
const context: PipelineContext = execution.context || {};
|
|
context.approval_result = approvalResult;
|
|
|
|
// Get remaining steps after the current waiting step
|
|
const stepsResult = await postgresClient.query<PipelineStep>(
|
|
`SELECT * FROM pipeline_steps
|
|
WHERE pipeline_id = $1 AND step_order > $2 AND is_active = true
|
|
ORDER BY step_order`,
|
|
[execution.pipeline_id, execution.current_step]
|
|
);
|
|
|
|
// Update execution to running
|
|
await postgresClient.query(
|
|
`UPDATE pipeline_executions SET status = 'running', context = $1 WHERE id = $2`,
|
|
[JSON.stringify(context), executionId]
|
|
);
|
|
|
|
// Mark the waiting step as completed
|
|
await this.updateStepLog(executionId, execution.current_step, 'completed', approvalResult, null, 0);
|
|
|
|
// Continue executing remaining steps
|
|
let finalStatus: PipelineStatus = 'completed';
|
|
let errorMessage: string | null = null;
|
|
|
|
for (const step of stepsResult.rows) {
|
|
await postgresClient.query(
|
|
`UPDATE pipeline_executions SET current_step = $1, context = $2 WHERE id = $3`,
|
|
[step.step_order, JSON.stringify(context), executionId]
|
|
);
|
|
|
|
const stepStart = Date.now();
|
|
await postgresClient.query(
|
|
`INSERT INTO pipeline_execution_steps (execution_id, step_order, step_type, step_name, status, started_at, input_data)
|
|
VALUES ($1, $2, $3, $4, 'running', NOW(), $5)`,
|
|
[executionId, step.step_order, step.step_type, step.name, JSON.stringify({ config: step.config })]
|
|
);
|
|
|
|
const executor = stepExecutors.get(step.step_type);
|
|
if (!executor) {
|
|
const err = `No executor for: ${step.step_type}`;
|
|
await this.updateStepLog(executionId, step.step_order, 'failed', null, err, Date.now() - stepStart);
|
|
if (step.on_failure === 'stop') { finalStatus = 'failed'; errorMessage = err; break; }
|
|
continue;
|
|
}
|
|
|
|
try {
|
|
const resolvedConfig = this.resolveTemplates(step.config, context);
|
|
const result = await executor({ ...step, config: resolvedConfig }, context, executionId);
|
|
const duration = Date.now() - stepStart;
|
|
|
|
if (result.waiting) {
|
|
await this.updateStepLog(executionId, step.step_order, 'waiting', result.output, null, duration);
|
|
finalStatus = 'waiting';
|
|
break;
|
|
}
|
|
|
|
if (result.success) {
|
|
if (result.output) Object.assign(context, result.output);
|
|
await this.updateStepLog(executionId, step.step_order, 'completed', result.output, null, duration);
|
|
} else {
|
|
await this.updateStepLog(executionId, step.step_order, 'failed', result.output, result.error || null, duration);
|
|
if (step.on_failure === 'stop') { finalStatus = 'failed'; errorMessage = result.error ?? null; break; }
|
|
}
|
|
} catch (err) {
|
|
const errMsg = err instanceof Error ? err.message : String(err);
|
|
await this.updateStepLog(executionId, step.step_order, 'failed', null, errMsg, Date.now() - stepStart);
|
|
if (step.on_failure === 'stop') { finalStatus = 'failed'; errorMessage = errMsg; break; }
|
|
}
|
|
}
|
|
|
|
await postgresClient.query(
|
|
`UPDATE pipeline_executions
|
|
SET status = $1, context = $2, completed_at = NOW(),
|
|
duration_ms = EXTRACT(EPOCH FROM (NOW() - started_at)) * 1000,
|
|
error_message = $3
|
|
WHERE id = $4`,
|
|
[finalStatus, JSON.stringify(context), errorMessage, executionId]
|
|
);
|
|
}
|
|
|
|
// ============================================================================
|
|
// Template Resolution
|
|
// ============================================================================
|
|
|
|
/**
|
|
* Recursively resolve {{...}} template variables in any value.
|
|
*/
|
|
resolveTemplates(value: any, context: PipelineContext): any {
|
|
if (typeof value === 'string') {
|
|
return this.resolveStringTemplate(value, context);
|
|
}
|
|
if (Array.isArray(value)) {
|
|
return value.map(v => this.resolveTemplates(v, context));
|
|
}
|
|
if (value && typeof value === 'object') {
|
|
const resolved: Record<string, any> = {};
|
|
for (const [k, v] of Object.entries(value)) {
|
|
resolved[k] = this.resolveTemplates(v, context);
|
|
}
|
|
return resolved;
|
|
}
|
|
return value;
|
|
}
|
|
|
|
private resolveStringTemplate(template: string, context: PipelineContext): string {
|
|
return template.replace(/\{\{([^}]+)\}\}/g, (match, path: string) => {
|
|
const value = this.getNestedValue(context, path.trim());
|
|
if (value === undefined || value === null) return '';
|
|
return String(value);
|
|
});
|
|
}
|
|
|
|
private getNestedValue(obj: any, path: string): any {
|
|
const parts = path.split('.');
|
|
let current = obj;
|
|
for (const part of parts) {
|
|
if (current == null) return undefined;
|
|
current = current[part];
|
|
}
|
|
return current;
|
|
}
|
|
|
|
// ============================================================================
|
|
// Condition Evaluation
|
|
// ============================================================================
|
|
|
|
evaluateConditions(conditions: TriggerCondition[], payload: Record<string, any>): boolean {
|
|
if (conditions.length === 0) return true;
|
|
return conditions.every(cond => this.evaluateCondition(cond, payload));
|
|
}
|
|
|
|
private evaluateCondition(cond: TriggerCondition, payload: Record<string, any>): boolean {
|
|
const fieldValue = this.getNestedValue(payload, cond.field);
|
|
|
|
switch (cond.operator) {
|
|
case 'equals':
|
|
return String(fieldValue) === String(cond.value);
|
|
case 'not_equals':
|
|
return String(fieldValue) !== String(cond.value);
|
|
case 'contains':
|
|
return fieldValue != null && String(fieldValue).toLowerCase().includes(String(cond.value).toLowerCase());
|
|
case 'not_contains':
|
|
return fieldValue == null || !String(fieldValue).toLowerCase().includes(String(cond.value).toLowerCase());
|
|
case 'in':
|
|
return Array.isArray(cond.value) && cond.value.some((v: any) => String(v) === String(fieldValue));
|
|
case 'not_in':
|
|
return !Array.isArray(cond.value) || !cond.value.some((v: any) => String(v) === String(fieldValue));
|
|
case 'regex':
|
|
try { return fieldValue != null && new RegExp(String(cond.value), 'i').test(String(fieldValue)); }
|
|
catch { return false; }
|
|
case 'exists':
|
|
return fieldValue != null && fieldValue !== '';
|
|
case 'not_exists':
|
|
return fieldValue == null || fieldValue === '';
|
|
default:
|
|
return false;
|
|
}
|
|
}
|
|
|
|
// ============================================================================
|
|
// Helpers
|
|
// ============================================================================
|
|
|
|
private async updateStepLog(
|
|
executionId: number,
|
|
stepOrder: number,
|
|
status: string,
|
|
outputData: any,
|
|
errorMessage: string | null,
|
|
durationMs: number
|
|
): Promise<void> {
|
|
await postgresClient.query(
|
|
`UPDATE pipeline_execution_steps
|
|
SET status = $1, output_data = $2, error_message = $3, duration_ms = $4, completed_at = NOW()
|
|
WHERE execution_id = $5 AND step_order = $6`,
|
|
[status, outputData ? JSON.stringify(outputData) : null, errorMessage, durationMs, executionId, stepOrder]
|
|
);
|
|
}
|
|
}
|
|
|
|
export const pipelineEngine = new PipelineEngine();
|