lorentz/wulf-pulse
1
0
Fork 0
Code Issues 1 Pull requests Projects Releases Packages Wiki Activity Actions
wulf-pulse/docs/webhook-pipeline-engine.md

22 KiB
Raw Permalink Blame History

Webhook Pipeline Engine

The Pipeline Engine is a DB-driven automation system that processes incoming webhooks through configurable multi-step pipelines. It supports ticket creation, cross-system enrichment, RMM quick jobs, AI analysis, notifications (Teams, Telegram, ntfy), and human-in-the-loop approvals.

Architecture

Webhook Source → Raw Log → Trigger Match → Pipeline Steps → Actions
     │                         │                │              │
  Datto RMM              conditions on       Filter,       Create ticket,
  Autotask               payload fields      Enrich,       Run quick job,
  Veeam                                      Transform,    Send notification,
  Manual                                     AI analyze,   Wait for approval,
                                             Delay         Update Autotask

When a webhook arrives at /api/webhooks/datto-rmm, the route:

  1. Logs the raw payload to datto_rmm_webhook_logs
  2. Calls pipelineEngine.processTrigger('datto_rmm', payload) (fire-and-forget)
  3. Returns 200 OK immediately

The engine then finds all active pipelines matching the trigger source and conditions, and executes each one sequentially.

Concepts

Pipelines

A pipeline is a named workflow triggered by a specific webhook source. Each pipeline has:

  • Trigger source — datto_rmm, autotask, veeam, or manual
  • Trigger conditions — JSON array of conditions that the incoming payload must match
  • Steps — ordered list of actions to execute
  • Active/inactive toggle — disabled pipelines are skipped

Steps

Each step has a type, a name, a JSON config, and failure handling. Steps execute in order. Each step can read from and write to a shared context object.

Context

The context is a JSONB object that accumulates data as steps execute:

  • context.trigger — the original webhook payload
  • Step outputs are merged into the top level (e.g., context.company_id, context.ticket_id)

Template Variables

Step configs support {{variable}} syntax that is resolved before execution:

  • {{trigger.device_hostname}} — field from the webhook payload
  • {{context.company_id}} — field set by a previous step
  • Nested paths work: {{trigger.nested.field}}
  • Templates resolve recursively through objects and arrays

Notification Channels

Channels are configured in the UI and referenced by ID in notify and approval steps. Each channel stores its type-specific credentials:

Type Config Fields
Teams webhook_url
Telegram bot_token, chat_id, parse_mode
ntfy server_url, topic, auth_token, default_priority
Webhook url, method, headers

Step Types

Logic

filter

Evaluate conditions against the current context. If conditions fail, the step fails (and with on_failure: stop, the pipeline stops).

{
  "conditions": [
    { "field": "trigger.alert_priority", "operator": "equals", "value": "CRITICAL" }
  ]
}

Operators: equals, not_equals, contains, not_contains, in, not_in, regex, exists, not_exists

transform

Map payload fields into context variables. All values are template-resolved.

{
  "mappings": {
    "device_hostname": "{{trigger.device_hostname}}",
    "alert_type": "{{trigger.alert_type}}",
    "site_name": "{{trigger.site_name}}"
  }
}

set_variable

Set a single context variable.

{
  "key": "severity",
  "value": "{{trigger.alert_priority}}"
}

delay

Wait before continuing to the next step.

{
  "seconds": 30
}

Enrichment

enrich_device

Look up a device from the local datto_rmm_devices table by UID. Outputs: device, device_found, device_hostname, device_os, device_ip, company_id, company_name.

{
  "lookup_by": "device_uid",
  "source_field": "{{trigger.device_uid}}"
}

enrich_company

Look up an Autotask company from datto_rmm_sites by site name or UID. Falls back to fuzzy match on companies table. Outputs: company_id, company_name, company_found.

{
  "lookup_by": "site_name",
  "source_field": "{{context.site_name}}"
}

enrich_ticket

Look up a ticket from the local tickets table. Outputs: ticket, ticket_found, ticket_id, ticket_number, ticket_title.

{
  "lookup_by": "ticket_number",
  "source_field": "{{context.ticket_number}}"
}

