Pipewatch

Monitoring, alerting and failed-run recovery for Zapier, Make, n8n and webhook automations.

Live demo: https://www.freelancerportfoliohub.com/jameslee/projects/pipewatch/index.html

Overview

Pipewatch watches the business automations that quietly move orders, leads and invoices between tools. Run history from every platform is normalised into one health model with three signals (failures, missed heartbeats and outcome checks), alerts go to a named owner, failed runs can be replayed from the step that broke, and every action lands in a hash-chained audit log.

This repository contains two parts:

• collector/, a Rust service that does the work: run-event ingest, health evaluation, the alert rule engine and notification fan-out, replay planning with a throttled recovery queue, and the audit log.
• The Next.js app at the root: the marketing site and the product UI it showcases. Product screens read from the collector over HTTP on every request.

Features

• Run history components: live run table with source badges and status, workflow health list with tag filters, step-level run timeline with Skip step / Edit payload / Replay, and a throttled bulk recovery queue with pause and resume
• Webhook ingest (POST /api/webhooks/:source): validated adapters for Zapier, Make, n8n Error Trigger and a generic signed webhook that report every payload problem with its JSON path; token or HMAC-SHA256 verification with a five-minute replay window and several signatures accepted during secret rotation; idempotent on (source, run ID), and a late "running" delivery never undoes a finished run
• Run model and health (collector/src/domain, health.rs): status and health vocabularies, the normalised RunEvent, and health from recent runs plus schedules (rate, daily at a local time with DST handled, or event-driven)
• Alert rules (collector/src/alerts): run failed after N retries, consecutive failures, failure rate, latency percentile, silence, missed daily deadline and volume drop against a multi-day median; alerts fire once, escalate to on-call, remind on a cooldown and resolve on their own
• Notifications (collector/src/notify): Slack Block Kit messages, plain-text and HTML email through a relay, and signed JSON webhooks, sent concurrently with retries on 429 and 5xx
• Replay service (collector/src/replay): replay plans that reuse stored step outputs, stable idempotency keys, payload diffs, drift detection that holds a run for review, and a recovery queue that runs one replay at a time with exponential backoff that honours Retry-After
• Audit log (collector/src/audit.rs): append-only, SHA-256 hash-chained entries with chain verification; replays, skipped steps, payload edits and escalations are recorded
• Heartbeats (/api/heartbeats/:workflow): ping URL for cron jobs and scripts

Tech stack

• Collector: Rust, axum on tokio, serde, chrono and chrono-tz for schedules, hmac/sha2 for signatures, idempotency keys and the audit chain, reqwest (rustls) for outbound notifications and replays
• Web: Next.js 15 App Router, React 19, TypeScript (strict), server-side fetch to the collector and a server action for replays
• Plain CSS design system in app/globals.css, Inter self-hosted

Getting started

You need Rust (stable) and Node 22 with pnpm. Run the collector and the web app in two terminals.

# Terminal 1: collector on :8080
cd collector
PIPEWATCH_WEBHOOK_SECRET=dev-secret cargo run

# Terminal 2: web app on :3000
pnpm install
cp .env.example .env.local   # API_URL=http://localhost:8080
pnpm dev

If the collector is not running, the product screens show an offline state and the marketing pages still render.

Configuration

