wulf-pulse/migrations/033_create_pipeline_engine_tables.sql

-- 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
        }
    }');