enrich_vspc

Query local Veeam VSPC data for a device's backup status. Searches backup agent jobs, server jobs, protected workloads, and active alarms. Outputs: vspc_found, vspc_last_job_status, vspc_last_success, vspc_hours_since_success, vspc_failure_message, vspc_restore_points, vspc_backed_up_size, vspc_alarm_count, vspc_failed_job_count, vspc_summary, vspc_agent_jobs, vspc_server_jobs, vspc_workloads, vspc_alarms.

{
  "lookup_by": "device_name",
  "source_field": "{{context.device_hostname}}"
}

Data

db_query

Run a parameterized read-only SQL query against local Postgres. Only SELECT statements are allowed — mutations are blocked. Outputs: {output_key} (rows or single row), {output_key}_count.

{
  "query": "SELECT status, COUNT(*) as count FROM veeam_backup_agent_jobs WHERE LOWER(name) LIKE LOWER($1) AND last_run > NOW() - INTERVAL '7 days' GROUP BY status",
  "params": ["%{{context.device_hostname}}%"],
  "output_key": "backup_trend",
  "single_row": false
}

Autotask Actions

create_ticket

Create a ticket in Autotask via the API. Numeric fields (companyID, ticketType, priority, queueID, etc.) are auto-converted. Outputs: ticket_id, ticket_number, created_ticket.

{
  "template": {
    "title": "[RMM {{context.alert_type}}] {{context.device_hostname}}",
    "description": "Alert: {{context.alert_message}}\nDevice: {{context.device_hostname}}\nSite: {{context.site_name}}",
    "companyID": "{{context.company_id}}",
    "ticketType": 2,
    "priority": 1,
    "queueID": 29682833,
    "status": 1
  }
}

update_ticket

Update fields on an existing Autotask ticket.

{
  "ticket_id": "{{context.ticket_id}}",
  "fields": {
    "priority": 4,
    "queueID": 29682833
  }
}

create_note

Add an internal note to an Autotask ticket.

{
  "ticket_id": "{{context.ticket_id}}",
  "title": "Pipeline Note",
  "body": "AI Analysis:\n{{context.ai_response}}",
  "note_type": 1,
  "publish": 1
}

AI

ai_analyze

Send a prompt to OpenAI or Anthropic. Uses AI settings from workflow_settings table. Outputs: ai_response, ai_provider, ai_model.

{
  "system_prompt": "You are an IT operations assistant.",
  "prompt": "Summarize this RMM alert for a technician:\n\nType: {{context.alert_type}}\nDevice: {{context.device_hostname}}\nMessage: {{context.alert_message}}",
  "provider": "openai",
  "model": "gpt-4o",
  "max_tokens": 500
}

Optionally reference a saved prompt template:

{
  "prompt_template_id": 1,
  "prompt": "..."
}

Notifications

notify

Send a notification to a configured channel. The channel is referenced by channel_id (from the Notification Channels UI).

{
  "channel_id": 1,
  "message": "RMM Alert: {{context.device_hostname}} - {{context.alert_message}}",
  "title": "RMM Alert"
}

For Teams, you can provide a custom Adaptive Card:

{
  "channel_id": 1,
  "card_template": {
    "type": "message",
    "attachments": [{ "contentType": "application/vnd.microsoft.card.adaptive", "content": { ... } }]
  }
}

For ntfy, you can set priority and title:

{
  "channel_id": 2,
  "message": "Alert on {{context.device_hostname}}",
  "title": "Critical Alert",
  "priority": "urgent"
}

approval

Send an approval request and pause the pipeline until a human responds. The response comes via a callback URL.

{
  "channel_id": 1,
  "message": "Auto-remediate {{context.device_hostname}}?",
  "options": ["Approve", "Reject", "Escalate"],
  "timeout_min": 60
}

When the approval is sent:

  • Teams — Adaptive Card with action buttons (each opens the callback URL)
  • Telegram — Message with inline keyboard buttons
  • ntfy — Push notification with action buttons

The callback URL is POST /api/pipelines/approval/{approval_id}?response=Approve. After the response, the pipeline resumes with context.approval_result containing the response data.

