# 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).

```json
{
  "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.

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

#### `set_variable`
Set a single context variable.

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

#### `delay`
Wait before continuing to the next step.

```json
{
  "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`.

```json
{
  "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`.

```json
{
  "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`.

```json
{
  "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`.

```json
{
  "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`.

```json
{
  "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`.

```json
{
  "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.

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

#### `create_note`
Add an internal note to an Autotask ticket.

```json
{
  "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`.

```json
{
  "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:
```json
{
  "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).

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

For Teams, you can provide a custom Adaptive Card:
```json
{
  "channel_id": 1,
  "card_template": {
    "type": "message",
    "attachments": [{ "contentType": "application/vnd.microsoft.card.adaptive", "content": { ... } }]
  }
}
```

For ntfy, you can set priority and title:
```json
{
  "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.

```json
{
  "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`.

```json
{
  "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`.

```json
{
  "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
```