Variable	Used by	Description
API_URL	web	Base URL of the collector (default http://localhost:8080)
PIPEWATCH_WEBHOOK_SECRET	collector	x-pipewatch-token for Zapier/Make/n8n, and the HMAC key for signed custom webhooks
PORT	collector	Listen port (default 8080)
SLACK_WEBHOOK_URL	collector	Slack incoming webhook for alert rules with a Slack channel
ALERT_WEBHOOK_URL	collector	Receives signed alert events as JSON
EMAIL_RELAY_URL	collector	Mail relay that accepts {to, subject, text, html}; EMAIL_RELAY_TOKEN is sent as a bearer token
REPLAY_WEBHOOK_URL	collector	Catch hook that re-runs a workflow from a replay plan; enables the recovery queue
PUBLIC_URL	collector	Base URL of the UI for links in notifications

Channels whose target is not configured are recorded as skipped in GET /api/alerts/deliveries.

Sending a run event

BODY='{"workflow":"payout-received","runId":"r_1042","status":"failed","startedAt":"2026-09-24T16:00:00Z"}'
TS=$(date +%s)
SIG=$(printf '%s' "$TS.$BODY" | openssl dgst -sha256 -hmac "$PIPEWATCH_WEBHOOK_SECRET" | awk '{print $2}')

curl -X POST http://localhost:8080/api/webhooks/webhook \
  -H "x-pipewatch-signature: t=$TS,v1=$SIG" \
  -H "Content-Type: application/json" \
  -d "$BODY"

Zapier, Make and n8n post to /api/webhooks/zapier, /make or /n8n with an x-pipewatch-token header instead.

Planning a replay

curl -X POST http://localhost:8080/api/runs/48219/replay \
  -H "Content-Type: application/json" \
  -d '{"fromStep":4,"actor":"Priya S."}'

Collector API

Method	Path	Description
POST	/api/webhooks/:source	Run-event ingest (zapier, make, n8n, webhook)
GET POST	/api/heartbeats/:workflow	Heartbeat ping; ?run= makes retries idempotent
GET	/api/workspace	Today's totals and the latest runs
GET	/api/workflows	Workflows with owner, schedule and health
GET	/api/runs, /api/runs/:number	Ingested runs; one run's timeline and stored payload
POST	/api/runs/:number/replay	Plan a replay and queue it unless it needs review
GET	/api/recovery/jobs	Live recovery queue; POST /api/recovery/pause etc.
GET	/api/alerts/rules	Alert rules with the sentences the rule editor shows
GET	/api/alerts/open, /deliveries	Firing alerts and the notification delivery log
GET	/api/audit	Audit feed with the chain re-verified on every read
GET	/healthz	Liveness

Testing

cd collector
cargo test                  # unit tests per module plus API integration tests (tests/api.rs)
cargo clippy --all-targets

pnpm typecheck && pnpm lint && pnpm build

Deploying

The collector ships with collector/Dockerfile, which builds a release binary and runs it as a non-root user on $PORT. On Railway, create one service from this repository with its root directory set to /collector (it builds from the Dockerfile) and a second service for the web app at the repository root, with API_URL pointing at the collector's private URL.

Project structure

collector/                  Rust service
  src/
    domain/                 runs, steps, workflows, schedules, RunEvent
    ingest/                 webhook auth, path-aware validator, source adapters
    health.rs               workflow health from runs and schedules
    store.rs                idempotent run store
    alerts/                 rule model, evaluation engine, rule sentences
    notify/                 Slack, email and webhook formatters, fan-out dispatcher
    replay/                 replay plans, recovery queue with backoff, webhook executor
    audit.rs                hash-chained audit log
    seed.rs                 demo workspace loaded from fixtures/
    monitor.rs, state.rs    alert wiring and shared state
    api/                    HTTP handlers
  fixtures/                 workspace, alert rules and audit seed data (JSON)
  tests/api.rs              API integration tests
  Dockerfile
app/
  page.tsx, product/, recovery/, security/, pricing/   marketing pages
  actions.ts                                          replay server action
  globals.css
components/
  runs/        RunHistory, RunHistoryTable, WorkflowHealthList, RunTimeline, RecoveryQueueView, AlertCard, *Panel
  alerts/      AlertRuleList, SlackNotification, AlertingPreview
  audit/       AuditLog
  marketing/   page sections (PageHero, Feature, InfoCards, CompareTable…)
  site/, ui/   header, footer, icons and primitives
lib/
  api.ts       collector client (server-side)
  status.ts    status labels and badge classes
  time.ts      clock, age and duration formatting
  data/        site and marketing content
types/         runs, alerts, audit, API responses
public/        favicon, fonts

Scripts

Script	Description
pnpm dev	Start the dev server (Turbopack)
pnpm build	Production build
pnpm start	Serve the production build
pnpm lint	ESLint
pnpm typecheck	TypeScript, no emit
pnpm format	Prettier
cargo run	Run the collector (in collector/)
cargo test	Collector tests (in collector/)