RMM Actions

rmm_quick_job

Run a Datto RMM quick job (automation component) on a device. Outputs: quick_job_result, job_uid.

{
  "device_uid": "{{context.device_uid}}",
  "component_uid": "comp-dns-flush-001",
  "job_name": "DNS Cache Flush",
  "variables": [
    { "name": "LogPath", "value": "C:\\Logs" }
  ]
}

To find available components, use GET /api/rmm/components.

rmm_get_job_results

Poll for quick job results. Outputs: job_results, job_status.

{
  "job_uid": "{{context.job_uid}}",
  "device_uid": "{{context.device_uid}}"
}

Failure Handling

Each step has an on_failure setting:

Value Behavior
stop Stop the pipeline, mark as failed (default)
continue Log the error and continue to the next step
skip_to Jump to a specific step number

API Reference

Pipelines

Method Endpoint Description
GET /api/pipelines List all pipelines (optional ?source=datto_rmm)
POST /api/pipelines Create a pipeline
GET /api/pipelines/{id} Get pipeline with steps and recent executions
PUT /api/pipelines/{id} Update pipeline settings
DELETE /api/pipelines/{id} Delete pipeline and all steps
GET /api/pipelines/{id}/steps List steps
POST /api/pipelines/{id}/steps Add a step
PUT /api/pipelines/{id}/steps Replace all steps (body: { steps: [...] })
GET /api/pipelines/{id}/executions Execution history (optional ?limit=50)
POST /api/pipelines/{id}/test Test with sample payload (body: { payload: {...} })

Notification Channels

Method Endpoint Description
GET /api/notification-channels List all channels
POST /api/notification-channels Create a channel
GET /api/notification-channels/{id} Get channel
PUT /api/notification-channels/{id} Update channel
DELETE /api/notification-channels/{id} Delete channel
POST /api/notification-channels/{id}/test Send test notification

Approval Callback

Method Endpoint Description
GET/POST /api/pipelines/approval/{id}?response=Approve Respond to approval request

Query params: response (required), by (optional — who approved).

RMM Components

Method Endpoint Description
GET /api/rmm/components List available Datto RMM automation components

UI Pages

Workflow Dashboard (/admin/workflow)

The main workflow page now includes two new navigation cards:

  • Webhook Pipelines — manage automation pipelines
  • Notification Channels — configure notification destinations

Pipeline List (/admin/workflow/pipelines)

  • View all pipelines with trigger source, step count, and active status
  • Toggle pipelines on/off
  • Create new pipelines
  • Delete pipelines

Pipeline Editor (/admin/workflow/pipelines/{id})

Four tabs:

  • Steps — visual step builder with drag ordering, inline JSON config editor, add/remove/reorder steps
  • Trigger — edit pipeline name, description, trigger source, and trigger conditions (JSON)
  • Test — paste a sample payload and run the pipeline in real-time, see step-by-step results
  • History — view recent execution results with status and timing

Notification Channels (/admin/workflow/channels)

  • Add channels: Teams (webhook URL), Telegram (bot token + chat ID), ntfy (topic + server), Generic Webhook (URL + method)
  • Edit and delete channels
  • Test button — sends a test notification to verify the channel works
  • Toggle channels active/inactive

Database Schema

notification_channels

Column Type Description
id SERIAL PK
name VARCHAR(200) Human-readable name
channel_type VARCHAR(20) teams, telegram, ntfy, webhook
config JSONB Type-specific credentials and settings
is_active BOOLEAN
created_at / updated_at TIMESTAMP

webhook_pipelines

Column Type Description
id SERIAL PK
name VARCHAR(200)
description TEXT
is_active BOOLEAN
trigger_source VARCHAR(50) datto_rmm, autotask, veeam, manual
trigger_conditions JSONB Array of {field, operator, value}
sort_order INTEGER Lower = higher priority
created_at / updated_at TIMESTAMP

pipeline_steps

