davepi-plugin-workflow

n8n/Zapier-style workflow automation for dAvePi. Workflows are tenant-scoped documents holding a React Flow-shaped nodes/edges graph: one trigger node, a chain of action nodes, optional condition branches. Only active workflows execute.

{
  "davepi": {
    "plugins": ["davepi-plugin-workflow"]
  }
}

Registering the plugin gives you two auto-generated resources (full REST/GraphQL/MCP/ Swagger surface, tenant isolation included):

• workflow — definitions. status is a state machine (draft → active ⇄ paused → archived); activate via PUT /api/v1/workflow/:id { "status": "active" }.
• workflow_run — the execution ledger: per-step input/output/error/timing. Engine-managed, read-only for clients, rows expire after WORKFLOW_RUN_TTL_DAYS (default 30).

Node types

| Type | Config | Notes | |------|--------|-------| | trigger_record | event (contact.created, contact.*, *), conditions?: [{ path, op, value }] | Fires on CRUD events in the owner's tenant. | | trigger_schedule | schedule (5-field cron), timezone? | Claimed atomically via nextRunAt CAS — multi-instance safe with no lock server. | | trigger_webhook | requireSignature?: boolean | Mints a per-workflow URL: POST /api/hooks/workflow/:id/:token. | | trigger_manual | — | POST /api/v1/workflow/:id/run (owner-auth'd; drafts allowed for testing). | | create_record | resource, version?, data | Runs the full write pipeline: field ACL, tenant stamping, lifecycle hooks, state-machine initial stamping, audit, bus events. | | update_record | resource, recordId, data | Owner-scoped; validates state-machine transitions; emits .updated / .transitioned. | | delete_record | resource, recordId | Soft-delete unless the schema opts out; beforeDelete hooks can refuse. | | http_request | url, method?, headers?, body?, allowFailure? | SSRF-guarded (framework urlValidator); redirects not followed; non-2xx fails the step unless allowFailure. | | condition | left, op, right? | Ops: eq ne gt gte lt lte contains exists not_exists. Outgoing edges carry branch: "true" / "false". |

Any string in a node's config may use {{path.to.value}} templates over { trigger, steps.<nodeId>, workflow, now }. A whole-string expression keeps the raw type ({{trigger.record.amount}} stays a number); embedded expressions stringify. No eval — plain own-property path walks only.

Execution model

• Every trigger funnels through a run row inserted first (queued) — a partial-unique (workflowId, dedupeKey) index makes concurrent claims race-safe.
• With davepi-plugin-queue installed and QUEUE_REDIS_URL set, runs execute durably via BullMQ (enqueued with attempts: 1 — action nodes are not idempotent, so failed runs are recorded, not replayed). Otherwise runs execute inline on the emitting process; a crash mid-run is flipped to failed / stale by the janitor after WORKFLOW_RUN_STALE_MS.
• Workflows execute as their owner with no roles ({ user_id, roles: [] }): plain owner-scope writes, zero ACL bypasses, even if an admin authored the workflow.
• Loop guard: engine writes emit bus events tagged origin: { plugin: 'workflow', workflowId, runId, depth }. Chained triggers execute at depth + 1, refused past WORKFLOW_MAX_DEPTH (default 3); a workflow never re-triggers itself; workflow.* / workflow_run.* / job.* events never match.

Inbound webhooks

POST /api/hooks/workflow/:workflowId/:token — the token is minted server-side when a workflow saves a trigger_webhook node (rotate by re-saving with the node removed and re-added). With requireSignature: true, senders must supply X-Davepi-Signature: sha256=<hmac-sha256(secret, rawBody)>. Optional X-Idempotency-Key header dedupes replays. Responses are 202 { runId } / 202 { duplicate: true }; every failure is a uniform 404.

Env vars

| Var | Default | | |-----|---------|---| | WORKFLOW_ENABLED | true | false = dormant (nothing registers). | | WORKFLOW_MAX_DEPTH | 3 | Trigger-chain depth cap. | | WORKFLOW_MAX_STEPS | 25 | Per-run step ceiling. | | WORKFLOW_RUN_TTL_DAYS | 30 | Run ledger retention. | | WORKFLOW_HTTP_TIMEOUT_MS | 10000 | http_request timeout (covers body consumption, not just headers). | | WORKFLOW_HTTP_MAX_RESPONSE_BYTES | 1048576 | http_request response-body cap; oversized responses fail the step. | | WORKFLOW_SWEEP_INTERVAL_MS | 60000 | Schedule sweeper / janitor cadence. | | WORKFLOW_STEP_IO_LIMIT | 8192 | Per-step input/output snapshot truncation. | | WORKFLOW_RUN_STALE_MS | 900000 | Stuck-running staleness window. | | WORKFLOW_HOOK_PATH | /api/hooks/workflow | Inbound webhook mount point. |

Timers never run under NODE_ENV=test; suites call plugin.sweepOnce() / plugin.janitorOnce() explicitly.

Known gaps (deliberate)

• Bulk paths skip hooks framework-wide — don't bulk-PUT /api/v1/workflow; the denormalized trigger columns would go stale (single-doc PUT re-derives them).
• MCP is read-only for both engine schemas (mcp: 'read-only'): MCP create/update are hook-exempt by framework design, and every workflow invariant (graph validation, credential minting, activation gating, the engine-managed run ledger) lives in hooks. Agents can list/inspect workflows and runs over MCP; edits go through REST/GraphQL.
• Sequential chains only in v1: no parallel branches, waits, or retries per node.

Programmatic API

const workflow = require('davepi-plugin-workflow');
await workflow.startRun({ workflow: doc, triggerKind: 'manual', triggerPayload: {}, depth: 0 });
workflow.sweepOnce();      // schedule tick (tests/ops)
workflow.janitorOnce();    // stale-run sweep
workflow.stop();           // detach timers + bus listener

Every framework dep is injectable via createPlugin(opts) for standalone testing — see test/ for the patterns.