-- Pipeline Engine Tables -- Webhook-triggered automation pipelines with multi-step execution, -- notification channels, and human-in-the-loop approvals. -- ============================================================================ -- Notification Channels — UI-configured notification providers -- ============================================================================ CREATE TABLE IF NOT EXISTS notification_channels ( id SERIAL PRIMARY KEY, name VARCHAR(200) NOT NULL, channel_type VARCHAR(20) NOT NULL, -- 'teams', 'telegram', 'ntfy', 'webhook' config JSONB NOT NULL DEFAULT '{}', -- type-specific: webhook_url, bot_token, chat_id, topic, etc. is_active BOOLEAN DEFAULT true, created_at TIMESTAMP DEFAULT NOW(), updated_at TIMESTAMP DEFAULT NOW() ); -- ============================================================================ -- Webhook Pipelines — workflow definitions -- ============================================================================ CREATE TABLE IF NOT EXISTS webhook_pipelines ( id SERIAL PRIMARY KEY, name VARCHAR(200) NOT NULL, description TEXT, is_active BOOLEAN DEFAULT true, trigger_source VARCHAR(50) NOT NULL, -- 'datto_rmm', 'autotask', 'veeam', 'manual' trigger_conditions JSONB NOT NULL DEFAULT '[]', -- array of {field, operator, value} sort_order INTEGER DEFAULT 0, created_at TIMESTAMP DEFAULT NOW(), updated_at TIMESTAMP DEFAULT NOW() ); -- ============================================================================ -- Pipeline Steps — ordered actions within a pipeline -- ============================================================================ CREATE TABLE IF NOT EXISTS pipeline_steps ( id SERIAL PRIMARY KEY, pipeline_id INTEGER NOT NULL REFERENCES webhook_pipelines(id) ON DELETE CASCADE, step_order INTEGER NOT NULL, step_type VARCHAR(50) NOT NULL, -- 'filter','transform','enrich_device','create_ticket','notify','approval','rmm_quick_job', etc. name VARCHAR(200) NOT NULL, config JSONB NOT NULL DEFAULT '{}', on_failure VARCHAR(20) DEFAULT 'stop', -- 'continue', 'stop', 'skip_to' skip_to_step INTEGER, is_active BOOLEAN DEFAULT true, timeout_ms INTEGER, created_at TIMESTAMP DEFAULT NOW(), updated_at TIMESTAMP DEFAULT NOW() ); -- ============================================================================ -- Pipeline Executions — runtime log -- ============================================================================ CREATE TABLE IF NOT EXISTS pipeline_executions ( id SERIAL PRIMARY KEY, pipeline_id INTEGER NOT NULL REFERENCES webhook_pipelines(id) ON DELETE CASCADE, trigger_source VARCHAR(50) NOT NULL, trigger_payload JSONB, status VARCHAR(20) NOT NULL DEFAULT 'pending', -- 'pending','running','waiting','completed','failed','skipped' current_step INTEGER, context JSONB NOT NULL DEFAULT '{}', -- accumulated data from steps started_at TIMESTAMP DEFAULT NOW(), completed_at TIMESTAMP, duration_ms INTEGER, error_message TEXT, created_at TIMESTAMP DEFAULT NOW() ); -- ============================================================================ -- Pipeline Execution Steps — per-step log -- ============================================================================ CREATE TABLE IF NOT EXISTS pipeline_execution_steps ( id SERIAL PRIMARY KEY, execution_id INTEGER NOT NULL REFERENCES pipeline_executions(id) ON DELETE CASCADE, step_order INTEGER NOT NULL, step_type VARCHAR(50) NOT NULL, step_name VARCHAR(200), status VARCHAR(20) NOT NULL DEFAULT 'pending', -- 'pending','running','completed','failed','waiting','skipped' input_data JSONB, output_data JSONB, started_at TIMESTAMP, completed_at TIMESTAMP, duration_ms INTEGER, error_message TEXT ); -- ============================================================================ -- Approval Requests — human-in-the-loop -- ============================================================================ CREATE TABLE IF NOT EXISTS approval_requests ( id SERIAL PRIMARY KEY, execution_id INTEGER NOT NULL REFERENCES pipeline_executions(id) ON DELETE CASCADE, step_order INTEGER NOT NULL, channel_id INTEGER REFERENCES notification_channels(id), message TEXT NOT NULL, options JSONB NOT NULL DEFAULT '["Approve","Reject"]', status VARCHAR(20) NOT NULL DEFAULT 'pending', -- 'pending','approved','rejected','timeout' responded_by TEXT, responded_at TIMESTAMP, response_data JSONB, expires_at TIMESTAMP, created_at TIMESTAMP DEFAULT NOW() ); -- ============================================================================ -- Indexes -- ============================================================================ CREATE INDEX IF NOT EXISTS idx_notification_channels_type ON notification_channels(channel_type, is_active); CREATE INDEX IF NOT EXISTS idx_webhook_pipelines_source ON webhook_pipelines(trigger_source, is_active, sort_order); CREATE INDEX IF NOT EXISTS idx_pipeline_steps_pipeline ON pipeline_steps(pipeline_id, step_order); CREATE INDEX IF NOT EXISTS idx_pipeline_executions_pipeline ON pipeline_executions(pipeline_id, created_at DESC); CREATE INDEX IF NOT EXISTS idx_pipeline_executions_status ON pipeline_executions(status, created_at DESC); CREATE INDEX IF NOT EXISTS idx_pipeline_execution_steps_exec ON pipeline_execution_steps(execution_id, step_order); CREATE INDEX IF NOT EXISTS idx_approval_requests_exec ON approval_requests(execution_id); CREATE INDEX IF NOT EXISTS idx_approval_requests_status ON approval_requests(status, expires_at); -- ============================================================================ -- SEED: Example pipeline — "RMM Alert → Autotask Ticket" -- ============================================================================ INSERT INTO webhook_pipelines (name, description, is_active, trigger_source, trigger_conditions, sort_order) VALUES ('RMM Alert → Autotask Ticket', 'Creates an Autotask ticket when a Datto RMM alert fires (triggered=True). Enriches with device and company data.', false, 'datto_rmm', '[{"field": "triggered", "operator": "equals", "value": "True"}]', 10); INSERT INTO pipeline_steps (pipeline_id, step_order, step_type, name, config) VALUES ((SELECT id FROM webhook_pipelines WHERE name = 'RMM Alert → Autotask Ticket'), 1, 'transform', 'Extract alert fields', '{ "mappings": { "alert_type": "{{trigger.alert_type}}", "alert_priority": "{{trigger.alert_priority}}", "alert_message": "{{trigger.alert_message_en}}", "device_hostname": "{{trigger.device_hostname}}", "device_uid": "{{trigger.device_uid}}", "site_name": "{{trigger.site_name}}", "site_uid": "{{trigger.site_uid}}", "device_ip": "{{trigger.device_ip}}", "device_os": "{{trigger.device_os}}", "last_user": "{{trigger.last_user}}" } }'), ((SELECT id FROM webhook_pipelines WHERE name = 'RMM Alert → Autotask Ticket'), 2, 'enrich_company', 'Lookup company from site', '{ "lookup_by": "site_name", "source_field": "{{context.site_name}}" }'), ((SELECT id FROM webhook_pipelines WHERE name = 'RMM Alert → Autotask Ticket'), 3, 'create_ticket', 'Create Autotask ticket', '{ "template": { "title": "[RMM {{context.alert_type}}] {{context.device_hostname}} - {{context.alert_message}}", "description": "Datto RMM Alert\n\nType: {{context.alert_type}}\nPriority: {{context.alert_priority}}\nDevice: {{context.device_hostname}} ({{context.device_ip}})\nOS: {{context.device_os}}\nSite: {{context.site_name}}\nLast User: {{context.last_user}}\n\nMessage:\n{{context.alert_message}}", "companyID": "{{context.company_id}}", "ticketType": 2, "ticketCategory": 3, "priority": 1, "queueID": 29682833 } }');