Column Type Description
id SERIAL PK
pipeline_id FK → webhook_pipelines
step_order INTEGER Execution order
step_type VARCHAR(50) See step types above
name VARCHAR(200) Human label
config JSONB Step-specific configuration
on_failure VARCHAR(20) stop, continue, skip_to
skip_to_step INTEGER Target step for skip_to
is_active BOOLEAN
timeout_ms INTEGER Max wait for approval steps

pipeline_executions

Column Type Description
id SERIAL PK
pipeline_id FK
trigger_source VARCHAR(50)
trigger_payload JSONB Raw webhook data
status VARCHAR(20) pending, running, waiting, completed, failed, skipped
current_step INTEGER
context JSONB Accumulated data from all steps
started_at / completed_at TIMESTAMP
duration_ms INTEGER
error_message TEXT

pipeline_execution_steps

Column Type Description
id SERIAL PK
execution_id FK
step_order INTEGER
step_type VARCHAR(50)
step_name VARCHAR(200)
status VARCHAR(20) pending, running, completed, failed, waiting, skipped
input_data / output_data JSONB
started_at / completed_at TIMESTAMP
duration_ms INTEGER
error_message TEXT

approval_requests

Column Type Description
id SERIAL PK
execution_id FK
step_order INTEGER
channel_id FK → notification_channels
message TEXT
options JSONB e.g., ["Approve", "Reject"]
status VARCHAR(20) pending, approved, rejected, timeout
responded_by TEXT
responded_at TIMESTAMP
response_data JSONB
expires_at TIMESTAMP

Example 1: Simple RMM Alert → Ticket

Pipeline: "RMM Alert → Autotask Ticket"
Trigger: datto_rmm WHERE triggered = "True"

Step 1: transform        → Extract device_uid, site_name, alert_message from payload
Step 2: enrich_company   → Lookup Autotask company from site_name
Step 3: create_ticket    → Create Autotask ticket with alert details

Example 2: Veeam Backup Failure → Full Diagnostic Pipeline

This is the flagship pipeline demonstrating the full power of the engine. When a Veeam backup failure alert arrives from Datto RMM, it:

  1. Extracts and enriches from multiple sources (RMM, VSPC, local DB)
  2. Runs a diagnostic PowerShell script on the affected device
  3. Feeds everything to AI for root cause analysis
  4. Creates a rich Autotask ticket with all findings
  5. Notifies the team via Teams
Pipeline: "Veeam Backup Failure → Smart Diagnostic Ticket"
Trigger: datto_rmm WHERE triggered = "True" AND alert_message contains "Veeam"

Step  1: transform           → Extract device_hostname, device_uid, site_name, alert fields
Step  2: enrich_device       → Lookup full device details from datto_rmm_devices
Step  3: enrich_company      → Lookup Autotask company from site name
Step  4: enrich_vspc         → Query VSPC for backup agent jobs, server jobs,
                                protected workloads, alarms, last success date
Step  5: db_query            → Backup failure trend: job status counts over last 7 days
Step  6: db_query            → Recent RMM alerts for this device (pattern detection)
Step  7: rmm_quick_job       → Run Veeam diagnostic PowerShell script on device:
                                - Check Veeam services (running/stopped)
                                - Check backup job status via VBR snap-in
                                - Check disk space on all drives
                                - Check Windows Event Log for Veeam errors (48h)
                                - Check for stuck Veeam processes (>48h)
                                - Test network connectivity to backup infrastructure
Step  8: delay               → Wait 60s for script execution
Step  9: rmm_get_job_results → Retrieve structured JSON diagnostic output
Step 10: ai_analyze          → Feed ALL data to AI:
                                "Given the alert, VSPC status, backup trends,
                                 recent alerts, and on-device diagnostics —
                                 what is the root cause? Is it recurring?
                                 What are the remediation steps?"
Step 11: create_ticket       → Create rich Autotask ticket with:
                                - VSPC backup status summary
                                - AI root cause analysis
                                - On-device diagnostic results
                                - Backup trend data
                                - Recent alert history
Step 12: create_note         → Add AI analysis as internal ticket note
Step 13: notify (Teams)      → Adaptive Card with summary + ticket link

What the ticket looks like

Instead of the generic alert ticket:

