Files
dashcaddy/dashcaddy-api/src/recipes/bundled-workflows.js
T
Krystie 503de258b8 [grade=pending] QA sprint: commit 103 at-risk files from multi-agent sprint work
Committed by Hermes autonomous QA sprint 2026-08-13.
These files were modified during the Aug 12 sprint but never committed.
2026-08-12 17:34:10 -07:00

666 lines
21 KiB
JavaScript

/**
* Bundled Workflows - Pre-configured automation templates
*
* Workflows attach to events (container-down, pre-update, resource-alert, scheduled)
* and execute a sequence of actions when triggered.
*/
const EventEmitter = require('events');
const fs = require('fs');
const path = require('path');
const { log } = require('../utils/logging');
const platformPaths = require('../../platform-paths');
const WORKFLOWS_FILE = process.env.WORKFLOWS_FILE || path.join(platformPaths.dataDir, 'workflows-config.json');
const WORKFLOW_HISTORY_FILE = process.env.WORKFLOW_HISTORY_FILE || path.join(platformPaths.dataDir, 'workflow-history.json');
/**
* Bundled workflow templates
*/
const BUNDLED_WORKFLOWS = {
'auto-restart-on-crash': {
id: 'auto-restart-on-crash',
name: 'Auto-Restart on Crash',
description: 'Automatically restart a container when it goes down',
trigger: 'container-down',
actions: [
{ type: 'docker-restart', containerId: '{{containerId}}' },
{ type: 'notify', message: 'Container {{containerId}} restarted automatically' }
]
},
'backup-before-update': {
id: 'backup-before-update',
name: 'Backup Before Update',
description: 'Create a backup before any app update',
trigger: 'pre-update',
actions: [
{ type: 'backup-create', appId: '{{appId}}', label: 'pre-update' },
{ type: 'notify', message: 'Backup created before updating {{appId}}' }
]
},
'health-check-on-interval': {
id: 'health-check-on-interval',
name: 'Periodic Health Check',
description: 'Run health checks every 15 minutes and alert if degraded',
trigger: 'scheduled',
interval: 15 * 60 * 1000, // 15 minutes
actions: [
{ type: 'health-check', target: '{{serviceId}}' },
// failingServices is set by healthCheckService when it throws (any
// service failed). It's a comma-joined string of failing service IDs.
// Previously this used {{serviceId}} which never resolved because
// no per-service ID is in scope at the workflow level — that's the
// DC-044 root-cause bug fix.
{ type: 'notify-on-failure', message: 'Health check failed for {{failingServices}}' }
]
},
'disk-space-alert': {
id: 'disk-space-alert',
name: 'Disk Space Alert',
description: 'Alert when disk usage exceeds 80%',
trigger: 'resource-alert',
condition: 'diskPercent > 80',
actions: [
{ type: 'notify', message: '⚠️ Disk usage at {{diskPercent}}% on {{host}}' }
]
},
'weekly-container-report': {
id: 'weekly-container-report',
name: 'Weekly Container Report',
description: 'Send a weekly summary of container status and resource usage',
trigger: 'scheduled',
interval: 7 * 24 * 60 * 60 * 1000, // weekly
actions: [
{ type: 'collect-metrics', period: '7d' },
{ type: 'notify', message: '{{report}}' }
]
}
};
/**
* WorkflowEngine - Executes bundled workflows
*/
class WorkflowEngine extends EventEmitter {
constructor(ctx) {
super();
this.ctx = ctx;
this.enabled = new Map();
this.history = [];
this.scheduledJobs = new Map();
this.loadConfig();
this.loadHistory();
this.startScheduledWorkflows();
}
/**
* Load enabled/disabled state for workflows
*/
loadConfig() {
try {
if (fs.existsSync(WORKFLOWS_FILE)) {
const data = JSON.parse(fs.readFileSync(WORKFLOWS_FILE, 'utf8'));
this.enabled = new Map(Object.entries(data.enabled || {}));
}
} catch (error) {
log.error('workflow', error, { operation: 'loadConfig' });
}
// Default all workflows to enabled if not explicitly set
for (const [id, workflow] of Object.entries(BUNDLED_WORKFLOWS)) {
if (!this.enabled.has(id)) {
this.enabled.set(id, true);
}
}
}
/**
* Save enabled/disabled state
*/
saveConfig() {
try {
const data = {
enabled: Object.fromEntries(this.enabled)
};
fs.writeFileSync(WORKFLOWS_FILE, JSON.stringify(data, null, 2));
} catch (error) {
log.error('workflow', error, { operation: 'saveConfig' });
}
}
/**
* Load execution history
*/
loadHistory() {
try {
if (fs.existsSync(WORKFLOW_HISTORY_FILE)) {
this.history = JSON.parse(fs.readFileSync(WORKFLOW_HISTORY_FILE, 'utf8'));
}
} catch (error) {
log.error('workflow', error, { operation: 'loadHistory' });
this.history = [];
}
}
/**
* Save execution history
*/
saveHistory() {
try {
fs.writeFileSync(WORKFLOW_HISTORY_FILE, JSON.stringify(this.history, null, 2));
} catch (error) {
log.error('workflow', error, { operation: 'saveHistory' });
}
}
/**
* Start scheduled workflows
*/
startScheduledWorkflows() {
for (const [id, workflow] of Object.entries(BUNDLED_WORKFLOWS)) {
if (workflow.trigger === 'scheduled' && workflow.interval) {
this.startScheduledWorkflow(id, workflow);
}
}
}
/**
* Start a scheduled workflow
*/
startScheduledWorkflow(workflowId, workflow) {
if (!this.enabled.get(workflowId)) return;
// Clear existing job if any
this.stopScheduledWorkflow(workflowId);
const job = setInterval(() => {
this.executeWorkflow(workflowId, { trigger: 'scheduled', timestamp: new Date().toISOString() })
.catch(err => log.error('workflow', err, { workflowId, phase: 'scheduled' }));
}, workflow.interval);
this.scheduledJobs.set(workflowId, job);
log.info('workflow', 'Scheduled workflow', { workflowId, intervalMs: workflow.interval });
}
/**
* Stop a scheduled workflow
*/
stopScheduledWorkflow(workflowId) {
if (this.scheduledJobs.has(workflowId)) {
clearInterval(this.scheduledJobs.get(workflowId));
this.scheduledJobs.delete(workflowId);
}
}
/**
* Execute a workflow by ID
*/
async executeWorkflow(workflowId, triggerData = {}) {
const workflow = BUNDLED_WORKFLOWS[workflowId];
if (!workflow) {
throw new Error(`Unknown workflow: ${workflowId}`);
}
if (!this.enabled.get(workflowId)) {
log.info('workflow', 'Workflow disabled, skipping', { workflowId });
return { skipped: true, reason: 'disabled' };
}
const executionId = `${workflowId}-${Date.now()}`;
const startTime = Date.now();
log.info('workflow', 'Executing workflow', { workflowId });
this.emit('workflow-start', { workflowId, executionId, triggerData });
const results = await this._runActions(workflow.actions, triggerData);
const duration = Date.now() - startTime;
const allSucceeded = results.every(r => r.success);
const historyEntry = {
executionId,
workflowId,
workflowName: workflow.name,
trigger: triggerData.trigger || 'manual',
timestamp: new Date().toISOString(),
duration,
success: allSucceeded,
results
};
this.history.push(historyEntry);
// Keep history to last 500 entries
if (this.history.length > 500) {
this.history = this.history.slice(-500);
}
this.saveHistory();
this.emit('workflow-complete', historyEntry);
log.info('workflow', 'Workflow completed', { workflowId, durationMs: duration, success: allSucceeded });
return historyEntry;
}
/**
* Run a sequence of actions and collect their results. Extracted from
* executeWorkflow so the per-action result threading (notify-on-failure
* gating) and the failingServices context surface can be unit-tested
* directly. executeWorkflow() is the production entry point; _runActions
* is an internal helper that callers shouldn't reach for.
*/
async _runActions(actions, triggerData = {}) {
const results = [];
const MAX_RETRIES = 3;
const RETRY_DELAY_MS = 2000;
for (let i = 0; i < actions.length; i++) {
const action = actions[i];
const previousResult = i > 0 ? results[i - 1] : null;
const actionContext = {
...triggerData,
previousResult,
failingServices: previousResult && previousResult.failingServices ? previousResult.failingServices : undefined,
};
// DC-093: Retry with exponential backoff for transient failures
let lastError = null;
let result = null;
let succeeded = false;
for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) {
try {
result = await this.executeAction(action, actionContext);
succeeded = true;
break;
} catch (error) {
lastError = error;
if (attempt < MAX_RETRIES) {
const delay = RETRY_DELAY_MS * Math.pow(2, attempt);
log.warn('workflow', `Action "${action.type}" failed (attempt ${attempt + 1}/${MAX_RETRIES + 1}), retrying in ${delay}ms`, { error: error.message });
await new Promise(resolve => setTimeout(resolve, delay));
}
}
}
if (succeeded) {
results.push({ action: action.type, success: true, result });
} else {
log.error('workflow', `Action "${action.type}" failed after ${MAX_RETRIES + 1} attempts`, { error: lastError.message });
results.push({
action: action.type,
success: false,
error: lastError.message,
failingServices: lastError.failingServices,
exhaustedRetries: MAX_RETRIES + 1,
});
}
}
return results;
}
/**
* Execute a single action
*/
async executeAction(action, context) {
switch (action.type) {
case 'docker-restart':
return this.restartContainer(this.interpolate(action.containerId, context));
case 'backup-create':
return this.createBackup(
this.interpolate(action.appId, context),
action.label
);
case 'notify':
return this.notify(
this.interpolate(action.message, context),
action.channel
);
case 'notify-on-failure':
// Only send if previous action failed (success: false). The
// previousResult is injected by executeWorkflow's loop. If there
// was no previous action, this is a no-op (returns skipped).
if (!context.previousResult || context.previousResult.success !== false) {
return { skipped: true, reason: 'no previous failure' };
}
return this.notify(
this.interpolate(action.message, context),
action.channel
);
case 'health-check':
return this.healthCheckService(this.interpolate(action.target, context));
case 'collect-metrics':
return this.collectMetrics(context.containerId, action.period);
default:
log.warn('workflow', 'Unknown action type', { actionType: action.type });
return { skipped: true, reason: `Unknown action type: ${action.type}` };
}
}
/**
* Interpolate template variables in a string
*/
interpolate(str, context) {
if (!str || typeof str !== 'string') return str;
return str.replace(/\{\{(\w+)\}\}/g, (match, key) => {
return context[key] !== undefined ? context[key] : match;
});
}
/**
* Health check action
*/
async healthCheckService(serviceId) {
if (!serviceId || serviceId === '{{serviceId}}') {
// Run health check on all services
const results = [];
const servicesStateManager = this.ctx.servicesStateManager;
if (servicesStateManager) {
// StateManager exposes async read() — never getState(). The previous
// call site used the wrong method name AND forgot to await, returning
// a Promise instead of an array; the `|| []` short-circuit then made
// every check silently no-op with "Health check failed" errors.
const services = await servicesStateManager.read().catch(() => []) || [];
for (const service of services) {
if (service.containerId) {
try {
const healthy = await this.checkContainerHealth(service.containerId);
results.push({ service: service.id, healthy });
} catch (e) {
results.push({ service: service.id, healthy: false, error: e.message });
}
}
}
}
// Surface failing service IDs so downstream notify-on-failure actions
// can interpolate `{{failingServices}}` into the alert message. Without
// this, templates like `Health check failed for {{serviceId}}` stay
// literal because there's no serviceId in scope.
const failing = results.filter(r => !r.healthy).map(r => r.service);
const healthy = results.filter(r => r.healthy).length;
const result = { checked: results.length, healthy, results, failing };
if (failing.length > 0) {
// Throw so the action's success:false path is taken and notify-on-failure fires.
const err = new Error(`Health check failed for ${failing.length} service(s): ${failing.join(', ')}`);
err.failingServices = failing;
err.workflowResult = result;
throw err;
}
return result;
}
// Single service check
const healthy = await this.checkContainerHealth(serviceId);
if (!healthy) {
const err = new Error(`Health check failed for ${serviceId}`);
err.failingServices = [serviceId];
err.workflowResult = { serviceId, healthy };
throw err;
}
return { serviceId, healthy };
}
/**
* Check if a container is healthy
*/
async checkContainerHealth(containerId) {
try {
const docker = this.ctx.docker?.client;
if (!docker) return false;
const container = docker.getContainer(containerId);
const info = await container.inspect();
// A container is healthy if it's running AND (it has no explicit
// health check OR its health check reports healthy/starting).
// info.State.Health is undefined when no HEALTHCHECK is declared.
// info.State.Health.Status is 'starting' | 'healthy' | 'unhealthy'
// when the health check IS declared.
if (!info.State || !info.State.Running) return false;
if (!info.State.Health) return true; // no health check defined → running = healthy
const status = info.State.Health.Status;
return status === 'healthy' || status === 'starting';
} catch (error) {
return false;
}
}
/**
* Docker restart action
*/
async restartContainer(containerId) {
const docker = this.ctx.docker?.client;
if (!docker) {
throw new Error('Docker client not available');
}
if (!containerId || containerId === '{{containerId}}') {
throw new Error('Container ID not provided');
}
log.info('workflow', 'Restarting container', { containerId });
const container = docker.getContainer(containerId);
await container.restart();
return { restarted: containerId };
}
/**
* Backup create action
*/
async createBackup(appId, label = 'workflow') {
const backupManager = this.ctx.backupManager;
if (!backupManager) {
throw new Error('Backup manager not available');
}
if (!appId || appId === '{{appId}}') {
throw new Error('App ID not provided');
}
log.info('workflow', 'Creating backup', { appId });
// Use backup manager's executeBackup if available
const backupName = `${appId}-${label}`;
const backupConfig = backupManager.config?.backups?.[appId];
if (backupConfig) {
const result = await backupManager.executeBackup(backupName, backupConfig);
return { backupId: result.backupId, appId, label };
}
// Fallback: trigger manual backup via backup manager
if (backupManager.executeBackup) {
const result = await backupManager.executeBackup(appId, {
include: ['config', 'data'],
schedule: 'manual'
});
return { backupId: result.backupId, appId, label };
}
throw new Error('Backup execution not available');
}
/**
* Notify action
*/
async notify(message, channel) {
const notification = this.ctx.notification;
if (!notification) {
log.warn('workflow', 'Notification manager not available');
return { notified: false, reason: 'no notification manager' };
}
log.info('workflow', 'Sending notification', { message });
notification.send('workflow', 'Workflow Notification', message, 'info');
return { notified: true, message };
}
/**
* Collect metrics action
*/
async collectMetrics(containerId, period = '7d') {
const resourceMonitor = this.ctx.resourceMonitor;
if (!resourceMonitor) {
throw new Error('Resource monitor not available');
}
// Get aggregated stats
const stats = resourceMonitor.getAllStats();
// Build report
let report = `# Weekly Container Report\n\n`;
report += `Generated: ${new Date().toLocaleString()}\n\n`;
for (const [id, info] of Object.entries(stats)) {
const agg = info.aggregated;
report += `## ${info.name || id}\n`;
report += `- Status: ${info.current?.status || 'unknown'}\n`;
if (agg) {
report += `- CPU: avg ${agg.cpu?.avg?.toFixed(1)}%, max ${agg.cpu?.max?.toFixed(1)}%\n`;
report += `- Memory: avg ${agg.memory?.avg?.toFixed(1)}%, max ${agg.memory?.max?.toFixed(1)}%\n`;
}
report += '\n';
}
return { report, containerCount: Object.keys(stats).length };
}
/**
* List all available workflows
*/
listWorkflows() {
return Object.entries(BUNDLED_WORKFLOWS).map(([id, workflow]) => ({
...workflow,
enabled: this.enabled.get(id) ?? true
}));
}
/**
* Enable or disable a workflow
*/
setWorkflowEnabled(workflowId, enabled) {
if (!BUNDLED_WORKFLOWS[workflowId]) {
throw new Error(`Unknown workflow: ${workflowId}`);
}
this.enabled.set(workflowId, enabled);
this.saveConfig();
// Handle scheduled workflows
const workflow = BUNDLED_WORKFLOWS[workflowId];
if (workflow.trigger === 'scheduled' && workflow.interval) {
if (enabled) {
this.startScheduledWorkflow(workflowId, workflow);
} else {
this.stopScheduledWorkflow(workflowId);
}
}
log.info('workflow', 'Workflow toggled', { workflowId, enabled });
return { workflowId, enabled };
}
/**
* Get execution history for a workflow
*/
getHistory(workflowId = null, limit = 50) {
let history = this.history;
if (workflowId) {
history = history.filter(h => h.workflowId === workflowId);
}
return history.slice(-limit).reverse();
}
/**
* Trigger workflows for a specific event
*/
async triggerForEvent(eventType, eventData) {
const matchingWorkflows = Object.entries(BUNDLED_WORKFLOWS)
.filter(([id, workflow]) => {
if (workflow.trigger !== eventType) return false;
if (!this.enabled.get(id)) return false;
// Check condition if specified
if (workflow.condition && eventData) {
try {
// Simple condition evaluation
const conditionMet = this.evaluateCondition(workflow.condition, eventData);
return conditionMet;
} catch (e) {
log.warn('workflow', 'Condition evaluation failed', { workflowId: id, error: e.message });
return false;
}
}
return true;
});
const results = [];
for (const [workflowId, workflow] of matchingWorkflows) {
try {
const result = await this.executeWorkflow(workflowId, {
trigger: eventType,
timestamp: new Date().toISOString(),
...eventData
});
results.push({ workflowId, success: true, result });
} catch (error) {
results.push({ workflowId, success: false, error: error.message });
}
}
return results;
}
/**
* Evaluate a simple condition string
*/
evaluateCondition(condition, data) {
// Simple condition like "diskPercent > 80"
// Supports: >, <, >=, <=, ==, !=
const match = condition.match(/^(\w+)\s*(>=|<=|==|!=|>|<)\s*(\S+)$/);
if (!match) return true;
const [, field, operator, value] = match;
const fieldValue = data[field];
if (fieldValue === undefined) return false;
const numValue = parseFloat(value);
const numFieldValue = parseFloat(fieldValue);
switch (operator) {
case '>': return numFieldValue > numValue;
case '<': return numFieldValue < numValue;
case '>=': return numFieldValue >= numValue;
case '<=': return numFieldValue <= numValue;
case '==': return fieldValue == value;
case '!=': return fieldValue != value;
default: return false;
}
}
/**
* Stop all scheduled workflows
*/
stop() {
for (const [workflowId] of this.scheduledJobs) {
this.stopScheduledWorkflow(workflowId);
}
log.info('workflow', 'All scheduled workflows stopped');
}
}
module.exports = { WorkflowEngine, BUNDLED_WORKFLOWS };