OpenFlow

A distributed workflow engine built in Go. Define task pipelines as JSON, trigger them via HTTP, and the platform runs them across a worker pool with dependency ordering, retries, and live status tracking.

What it does

You POST a workflow definition like this:

{
  "name": "Invoice Pipeline",
  "tasks": [
    { "id": "validate-payment", "type": "payment" },
    { "id": "generate-pdf",     "type": "pdf",   "depends_on": ["validate-payment"] },
    { "id": "send-email",       "type": "email", "depends_on": ["generate-pdf"] },
    { "id": "notify-slack",     "type": "slack", "depends_on": ["generate-pdf"] }
  ]
}

The platform resolves the dependency graph, publishes root tasks to a queue, and workers execute them concurrently. When a task finishes, the next unlocked tasks are published automatically. Each task retries up to 5 times on failure. Tasks that exhaust retries go to a dead-letter queue.

The full run in the example above finishes in under 100ms. send-email and notify-slack run in parallel since both depend only on generate-pdf.

Stack

Component	Role
Go + Gin	API server, worker pool, scheduler
PostgreSQL (pgx)	Workflow definitions, run state, audit trail
NATS JetStream	Task queue and dead-letter queue
Redis	Distributed lock for the cron scheduler
Prometheus	Metrics
Docker	Local dev infrastructure

Architecture

The API writes run state to Postgres and publishes root tasks to NATS. Workers pull from NATS, execute the handler, update the task run status, then publish any newly unblocked downstream tasks. The scheduler polls the schedules table every 30 seconds and uses a Redis lock to prevent duplicate triggers across API replicas.

Performance

20 virtual users, 30 second run against POST /api/workflows/:id/run:

• p95 response time: 47ms
• 0 failed requests
• 46.8 req/s throughput

Running locally

You need Docker and Go 1.21+.

git clone https://github.com/mihir-dixit2k27/openflow
cd openflow
cp .env.example .env

Start infrastructure:

docker compose up postgres redis nats -d

Apply schema:

docker exec -i openflow-postgres-1 psql -U openflow -d openflow < scripts/migrations/001_init.up.sql

Start the API and worker (two terminals):

go run ./cmd/api      # terminal 1
go run ./cmd/worker   # terminal 2

Or use the scripts:

bash scripts/start.sh   # starts everything
bash scripts/demo.sh    # runs a full end-to-end workflow

Open http://localhost:8080.

API

POST   /auth/register
POST   /auth/login

GET    /api/projects
POST   /api/projects

POST   /api/projects/:id/workflows
GET    /api/projects/:id/workflows
GET    /api/workflows/:id
POST   /api/workflows/:id/run

GET    /api/runs/:id
GET    /api/runs/:id/tasks
POST   /api/tasks/:id/retry

POST   /api/workflows/:id/schedules
GET    /health/live
GET    /health/ready
GET    /metrics

Observability

Prometheus metrics at /metrics:

workflow_runs_total
task_success_total
task_failure_total
workflow_duration_seconds
queue_size
worker_active
api_latency_seconds

Start Grafana: docker compose up grafana -d

Adding task types

Register a handler in internal/worker/pool.go:

p.Register("my-type", func(ctx context.Context, cfg map[string]string) error {
    // your logic here
    return nil
})

Any workflow task with "type": "my-type" will call your function.

License

MIT