"A Veeam Backup & Replication monitoring policy reported a backup job as missing or stalled for device pgbvsywnp01."

The pipeline produces a ticket like:

[Veeam Backup Failure] pgbvsywnp01 - V-Systems - Main Office

VSPC Backup Status

Agent Jobs: 2 (1 success, 1 failed, 0 warning) Latest Job: "pgbvsywnp01 Backup" — Failed at 2026-02-20 19:30:00 Failure: "Failed to process disk 0 of VM. Error: The backup infrastructure..." Last Success: 2026-02-19 03:15:00 (40h ago) Active Alarms: 1

AI Root Cause Analysis

Root Cause: The Veeam Backup Service (VeeamBackupSvc) is stopped on the device. This was likely caused by a Windows Update that restarted the server but the Veeam services did not auto-start due to a delayed start configuration...

Remediation Steps:

  1. Start the VeeamBackupSvc service
  2. Set startup type to Automatic (not Delayed Start)
  3. Trigger a manual backup run to verify
  4. Monitor for 24h to confirm resolution

On-Device Diagnostics

  • Services: VeeamBackupSvc STOPPED, VeeamBrokerSvc Running
  • Disk: C: 45% used (55GB free), D: 78% used (220GB free)
  • Event Log: 3 Veeam errors in last 48h
  • Network: SQL server reachable, REST API port open

Diagnostic PowerShell Script

The script at scripts/rmm-diagnostics/veeam-backup-diagnostic.ps1 must be uploaded to Datto RMM as a component. It checks:

Check What it does
Veeam Services Checks 11 Veeam service names, reports stopped critical services
Backup Jobs Loads VBR PowerShell snap-in, gets all jobs with last session status
Disk Space All fixed drives, flags >90% used
Event Logs Veeam Backup, Veeam Agent, Application log — errors in last 48h
Processes Running Veeam processes, flags any >48h (stuck)
Network Tests SQL server connectivity, Veeam service ports (9392, 9419, 6180)

Output is structured JSON with checks, issues_found, recommendations, and a severity rating (OK/WARNING/CRITICAL).

Activation Steps

  1. Upload veeam-backup-diagnostic.ps1 to Datto RMM as a component
  2. Copy the component UID
  3. Edit pipeline step 7 → replace REPLACE_WITH_COMPONENT_UID with the real UID
  4. Create a notification channel (Teams webhook) at /admin/workflow/channels
  5. Update step 13 channel_id to match
  6. Toggle the pipeline active

File Structure

lib/
  types/
    pipeline.ts                    # TypeScript types
  services/
    pipeline-engine.ts             # Core engine
    pipeline-steps/
      index.ts                     # Registers all executors
      filter.ts
      transform.ts
      set-variable.ts
      delay.ts
      enrich-device.ts
      enrich-company.ts
      enrich-ticket.ts
      enrich-vspc.ts               # Veeam VSPC backup status lookup
      db-query.ts                  # Parameterized SQL queries
      create-ticket.ts
      update-ticket.ts
      create-note.ts
      ai-analyze.ts
      notify.ts
      approval.ts
      rmm-quick-job.ts

scripts/
  rmm-diagnostics/
    veeam-backup-diagnostic.ps1    # Veeam diagnostic script for RMM quick job

app/
  api/
    pipelines/
      route.ts                     # List + create pipelines
      [id]/
        route.ts                   # Get/update/delete pipeline
        steps/route.ts             # Manage steps
        executions/route.ts        # Execution history
        test/route.ts              # Test with sample payload
      approval/
        [id]/route.ts              # Approval callback
    notification-channels/
      route.ts                     # List + create channels
      [id]/
        route.ts                   # Get/update/delete channel
        test/route.ts              # Send test notification
    rmm/
      components/route.ts          # List RMM components
  admin/
    workflow/
      pipelines/
        page.tsx                   # Pipeline list
        [id]/page.tsx              # Pipeline editor
      channels/
        page.tsx                   # Channel management

components/
  admin/
    pipeline/
      StepConfigEditor.tsx         # Visual step config editors

migrations/
  033_create_pipeline_engine_tables.sql
  034_seed_veeam_backup_failure_pipeline.sql