# Orion — complete documentation
# Generated from the mdBook sources at https://github.com/GoPlasmatic/Orion/tree/main/docs/src


---
source: ./introduction.md
---

<div class="hero-logo">
  <img src="images/plasmatic-logo.png" alt="Plasmatic Logo">
  <h1>Orion</h1>
  <p>Deploy high-performance, governed microservices as JSON workflows—without writing boilerplate.</p>
</div>

**Orion** is an API-first declarative services runtime written in Rust. Instead of writing, containerizing, and operating a new microservice for every piece of business logic, you simply **declare** what the service should do as a JSON workflow, and Orion runs it.

Every workflow is executed with enterprise-grade architectural governance—including observability, rate limiting, circuit breakers, caching, input validation, and versioning—built directly into the runtime, not bolted on. Build workflows yourself or let an AI generate them; either way, they run under the same production-grade guarantees.

### Why Orion?

Developers spend too much time building the same boilerplate for microservices—setting up HTTP servers, configuring database connection pools, writing Prometheus exporters, integrating OpenTelemetry, and coding retry loops or circuit breakers. Orion eliminates this overhead entirely.

* **⚡ Zero Boilerplate:** Go from idea to a live REST/Kafka service in seconds. No Dockerfiles, no CI pipelines, and no server boilerplates.
* **🛡️ Built-in Governance:** Out-of-the-box support for rate limiting, circuit breakers, timeouts, caching, and payload validation.
* **🤖 AI-Native & Safe:** Structured JSON workflows are exceptionally easy for LLMs to generate. Safe rollout pipelines (validation, draft/dry-run, rollout percentage, rollbacks) ensure AI-generated code never breaks production.
* **🦀 Rust Performance:** Built on Tokio and Axum. Achieves **6,000+ requests/sec** per instance with single-digit millisecond latency and a tiny memory footprint.
* **🧩 In-Process Composition:** Call other workflows in-process like functions with zero network round-trip overhead.

This site is the deep reference and how-to guide — new here? [**Install Orion and ship your first service**](./tutorials/cli-setup.md) in a couple of minutes.

<div class="themed-media">
  <video class="media-dark" controls muted playsinline preload="metadata" src="videos/ui-quickstart-dark.webm"></video>
  <video class="media-light" controls muted playsinline preload="metadata" src="videos/ui-quickstart-light.webm"></video>
</div>
<span class="asciinema-caption">▶ Zero to a live service in under a minute, without writing code — in <a href="getting-started/console.html">the Orion console</a>. Prefer a terminal? The same flow over plain HTTP:</span>

<div class="asciinema-player" data-cast="casts/quickstart.cast"></div>
<span class="asciinema-caption">From zero to a live, governed service — business logic as JSON, deployed over plain HTTP. ▶ Click to play.</span>

## Three Primitives

You build services in Orion with three things:

```orion-diagram
{
  "direction": "LR",
  "nodes": [
    { "id": "Channel",   "label": "Channel",   "sublabel": "endpoint", "type": "channel" },
    { "id": "Workflow",  "label": "Workflow",  "sublabel": "logic",    "type": "service" },
    { "id": "Connector", "label": "Connector", "sublabel": "external", "type": "datastore" }
  ],
  "edges": [
    { "from": "Channel",  "to": "Workflow" },
    { "from": "Workflow", "to": "Connector" }
  ]
}
```

| Primitive | What it is | Example |
|-----------|-----------|---------|
| **Channel** | A service endpoint: sync (REST, HTTP) or async (Kafka) | `POST /orders`, `GET /users/{id}`, Kafka topic `order.placed` |
| **Workflow** | A pipeline of tasks that defines what the service does | Parse → validate → enrich → transform → respond |
| **Connector** | A named connection to an external system, with auth and retries | Stripe API, PostgreSQL, MongoDB, Elasticsearch, Redis, Kafka cluster |

**Design-time:** define channels, build workflows, configure connectors, test with dry-run, manage versions — all through the admin API. **Runtime:** Orion routes traffic to channels, executes workflows, calls connectors, and handles observability automatically. See [**Dev & Prod Environments**](./topology/environments.md) for how the same binary serves both planes.

## Start here

- [**CLI Setup**](./tutorials/cli-setup.md) — install Orion and ship your first service in a couple of minutes.
- [**MCP Server Setup**](./tutorials/mcp-setup.md) — give an AI assistant full Orion context so it generates valid workflows.
- [**Use Cases & Patterns**](./tutorials/use-cases.md) — complete, tested examples for classification, transformation, routing, and CI/CD.

## Build workflows

- [**Workflow Reference**](./reference/workflows.md) — the workflow & task JSON schema, conditions, error handling, lifecycle, and rollout.
- [**Function Reference**](./reference/functions.md) — every built-in task function and its exact `input` schema.
- [**Admin API**](./api/admin.md) & [**Data API**](./api/data.md) — the full REST surface for managing and calling services.
- [**Configuration**](./configuration/reference.md) — config file, environment variables, database backends, and deployment.

## How it works

- [**Architecture Overview**](./architecture/overview.md) — channels, workflows, the engine, hot-reload, and the request-processing flow.
- **Production features**, all built in and configurable per channel: [Observability](./features/observability.md), [Resilience](./features/resilience.md), [Security](./features/security.md), [Scalability](./features/scalability.md), [Deployability](./features/deployability.md), [Extensibility](./features/extensibility.md), [Availability](./features/availability.md), and [Maintainability](./features/maintainability.md).


---
source: ./comparison.md
---

# Is Orion Right for You?

Orion occupies a specific spot: **declarative business logic served as
governed, high-performance services** — below full workflow orchestrators,
above API gateways, beside (not inside) your application code. This page maps
the neighbors honestly; several of these tools are the better choice for their
home turf, and pairing them with Orion is often the right answer.

## The short answer

| If you need to... | Orion? | Reach for |
|---|:-:|---|
| Turn business rules into live REST/Kafka services | **Yes** | — |
| Let AI generate and manage business logic safely | **Yes** | — |
| Replace a handful of single-purpose microservices | **Yes** | — |
| Orchestrate long-running jobs (hours/days), human approvals | No | Temporal, Airflow, BPMN engines |
| Run a full API gateway with a plugin ecosystem | No | Kong, Envoy (Orion runs happily behind them) |
| A RETE rule engine with complex fact networks | No | Drools |
| Manage services from a browser dashboard | **Yes** | [Orion UI](https://github.com/GoPlasmatic/Orion-ui), the admin dashboard |
| Embed a workflow engine inside your app | No | [dataflow-rs](https://github.com/GoPlasmatic/dataflow-rs), the library Orion is built on |
| General-purpose compute (ML, media processing) | No | Your own services / serverless |

## vs. writing another microservice

The baseline alternative. A new microservice means an HTTP server, connection
pools, metrics exporters, tracing integration, retry loops, a Dockerfile, a CI
pipeline, and a deployment — before the first line of business logic. In
Orion the business logic is a JSON document deployed with one API call, and
rate limiting, circuit breakers, validation, versioning, and observability are
runtime guarantees rather than per-service chores.

The trade: your logic must be expressible as a pipeline of Orion's task
functions and JSONLogic. Parse → validate → enrich → call systems → transform
→ respond fits naturally; bespoke algorithms and heavy computation do not
(that's what [`http_call`](./reference/functions.md) to a real service is
for).

## vs. Temporal / Airflow / BPMN engines

Orchestrators own **durable, long-running, stateful** execution: workflows
that sleep for days, survive restarts mid-run, and wait on humans. Orion
workflows are **stateless request pipelines** — they execute in milliseconds
inside a request or event, and durability lives in your databases and topics,
not in the engine.

Choose an orchestrator for saga patterns, human-in-the-loop approvals, and
scheduled batch DAGs. Choose Orion for request-response services and
event-stream processing where those engines' latency and operational weight
are the wrong fit. They compose: a Temporal activity can call an Orion
channel, and an Orion workflow can kick off a Temporal run via `http_call`.

## vs. Kong / Envoy / API gateways

A gateway **proxies and polices** traffic on its way to your services; it
deliberately doesn't implement them. Orion **is** the service: the request
terminates in a workflow that executes your logic. The overlap (rate
limiting, auth, CORS) exists because Orion channels are self-sufficient — but
in front of a fleet, a gateway still earns its place, with Orion as an
upstream that happens to need far fewer of the gateway's crutches.

## vs. Drools and RETE rule engines

Orion evaluates [JSONLogic](https://jsonlogic.com) conditions — compiled at
engine build time, fast, deterministic, and trivially generatable by LLMs.
What it is not: a RETE engine doing incremental matching over a working
memory of thousands of interdependent facts. If your problem is "which of
10,000 rules fire as facts accumulate," use Drools. If it's "does this
request satisfy these conditions, then transform and route it," Orion's model
is simpler to write, review, version, and roll back.

## vs. n8n / Zapier / Make

Visual automation tools optimize for **person-builds-integration-quickly**:
huge app catalogs, drag-and-drop, hosted convenience. Orion optimizes for
**production service traffic**: thousands of requests per second, P99 in
milliseconds, versioned rollouts, circuit breakers, Prometheus metrics — and
an API-first design where definitions are JSON in your repo, not boxes in a
canvas ([Orion UI](https://github.com/GoPlasmatic/Orion-ui) adds a dashboard
for managing and visualizing them, but the API stays the source of truth).
If a workflow runs a few times an hour and touches 40 SaaS apps, use the
automation tools. If it *is* one of your services, run it on Orion.

## vs. embedding dataflow-rs

Orion is the **runtime** built on the
[dataflow-rs](https://github.com/GoPlasmatic/dataflow-rs) engine. If you want
workflow execution *inside* your own Rust application — no server, no admin
API, no lifecycle management — embed dataflow-rs directly. Orion is what you
deploy when you want the engine plus channels, connectors, versioning,
governance, and an admin API as a standing service.

---

Convinced, or at least curious? [Install & ship your first service](./tutorials/cli-setup.md)
in a couple of minutes.


---
source: ./tutorials/cli-setup.md
---

# Install & First Service

Get Orion running on your machine in under a minute.

> Prefer point-and-click? [**The Console (Orion UI)**](../getting-started/console.md) walks
> the same zero-to-live-service flow entirely in the browser — with a demo video.

## Installation

Choose your preferred method:

**Homebrew** (macOS and Linux):

```bash
brew install GoPlasmatic/tap/orion-server
```

**Shell installer** (Linux/macOS):

```bash
curl --proto '=https' --tlsv1.2 -LsSf https://github.com/GoPlasmatic/Orion/releases/latest/download/orion-server-installer.sh | sh
```

**PowerShell** (Windows):

```powershell
powershell -ExecutionPolicy ByPass -c "irm https://github.com/GoPlasmatic/Orion/releases/latest/download/orion-server-installer.ps1 | iex"
```

**Docker:**

```bash
docker run -p 8080:8080 ghcr.io/goplasmatic/orion:latest
```

**From source** (requires Rust 1.88+):

```bash
cargo install --git https://github.com/GoPlasmatic/Orion
```

## First Run

Start Orion with default settings (SQLite, port 8080):

```bash
orion-server
```

Verify it's running:

```bash
curl -s http://localhost:8080/health
```

```json
{
  "status": "ok",
  "version": "0.2.0",
  "uptime_seconds": 5,
  "workflows_loaded": 0,
  "components": {
    "database": "ok",
    "engine": "ok"
  }
}
```

Swagger UI is available at [http://localhost:8080/docs](http://localhost:8080/docs).

## Configuration

Create a config file for custom settings:

```bash
orion-server -c config.toml
```

Or use environment variables for individual overrides:

```bash
ORION_SERVER__PORT=9090 \
ORION_LOGGING__FORMAT=json \
orion-server
```

Common configuration scenarios:

```bash
# Use PostgreSQL instead of SQLite
ORION_STORAGE__URL="postgres://user:pass@localhost/orion" orion-server

# Enable admin authentication
ORION_ADMIN_AUTH__ENABLED=true \
ORION_ADMIN_AUTH__API_KEY="your-secret-key" \
orion-server

# Enable metrics and tracing
ORION_METRICS__ENABLED=true \
ORION_TRACING__ENABLED=true \
ORION_TRACING__OTLP_ENDPOINT="http://localhost:4317" \
orion-server
```

Validate a config file without starting the server:

```bash
orion-server validate-config -c config.toml
```

## Create Your First Service

> **In a hurry?** One command deploys a first service (workflow + channel,
> activated) against your running instance and sends a test request:
>
> ```bash
> curl -fsSL https://raw.githubusercontent.com/GoPlasmatic/Orion/main/examples/quickstart.sh | bash
> ```
>
> The steps below build a service by hand so you can see each moving part.

**1. Create a workflow:**

```bash
curl -s -X POST http://localhost:8080/api/v1/admin/workflows \
  -H "Content-Type: application/json" \
  -d '{
    "workflow_id": "hello-world",
    "name": "Hello World",
    "condition": true,
    "tasks": [
      { "id": "parse", "name": "Parse", "function": {
          "name": "parse_json", "input": { "source": "payload", "target": "req" }
      }},
      { "id": "greet", "name": "Greet", "function": {
          "name": "map", "input": { "mappings": [
            { "path": "data.req.greeting", "logic": {
              "cat": ["Hello, ", { "var": "data.req.name" }, "!"]
            }}
          ]}
      }}
    ]
  }'
```

**2. Activate the workflow:**

```bash
curl -s -X PATCH http://localhost:8080/api/v1/admin/workflows/hello-world/status \
  -H "Content-Type: application/json" -d '{"status": "active"}'
```

**3. Create and activate a channel:**

```bash
curl -s -X POST http://localhost:8080/api/v1/admin/channels \
  -H "Content-Type: application/json" \
  -d '{ "channel_id": "hello", "name": "hello", "channel_type": "sync",
        "protocol": "rest", "route_pattern": "/hello",
        "methods": ["POST"], "workflow_id": "hello-world" }'

curl -s -X PATCH http://localhost:8080/api/v1/admin/channels/hello/status \
  -H "Content-Type: application/json" -d '{"status": "active"}'
```

**4. Test it:**

```bash
curl -s -X POST http://localhost:8080/api/v1/data/hello \
  -H "Content-Type: application/json" \
  -d '{ "data": { "name": "World" } }'
```

```json
{
  "status": "ok",
  "data": { "req": { "name": "World", "greeting": "Hello, World!" } },
  "errors": []
}
```

## Orion CLI

The [Orion CLI](https://github.com/GoPlasmatic/Orion-cli) provides a command-line interface and MCP server for managing Orion. No curl commands needed.

**Homebrew** (macOS and Linux):

```bash
brew install GoPlasmatic/tap/orion-cli
```

**Shell installer** (Linux/macOS):

```bash
curl --proto '=https' --tlsv1.2 -LsSf https://github.com/GoPlasmatic/Orion-cli/releases/latest/download/orion-cli-installer.sh | sh
```

**PowerShell** (Windows):

```powershell
powershell -ExecutionPolicy ByPass -c "irm https://github.com/GoPlasmatic/Orion-cli/releases/latest/download/orion-cli-installer.ps1 | iex"
```

**From source** (requires Rust 1.88+):

```bash
cargo install --git https://github.com/GoPlasmatic/Orion-cli
```

**Usage:**

```bash
orion-cli config set-server http://localhost:8080
orion-cli health
orion-cli workflows list
orion-cli channels list
orion-cli send hello -d '{ "data": { "name": "World" } }'
```

The full lifecycle — create, activate, dry-run, then send live data — in one tool:

<div class="asciinema-player" data-cast="casts/cli-lifecycle.cast"></div>
<span class="asciinema-caption">▶ Click to play. Dry-run testing and live traffic flow through the same workflow.</span>

See the [CLI reference](https://github.com/GoPlasmatic/Orion-cli) for the full command list, or set up the [MCP Server](./mcp-setup.md) for AI assistant integration.

## Next Steps

- Connect a real database: [Your First Connector](../getting-started/first-connector.md)
- Browse the [API Reference](../api/admin.md) for all available endpoints
- Explore [Production Features](../features/observability.md) for observability, security, and resilience
- See the [Config Reference](../configuration/reference.md) for all configuration options


---
source: ./getting-started/console.md
---

# The Console (Orion UI)

Orion itself is API-first — everything in these docs is plain HTTP. [**Orion UI**](https://github.com/GoPlasmatic/Orion-ui)
is the operations console on top of that API: live dashboards, a system map of every
channel → workflow → connector, workflow logic visualization, trace drill-downs, and a
data console for firing test requests. Everything the admin API can do, point-and-click.

## Zero to a live service — no code

The full creation loop in under a minute: import a workflow (paste → validate →
**dry-run** → activate), watch its logic render as a graph, give it an endpoint with the
channel form, send a request from the Data Console, and see the service on the System Map.

<div class="themed-media">
  <video class="media-dark" controls muted playsinline preload="metadata" src="../videos/ui-quickstart-dark.webm"></video>
  <video class="media-light" controls muted playsinline preload="metadata" src="../videos/ui-quickstart-light.webm"></video>
</div>
<span class="asciinema-caption">▶ Click to play. The same flow as the <a href="../tutorials/cli-setup.html">terminal quickstart</a> — as clicks instead of curl.</span>

## Run it

The UI ships as a production-ready container image (nginx, multi-arch):

```bash
docker run -p 8081:8080 \
  -e ORION_URL=http://host.docker.internal:8080 \
  ghcr.io/goplasmatic/orion-ui:latest
```

Open `http://localhost:8081` — `ORION_URL` points at your Orion server and the bundled
nginx reverse-proxies all `/api/` requests to it. Developing against a local checkout?
`npm install && npm run dev` in the [Orion-ui repo](https://github.com/GoPlasmatic/Orion-ui)
does the same via the Vite dev server. A `docker-compose.yml` that brings up server + UI
together is in that repo as well.

## What you get

### Operations dashboard

Live request rate, error rate, latency percentiles, outcomes by channel, top channels,
recent traces, and anything that needs attention (open circuit breakers, idle channels,
recent failures) — for the whole instance, at a glance.

<div class="themed-media">
  <img class="media-dark" src="../images/ui-operations-dark.png" alt="Operations dashboard — request rate, error rate, latency percentiles, outcomes by channel, top channels, and recent traces">
  <img class="media-light" src="../images/ui-operations-light.png" alt="Operations dashboard — request rate, error rate, latency percentiles, outcomes by channel, top channels, and recent traces">
</div>

### System Map

Pick any channel and trace it through the workflow it runs, the channels it calls
in-process, and the connectors it touches — as a live topology graph. Every node links to
its detail page.

<div class="themed-media">
  <img class="media-dark" src="../images/ui-system-map-dark.png" alt="System Map — a channel traced through its workflow and connectors as a topology graph">
  <img class="media-light" src="../images/ui-system-map-light.png" alt="System Map — a channel traced through its workflow and connectors as a topology graph">
</div>

### Workflow logic, visualized

Workflows are managed through a guided import wizard — paste JSON (often AI-generated),
validate it, import as a draft, **dry-run it against a sample payload**, then activate.
On the detail page, each task's JSONLogic renders as a flow graph, with tabs for
relationships, dry-run testing, version history, and the raw JSON.

<div class="themed-media">
  <img class="media-dark" src="../images/ui-workflow-dag-dark.png" alt="Workflow detail — task explorer with the selected task's JSONLogic rendered as a flow graph">
  <img class="media-light" src="../images/ui-workflow-dag-light.png" alt="Workflow detail — task explorer with the selected task's JSONLogic rendered as a flow graph">
</div>

### Data Console

Send test requests to any channel — sync or async, with optional per-task profiling —
and inspect the response, the request profile (per-function and per-connector timings),
and the resulting trace, one click away.

<div class="themed-media">
  <img class="media-dark" src="../images/ui-console-dark.png" alt="Data Console — send a test request to a channel and inspect the response, per-task timings, and trace">
  <img class="media-light" src="../images/ui-console-light.png" alt="Data Console — send a test request to a channel and inspect the response, per-task timings, and trace">
</div>

Also in the console: channel and connector management with lifecycle actions
(draft → active → archived), circuit-breaker monitoring and reset, the audit log, trace
search and drill-down, and a command palette (<kbd>⌘K</kbd>) for jumping anywhere.

> All visuals on this page are generated from a live instance by the
> [recording pipeline](https://github.com/GoPlasmatic/Orion/tree/main/docs/recordings)
> — re-run `record-ui.sh` and they regenerate.


---
source: ./getting-started/first-connector.md
---

# Your First Connector

The [first service](../tutorials/cli-setup.md#create-your-first-service) you
built transformed data in-process. Real services talk to databases. This
tutorial connects Orion to PostgreSQL and builds a service that **writes an
order and reads back the customer's history** — using the portable data
dialect (`data_write` / `data_query`), so the same workflow JSON would run
unchanged against MongoDB or Elasticsearch.

A **connector** is a named, reusable connection to an external system:
configured once through the admin API, referenced by name from any workflow,
with credentials, pooling, retries, and circuit breaking handled by the
runtime.

> **Fast path:** this tutorial is shipped as a runnable example. From a repo
> clone: `cd examples/postgres-orders && docker compose up -d && cd .. &&
> ./deploy.sh postgres-orders`. The steps below build the same thing by hand.

## 1. Start a database

From the repo's [`examples/postgres-orders/`](https://github.com/GoPlasmatic/Orion/tree/main/examples/postgres-orders)
directory (which ships a compose file and seed data — two `customers`, three
`orders`):

```bash
docker compose up -d postgres
```

Then start Orion, telling it where the database is:

```bash
ORDERS_DB_URL=postgres://orion:orion@localhost:5432/orion_orders orion-server
```

## 2. Create the connector

```bash
curl -s -X POST http://localhost:8080/api/v1/admin/connectors \
  -H 'Content-Type: application/json' \
  -d '{
    "id": "orders-db",
    "name": "orders-db",
    "connector_type": "db",
    "config": {
      "type": "db",
      "connection_string": "${ORDERS_DB_URL:-postgres://orion:orion@postgres:5432/orion_orders}",
      "driver": "postgres",
      "max_connections": 5,
      "operations": { "delete": false }
    }
  }'
```

Two details here are worth copying into every real deployment:

- **The connection string is an environment reference.** `${ORDERS_DB_URL:-…}`
  is substituted from the *server's* environment when the connector loads —
  the saved config carries no credentials, and the same JSON works in every
  environment.
- **`"operations": { "delete": false }` makes the connector delete-proof.**
  Operation gates (`read` / `insert` / `update` / `delete` / `upsert` /
  `raw_write`) are enforced at the connector, regardless of what any workflow
  asks for.

The connector is live immediately — no activation step, and the registry
hot-reloads on every connector change.

## 3. Create the workflow

Three tasks: parse the request, insert the order, read back the customer with
their order history.

```bash
curl -s -X POST http://localhost:8080/api/v1/admin/workflows \
  -H 'Content-Type: application/json' \
  -d '{
    "workflow_id": "record-order",
    "name": "Record Order",
    "condition": true,
    "tasks": [
      { "id": "parse", "name": "Parse payload",
        "function": { "name": "parse_json", "input": { "source": "payload", "target": "req" } } },
      { "id": "record", "name": "Insert the order",
        "function": { "name": "data_write", "input": {
          "connector": "orders-db",
          "op": "insert",
          "target": "orders",
          "values": {
            "customer_id": { "param": "customer_id" },
            "item":        { "param": "item" },
            "total":       { "param": "total" }
          },
          "params": {
            "customer_id": { "var": "data.req.customer_id" },
            "item":        { "var": "data.req.item" },
            "total":       { "var": "data.req.total" }
          },
          "returning": ["id"],
          "output": "data.created"
        } } },
      { "id": "history", "name": "Fetch customer with order history",
        "function": { "name": "data_query", "input": {
          "connector": "orders-db",
          "query": {
            "source": "customers",
            "filter": { "==": [{ "field": "id" }, { "param": "customer_id" }] },
            "include": { "orders": { "fields": ["id", "item", "total"], "limit": 10 } }
          },
          "params": { "customer_id": { "var": "data.req.customer_id" } },
          "schema": {
            "entities": {
              "customers": {
                "relations": {
                  "orders": { "to": "orders", "kind": "has_many", "local": "id", "foreign": "customer_id" }
                }
              },
              "orders": {}
            }
          },
          "output": "data.customer"
        } } }
    ]
  }'

curl -s -X PATCH http://localhost:8080/api/v1/admin/workflows/record-order/status \
  -H 'Content-Type: application/json' -d '{"status":"active"}'
```

How the pieces fit:

- **`{ "param": "total" }` marks a value slot.** The `params` map is the only
  place request data enters a query, and every resolved value is a bound
  parameter — never string-interpolated. The dialect is injection-safe by
  construction.
- **The inline `schema` declares `customers has_many orders`.** That relation
  is what powers `"include": { "orders": … }` (and `some`/`all`/`none`
  filters over related records). Without a schema the dialect runs in
  identity mode — names pass through as-is.
- **`"returning": ["id"]`** captures the generated key from the insert.

## 4. Expose it as a service

```bash
curl -s -X POST http://localhost:8080/api/v1/admin/channels \
  -H 'Content-Type: application/json' \
  -d '{ "channel_id": "record-order", "name": "record-order",
        "channel_type": "sync", "protocol": "rest",
        "route_pattern": "/record-order", "methods": ["POST"],
        "workflow_id": "record-order" }'

curl -s -X PATCH http://localhost:8080/api/v1/admin/channels/record-order/status \
  -H 'Content-Type: application/json' -d '{"status":"active"}'
```

## 5. Call it

```bash
curl -s -X POST http://localhost:8080/api/v1/data/record-order \
  -H 'Content-Type: application/json' \
  -d '{ "data": { "customer_id": 1, "item": "Difference Engine Blueprint", "total": 4200.00 } }'
```

```json
{
  "status": "ok",
  "data": {
    "created": { "rows_affected": 1, "returning": [{ "id": 4 }] },
    "customer": [{
      "id": 1, "name": "Ada Lovelace", "email": "ada@example.com",
      "orders": [
        { "id": 1, "item": "Analytical Engine Manual", "total": 120.0 },
        { "id": 2, "item": "Punch Card Set", "total": 35.5 },
        { "id": 4, "item": "Difference Engine Blueprint", "total": 4200.0 }
      ]
    }]
  }
}
```

One request: a parameterized insert, then a relation-hydrated read — through a
pooled, circuit-broken, delete-proof connector you configured with one API
call.

## Switching backends

Nothing in the workflow above is Postgres-specific. Point `orders-db` at
MySQL or SQLite and it renders different SQL; point it at a MongoDB or
Elasticsearch connector and the same envelope renders a `find` filter or a
Query DSL search. Switching backends is a connector change, not a rewrite —
see the [Portable Data Dialect](../reference/data-dialect.md) for the full
vocabulary and per-backend notes.

## Next steps

- [Portable Data Dialect](../reference/data-dialect.md) — operators, schema
  registry, relations, write envelopes, safety guards
- [Connectors & Extensibility](../features/extensibility.md) — HTTP, cache,
  MongoDB, Elasticsearch, and Kafka connectors; auth, retries, circuit
  breakers
- [Use Cases & Patterns](../tutorials/use-cases.md) — larger worked examples


---
source: ./tutorials/mcp-setup.md
---

# MCP Server Setup

Orion includes an MCP (Model Context Protocol) server that lets AI assistants like Claude manage workflows, channels, and connectors through natural language. The MCP server wraps the Orion admin API, giving AI agents full control over your Orion instance.

<div class="asciinema-player" data-cast="casts/mcp.cast"></div>
<span class="asciinema-caption">▶ Click to play. A real stdio JSON-RPC session: handshake, tool discovery, then a live tool call.</span>

## Prerequisites

- A running Orion instance (see [Install & First Service](./cli-setup.md))
- An MCP-compatible client (Claude Code, Claude Desktop, or other MCP clients)

> No MCP client? Use the [Prompt Pack](../getting-started/prompt-pack.md) — a
> self-contained context block that lets any LLM drive Orion through the plain
> REST API.

## Configuration

Add the Orion MCP server to your MCP client configuration.

**Claude Code:** add to your `.claude/settings.json` or project-level `.mcp.json`:

```json
{
  "mcpServers": {
    "orion": {
      "command": "orion-cli",
      "args": ["mcp", "serve"],
      "env": {
        "ORION_SERVER_URL": "http://localhost:8080"
      }
    }
  }
}
```

**Claude Desktop:** add to your Claude Desktop configuration file:

```json
{
  "mcpServers": {
    "orion": {
      "command": "orion-cli",
      "args": ["mcp", "serve"],
      "env": {
        "ORION_SERVER_URL": "http://localhost:8080"
      }
    }
  }
}
```

If admin authentication is enabled on your Orion instance, include the API key:

```json
{
  "env": {
    "ORION_SERVER_URL": "http://localhost:8080",
    "ORION_API_KEY": "your-secret-key"
  }
}
```

## Available Tools

The MCP server exposes **46 tools** covering the full Orion API. Tool names follow a
`<resource>_<action>` convention:

| Category | Tools |
|----------|-------|
| **Health** | `health_check` |
| **Engine** | `engine_status`, `engine_reload` |
| **Workflows** | `workflows_list`, `workflows_get`, `workflows_create`, `workflows_update`, `workflows_delete`, `workflows_activate`, `workflows_archive`, `workflows_test`, `workflows_validate`, `workflows_rollout`, `workflows_versions`, `workflows_create_version`, `workflows_export`, `workflows_import` |
| **Channels** | `channels_list`, `channels_get`, `channels_create`, `channels_update`, `channels_delete`, `channels_activate`, `channels_archive`, `channels_versions`, `channels_create_version`, `channels_import` |
| **Connectors** | `connectors_list`, `connectors_get`, `connectors_create`, `connectors_update`, `connectors_delete`, `connectors_enable`, `connectors_disable`, `connectors_import` |
| **Circuit Breakers** | `circuit_breakers_list`, `circuit_breaker_reset` |
| **Data** | `data_send_sync`, `data_send_async` |
| **Traces** | `traces_list`, `traces_get` |
| **Functions** | `functions_list` |
| **Audit Logs** | `audit_logs_list` |
| **Backups** | `backups_create`, `backups_list` |
| **Metrics** | `get_metrics` |

## Usage Example

Once configured, you can use natural language to manage Orion:

> "Create a workflow that parses incoming orders, flags any over $10,000, and adds a risk level field. Then create a channel called 'orders' that uses it."

The AI assistant will use the MCP tools to:

1. Create the workflow via `workflows_create`
2. Test it with sample data via `workflows_test`
3. Activate it via `workflows_activate`
4. Create the channel via `channels_create`
5. Activate the channel via `channels_activate`, then apply with `engine_reload`

## Troubleshooting

**MCP server not connecting:**
- Verify Orion is running: `curl http://localhost:8080/health`
- Check the `ORION_SERVER_URL` environment variable is set correctly
- Ensure `orion-cli` is in your PATH

**Authentication errors:**
- Verify `ORION_API_KEY` matches your Orion instance's `admin_auth.api_key`
- Check that admin auth is enabled on the Orion instance if you're passing a key


---
source: ./tutorials/claude-code.md
---

# Orion + Claude Code

The fastest way to experience "AI writes services, not code": connect [Claude
Code](https://claude.com/claude-code) to Orion through the [MCP
server](./mcp-setup.md) and describe the service you want. Claude drafts the
workflow, dry-runs it against sample data, activates it, and wires up the
endpoint — while Orion's lifecycle rules (draft → test → activate, immutable
versions, instant rollback) keep every step reversible.

This page is a 10-minute guided session. It assumes nothing beyond a running
Orion instance.

## Setup

1. **Run Orion** (see [Install & First Service](./cli-setup.md)):

   ```bash
   brew install GoPlasmatic/tap/orion-server && orion-server
   ```

2. **Install the CLI**, which contains the MCP server:

   ```bash
   brew install GoPlasmatic/tap/orion-cli
   ```

3. **Register the MCP server with Claude Code** — one command:

   ```bash
   claude mcp add orion --env ORION_SERVER_URL=http://localhost:8080 -- orion-cli mcp serve
   ```

   Or, to share the config with your team, commit a `.mcp.json` at the project
   root instead:

   ```json
   {
     "mcpServers": {
       "orion": {
         "command": "orion-cli",
         "args": ["mcp", "serve"],
         "env": { "ORION_SERVER_URL": "http://localhost:8080" }
       }
     }
   }
   ```

4. **Verify:** start `claude` and run `/mcp` — you should see `orion` listed
   with 46 tools.

## Build a service by describing it

Paste this into Claude Code:

> Create an Orion workflow called `order-triage` that parses incoming orders,
> flags any order over $10,000 with an alert message, and adds a
> `risk_level` field ("high" above 10000, "normal" otherwise). Test it with a
> realistic sample order **before** activating. Then create a REST channel
> `POST /orders` that uses it, activate everything, and send a $25,000 test
> order through it.

Watch the tool calls as Claude works. The sequence it follows is the same safe
path you'd follow by hand:

1. `workflows_create` — the logic lands as a **draft**; nothing serves traffic yet
2. `workflows_test` — dry-run with sample data; the response shows which tasks
   ran and what each produced
3. `workflows_activate` — draft goes live; the engine hot-reloads
4. `channels_create` + `channels_activate` — `POST /orders` now routes to it
5. `data_send_sync` — the test order comes back flagged, with the alert message

Your service is live. Total code written: zero.

## Inspect and operate

Everything Orion records is queryable through the same session:

> Show me the recent traces for the orders channel. What did each task do on
> the last request?

> Is the engine healthy? How many workflows and channels are active?

Claude answers from `traces_list` / `traces_get` / `engine_status` — real
observability data, not summaries of what it *thinks* it deployed.

## Change it safely

Orion's governance makes iteration safe to delegate. Try:

> Lower the flag threshold to $5,000. Create a new version, dry-run it with an
> order at $7,500, and roll it out to 10% of traffic first.

Active versions are immutable, so Claude must create a new version
(`workflows_create_version`), test it, and use `workflows_rollout` for the
canary. If anything looks wrong:

> Roll the orders workflow back to the previous version.

— one tool call, instant.

## Where to go next

- The full tool catalog and client configs (Claude Desktop, Cursor, HTTP
  transport): [MCP Server Setup](./mcp-setup.md)
- No MCP client available? The [Prompt Pack](../getting-started/prompt-pack.md)
  gives any LLM the same powers over the plain REST API.
- Ready-made prompts for common services (webhook transforms, enrichment,
  routing): [Use Cases & Patterns](./use-cases.md)


---
source: ./getting-started/prompt-pack.md
---

# Prompt Pack (any LLM)

The [MCP server](../tutorials/mcp-setup.md) is the richest way to give an AI
assistant control of Orion — 46 tools, live schema discovery, no prompt
engineering. But it needs an MCP-capable client.

This page is the zero-install alternative: **paste the block below into any
LLM** (as a system prompt, a project instruction, or just the first message)
and it has enough context to write valid workflows and deploy them through
Orion's plain REST API.

````text
You are working with Orion, an API-first declarative services runtime.
Services are defined as JSON over a REST admin API — no code, no deploys.
Base URL: http://localhost:8080 (adjust if told otherwise).

## The three primitives

- WORKFLOW: a pipeline of tasks (the business logic)
- CHANNEL: a service endpoint (REST route or Kafka topic) that routes to a workflow
- CONNECTOR: a named connection to an external system (HTTP API, SQL database,
  MongoDB, Elasticsearch, Redis cache, Kafka), referenced from tasks by name

## Workflow JSON

{
  "workflow_id": "kebab-case-id",
  "name": "Human Name",
  "condition": true,                    // or a JSONLogic expression
  "tasks": [
    {
      "id": "task-id",
      "name": "Task name",
      "condition": { ">": [{ "var": "data.order.total" }, 10000] },   // optional; JSONLogic
      "function": { "name": "<function>", "input": { ... } }
    }
  ]
}

Data flow: requests arrive as { "data": <payload> }. A parse_json task with
input { "source": "payload", "target": "order" } places the payload at
data.order; downstream JSONLogic reads it via { "var": "data.order.<field>" }
and map tasks write via paths like "data.order.flagged". The final data
context is returned to the caller.

Built-in task functions (get exact input schemas from
GET /api/v1/admin/functions — always check before using one):
parse_json, parse_xml, filter, map, validation, http_call, channel_call,
data_query, data_write (portable, backend-neutral DB read/write — preferred),
db_read, db_write (raw SQL escape hatch), cache_read, cache_write, mongo_read,
publish_json, publish_xml, publish_kafka, log.

## Channel JSON

{
  "channel_id": "orders", "name": "orders",
  "channel_type": "sync",              // sync | async
  "protocol": "rest",                  // rest | http | kafka
  "route_pattern": "/orders",          // REST; supports /orders/{id} params
  "methods": ["POST"],
  "workflow_id": "kebab-case-id"
}

## Lifecycle rules (important)

1. Create returns a DRAFT. Nothing serves traffic until activated.
2. Test drafts before activating:
   POST /api/v1/admin/workflows/{id}/test  with body { "data": { ... } }
   → returns an execution trace (which tasks ran / were skipped).
3. Activate: PATCH /api/v1/admin/workflows/{id}/status  {"status":"active"}
   (same for channels). The engine hot-reloads; no restart.
4. Active versions are IMMUTABLE. To change one, create a new version and
   activate it; rollback = re-activating a previous version.

## Core API calls

POST  /api/v1/admin/workflows                      create workflow (draft)
POST  /api/v1/admin/workflows/{id}/test            dry-run with sample data
PATCH /api/v1/admin/workflows/{id}/status          {"status":"active"}
POST  /api/v1/admin/channels                       create channel (draft)
PATCH /api/v1/admin/channels/{id}/status           {"status":"active"}
POST  /api/v1/admin/connectors                     create connector (live immediately)
GET   /api/v1/admin/functions                      function input schemas
GET   /api/v1/openapi.json                         full OpenAPI 3.1 spec
POST  /api/v1/data/{route}                         call a deployed service
POST  /api/v1/data/{route}/async                   async: returns trace_id
GET   /api/v1/data/traces/{trace_id}               poll async result

## Working style

- Always: create draft → test with realistic sample data → activate.
- Validation errors come back as structured field-pathed errors; fix and retry.
- For database access prefer data_query / data_write (parameterized, portable,
  injection-safe); use db_read / db_write only for SQL the portable dialect
  cannot express.
- Connector configs support ${VAR} / ${VAR:-default} environment references —
  never embed real credentials in JSON.
````

A worked end-to-end session using exactly these calls is in
[Install & First Service](../tutorials/cli-setup.md#create-your-first-service);
the deeper references the LLM (or you) may need are the
[Workflow Reference](../reference/workflows.md),
[Function Reference](../reference/functions.md), and
[Portable Data Dialect](../reference/data-dialect.md).

> **Tip:** the docs site also serves [`llms.txt`](https://goplasmatic.github.io/Orion/llms.txt)
> (a machine-readable index) and [`llms-full.txt`](https://goplasmatic.github.io/Orion/llms-full.txt)
> (the entire documentation as one file) — point your tools at those for full
> documentation context.


---
source: ./architecture/overview.md
---

# Architecture Overview

## Three Primitives

Services in Orion are composed from three building blocks:

| Primitive | Role | Examples |
|-----------|------|----------|
| **Channel** | Service endpoint: sync (REST, HTTP) or async (Kafka) | `POST /orders`, `GET /users/{id}`, Kafka topic `order.placed` |
| **Workflow** | Pipeline of tasks that defines what the service does | Parse → validate → enrich → transform → respond |
| **Connector** | Named connection to an external system with auth and retries | Stripe API, PostgreSQL, Redis, Kafka cluster |

Channels receive traffic. Workflows process it. Connectors reach out to external systems. Everything else (rate limiting, metrics, circuit breakers, versioning) is handled by the platform.

## Deployment Topology

### Before Orion

Every piece of business logic is its own service to build, deploy, and operate, each with its own infrastructure stack:

```orion-diagram
{
  "direction": "TB",
  "groups": [
    { "id": "g_svc1", "label": "Pricing Service (Auto-Scaled)" },
    { "id": "g_svc2", "label": "Fraud Service (Auto-Scaled)" },
    { "id": "g_svc3", "label": "Routing Service (Auto-Scaled)" },
    { "id": "g_svc4", "label": "Notification Service (Auto-Scaled)" },
    { "id": "g_obs", "label": "Observability Stack" },
    { "id": "g_ci", "label": "CI/CD per Service (x4)" },
    { "id": "g_inf", "label": "Shared Platform Infrastructure" }
  ],
  "nodes": [
    { "id": "cdn", "label": "CDN / Edge Cache", "type": "gateway", "shape": "cloud" },
    { "id": "lb", "label": "Load Balancer", "type": "gateway", "shape": "hexagon" },
    { "id": "gw", "label": "API Gateway", "sublabel": "auth · throttle · routing", "type": "gateway", "shape": "hexagon" },

    { "id": "sm1", "label": "Mesh Proxy", "type": "gateway", "group": "g_svc1" },
    { "id": "p1", "label": "Container", "type": "service", "group": "g_svc1" },
    { "id": "p1h", "label": "Health Check", "type": "channel", "group": "g_svc1" },
    { "id": "p1l", "label": "Log Agent", "type": "observability", "group": "g_svc1" },
    { "id": "p1m", "label": "Metrics Agent", "type": "observability", "group": "g_svc1" },

    { "id": "sm2", "label": "Mesh Proxy", "type": "gateway", "group": "g_svc2" },
    { "id": "f1", "label": "Container", "type": "service", "group": "g_svc2" },
    { "id": "f1h", "label": "Health Check", "type": "channel", "group": "g_svc2" },
    { "id": "f1l", "label": "Log Agent", "type": "observability", "group": "g_svc2" },
    { "id": "f1m", "label": "Metrics Agent", "type": "observability", "group": "g_svc2" },

    { "id": "sm3", "label": "Mesh Proxy", "type": "gateway", "group": "g_svc3" },
    { "id": "r1", "label": "Container", "type": "service", "group": "g_svc3" },
    { "id": "r1h", "label": "Health Check", "type": "channel", "group": "g_svc3" },
    { "id": "r1l", "label": "Log Agent", "type": "observability", "group": "g_svc3" },
    { "id": "r1m", "label": "Metrics Agent", "type": "observability", "group": "g_svc3" },

    { "id": "sm4", "label": "Mesh Proxy", "type": "gateway", "group": "g_svc4" },
    { "id": "n1", "label": "Container", "type": "service", "group": "g_svc4" },
    { "id": "n1h", "label": "Health Check", "type": "channel", "group": "g_svc4" },
    { "id": "n1l", "label": "Log Agent", "type": "observability", "group": "g_svc4" },
    { "id": "n1m", "label": "Metrics Agent", "type": "observability", "group": "g_svc4" },

    { "id": "db", "label": "Database", "sublabel": "Primary + Replica", "type": "datastore" },
    { "id": "cache", "label": "Cache Cluster", "type": "datastore" },
    { "id": "mq", "label": "Message Queue", "type": "datastore", "shape": "queue" },
    { "id": "smtp", "label": "Email Service", "type": "datastore", "shape": "cloud" },

    { "id": "log", "label": "Log Aggregator", "type": "observability", "group": "g_obs" },
    { "id": "met", "label": "Metrics Backend", "type": "observability", "group": "g_obs" },
    { "id": "trc", "label": "Trace Collector", "type": "observability", "group": "g_obs" },
    { "id": "dash", "label": "Dashboards & Alerts", "type": "observability", "group": "g_obs" },

    { "id": "reg", "label": "Container Registry", "type": "ci", "group": "g_ci" },
    { "id": "pipe", "label": "Build Pipeline", "type": "ci", "group": "g_ci" },
    { "id": "deploy", "label": "Deploy Controller", "type": "ci", "group": "g_ci" },
    { "id": "canary", "label": "Canary / Rollout", "type": "ci", "group": "g_ci" },

    { "id": "sr", "label": "Service Registry", "type": "infra", "group": "g_inf" },
    { "id": "sec", "label": "Secret Manager", "type": "infra", "group": "g_inf" },
    { "id": "cfg", "label": "Config Server", "type": "infra", "group": "g_inf" },
    { "id": "cert", "label": "Certificate Manager", "type": "infra", "group": "g_inf" }
  ],
  "edges": [
    { "from": "cdn", "to": "lb" }, { "from": "lb", "to": "gw" },
    { "from": "gw", "to": "sm1" }, { "from": "gw", "to": "sm2" }, { "from": "gw", "to": "sm3" }, { "from": "gw", "to": "sm4" },
    { "from": "sm1", "to": "p1" }, { "from": "sm2", "to": "f1" }, { "from": "sm3", "to": "r1" }, { "from": "sm4", "to": "n1" },
    { "from": "p1", "to": "p1h" }, { "from": "p1", "to": "p1l" }, { "from": "p1", "to": "p1m" },
    { "from": "f1", "to": "f1h" }, { "from": "f1", "to": "f1l" }, { "from": "f1", "to": "f1m" },
    { "from": "r1", "to": "r1h" }, { "from": "r1", "to": "r1l" }, { "from": "r1", "to": "r1m" },
    { "from": "n1", "to": "n1h" }, { "from": "n1", "to": "n1l" }, { "from": "n1", "to": "n1m" },
    { "from": "p1", "to": "db" }, { "from": "f1", "to": "cache" }, { "from": "r1", "to": "mq" }, { "from": "n1", "to": "smtp" },
    { "from": "p1l", "to": "log" }, { "from": "f1l", "to": "log" }, { "from": "r1l", "to": "log" }, { "from": "n1l", "to": "log" },
    { "from": "p1m", "to": "met" }, { "from": "f1m", "to": "met" }, { "from": "r1m", "to": "met" }, { "from": "n1m", "to": "met" },
    { "from": "log", "to": "dash" }, { "from": "met", "to": "dash" }, { "from": "trc", "to": "dash" }
  ]
}
```

**4 services x (code + Dockerfile + CI pipeline + health checks + metrics agent + log agent + sidecar proxy + scaling policy + secret config + canary rollout) = dozens of components to build, wire, and keep running.**

### After Orion

One Orion instance replaces all four:

```orion-diagram
{
  "direction": "LR",
  "groups": [ { "id": "g_orion", "label": "Orion Runtime" } ],
  "nodes": [
    { "id": "clients", "label": "Clients", "type": "gateway", "shape": "actor" },
    { "id": "ch1", "label": "/pricing", "type": "channel", "group": "g_orion" },
    { "id": "ch2", "label": "/fraud", "type": "channel", "group": "g_orion" },
    { "id": "ch3", "label": "/routing", "type": "channel", "group": "g_orion" },
    { "id": "ch4", "label": "/notify", "type": "channel", "group": "g_orion" },
    { "id": "wf1", "label": "workflow", "type": "service", "group": "g_orion" },
    { "id": "wf2", "label": "workflow", "type": "service", "group": "g_orion" },
    { "id": "wf3", "label": "workflow", "type": "service", "group": "g_orion" },
    { "id": "wf4", "label": "workflow", "type": "service", "group": "g_orion" },
    { "id": "db", "label": "Database", "type": "datastore" },
    { "id": "rd", "label": "Redis", "type": "datastore" },
    { "id": "kf", "label": "Kafka", "type": "datastore", "shape": "queue" },
    { "id": "smtp", "label": "SMTP", "type": "datastore", "shape": "cloud" }
  ],
  "edges": [
    { "from": "clients", "to": "ch1" }, { "from": "clients", "to": "ch2" }, { "from": "clients", "to": "ch3" }, { "from": "clients", "to": "ch4" },
    { "from": "ch1", "to": "wf1" }, { "from": "ch2", "to": "wf2" }, { "from": "ch3", "to": "wf3" }, { "from": "ch4", "to": "wf4" },
    { "from": "wf1", "to": "db" }, { "from": "wf2", "to": "rd" }, { "from": "wf3", "to": "kf" }, { "from": "wf4", "to": "smtp" }
  ]
}
```

No API gateway needed. Governance is built in. One binary to deploy.

**The best of both worlds:** each channel and workflow is independently versioned, testable, and deployable. The modularity of microservices with the operational simplicity of a monolith.

## Deploy Anywhere

```orion-diagram
{
  "direction": "LR",
  "groups": [
    { "id": "g_standalone", "label": "Standalone" },
    { "id": "g_sidecar", "label": "Sidecar" },
    { "id": "g_docker", "label": "Docker" }
  ],
  "nodes": [
    { "id": "s1", "label": "./orion-server", "sublabel": "That's it.", "type": "infra", "group": "g_standalone" },
    { "id": "app", "label": "Your App", "type": "service", "group": "g_sidecar" },
    { "id": "sc", "label": "Orion", "type": "infra", "group": "g_sidecar" },
    { "id": "d1", "label": "docker run", "sublabel": "orion:latest", "type": "observability", "group": "g_docker" }
  ],
  "edges": [
    { "from": "app", "to": "sc" }
  ]
}
```

Single binary. SQLite by default, no database to provision, no runtime dependencies. Need more scale? Swap to **PostgreSQL** or **MySQL** by changing `storage.url`. No rebuild needed.

Same channel definitions work in any topology: run everything in one instance, split channels across instances with include/exclude filters, or deploy as sidecars.

## Request Processing Flow

```orion-diagram
{
  "direction": "LR",
  "nodes": [
    { "id": "req", "label": "HTTP Request", "type": "service" },
    { "id": "router", "label": "Axum Router", "type": "gateway" },
    { "id": "handler", "label": "Data Route Handler", "type": "gateway" },
    { "id": "resolve", "label": "Route Resolution", "sublabel": "pattern match → channel", "type": "gateway" },
    { "id": "registry", "label": "Channel Registry", "sublabel": "dedup · rate limit · validation", "type": "ci" },
    { "id": "engine", "label": "Engine", "sublabel": "RwLock<Arc<Engine>>", "type": "observability" },
    { "id": "matcher", "label": "Workflow Matcher", "sublabel": "JSONLogic + rollout", "type": "gateway" },
    { "id": "pipeline", "label": "Task Pipeline", "sublabel": "ordered execution", "type": "gateway" },
    { "id": "resp", "label": "JSON Response", "type": "channel" }
  ],
  "edges": [
    { "from": "req", "to": "router" }, { "from": "router", "to": "handler" }, { "from": "handler", "to": "resolve" },
    { "from": "resolve", "to": "registry" }, { "from": "registry", "to": "engine" }, { "from": "engine", "to": "matcher" },
    { "from": "matcher", "to": "pipeline" }, { "from": "pipeline", "to": "resp" }
  ]
}
```

1. **Route Resolution:** REST pattern matching finds the channel, or falls back to name lookup
2. **Channel Registry:** enforces deduplication, rate limits, input validation, backpressure, and checks the response cache
3. **Engine:** the workflow engine sits behind a double-Arc (`Arc<RwLock<Arc<Engine>>>`) allowing zero-downtime swaps
4. **Workflow Matcher:** evaluates JSONLogic conditions and rollout percentages to pick the right workflow
5. **Task Pipeline:** executes functions in order (parse, map, filter, http_call, db_read, etc.)

## Sync and Async

```
Sync     POST /api/v1/data/{channel}         → immediate response
Async    POST /api/v1/data/{channel}/async   → returns trace_id, poll later

REST     GET /api/v1/data/orders/{id}        → matched by route pattern
Kafka    topic: order.placed                 → consumed automatically
```

Sync channels respond immediately. Async channels return a trace ID; poll `GET /api/v1/data/traces/{id}` for results. Kafka channels consume from topics configured in the DB or config file.

**Bridging is a pattern, not a feature.** A sync workflow can `publish_kafka` and return 202. An async channel picks it up from there.

## Service Composition

Most platforms require HTTP calls between services, adding latency, failure modes, and serialization overhead. Orion's `channel_call` invokes another channel's workflow **in-process** with zero network round-trip:

```orion-diagram
{
  "direction": "TB",
  "groups": [ { "id": "wf", "label": "order-processing workflow · POST /orders" } ],
  "nodes": [
    { "id": "p", "label": "parse_json", "sublabel": "extract order data", "type": "service", "group": "wf" },
    { "id": "c1", "label": "channel_call", "type": "service", "group": "wf" },
    { "id": "c2", "label": "channel_call", "type": "service", "group": "wf" },
    { "id": "m", "label": "map", "sublabel": "compute pricing", "type": "service", "group": "wf" },
    { "id": "pub", "label": "publish_json", "sublabel": "return combined result", "type": "service", "group": "wf" },
    { "id": "inv", "label": "inventory-check channel", "type": "channel" },
    { "id": "cust", "label": "customer-lookup channel", "type": "channel" }
  ],
  "edges": [
    { "from": "p", "to": "c1" }, { "from": "c1", "to": "c2" }, { "from": "c2", "to": "m" }, { "from": "m", "to": "pub" },
    { "from": "c1", "to": "inv", "label": "in-process", "style": "dashed" },
    { "from": "c2", "to": "cust", "label": "in-process", "style": "dashed" }
  ]
}
```

Each composed channel has its own workflow, versioning, and governance, but calls between them are function calls, not network hops. Cycle detection prevents infinite recursion.

## Built-in Task Functions

Two sources contribute task functions, all compiled into every binary:

**From dataflow-rs 3.0 (workflow primitives):**

| Function | Description |
|----------|-------------|
| `parse_json` | Parse payload into the data context |
| `parse_xml` | Parse XML payloads into structured JSON |
| `filter` | Allow or halt processing based on JSONLogic conditions |
| `map` | Transform and reshape JSON using JSONLogic expressions |
| `validation` | Enforce required fields and constraints |
| `publish_json` / `publish_xml` | Serialize data to JSON or XML output |
| `log` | Emit structured log entries |

**From Orion (connector- and channel-backed handlers):**

| Function | Description |
|----------|-------------|
| `http_call` | Invoke downstream APIs via an HTTP connector |
| `channel_call` | Invoke another channel's workflow in-process |
| `data_query` / `data_write` | Portable, backend-neutral queries and mutations against SQL, MongoDB, or Elasticsearch connectors (see the [Portable Data Dialect](../reference/data-dialect.md)) |
| `db_read` / `db_write` | Execute raw SQL against a SQL connector, return rows / affected count |
| `cache_read` / `cache_write` | Read/write to an in-memory or Redis cache connector |
| `mongo_read` | Query MongoDB collections with raw find() filters |
| `publish_kafka` | Publish messages via a Kafka connector |

The Orion handlers have machine-readable input schemas surfaced at `GET /api/v1/admin/functions`, and workflow create/update calls validate `function.input` against those schemas with field-pathed `FieldError`s before the workflow can be activated.


---
source: ./architecture/characteristics.md
---

# Architectural Characteristics

Orion provides production-grade capabilities across eight architectural dimensions. Each subcategory below links to its detailed documentation.

> **C** Creational · **S** Structural · **B** Behavioral
>
> *Click a node to expand its capabilities. Click any leaf node to jump to its documentation.*

<div id="mindmap" class="d3-diagram" style="height:600px"></div>

---

## Observability — S

| Area | Capabilities |
|------|-------------|
| [Structured Logging](../features/observability.md#structured-logging) | JSON & pretty-print formats · Configurable log levels · Per-request context · Per-crate filtering |
| [Prometheus Metrics](../features/observability.md#prometheus-metrics) | Request counters & error rates · Latency histograms · Circuit breaker metrics · Rate limit rejections |
| [Distributed Tracing](../features/observability.md#distributed-tracing) | W3C Trace Context · OpenTelemetry OTLP export · Configurable sampling rate · Per-task span tracking |
| [Health Monitoring](../features/observability.md#health-monitoring) | Component-level health checks · Automatic degradation · Request ID propagation · Kubernetes liveness & readiness probes |

## Resilience — S

| Area | Capabilities |
|------|-------------|
| [Circuit Breakers](../features/resilience.md#circuit-breakers) | Lock-free state machine · Per-connector isolation · Auto-recovery after cooldown · Admin API to inspect & reset |
| [Retry & Backoff](../features/resilience.md#retry-backoff) | Exponential backoff (capped 60 s) · Configurable max retries · Retryable error detection |
| [Timeouts](../features/resilience.md#timeouts) | Per-channel enforcement · Workflow execution limits · Per-connector query timeout |
| [Fault Tolerance](../features/resilience.md#fault-tolerance) | Graceful shutdown (SIGTERM/SIGINT) · Connection draining · Dead letter queue with retry · Panic recovery middleware |

## Security — B

| Area | Capabilities |
|------|-------------|
| [Secret Management](../features/security.md#secret-management) | Auto-masked API responses · Credential isolation via connectors |
| [Input Validation](../features/security.md#input-validation) | Per-channel JSONLogic rules · Payload size limits · Header & query param access |
| [Network Security](../features/security.md#network-security) | SSRF protection (private IP blocking) · TLS/HTTPS support · Security headers (CSP, X-Frame-Options) |
| [Access Control](../features/security.md#access-control) | Admin API authentication · Per-channel CORS enforcement · Origin allowlist |
| [Data Safety](../features/security.md#data-safety) | Parameterized SQL queries · Injection protection · URL validation |

## Scalability — C

| Area | Capabilities |
|------|-------------|
| [Rate Limiting](../features/scalability.md#rate-limiting) | Token bucket algorithm · Per-client keying via JSONLogic · Platform & per-channel limits |
| [Backpressure](../features/scalability.md#backpressure) | Semaphore concurrency limits · 503 load shedding · Per-channel configuration |
| [Async Processing](../features/scalability.md#async-processing) | Multi-worker trace queue · Bounded buffer channels · DLQ retry processor |
| [Horizontal Scaling](../features/scalability.md#horizontal-scaling) | Stateless instances · Channel include/exclude filters · Multi-database backends |

## Deployability — C

| Area | Capabilities |
|------|-------------|
| [Packaging](../features/deployability.md#packaging) | Single binary · SQLite, PostgreSQL, MySQL · Minimal footprint |
| [Containerization](../features/deployability.md#containerization) | Multi-stage Docker build · Non-root execution · Built-in health probes |
| [Configuration](../features/deployability.md#configuration) | TOML + env var overrides · Sensible defaults · Runtime configuration |
| [Distribution](../features/deployability.md#distribution) | Homebrew tap · Shell & PowerShell installers · Multi-platform binaries |

## Extensibility — S

| Area | Capabilities |
|------|-------------|
| [Connectors](../features/extensibility.md#connectors) | HTTP & Webhooks · Kafka pub/sub · Database (SQL) · Cache (Memory & Redis) · Storage (S3/GCS) · MongoDB (NoSQL) |
| [Custom Functions](../features/extensibility.md#custom-functions) | Async function handlers · Built-in function library · JSONLogic expressions |
| [Channel Protocols](../features/extensibility.md#channel-protocols) | REST with route matching (sync) · Simple HTTP (sync) · Kafka (async) |

## Availability — C

| Area | Capabilities |
|------|-------------|
| [Hot-Reload](../features/availability.md#hot-reload) | Zero-downtime engine swap · Channel registry rebuild · Kafka consumer restart |
| [Canary Rollouts](../features/availability.md#canary-rollouts) | Percentage-based traffic split · Gradual migration · Instant rollback |
| [Versioning](../features/availability.md#versioning) | Draft / Active / Archived lifecycle · Multi-version history · Workflow import & export |
| [Performance](../features/availability.md#performance) | Response caching · Request deduplication · Connection pool caching |

## Maintainability — B

| Area | Capabilities |
|------|-------------|
| [Admin APIs](../features/maintainability.md#admin-apis) | Full CRUD for all entities · Version management · Engine control · OpenAPI / Swagger UI |
| [CI/CD Integration](../features/maintainability.md#cicd-integration) | Bulk import & export · Pre-deploy validation · GitOps-friendly |
| [Testing](../features/maintainability.md#testing) | Dry-run execution · Workflow validation · Step-by-step traces |
| [Operations](../features/maintainability.md#operations) | Audit logging · Database backup & restore · Config validation CLI |

<script src="https://d3js.org/d3.v7.min.js"></script>
<script>
(function () {
  var container = document.getElementById('mindmap');
  if (!container) return;

  var isDark = document.documentElement.classList.contains('coal')
    || document.documentElement.classList.contains('navy')
    || document.documentElement.classList.contains('ayu');

  var TYPE_COLORS = { C: '#119FCD', S: '#4CBD97', B: '#FFD167' };

  /* ═══ DATA — inlined from original features.json with doc links ═══ */
  var categories = [
    {
      name: 'Observability', type: 'S',
      color: isDark ? '#102838' : '#D0EBF5', border: '#119FCD',
      children: [
        { name: 'Structured Logging', link: '../features/observability.html#structured-logging',
          children: ['JSON & pretty-print formats', 'Configurable log levels', 'Per-request context', 'Per-crate filtering'] },
        { name: 'Prometheus Metrics', link: '../features/observability.html#prometheus-metrics',
          children: ['Request counters & error rates', 'Latency histograms', 'Circuit breaker metrics', 'Rate limit rejections'] },
        { name: 'Distributed Tracing', link: '../features/observability.html#distributed-tracing',
          children: ['W3C Trace Context', 'OpenTelemetry OTLP export', 'Configurable sampling rate', 'Per-task span tracking'] },
        { name: 'Health Monitoring', link: '../features/observability.html#health-monitoring',
          children: ['Component-level health checks', 'Automatic degradation', 'Request ID propagation', 'Kubernetes probes'] }
      ]
    },
    {
      name: 'Resilience', type: 'S',
      color: isDark ? '#0F2238' : '#D5EDF8', border: '#7DD3FC',
      children: [
        { name: 'Circuit Breakers', link: '../features/resilience.html#circuit-breakers',
          children: ['Lock-free state machine', 'Per-connector isolation', 'Auto-recovery after cooldown', 'Admin API to inspect & reset'] },
        { name: 'Retry & Backoff', link: '../features/resilience.html#retry-backoff',
          children: ['Exponential backoff (capped 60s)', 'Configurable max retries', 'Retryable error detection'] },
        { name: 'Timeouts', link: '../features/resilience.html#timeouts',
          children: ['Per-channel enforcement', 'Workflow execution limits', 'Per-connector query timeout'] },
        { name: 'Fault Tolerance', link: '../features/resilience.html#fault-tolerance',
          children: ['Graceful shutdown (SIGTERM/SIGINT)', 'Connection draining', 'Dead letter queue with retry', 'Panic recovery middleware'] }
      ]
    },
    {
      name: 'Security', type: 'B',
      color: isDark ? '#251520' : '#F5D5DD', border: '#EF476F',
      children: [
        { name: 'Secret Management', link: '../features/security.html#secret-management',
          children: ['Auto-masked API responses', 'Credential isolation via connectors'] },
        { name: 'Input Validation', link: '../features/security.html#input-validation',
          children: ['Per-channel JSONLogic rules', 'Payload size limits', 'Header & query param access'] },
        { name: 'Network Security', link: '../features/security.html#network-security',
          children: ['SSRF protection (private IP blocking)', 'TLS/HTTPS support', 'Security headers'] },
        { name: 'Access Control', link: '../features/security.html#access-control',
          children: ['Admin API authentication', 'Per-channel CORS enforcement', 'Origin allowlist'] },
        { name: 'Data Safety', link: '../features/security.html#data-safety',
          children: ['Parameterized SQL queries', 'Injection protection', 'URL validation'] }
      ]
    },
    {
      name: 'Scalability', type: 'C',
      color: isDark ? '#0F2530' : '#CCE8F0', border: '#119FCD',
      children: [
        { name: 'Rate Limiting', link: '../features/scalability.html#rate-limiting',
          children: ['Token bucket algorithm', 'Per-client keying via JSONLogic', 'Platform & per-channel limits'] },
        { name: 'Backpressure', link: '../features/scalability.html#backpressure',
          children: ['Semaphore concurrency limits', '503 load shedding', 'Per-channel configuration'] },
        { name: 'Async Processing', link: '../features/scalability.html#async-processing',
          children: ['Multi-worker trace queue', 'Bounded buffer channels', 'DLQ retry processor'] },
        { name: 'Horizontal Scaling', link: '../features/scalability.html#horizontal-scaling',
          children: ['Stateless instances', 'Channel include/exclude filters', 'Multi-database backends'] }
      ]
    },
    {
      name: 'Deployability', type: 'C',
      color: isDark ? '#201E10' : '#F5EDD0', border: '#FFD167',
      children: [
        { name: 'Packaging', link: '../features/deployability.html#packaging',
          children: ['Single binary', 'SQLite, PostgreSQL, MySQL', 'Minimal footprint'] },
        { name: 'Containerization', link: '../features/deployability.html#containerization',
          children: ['Multi-stage Docker build', 'Non-root execution', 'Built-in health probes'] },
        { name: 'Configuration', link: '../features/deployability.html#configuration',
          children: ['TOML + env var overrides', 'Sensible defaults', 'Runtime configuration'] },
        { name: 'Distribution', link: '../features/deployability.html#distribution',
          children: ['Homebrew tap', 'Shell & PowerShell installers', 'Multi-platform binaries'] }
      ]
    },
    {
      name: 'Extensibility', type: 'S',
      color: isDark ? '#0F2822' : '#D0F0E5', border: '#4CBD97',
      children: [
        { name: 'Connectors', link: '../features/extensibility.html#connectors',
          children: ['HTTP & Webhooks', 'Kafka pub/sub', 'Database (SQL)', 'Cache (Memory & Redis)', 'Storage (S3/GCS)', 'MongoDB (NoSQL)'] },
        { name: 'Custom Functions', link: '../features/extensibility.html#custom-functions',
          children: ['Async function handlers', 'Built-in function library', 'JSONLogic expressions'] },
        { name: 'Channel Protocols', link: '../features/extensibility.html#channel-protocols',
          children: ['REST with route matching (sync)', 'Simple HTTP (sync)', 'Kafka (async)'] }
      ]
    },
    {
      name: 'Availability', type: 'C',
      color: isDark ? '#102520' : '#D0EDE5', border: '#4CBD97',
      children: [
        { name: 'Hot-Reload', link: '../features/availability.html#hot-reload',
          children: ['Zero-downtime engine swap', 'Channel registry rebuild', 'Kafka consumer restart'] },
        { name: 'Canary Rollouts', link: '../features/availability.html#canary-rollouts',
          children: ['Percentage-based traffic split', 'Gradual migration', 'Instant rollback'] },
        { name: 'Versioning', link: '../features/availability.html#versioning',
          children: ['Draft / Active / Archived lifecycle', 'Multi-version history', 'Workflow import & export'] },
        { name: 'Performance', link: '../features/availability.html#performance',
          children: ['Response caching', 'Request deduplication', 'Connection pool caching'] }
      ]
    },
    {
      name: 'Maintainability', type: 'B',
      color: isDark ? '#151E28' : '#D8E5ED', border: '#7FAFC0',
      children: [
        { name: 'Admin APIs', link: '../features/maintainability.html#admin-apis',
          children: ['Full CRUD for all entities', 'Version management', 'Engine control', 'OpenAPI / Swagger UI'] },
        { name: 'CI/CD Integration', link: '../features/maintainability.html#cicd-integration',
          children: ['Bulk import & export', 'Pre-deploy validation', 'GitOps-friendly'] },
        { name: 'Testing', link: '../features/maintainability.html#testing',
          children: ['Dry-run execution', 'Workflow validation', 'Step-by-step traces'] },
        { name: 'Operations', link: '../features/maintainability.html#operations',
          children: ['Audit logging', 'Database backup & restore', 'Config validation CLI'] }
      ]
    }
  ];

  /* ═══ BUILD TREE ═══ */
  function makeNode(cat) {
    return {
      _id: cat.name, name: cat.name, color: cat.color, border: cat.border,
      catType: cat.type, lvl: 1, _c: null,
      children: cat.children.map(function (sub) {
        return {
          _id: cat.name + '/' + sub.name, name: sub.name, link: sub.link,
          color: cat.color, border: cat.border, lvl: 2,
          children: null,
          _c: (sub.children || []).map(function (leaf) {
            return {
              _id: cat.name + '/' + sub.name + '/' + leaf, name: leaf,
              color: cat.color, border: cat.border, lvl: 3, link: sub.link
            };
          })
        };
      })
    };
  }

  var cats = categories.map(makeNode);
  var leftCats = cats.slice(0, 4);
  var rightCats = cats.slice(4);

  /* ═══ SVG ═══ */
  var rect = container.getBoundingClientRect();
  var W = rect.width || 800, H = rect.height || 600;
  var svg = d3.select(container).append('svg')
    .attr('width', '100%').attr('height', '100%')
    .style('display', 'block').style('cursor', 'grab');
  var g = svg.append('g');
  var zoomBeh = d3.zoom().scaleExtent([0.1, 3]).on('zoom', function (e) { g.attr('transform', e.transform); });
  svg.call(zoomBeh);
  svg.on('mousedown.cursor', function () { svg.style('cursor', 'grabbing'); });
  svg.on('mouseup.cursor mouseleave.cursor', function () { svg.style('cursor', 'grab'); });

  var defs = g.append('defs');
  [['shadow', 2, 4, 0.10], ['shadow-sm', 1, 2.5, 0.06]].forEach(function (s) {
    var f = defs.append('filter').attr('id', s[0])
      .attr('x', '-25%').attr('y', '-25%').attr('width', '150%').attr('height', '150%');
    f.append('feDropShadow').attr('dx', 0).attr('dy', s[1])
      .attr('stdDeviation', s[2]).attr('flood-color', 'rgba(0,0,0,' + s[3] + ')');
  });

  var gLinks = g.append('g');
  var gBadges = g.append('g');
  var gNodes = g.append('g');

  /* ═══ LAYOUT ═══ */
  var linkH = d3.linkHorizontal().x(function (d) { return d[0]; }).y(function (d) { return d[1]; });
  var SIB = 36, DEPTH = 185, CAT_GAP = 50;

  function computeAll() {
    var nodes = [], links = [], badges = [];
    nodes.push({ _id: '__center__', name: 'Orion', x: 0, y: 0, lvl: 0, parentId: null });

    function doSide(catArr, xSign) {
      var metas = catArr.map(function (cat) {
        var root = d3.hierarchy(cat, function (d) { return d.children; });
        d3.tree().nodeSize([SIB, DEPTH])(root);
        var ys = root.descendants().map(function (d) { return d.x; });
        return { cat: cat, root: root, minY: d3.min(ys), maxY: d3.max(ys), h: d3.max(ys) - d3.min(ys) };
      });

      var cursor = 0, offsets = [];
      metas.forEach(function (m) { offsets.push(cursor - m.minY); cursor += m.h + CAT_GAP; });
      var totalH = cursor - CAT_GAP;
      var yShift = -totalH / 2;

      metas.forEach(function (m, mi) {
        var yOff = offsets[mi] + yShift;
        m.root.descendants().forEach(function (d) {
          var data = d.data;
          var sx = xSign * (d.y + 160);
          var sy = d.x + yOff;
          var parentId = d.depth === 0 ? '__center__' : d.parent.data._id;
          var px = d.depth === 0 ? 0 : xSign * (d.parent.y + 160);
          var py = d.depth === 0 ? 0 : d.parent.x + yOff;

          nodes.push({
            _id: data._id, name: data.name, x: sx, y: sy,
            lvl: data.lvl, color: data.color, border: data.border,
            catType: data.catType, parentId: parentId, link: data.link,
            hasKids: !!(data.children || data._c),
            expanded: !!data.children, data: data
          });
          links.push({ _id: 'l:' + data._id, sourceId: parentId, sx: px, sy: py, tx: sx, ty: sy });

          if (data.lvl === 1 && data.catType)
            badges.push({ _id: 'b:' + data._id, type: data.catType, x: (px + sx) / 2, y: (py + sy) / 2 });
        });
      });
    }

    doSide(leftCats, -1);
    doSide(rightCats, 1);
    return { nodes: nodes, links: links, badges: badges };
  }

  /* ═══ RENDER ═══ */
  function render(animate) {
    var computed = computeAll();
    var nodes = computed.nodes, links = computed.links, badges = computed.badges;
    var dur = animate ? 400 : 0;
    var t = dur ? d3.transition().duration(dur).ease(d3.easeCubicOut) : null;
    function tr(s) { return t ? s.transition(t) : s; }

    // Links
    var lS = gLinks.selectAll('.mm-link').data(links, function (d) { return d._id; });
    tr(lS.exit()).attr('opacity', 0).remove();
    var lE = lS.enter().append('path').attr('class', 'mm-link').attr('opacity', 0)
      .attr('fill', 'none').attr('stroke', isDark ? '#3D6B7D' : '#7FAFC0')
      .attr('stroke-width', 1.8).attr('stroke-linecap', 'round')
      .attr('d', function () { return linkH({ source: [0, 0], target: [0, 0] }); });
    tr(lE.merge(lS))
      .attr('d', function (d) { return linkH({ source: [d.sx, d.sy], target: [d.tx, d.ty] }); })
      .attr('opacity', 0.5);

    // Badges
    var bS = gBadges.selectAll('.type-badge').data(badges, function (d) { return d._id; });
    bS.exit().remove();
    var bE = bS.enter().append('g').attr('class', 'type-badge');
    bE.append('circle'); bE.append('text').attr('text-anchor', 'middle').attr('dominant-baseline', 'central').attr('pointer-events', 'none');
    var bA = bE.merge(bS);
    bA.each(function (d) {
      var el = d3.select(this), col = TYPE_COLORS[d.type] || '#999';
      el.select('circle').attr('r', 10).attr('fill', isDark ? '#07111A' : '#F4F8FB').attr('stroke', col).attr('stroke-width', 1.5).attr('stroke-dasharray', '3 2');
      el.select('text').attr('font-size', 8).attr('font-weight', 700).attr('fill', col).attr('font-family', "'Inter',system-ui,sans-serif").text(d.type);
    });
    tr(bA).attr('transform', function (d) { return 'translate(' + d.x + ',' + d.y + ')'; }).attr('opacity', 1);

    // Nodes
    var nS = gNodes.selectAll('.mm-node').data(nodes, function (d) { return d._id; });
    tr(nS.exit()).attr('opacity', 0).remove();
    var nE = nS.enter().append('g').attr('class', 'mm-node').attr('opacity', 0)
      .attr('transform', 'translate(0,0)');
    nE.append('rect').attr('class', 'node-bg');
    nE.append('text').attr('text-anchor', 'middle').attr('dominant-baseline', 'central');
    nE.append('text').attr('class', 'expand-hint').attr('dominant-baseline', 'central').attr('pointer-events', 'none');
    var nA = nE.merge(nS);

    nA.each(function (d) {
      var el = d3.select(this);
      var L = d.lvl;
      var fs = [15, 12, 10.5, 9.5][L];
      var fw = [700, 700, 600, 500][L];
      var tc = L === 0 ? '#ECF4F8' : (isDark ? '#ECF4F8' : '#07111A');
      var txt = el.select('text:not(.expand-hint)').attr('font-size', fs).attr('font-weight', fw).attr('fill', tc)
        .attr('font-family', "'Inter',system-ui,sans-serif").text(d.name);
      var tw = txt.node().getComputedTextLength();
      var px = [26, 12, 9, 7][L], py = [11, 7, 5, 4][L];
      var rw = tw + px * 2, rh = fs + py * 2, rx = [8, 5, 4, 4][L];

      function blendWhite(hex, amount) {
        var c = d3.color(hex);
        return d3.rgb(
          Math.round(c.r + (255 - c.r) * amount),
          Math.round(c.g + (255 - c.g) * amount),
          Math.round(c.b + (255 - c.b) * amount)
        ) + '';
      }
      function blendDark(hex, amount) {
        var c = d3.color(hex);
        return d3.rgb(
          Math.round(c.r * (1 - amount)),
          Math.round(c.g * (1 - amount)),
          Math.round(c.b * (1 - amount))
        ) + '';
      }

      var fill, stroke;
      if (L === 0) { fill = '#119FCD'; stroke = '#0D7FA3'; }
      else if (L === 1) { fill = d.color; stroke = d.border; }
      else if (L === 2) { fill = isDark ? blendDark(d.color, 0.3) : blendWhite(d.color, 0.6); stroke = d.border; }
      else { fill = isDark ? blendDark(d.color, 0.5) : blendWhite(d.color, 0.78); stroke = d.border; }

      el.select('rect').attr('x', -rw / 2).attr('y', -rh / 2).attr('width', rw).attr('height', rh)
        .attr('rx', rx).attr('ry', rx).attr('fill', fill).attr('stroke', stroke)
        .attr('stroke-width', L === 0 ? 2.5 : L === 1 ? 1.8 : 1.2)
        .attr('stroke-dasharray', L === 3 ? '4 3' : null)
        .attr('filter', L <= 1 ? 'url(#shadow)' : L === 2 ? 'url(#shadow-sm)' : null);

      // Chevron hint for collapsed nodes
      var hint = el.select('.expand-hint');
      var collapsed = d.hasKids && !d.expanded;
      if (collapsed && L > 0) {
        var isLeft = d.x < 0;
        hint.attr('x', isLeft ? -rw / 2 - 4 : rw / 2 + 4).attr('y', 1)
          .attr('text-anchor', isLeft ? 'end' : 'start')
          .attr('font-size', 12).attr('font-weight', 600)
          .attr('fill', d.border || (isDark ? '#3D6B7D' : '#7FAFC0'))
          .attr('font-family', "'Inter',system-ui,sans-serif")
          .text(isLeft ? '\u2039' : '\u203a').attr('opacity', 0.7);
      } else {
        hint.attr('opacity', 0).text('');
      }

      // Cursor style for clickable nodes
      el.style('cursor', (d.hasKids || d.link) ? 'pointer' : 'default');
    });

    tr(nA).attr('transform', function (d) { return 'translate(' + d.x + ',' + d.y + ')'; }).attr('opacity', 1);

    // Z-order
    gLinks.each(function () { g.node().insertBefore(this, g.node().firstChild); });
    gBadges.raise();
    gNodes.raise();

    // Click handlers — expand/collapse or navigate to link
    nA.on('click', null);
    nA.filter(function (d) { return d.hasKids; }).on('click', function (event, d) {
      event.stopPropagation();
      var data = d.data;
      if (data.children) { data._c = data.children; data.children = null; }
      else { data.children = data._c; data._c = null; }
      render(true);
    });
    nA.filter(function (d) { return d.link && !d.hasKids; }).on('click', function (event, d) {
      event.stopPropagation();
      window.location.href = d.link;
    });
    // Also allow middle-click / ctrl-click to open link in new tab for lvl 2 nodes with links
    nA.filter(function (d) { return d.link; }).on('auxclick', function (event, d) {
      if (event.button === 1) window.open(d.link, '_blank');
    });
  }

  /* ═══ FIT ═══ */
  function fitToWidth() {
    var box = g.node().getBBox();
    if (!box.width || !box.height) return;
    var pad = 30;
    var cr = container.getBoundingClientRect();
    var cw = cr.width, ch = cr.height;
    var s = Math.min((cw - pad * 2) / box.width, (ch - pad * 2) / box.height);
    var cx = box.x + box.width / 2;
    var cy = box.y + box.height / 2;
    svg.call(zoomBeh.transform, d3.zoomIdentity.translate(cw / 2 - cx * s, ch / 2 - cy * s).scale(s));
  }

  render(false);
  requestAnimationFrame(fitToWidth);

  svg.on('dblclick.zoom', null);
  svg.on('dblclick', function () {
    var box = g.node().getBBox();
    if (!box.width || !box.height) return;
    var pad = 30;
    var cr = container.getBoundingClientRect();
    var cw = cr.width, ch = cr.height;
    var s = Math.min((cw - pad * 2) / box.width, (ch - pad * 2) / box.height);
    var cx = box.x + box.width / 2;
    var cy = box.y + box.height / 2;
    svg.transition().duration(400).ease(d3.easeCubicOut)
      .call(zoomBeh.transform, d3.zoomIdentity.translate(cw / 2 - cx * s, ch / 2 - cy * s).scale(s));
  });

  /* ── Hint ── */
  var hint = document.createElement('div');
  hint.className = 'diagram-hint';
  hint.textContent = 'Scroll to zoom \u00b7 Drag to pan \u00b7 Click to expand \u00b7 Double-click to reset';
  container.appendChild(hint);

  /* \u2550\u2550\u2550 FULLSCREEN \u2550\u2550\u2550 */
  function walk(node, fn) {
    fn(node);
    var kids = node.children || node._c;
    if (kids) kids.forEach(function (k) { walk(k, fn); });
  }
  function snapshot() {
    var snap = {};
    cats.forEach(function (c) { walk(c, function (n) { snap[n._id] = !!n.children; }); });
    return snap;
  }
  // snap === null \u2192 expand everything; otherwise restore each node to its saved state
  function setExpanded(snap) {
    cats.forEach(function (c) { walk(c, function (n) {
      var kids = n.children || n._c;
      if (!kids) return; // leaf
      if (snap === null || snap[n._id]) { n.children = kids; n._c = null; }
      else { n._c = kids; n.children = null; }
    }); });
  }

  var savedSnap = null;
  var fsBtn = document.createElement('button');
  fsBtn.type = 'button';
  fsBtn.className = 'mm-fs-btn';
  fsBtn.setAttribute('aria-label', 'Toggle fullscreen');
  fsBtn.textContent = '\u2922 Fullscreen';
  container.appendChild(fsBtn);

  function onResize() { requestAnimationFrame(fitToWidth); }

  function setFullscreen(on) {
    if (on === container.classList.contains('is-fullscreen')) return;
    if (on) {
      savedSnap = snapshot();
      setExpanded(null);
      window.addEventListener('resize', onResize);
    } else {
      if (savedSnap) setExpanded(savedSnap);
      window.removeEventListener('resize', onResize);
    }
    container.classList.toggle('is-fullscreen', on);
    document.body.style.overflow = on ? 'hidden' : '';
    fsBtn.textContent = on ? '\u2921 Exit' : '\u2922 Fullscreen';
    fsBtn.setAttribute('aria-label', on ? 'Exit fullscreen' : 'Toggle fullscreen');
    render(false);
    requestAnimationFrame(fitToWidth);
  }

  fsBtn.addEventListener('click', function (e) {
    e.stopPropagation();
    setFullscreen(!container.classList.contains('is-fullscreen'));
  });
  document.addEventListener('keydown', function (e) {
    if (e.key === 'Escape') setFullscreen(false);
  });

})();
</script>


---
source: ./topology/environments.md
---

# Dev & Prod Environments

Orion's architecture reads differently depending on which way you approach it.
This page contrasts the two:

1. **Design time (dev)** — how channels, workflows, and connectors are *authored*,
   often by an LLM driving the Admin API through the CLI/MCP.
2. **Run time (prod)** — how production traffic is *served*, API- and event-driven,
   against a reliable, highly-available backend.

The split matters: **design time is human/AI-in-the-loop and write-heavy**, while
**run time is machine-driven and read/execute-heavy**. The same single binary
(`orion-server`) serves both planes — only the configuration and topology change.

> The building blocks referenced throughout this page — **Channel**, **Workflow**,
> and **Connector** — are introduced in
> [Architecture Overview → Three Primitives](../architecture/overview.md#three-primitives).
> Every object follows a `draft → active → archived` lifecycle and is independently
> versioned.

---

## Design-Time Architecture (Dev Environment)

At design time, an **LLM** (or a human operator) authors and validates services
through Orion's **Admin API**. The loop is: *generate → validate → dry-run →
activate → hot-reload* — no redeploy, no container rebuild.

```orion-diagram
{
  "direction": "TB",
  "groups": [
    { "id": "authors", "label": "🧑‍💻 Authors" },
    { "id": "clients", "label": "Design-Time Clients" },
    { "id": "orion", "label": "🟦 Orion (single binary — dev)" },
    { "id": "store", "label": "Embedded Storage (dev default)" }
  ],
  "nodes": [
    { "id": "LLM", "label": "🤖 LLM / AI Agent", "sublabel": "(Claude, etc.)", "type": "observability", "group": "authors", "shape": "actor" },
    { "id": "DEV", "label": "👤 Developer / Operator", "type": "observability", "group": "authors", "shape": "actor" },

    { "id": "CLI", "label": "Orion CLI", "sublabel": "orion-server lint / dry-run\nvalidate-config / migrate", "type": "service", "group": "clients" },
    { "id": "MCP", "label": "MCP Server", "sublabel": "tool-calls from LLM", "type": "service", "group": "clients" },
    { "id": "HTTP", "label": "HTTP client / Swagger UI", "sublabel": "GET /docs", "type": "service", "group": "clients" },
    { "id": "GIT", "label": "GitOps / CI-CD", "sublabel": "export → commit → import", "type": "service", "group": "clients" },

    { "id": "ADMIN", "label": "Admin API  /api/v1/admin/*", "sublabel": "• /workflows  (CRUD, versions, rollout)\n• /channels   (CRUD, versions, status)\n• /connectors (CRUD, circuit-breakers)\n• /engine     (status, reload)\n• /audit-logs · /backups · /functions", "type": "accent", "group": "orion" },
    { "id": "VALIDATE", "label": "Validation & Dry-Run", "sublabel": "POST /workflows/validate\nPOST /workflows/{id}/test\n(schema check + step traces + field-pathed errors)", "type": "accent", "group": "orion" },
    { "id": "LIFECYCLE", "label": "Lifecycle & Versioning", "sublabel": "draft → active → archived\n(single-draft + active-immutable\nenforced by DB triggers)", "type": "accent", "group": "orion" },
    { "id": "RELOAD", "label": "Engine Hot-Reload", "sublabel": "Arc<RwLock<Arc<Engine>>>\nzero-downtime swap on activate", "type": "accent", "group": "orion" },

    { "id": "SQLITE", "label": "SQLite", "sublabel": "orion.db — single file\nzero external dependencies\nembedded migrations", "type": "datastore", "group": "store" }
  ],
  "edges": [
    { "from": "ADMIN", "to": "VALIDATE" },
    { "from": "ADMIN", "to": "LIFECYCLE" },
    { "from": "LIFECYCLE", "to": "RELOAD", "label": "activate / archive / delete" },
    { "from": "LLM", "to": "CLI" },
    { "from": "LLM", "to": "MCP" },
    { "from": "DEV", "to": "CLI" },
    { "from": "DEV", "to": "HTTP" },
    { "from": "DEV", "to": "GIT" },
    { "from": "CLI", "to": "ADMIN" },
    { "from": "MCP", "to": "ADMIN" },
    { "from": "HTTP", "to": "ADMIN" },
    { "from": "GIT", "to": "ADMIN" },
    { "from": "LIFECYCLE", "to": "SQLITE" },
    { "from": "VALIDATE", "to": "SQLITE", "label": "read schemas", "style": "dashed" },
    { "from": "RELOAD", "to": "SQLITE", "label": "rebuild from DB", "style": "dashed" }
  ]
}
```

### What makes the dev loop fast

- **AI-authored, governed by the platform.** *"AI generates workflows, Orion
  provides the governance."* Whatever the LLM produces, the platform enforces
  health checks, metrics, retries, and validation.
- **Dry-run before you ship.** `POST /workflows/{id}/test` runs the pipeline on a
  sample payload and returns per-task timing and field-pathed validation errors
  (e.g. `tasks[0].function.input.connector`).
- **Drafts are free.** Creating/updating drafts never touches the live engine —
  only `activate` / `archive` / `delete` / `rollout` trigger an audit-logged
  **hot-reload** (the inner `Arc<Engine>` is swapped while readers keep the old).
- **GitOps-friendly.** Workflows and channels are plain JSON: `export` → commit →
  PR review → `validate` → `import as drafts` → `test` → `activate`.
- **Zero-dependency local.** The dev box runs the **single binary against embedded
  SQLite** — `./orion-server` and you have the full Admin API + Swagger UI at
  `/docs`.

See also: [Maintainability](../features/maintainability.md) (Admin APIs, CI/CD,
dry-run) and the [Admin API reference](../api/admin.md).

---

## Run-Time Architecture (Prod Environment)

In production the picture inverts: traffic is **API- and event-driven**, served by
**stateless Orion instances** behind a load balancer, backed by a **reliable,
highly-available database** plus optional Redis / MongoDB / Kafka / OTel.

```orion-diagram
{
  "direction": "TB",
  "groups": [
    { "id": "ingress", "label": "Ingress" },
    { "id": "inst", "label": "🟦 Orion Fleet · stateless instance (orion-server)" },
    { "id": "data", "label": "Reliable Backends" },
    { "id": "obs", "label": "Observability" }
  ],
  "nodes": [
    { "id": "REST", "label": "REST / HTTP clients", "sublabel": "POST /api/v1/data/{channel}\nGET  /api/v1/data/orders/{id}", "type": "service", "group": "ingress" },
    { "id": "ASYNC", "label": "Async clients", "sublabel": "POST .../{channel}/async\n→ trace_id, poll later", "type": "service", "group": "ingress" },
    { "id": "PRODUCER", "label": "Kafka producers", "sublabel": "(event streams)", "type": "service", "group": "ingress" },

    { "id": "LB", "label": "⚖️ Load Balancer / API Gateway", "sublabel": "TLS, health probes /readyz /healthz", "type": "gateway", "shape": "hexagon" },

    { "id": "MW", "label": "Middleware stack", "sublabel": "panic-recovery · OTel · metrics · admin-auth\n· rate-limit · body-limit · compression\n· security-headers · request-id · CORS", "type": "accent", "group": "inst" },
    { "id": "REGISTRY", "label": "Channel Registry  (in-memory)", "sublabel": "route table · dedup · rate limiter\n· input validation (JSONLogic)\n· backpressure semaphore · response cache", "type": "accent", "group": "inst" },
    { "id": "ENGINE", "label": "Dataflow Engine", "sublabel": "Arc<RwLock<Arc<Engine>>>\nworkflow matcher (JSONLogic + rollout %)\n→ ordered task pipeline", "type": "accent", "group": "inst" },
    { "id": "FUNCS", "label": "Custom Functions", "sublabel": "http_call · channel_call\ndb_read/db_write · cache_read/write\nmongo_read · publish_kafka", "type": "accent", "group": "inst" },
    { "id": "KAFKAC", "label": "Kafka Consumer", "sublabel": "(topic → channel)", "type": "accent", "group": "inst" },
    { "id": "QUEUE", "label": "Async Trace Queue + DLQ", "sublabel": "(bounded, multi-worker, auto-retry)", "type": "accent", "group": "inst", "shape": "queue" },

    { "id": "PG", "label": "PostgreSQL / MySQL", "sublabel": "primary + replica (HA)\nworkflows · channels · connectors\ntraces · trace_dlq · audit_logs", "type": "datastore", "group": "data" },
    { "id": "REDIS", "label": "Redis", "sublabel": "cache · dedup · response cache", "type": "datastore", "group": "data" },
    { "id": "MONGO", "label": "MongoDB", "sublabel": "document reads", "type": "datastore", "group": "data" },
    { "id": "KAFKA", "label": "Kafka cluster", "sublabel": "ingest · egress · DLQ", "type": "datastore", "group": "data", "shape": "queue" },
    { "id": "EXT", "label": "External HTTP services", "sublabel": "(Stripe, internal APIs, webhooks)\n+ circuit breakers + retries", "type": "datastore", "group": "data", "shape": "cloud" },

    { "id": "OTEL", "label": "OpenTelemetry Collector", "sublabel": "OTLP / W3C trace context", "type": "observability", "group": "obs" },
    { "id": "PROM", "label": "Prometheus → Grafana", "sublabel": "scrape /metrics", "type": "observability", "group": "obs" }
  ],
  "edges": [
    { "from": "REST", "to": "LB" },
    { "from": "ASYNC", "to": "LB" },
    { "from": "PRODUCER", "to": "KAFKA" },
    { "from": "LB", "to": "MW" },
    { "from": "MW", "to": "REGISTRY" },
    { "from": "REGISTRY", "to": "ENGINE" },
    { "from": "ENGINE", "to": "FUNCS" },
    { "from": "KAFKAC", "to": "ENGINE" },
    { "from": "ENGINE", "to": "QUEUE" },
    { "from": "KAFKA", "to": "KAFKAC" },
    { "from": "FUNCS", "to": "PG", "label": "db_read / db_write" },
    { "from": "FUNCS", "to": "REDIS", "label": "cache_read / write" },
    { "from": "FUNCS", "to": "MONGO", "label": "mongo_read" },
    { "from": "FUNCS", "to": "KAFKA", "label": "publish_kafka" },
    { "from": "FUNCS", "to": "EXT", "label": "http_call" },
    { "from": "REGISTRY", "to": "REDIS", "label": "dedup / cache", "style": "dashed" },
    { "from": "ENGINE", "to": "PG", "label": "load config · persist traces" },
    { "from": "QUEUE", "to": "PG", "label": "store traces / DLQ" },
    { "from": "ENGINE", "to": "OTEL", "label": "spans", "style": "dashed" },
    { "from": "ENGINE", "to": "PROM", "label": "scrape", "style": "dashed" }
  ]
}
```

The data-plane request pipeline (route resolution → channel registry → engine →
task pipeline → response) is detailed in
[Architecture Overview → Request Processing Flow](../architecture/overview.md#request-processing-flow).

### What makes prod reliable

- **Stateless, horizontally scalable.** Instances hold no session state; scale out
  behind a load balancer. Use `include`/`exclude` channel filters to shard
  workloads across instances.
- **Reliable DB with HA.** Point `storage.url` at **PostgreSQL** (or MySQL) with a
  primary/replica setup. The *same* binary that ran on embedded SQLite in dev now
  speaks Postgres — backend is auto-detected from the URL scheme; migrations are
  embedded per backend.
- **Resilience built in.** Per-connector **circuit breakers** (auto-recovery),
  **retries** with exponential backoff (capped at 60s), per-channel / per-query /
  per-request **timeouts**, **backpressure** semaphores (503 load shedding), and a
  **Dead Letter Queue** with automatic retry for failed async traces.
- **Zero-downtime changes.** Hot-reload swaps the engine without dropping
  in-flight requests; canary **rollouts** split traffic by percentage with instant
  rollback.
- **In-process composition.** `channel_call` invokes another channel's workflow
  **in-process** — no network round-trip, with cycle/depth detection.
- **Full observability.** Structured JSON logs, Prometheus metrics at `/metrics`,
  OpenTelemetry spans over OTLP with W3C trace-context propagated even through
  Kafka headers, and `/health` · `/healthz` · `/readyz` for orchestrators.
- **Security.** Secret masking, SSRF protection (private-IP blocking), TLS,
  security headers, admin-API auth, parameterized SQL, JSONLogic input validation.

See also: [Resilience](../features/resilience.md),
[Scalability](../features/scalability.md),
[Availability](../features/availability.md), and
[Observability](../features/observability.md).

---

## Dev → Prod: One Binary, Two Topologies

The key design decision is that **nothing about the artifacts changes between
environments** — only the configuration and the backends they point at.

```orion-diagram
{
  "direction": "LR",
  "groups": [
    { "id": "dev", "label": "DEV — design time" },
    { "id": "prod", "label": "PROD — run time" }
  ],
  "nodes": [
    { "id": "D1", "label": "LLM / Operator", "type": "observability", "group": "dev", "shape": "actor" },
    { "id": "D2", "label": "Admin API", "sublabel": "(CLI · MCP · Swagger)", "type": "accent", "group": "dev" },
    { "id": "D3", "label": "SQLite", "sublabel": "single file", "type": "datastore", "group": "dev" },

    { "id": "ARTIFACTS", "label": "📦 Versioned JSON artifacts", "sublabel": "channels · workflows · connectors\n(export → git → import)", "type": "infra" },

    { "id": "P1", "label": "API · Kafka traffic", "type": "service", "group": "prod" },
    { "id": "P2", "label": "Orion fleet", "sublabel": "(N stateless instances)", "type": "service", "group": "prod" },
    { "id": "P3", "label": "PostgreSQL / MySQL", "sublabel": "+ Redis · Mongo · Kafka · OTel", "type": "datastore", "group": "prod" }
  ],
  "edges": [
    { "from": "D1", "to": "D2" },
    { "from": "D2", "to": "D3" },
    { "from": "P1", "to": "P2" },
    { "from": "P2", "to": "P3" },
    { "from": "D3", "to": "ARTIFACTS", "label": "export / import · GitOps · CI-CD" },
    { "from": "ARTIFACTS", "to": "P1", "label": "validate → import drafts → activate" }
  ]
}
```

| Dimension | Dev (design time) | Prod (run time) |
|-----------|-------------------|-----------------|
| **Primary driver** | LLM / operator via Admin API | API & Kafka traffic on the data plane |
| **Dominant traffic** | Write-heavy (author, validate, activate) | Read/execute-heavy (serve requests) |
| **Entry points** | CLI · MCP · Swagger UI · GitOps | REST · async · Kafka behind a load balancer |
| **Database** | Embedded **SQLite** (`orion.db`, zero deps) | **PostgreSQL / MySQL** with HA (primary + replica) |
| **Optional backends** | usually none | Redis, MongoDB, Kafka, OTel collector, Prometheus |
| **Scaling** | single local instance | N stateless instances, horizontally scaled |
| **Config** | minimal / defaults, `./orion-server` | TLS, admin-auth, rate limits, tracing, pools tuned |
| **Engine reload** | constant (every activate) | controlled, audit-logged, zero-downtime |

---

## Deployment Notes

The same `orion-server` binary ships everywhere — all three DB backends, Kafka
producer/consumer, OTLP export, TLS, and Swagger UI are compiled in (no feature
flags, no plugins). Configuration is a TOML file (`-c config.toml`) with every key
overridable via `ORION_SECTION__KEY` environment variables
(e.g. `ORION_STORAGE__URL=postgres://…`).

For packaging, containerization, multi-instance scaling, and the full set of config
keys, see [Deployability](../features/deployability.md) and the
[Configuration Reference](../configuration/reference.md).


---
source: ./features/observability.md
---

# Observability

Orion provides structured logging, Prometheus metrics, distributed tracing, and health monitoring out of the box. No sidecars, no agents. Everything runs inside the single binary.

All of it surfaces visually in [the Orion console](../getting-started/console.md) — live request rate, error rate, latency percentiles, outcomes by channel, and trace drill-downs:

<div class="themed-media">
  <img class="media-dark" src="../images/ui-operations-dark.png" alt="Operations dashboard — request rate, error rate, latency percentiles, outcomes by channel, top channels, and recent traces for a live Orion instance">
  <img class="media-light" src="../images/ui-operations-light.png" alt="Operations dashboard — request rate, error rate, latency percentiles, outcomes by channel, top channels, and recent traces for a live Orion instance">
</div>

## Structured Logging

Orion emits structured logs in JSON or pretty-printed format, configurable at runtime:

```toml
[logging]
level = "info"        # trace, debug, info, warn, error
format = "pretty"     # pretty or json
```

**JSON format** is recommended for production. It integrates directly with log aggregators like Loki, Datadog, or CloudWatch:

```bash
ORION_LOGGING__FORMAT=json
ORION_LOGGING__LEVEL=info
```

**Per-crate filtering** with `RUST_LOG` gives fine-grained control:

```bash
RUST_LOG=orion=debug,tower_http=warn,sqlx=warn
```

| Level | Usage |
|-------|-------|
| `error` | Failures that need attention |
| `warn` | Degraded behavior (circuit breakers, retries) |
| `info` | Request lifecycle, engine reloads, startup/shutdown |
| `debug` | Detailed processing, SQL queries, connector calls |
| `trace` | Fine-grained internal state |

Every request carries a UUID `x-request-id` header. Pass your own or let Orion generate one. The ID propagates through logs and responses for end-to-end correlation.

## Prometheus Metrics

Enable metrics and scrape at `GET /metrics` (Prometheus text format):

```toml
[metrics]
enabled = true
```

| Metric | Type | Labels | Description |
|--------|------|--------|-------------|
| `messages_total` | Counter | `channel`, `status` | Total messages processed |
| `message_duration_seconds` | Histogram | `channel` | Processing latency |
| `active_workflows` | Gauge | — | Workflows loaded in engine |
| `errors_total` | Counter | `type` | Errors encountered |
| `http_requests_total` | Counter | `method`, `path`, `status` | HTTP requests served |
| `http_request_duration_seconds` | Histogram | `method`, `path`, `status` | HTTP request latency |
| `db_query_duration_seconds` | Histogram | `operation` | Database query latency |
| `engine_reloads_total` | Counter | `status` | Engine reload events |
| `engine_reload_duration_seconds` | Histogram | — | Engine reload latency |
| `circuit_breaker_trips_total` | Counter | `connector`, `channel` | Circuit breaker trip events |
| `circuit_breaker_rejections_total` | Counter | `connector`, `channel` | Requests rejected by open breakers |
| `channel_executions_total` | Counter | `channel` | Channel invocations |
| `rate_limit_rejections_total` | Counter | `client` | Rate-limited requests |

## Distributed Tracing

Enable OpenTelemetry trace export with OTLP gRPC:

```toml
[tracing]
enabled = true
otlp_endpoint = "http://localhost:4317"
service_name = "orion"
sample_rate = 1.0    # 0.0 (none) to 1.0 (all)
```

- **W3C Trace Context** extraction and propagation: incoming `traceparent` headers are respected
- Per-request spans with channel, workflow, and task attributes
- OTLP gRPC export to Jaeger, Tempo, or any compatible collector
- Configurable sampling rate for production use
- Trace context injected into outbound `http_call` requests for full distributed traces

## Health Monitoring

Orion exposes three health endpoints for different operational needs.

**Component health:** `GET /health` returns component-level status with automatic degradation detection:

```json
{
  "status": "ok",
  "version": "0.2.0",
  "uptime_seconds": 3600,
  "workflows_loaded": 42,
  "components": {
    "database": "ok",
    "engine": "ok"
  }
}
```

The health check tests the database with `SELECT 1` and verifies engine availability with a configurable lock timeout. If either check fails, the endpoint returns `503 Service Unavailable` with `"status": "degraded"`.

**Kubernetes probes:**

| Endpoint | Purpose | Behavior |
|----------|---------|----------|
| `GET /healthz` | Liveness probe | Always returns 200. If the process is running, it's alive |
| `GET /readyz` | Readiness probe | Returns 200 only when DB is reachable, engine is loaded, and startup is complete; 503 otherwise |

```yaml
livenessProbe:
  httpGet:
    path: /healthz
    port: 8080
  initialDelaySeconds: 5
  periodSeconds: 10
readinessProbe:
  httpGet:
    path: /readyz
    port: 8080
  initialDelaySeconds: 5
  periodSeconds: 5
```

**Engine status:** `GET /api/v1/admin/engine/status` returns a detailed breakdown:

```json
{
  "version": "0.2.0",
  "uptime_seconds": 3600,
  "workflows_count": 42,
  "active_workflows": 38,
  "channels": ["orders", "events", "alerts"]
}
```


---
source: ./features/resilience.md
---

# Resilience

Orion protects your services from cascading failures, transient errors, and overload with circuit breakers, automatic retries, timeouts, and graceful degradation, all built into the runtime.

## Circuit Breakers

Every connector gets automatic circuit breaker protection. When failures exceed a threshold, the breaker opens and short-circuits requests to prevent cascading failures.

| State | Behavior |
|-------|----------|
| **Closed** | Normal operation; requests flow through |
| **Open** | Requests rejected immediately (503) after failure threshold exceeded |
| **Half-Open** | After cooldown, one probe request allowed to test recovery |

The circuit breaker uses a lock-free state machine with per-connector isolation. Configure globally:

```toml
[engine.circuit_breaker]
enabled = true
failure_threshold = 5          # Failures before tripping the breaker
recovery_timeout_secs = 30     # Cooldown before half-open probe
max_breakers = 10000           # Max breaker instances (LRU eviction)
```

Inspect and reset breakers via the admin API:

```bash
# List all circuit breaker states
curl -s http://localhost:8080/api/v1/admin/connectors/circuit-breakers

# Reset a specific breaker
curl -s -X POST http://localhost:8080/api/v1/admin/connectors/circuit-breakers/{key}
```

## Retry-Backoff

All HTTP connectors support automatic retries with exponential backoff, capped at 60 seconds:

```json
{
  "retry": {
    "max_retries": 5,
    "retry_delay_ms": 500
  }
}
```

Delay doubles on each retry: 500ms → 1s → 2s → 4s → ... → capped at 60s.

Retries are configured per-connector, so each external service can have its own retry policy. The mechanism automatically detects retryable errors (network failures, 5xx responses) and skips non-retryable ones (4xx client errors).

All connector types (HTTP, DB, Cache, MongoDB, Storage) support the same retry configuration.

## Timeouts

Timeouts are enforced at multiple levels to prevent runaway requests:

**Per-channel timeout:** set in the channel's `config_json` to limit workflow execution time:

```json
{
  "timeout_ms": 5000
}
```

If the workflow exceeds this limit, the request returns `504 Gateway Timeout`.

**Per-connector query timeout:** for database connectors:

```json
{
  "query_timeout_ms": 30000,
  "connect_timeout_ms": 5000
}
```

**Global HTTP timeout:** for the shared HTTP client used by `http_call`:

```toml
[engine]
global_http_timeout_secs = 30
```

**Engine lock timeouts:** prevent health checks and reloads from blocking indefinitely:

```toml
[engine]
health_check_timeout_secs = 2
reload_timeout_secs = 10
```

## Fault Tolerance

**Graceful shutdown:** Orion handles `SIGTERM` and `SIGINT` with a controlled shutdown sequence:

1. HTTP server stops accepting new connections
2. In-flight requests drain (configurable via `shutdown_drain_secs`, default 30s)
3. Kafka consumer (if enabled) is signaled to stop
4. Trace cleanup task is stopped
5. DLQ retry consumer is stopped
6. Async trace queue drains with timeout
7. OpenTelemetry spans are flushed (if enabled)
8. Process exits

**Dead letter queue:** failed async traces are stored in the `trace_dlq` database table with automatic retry:

```toml
[queue]
dlq_retry_enabled = true
dlq_max_retries = 5
dlq_poll_interval_secs = 30
```

For Kafka, failed messages can also be routed to a configurable DLQ topic:

```toml
[kafka.dlq]
enabled = true
topic = "orion-dlq"
```

**Fault-tolerant pipelines:** set `continue_on_error: true` on a workflow to keep the task pipeline running even if individual tasks fail. Errors are collected in the response rather than halting execution:

```json
{
  "status": "ok",
  "data": { "req": { "action": "test-call" } },
  "errors": [
    { "code": "TASK_ERROR", "task_id": "call", "message": "HTTP request failed..." }
  ]
}
```

**Panic recovery:** the outermost middleware layer (`CatchPanicLayer`) catches panics in any handler, returning a 500 response instead of crashing the process.


---
source: ./features/security.md
---

# Security

Orion enforces security at every layer: secrets are isolated in connectors, inputs are validated before processing, network requests are checked for SSRF, and admin endpoints are protected by authentication.

## Secret Management

Sensitive fields are automatically masked in all API responses. Fields named `token`, `password`, `key`, `secret`, `api_key`, and `connection_string` are returned as `"******"`. Secrets are stored but never exposed through the API.

```bash
# Create a connector with real credentials
curl -s -X POST http://localhost:8080/api/v1/admin/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "payments-api",
    "connector_type": "http",
    "config": {
      "type": "http",
      "url": "https://api.stripe.com/v1",
      "auth": { "type": "bearer", "token": "sk-live-secret-token" }
    }
  }'

# Read it back (secrets are masked)
curl -s http://localhost:8080/api/v1/admin/connectors/<id>
# auth.token → "******"
```

Workflows reference connectors by name (`"connector": "payments-api"`). They never see or embed actual credentials. This means AI-generated workflows can be safely created and shared without risk of credential exposure.

## Input Validation

Each channel can define JSONLogic validation rules evaluated against incoming requests before workflow execution:

```json
{
  "validation_logic": {
    "and": [
      { "!!": [{ "var": "data.order_id" }] },
      { ">": [{ "var": "data.amount" }, 0] }
    ]
  }
}
```

If validation fails, the request is rejected with `400 Bad Request` before any workflow logic runs.

Validation rules have access to:

- `data.*`: request body fields
- `headers.*`: HTTP headers
- `query.*`: query string parameters
- `path.*`: path parameters (for REST channels)

**Payload size limits** are enforced globally to prevent oversized requests:

```toml
[ingest]
max_payload_size = 1048576   # 1 MB
```

## Network Security

**SSRF protection:** HTTP connectors validate URLs to prevent Server-Side Request Forgery. By default, requests to private/internal IP addresses (RFC 1918, loopback, link-local) are blocked:

```json
{
  "name": "external-api",
  "connector_type": "http",
  "config": {
    "type": "http",
    "url": "https://api.example.com",
    "allow_private_urls": false
  }
}
```

Set `allow_private_urls: true` only when calling internal services.

**TLS/HTTPS:** enable TLS termination in the server:

```toml
[server.tls]
enabled = true
cert_path = "cert.pem"
key_path = "key.pem"
```

**Security headers:** set on all responses:

| Header | Value |
|--------|-------|
| `X-Content-Type-Options` | `nosniff` |
| `X-Frame-Options` | `DENY` |
| `Content-Security-Policy` | `default-src 'none'; frame-ancestors 'none'` |
| `Referrer-Policy` | `strict-origin-when-cross-origin` |
| `Permissions-Policy` | `camera=(), microphone=(), geolocation=()` |
| `Strict-Transport-Security` | Set when TLS is enabled |

## Access Control

**Admin API authentication:** protect admin endpoints with bearer token or API key:

```toml
[admin_auth]
enabled = true
api_keys = ["your-secret-key"]   # Any number of accepted keys; any match authorises a request
# header = "Authorization"       # Bearer format (default)
# header = "X-API-Key"           # Raw key format
```

When `header` is `"Authorization"`, the key is expected as `Bearer <key>`. For any other header name, the raw key value is matched directly.

```bash
# Bearer token
curl -H "Authorization: Bearer your-secret-key" \
  http://localhost:8080/api/v1/admin/workflows

# API key via custom header
curl -H "X-API-Key: your-secret-key" \
  http://localhost:8080/api/v1/admin/workflows
```

**Per-channel CORS:** configure allowed origins per channel in `config_json`:

```json
{
  "cors": {
    "allowed_origins": ["https://app.example.com", "https://admin.example.com"]
  }
}
```

Global CORS defaults are configured in the server config:

```toml
[cors]
allowed_origins = ["*"]    # Global default
```

## Data Safety

**Parameterized SQL queries:** the `db_read` and `db_write` functions use parameterized queries to prevent SQL injection:

```json
{
  "function": {
    "name": "db_read",
    "input": {
      "connector": "orders-db",
      "query": "SELECT * FROM orders WHERE customer_id = $1",
      "params": [{ "var": "data.customer_id" }],
      "output": "data.orders"
    }
  }
}
```

Values are always passed as parameters, never interpolated into SQL strings.

**The portable dialect is injection-safe by construction:** `data_query` and
`data_write` never accept SQL or query text at all — message data enters only
through the `params` map, and every resolved value becomes a bound parameter
(SQL), a document value (MongoDB), or a script parameter (Elasticsearch
painless scripts, where field names *and* values travel as params). Identifiers
come only from the envelope and schema and are quoted per dialect. An optional
schema allowlist (`"unmapped": "reject"`) restricts which entities and columns
workflows can touch, with per-column `queryable`/`writable` flags. See the
[Portable Data Dialect](../reference/data-dialect.md) reference.

**Write-safety guards:** an unfiltered `data_write` update/delete is rejected
unless the call carries `"all": true` **and** the server enables
`write.allow_unfiltered` — a double opt-in against accidental table truncation.
Bulk inserts over `write.max_rows` are rejected, never silently truncated.

**Per-connector operation gates:** a `db` or `es` connector's config can
disable operation types outright — `operations: { "read", "insert", "update",
"delete", "upsert", "raw_write" }`, all defaulting to allowed. A gated call
fails with a validation error naming the op and connector, so a connector can
be made read-only (or insert-only, delete-proof, …) in configuration,
regardless of what any workflow asks for:

```json
{
  "name": "orders-db-readonly",
  "connector_type": "db",
  "config": {
    "type": "db",
    "connection_string": "postgres://…",
    "operations": { "insert": false, "update": false, "delete": false, "upsert": false, "raw_write": false }
  }
}
```

**URL validation:** connector URLs are validated at creation time. Combined with SSRF protection, this prevents workflows from making requests to unexpected destinations.

**Injection protection:** JSONLogic expressions are evaluated in a sandboxed environment. User-supplied data cannot escape the data context or execute arbitrary code.


---
source: ./features/scalability.md
---

# Scalability

Orion handles high-throughput workloads with token-bucket rate limiting, semaphore-based backpressure, async processing queues, and stateless horizontal scaling, all configurable per channel.

## Rate Limiting

Rate limiting operates at two levels: **platform-wide** (all requests) and **per-channel** (individual service endpoints).

**Platform-level:** enable in config:

```toml
[rate_limit]
enabled = true
default_rps = 100
default_burst = 50

[rate_limit.endpoints]
admin_rps = 50
data_rps = 200
```

**Per-channel:** configure in the channel's `config_json`:

```json
{
  "rate_limit": {
    "requests_per_second": 100,
    "burst": 50
  }
}
```

Rate limiting uses the **token bucket algorithm**: tokens replenish at the configured rate, and burst allows short spikes above the steady-state limit. When the bucket is empty, requests receive `429 Too Many Requests`.

**Per-client keying:** use JSONLogic to compute rate limit keys from request data, enabling per-user or per-tenant limits:

```json
{
  "rate_limit": {
    "requests_per_second": 10,
    "burst": 5,
    "key_logic": { "var": "headers.x-api-key" }
  }
}
```

Rate limiter state is per-instance (in-memory). In multi-instance deployments, divide the configured RPS by the number of instances to approximate global limits, or use sticky sessions at the load balancer.

## Backpressure

Semaphore-based concurrency limits prevent any single channel from overwhelming the system:

```json
{
  "backpressure": {
    "max_concurrent": 200
  }
}
```

When all semaphore permits are taken, additional requests receive `503 Service Unavailable` immediately. This is load shedding. The system sheds excess load rather than queuing unboundedly, which protects latency for requests that are admitted.

Each channel has its own independent backpressure semaphore, so a spike in one channel doesn't affect others.

## Async Processing

For workloads that don't need immediate responses, Orion supports async processing via a bounded trace queue:

```bash
# Submit for async processing (returns immediately with a trace ID)
curl -s -X POST http://localhost:8080/api/v1/data/orders/async \
  -H "Content-Type: application/json" \
  -d '{ "data": { "order_id": "ORD-123" } }'

# Poll for the result
curl -s http://localhost:8080/api/v1/data/traces/{trace-id}
```

The queue is backed by `tokio::sync::mpsc` channels with configurable concurrency:

```toml
[queue]
workers = 4                       # Concurrent trace workers
buffer_size = 1000                # Channel buffer for pending traces
processing_timeout_ms = 60000     # Per-trace processing timeout
max_result_size_bytes = 1048576   # Max size of trace result (1 MB)
max_queue_memory_bytes = 104857600  # Max memory for queued traces (100 MB)
```

Failed traces go to the **dead letter queue** with automatic retry:

```toml
[queue]
dlq_retry_enabled = true
dlq_max_retries = 5
dlq_poll_interval_secs = 30
```

Completed traces are cleaned up automatically based on retention policy:

```toml
[queue]
trace_retention_hours = 72
trace_cleanup_interval_secs = 3600
```

## Horizontal Scaling

Orion is designed for **single-instance simplicity** with **multi-instance capability**. Each instance is stateless; all persistent data lives in the shared database.

**What works across instances:**

| Component | How It Works |
|-----------|-------------|
| Database | All instances share the same database (PostgreSQL or MySQL recommended) |
| Kafka consumers | Consumer groups handle partition assignment automatically |
| Traces | Stored in the shared database; queries return consistent results |
| Workflows & Channels | Definitions live in the database; all instances load the same set |
| Audit logs | Stored in the shared database regardless of which instance handles the request |

### Per-Instance State

The following components use in-memory state that is local to each instance:

| Component | Impact | Workaround |
|-----------|--------|------------|
| **Rate Limiting** | 3 instances at 100 RPS = 300 RPS effective global limit | Sticky sessions; divide configured RPS by instance count |
| **Request Deduplication** | Same idempotency key on two instances → processed twice | Sticky sessions, or Redis-backed dedup store |
| **Response Caching** | Lower cache hit rates (each instance has a cold cache) | Sticky sessions, or Redis-backed cache connector |
| **Circuit Breakers** | One instance may trip while others keep sending | Acceptable; monitor `/health` on each instance |
| **Engine State** | `POST /admin/engine/reload` only reloads the receiving instance | Script reload to hit all instances (see below) |

**Reload all instances:**

```bash
for host in $INSTANCE_HOSTS; do
  curl -X POST "http://$host:8080/api/v1/admin/engine/reload" \
    -H "Authorization: Bearer $API_KEY"
done
```

Alternatively, use a rolling restart strategy with your orchestrator (e.g., Kubernetes rolling deployment).

### Topology Control

Use channel include/exclude filters to run different Orion instances for different channel groups:

```toml
# Instance A: order processing
[channels]
include = ["orders.*", "payments.*"]

# Instance B: analytics and reporting
[channels]
include = ["analytics.*", "reports.*"]
```

This enables microservice-style deployment where each instance handles a subset of channels, all sharing the same database.

### Database Backend Recommendations

| Backend | Single Instance | Multiple Instances | Notes |
|---------|:-:|:-:|-------|
| **SQLite** | Recommended | Not recommended | WAL mode supports concurrent reads but only one writer. File-based, cannot be shared across hosts. |
| **PostgreSQL** | Supported | Recommended | Full multi-connection support. Use connection pooling (PgBouncer) for many instances. |
| **MySQL** | Supported | Supported | Ensure `READ-COMMITTED` isolation for best concurrency. |

For multi-instance deployments, use PostgreSQL with connection pooling (PgBouncer). Script engine reloads to broadcast to all instances after workflow or channel changes.


---
source: ./features/deployability.md
---

# Deployability

Orion ships as a single binary with embedded migrations, sensible defaults, and multiple installation methods. No runtime dependencies. You're up and running immediately.

## Packaging

Orion compiles into a **single binary** (`orion-server`) with everything built in:

- All three database backends (SQLite, PostgreSQL, MySQL) with embedded migrations
- Kafka producer and consumer
- OpenTelemetry trace export
- TLS termination
- Swagger UI at `/docs`

No feature flags, no plugins, no shared libraries to manage. The binary auto-detects the database backend from the URL scheme at startup.

| Capability | Configuration | Default |
|-----------|--------------|---------|
| Database backend | `storage.url` scheme | SQLite |
| Kafka | `kafka.enabled` | Disabled |
| OpenTelemetry | `tracing.enabled` | Disabled |
| TLS/HTTPS | `server.tls.enabled` | Disabled |
| Swagger UI | Always at `/docs` | Enabled |
| Metrics | `metrics.enabled` | Disabled |

With the default SQLite backend, there are zero external dependencies.

## Containerization

Orion uses a multi-stage Docker build for minimal image size:

```bash
docker build -t orion .
```

The Dockerfile uses `rust:1.93-slim` for building and `debian:trixie-slim` for the runtime image. Key characteristics:

- **Non-root execution:** the container runs as a non-root user
- **Built-in health probes:** `/healthz` (liveness) and `/readyz` (readiness) are available without additional configuration

**Docker with SQLite persistence:** SQLite stores data in a local file. Without a persistent volume, data is lost when the container restarts:

```bash
docker run -p 8080:8080 \
  -v orion-data:/app/data \
  -e ORION_STORAGE__URL=sqlite:/app/data/orion.db \
  orion
```

With Docker Compose:

```yaml
services:
  orion:
    image: orion
    ports:
      - "8080:8080"
    environment:
      ORION_STORAGE__URL: sqlite:/app/data/orion.db
    volumes:
      - orion-data:/app/data

volumes:
  orion-data:
```

For production, PostgreSQL or MySQL is recommended. Point `ORION_STORAGE__URL` to your database server.

## Configuration

All settings have sensible defaults. You can run Orion with no config file at all. `orion-server` just works.

**TOML config file:** pass with `-c`:

```bash
orion-server -c config.toml
```

**Environment variable overrides:** use double-underscore nesting to override any setting:

```bash
ORION_SERVER__PORT=9090
ORION_STORAGE__URL="postgres://user:pass@localhost/orion"
ORION_KAFKA__ENABLED=true
ORION_LOGGING__FORMAT=json
ORION_ADMIN_AUTH__ENABLED=true
ORION_ADMIN_AUTH__API_KEY="your-secret-key"
```

Environment variables take precedence over the config file, making it easy to customize per environment without changing files.

**Runtime configuration:** channels carry their own runtime config (rate limits, timeouts, CORS, validation) via `config_json`, changeable through the admin API without restarts.

## Distribution

Orion is available through multiple installation methods:

**Homebrew** (macOS and Linux):

```bash
brew install GoPlasmatic/tap/orion-server
```

**Shell installer** (Linux/macOS):

```bash
curl --proto '=https' --tlsv1.2 -LsSf https://github.com/GoPlasmatic/Orion/releases/latest/download/orion-server-installer.sh | sh
```

**PowerShell installer** (Windows):

```powershell
powershell -ExecutionPolicy ByPass -c "irm https://github.com/GoPlasmatic/Orion/releases/latest/download/orion-server-installer.ps1 | iex"
```

**Docker:**

```bash
docker run -p 8080:8080 ghcr.io/goplasmatic/orion:latest
```

**From source:**

```bash
cargo install --git https://github.com/GoPlasmatic/Orion
```

Multi-platform binaries are published for Linux (x86_64, aarch64), macOS (x86_64, aarch64), and Windows (x86_64).


---
source: ./features/extensibility.md
---

# Extensibility

Orion integrates with external systems through connectors, exposes custom logic via async function handlers, and supports multiple channel protocols for different ingestion patterns.

## Connectors

Connectors are named external service configurations. Secrets stay in connectors, out of your workflows.

### Authentication

Three auth schemes are supported:

| Auth Type | Fields | Example |
|-----------|--------|---------|
| `bearer` | `token` | `{ "type": "bearer", "token": "sk-..." }` |
| `basic` | `username`, `password` | `{ "type": "basic", "username": "user", "password": "pass" }` |
| `apikey` | `header`, `key` | `{ "type": "apikey", "header": "X-API-Key", "key": "abc123" }` |

### Header Precedence

When `http_call` builds a request, headers are applied in this order (later layers override earlier ones):

| Priority | Source | Example |
|----------|--------|---------|
| 1 (lowest) | Connector default headers | `"headers": {"x-source": "orion"}` in connector config |
| 2 | Connector auth | Bearer token, Basic auth, API key |
| 3 | Default `content-type` | `application/json` (only when a body is present) |
| 4 (highest) | Task-level headers | `"headers": {"content-type": "text/xml"}` in the task input |

Task-level headers always win. This means a workflow developer can override `content-type`, `authorization`, or any other header set by the connector.

### Secret Masking

Sensitive fields (`token`, `password`, `key`, `secret`, `api_key`, `connection_string`) are automatically masked as `"******"` in all API responses. Secrets are stored but never exposed through the API. Workflows reference connectors by name; they never see or embed actual credentials.

### Pulling Secrets from the Environment

Any string field inside a connector's `config` may use an `env://VAR_NAME` reference instead of a literal value. The resolver runs once when the connector is loaded:

```json
{
  "name": "payments-api",
  "connector_type": "http",
  "config": {
    "type": "http",
    "url": "https://api.stripe.com/v1",
    "auth": { "type": "bearer", "token": "env://STRIPE_API_KEY" }
  }
}
```

If `STRIPE_API_KEY` is not set in the process environment, startup (or the create/update call) fails with a structured error pointing at the field — production credentials never have to be POSTed into the admin API or stored in the database. The same `env://` scheme works on every string field in every connector type.

### HTTP Connector

REST API calls, webhooks, and external service integration:

```json
{
  "name": "payments-api",
  "connector_type": "http",
  "config": {
    "type": "http",
    "url": "https://api.stripe.com/v1",
    "auth": { "type": "bearer", "token": "sk-..." },
    "headers": { "x-source": "orion" },
    "retry": { "max_retries": 3, "retry_delay_ms": 1000 },
    "max_response_size": 10485760,
    "allow_private_urls": false
  }
}
```

| Field | Default | Description |
|-------|---------|-------------|
| `url` | required | Base URL for all requests |
| `method` | `""` | Default HTTP method |
| `headers` | `{}` | Default headers applied to every request |
| `auth` | `null` | Authentication config (bearer, basic, or apikey) |
| `retry` | 3 retries, 1000ms | Retry with exponential backoff |
| `max_response_size` | 10 MB | Maximum response body size to prevent OOM |
| `allow_private_urls` | `false` | Allow requests to private/internal IPs (SSRF protection) |

### Kafka Connector

Produce to Kafka topics:

```json
{
  "name": "event-bus",
  "connector_type": "kafka",
  "config": {
    "type": "kafka",
    "brokers": ["kafka1:9092", "kafka2:9092"],
    "topic": "events",
    "group_id": "orion-producer"
  }
}
```

Use the `publish_kafka` task function with optional JSONLogic for dynamic keys and values:

```json
{
  "function": {
    "name": "publish_kafka",
    "input": {
      "connector": "event-bus",
      "topic": "processed-orders",
      "key_logic": { "var": "data.order_id" }
    }
  }
}
```

| Field | Required | Description |
|-------|----------|-------------|
| `connector` | Yes | Kafka connector name |
| `topic` | Yes | Target topic |
| `key_logic` | No | JSONLogic expression for partition key |
| `value_logic` | No | JSONLogic expression for message value (default: `message.data`) |

**Kafka consumer configuration:** map topics to channels in your config file:

```toml
[kafka]
enabled = true
brokers = ["localhost:9092"]
group_id = "orion"

[[kafka.topics]]
topic = "incoming-orders"
channel = "orders"
```

Async channels with `protocol: "kafka"` can also register topics via the API (DB-driven). Config-file and DB-driven topics are merged; duplicates are deduplicated with config-file entries taking precedence. The consumer restarts automatically on engine reload when the topic set changes.

**Metadata injection:** Kafka metadata is automatically injected into every message:

| Field | Description |
|-------|-------------|
| `kafka_topic` | Source topic name |
| `kafka_key` | Message key (if present) |
| `kafka_partition` | Partition number |
| `kafka_offset` | Offset within partition |

Access these in workflows via `{ "var": "metadata.kafka_topic" }`.

**Dead letter queue:** failed messages are routed to a configurable DLQ topic:

```toml
[kafka.dlq]
enabled = true
topic = "orion-dlq"
```

**Consumer settings:**

| Config | Default | Description |
|--------|---------|-------------|
| `kafka.processing_timeout_ms` | `60000` | Per-message processing timeout |
| `kafka.max_inflight` | `100` | Max in-flight messages |
| `kafka.lag_poll_interval_secs` | `30` | Consumer lag polling interval |

### Database Connector (SQL)

Parameterized SQL queries against PostgreSQL, MySQL, or SQLite:

```json
{
  "name": "orders-db",
  "connector_type": "db",
  "config": {
    "type": "db",
    "connection_string": "postgres://user:pass@db-host:5432/orders",
    "driver": "postgres",
    "max_connections": 10,
    "connect_timeout_ms": 5000,
    "query_timeout_ms": 30000
  }
}
```

| Field | Default | Description |
|-------|---------|-------------|
| `connection_string` | required | Database URL (auto-masked in API responses) |
| `driver` | `"postgres"` | Driver type: `postgres`, `mysql`, or `sqlite` |
| `max_connections` | `null` | Connection pool max size |
| `connect_timeout_ms` | `null` | Connection establishment timeout |
| `query_timeout_ms` | `null` | Individual query timeout |
| `retry` | 3 retries, 1000ms | Retry with exponential backoff |
| `operations` | all allowed | [Operation gates](#operation-gates) — en/disable read / insert / update / delete / upsert / raw_write |

Two ways to talk to it:

- **Portable dialect** — `data_query` / `data_write` express backend-neutral
  queries and mutations that run unchanged against SQL, MongoDB, or
  Elasticsearch connectors. See the
  [Portable Data Dialect](../reference/data-dialect.md) reference.
- **Raw SQL** — `db_read` for SELECT (returns rows as JSON array) and
  `db_write` for INSERT/UPDATE/DELETE (returns affected count):

```json
{
  "function": {
    "name": "db_read",
    "input": {
      "connector": "orders-db",
      "query": "SELECT * FROM orders WHERE customer_id = $1",
      "params": [{ "var": "data.customer_id" }],
      "output": "data.orders"
    }
  }
}
```

### Operation Gates

`db` and `es` connectors carry per-operation gates — everything defaults to
allowed, and disabling an operation rejects the call with a validation error
naming the op and connector, regardless of what a workflow asks for:

```json
{
  "name": "orders-db-readonly",
  "connector_type": "db",
  "config": {
    "type": "db",
    "connection_string": "postgres://user:pass@db-host:5432/orders",
    "operations": { "insert": false, "update": false, "delete": false, "upsert": false, "raw_write": false }
  }
}
```

| Gate | Default | Blocks |
|------|---------|--------|
| `read` | `true` | `data_query`, `db_read`, `mongo_read` |
| `insert` / `update` / `delete` / `upsert` | `true` | The matching `data_write` operation |
| `raw_write` | `true` | The raw-SQL `db_write` escape hatch (raw SQL cannot be classified per-op) |

To make a connector fully delete-proof, disable both `delete` and `raw_write`.

### Cache Connector

In-memory or Redis cache for lookups, session state, and temporary storage:

```json
{
  "name": "session-cache",
  "connector_type": "cache",
  "config": {
    "type": "cache",
    "backend": "redis",
    "url": "redis://localhost:6379",
    "default_ttl_secs": 300,
    "max_connections": 10
  }
}
```

| Field | Default | Description |
|-------|---------|-------------|
| `backend` | required | `"redis"` or `"memory"` |
| `url` | required (redis) | Redis connection URL |
| `default_ttl_secs` | `null` | Default TTL for cache entries |
| `max_connections` | `null` | Connection pool max size |
| `retry` | 3 retries, 1000ms | Retry with exponential backoff |

Use `cache_read` and `cache_write` in workflows:

```json
{
  "function": {
    "name": "cache_write",
    "input": {
      "connector": "session-cache",
      "key": "session:user123",
      "value": { "var": "data.session" },
      "ttl_secs": 3600
    }
  }
}
```

### Storage Connector (S3/GCS)

S3, GCS, or local filesystem for file operations:

```json
{
  "name": "uploads",
  "connector_type": "storage",
  "config": {
    "type": "storage",
    "provider": "s3",
    "bucket": "my-uploads",
    "region": "us-east-1",
    "base_path": "/data"
  }
}
```

| Field | Default | Description |
|-------|---------|-------------|
| `provider` | required | Storage provider: `s3`, `gcs`, `local` |
| `bucket` | `null` | Bucket or container name |
| `region` | `null` | Cloud region |
| `base_path` | `null` | Base path prefix for all operations |
| `retry` | 3 retries, 1000ms | Retry with exponential backoff |

### MongoDB Connector (NoSQL)

MongoDB uses a `db` connector with a `mongodb://` connection string:

```json
{
  "name": "analytics-db",
  "connector_type": "db",
  "config": {
    "type": "db",
    "connection_string": "mongodb://localhost:27017",
    "driver": "mongodb"
  }
}
```

The portable `data_query` / `data_write` functions run against it unchanged
(pass a `database` field in the task input). For raw `find()` filters, use
`mongo_read`:

```json
{
  "function": {
    "name": "mongo_read",
    "input": {
      "connector": "analytics-db",
      "database": "analytics",
      "collection": "events",
      "filter": { "user_id": { "var": "data.user_id" } },
      "output": "data.events"
    }
  }
}
```

### Elasticsearch Connector

A REST endpoint driven by the portable dialect: `data_query` renders an ES
Query DSL `_search` body; `data_write` renders `_bulk` /
`_update_by_query` / `_delete_by_query` / `_update` calls. Executed via the
shared HTTP client — no dedicated ES driver:

```json
{
  "name": "search-cluster",
  "connector_type": "es",
  "config": {
    "type": "es",
    "url": "http://localhost:9200",
    "auth": { "type": "apikey", "header": "Authorization", "key": "ApiKey ..." },
    "request_timeout_ms": 10000
  }
}
```

| Field | Default | Description |
|-------|---------|-------------|
| `url` | required | Base URL of the cluster, e.g. `http://localhost:9200` |
| `auth` | `null` | Authentication config (bearer, basic, or apikey) |
| `request_timeout_ms` | `null` | Per-request timeout |
| `retry` | 3 retries, 1000ms | Retry with exponential backoff |
| `allow_private_urls` | `false` | Allow private/internal IPs (SSRF protection) |
| `operations` | all allowed | [Operation gates](#operation-gates) |

ES-specific dialect semantics (the `_id` schema rename, forced refresh for
read-your-writes, capability limits) are documented in the
[Portable Data Dialect](../reference/data-dialect.md#elasticsearch-notes)
reference.

## Custom Functions

Orion provides 10 async function handlers that can be used in workflow tasks:

| Function | Description |
|----------|-------------|
| `http_call` | Call external APIs via HTTP connectors |
| `channel_call` | Invoke another channel's workflow in-process (no HTTP round-trip) |
| `data_query` | Portable, backend-neutral query against SQL / MongoDB / Elasticsearch |
| `data_write` | Portable, backend-neutral insert/update/delete/upsert against SQL / MongoDB / Elasticsearch |
| `db_read` | Execute raw SELECT queries against SQL connectors |
| `db_write` | Execute raw INSERT/UPDATE/DELETE against SQL connectors |
| `cache_read` | Read from memory or Redis cache connectors |
| `cache_write` | Write to memory or Redis cache connectors |
| `mongo_read` | Query MongoDB collections with raw find() filters |
| `publish_kafka` | Produce messages to Kafka topics |

In addition to the Orion-specific handlers, the dataflow-rs 3.0 engine contributes a **built-in function library** for parsing, transformation, and output:

| Function | Description |
|----------|-------------|
| `parse_json` | Parse raw payload into structured data |
| `parse_xml` | Parse XML payload into structured data |
| `filter` | Filter arrays using JSONLogic conditions |
| `map` | Transform data with field mappings |
| `validation` | Validate data against JSONLogic rules |
| `publish_json` | Serialize the data context into a JSON response body |
| `publish_xml` | Serialize the data context into an XML response body |
| `log` | Log data at a specified level |

The Orion-specific handlers have machine-readable input schemas surfaced at `GET /api/v1/admin/functions`; workflow create/update calls validate `function.input` against those schemas with field-pathed errors before the workflow can be activated.

**JSONLogic expressions** power all conditions and dynamic values. Use `{ "var": "data.field" }` to reference data, `{ "cat": [...] }` for string concatenation, arithmetic operators, and more. Dynamic paths (`path_logic`) and bodies (`body_logic`) let you compute URLs and request payloads from message data. Under the hood, datalogic-rs 5 compiles each JSONLogic expression once at engine-construction time and evaluates it via arena-mode dispatch, so per-request cost is constant regardless of expression complexity.

## Channel Protocols

Channels support three protocol modes:

### REST (Sync)

REST channels define route patterns for RESTful API routing with method and path matching:

```json
{
  "name": "order-detail",
  "channel_type": "sync",
  "protocol": "rest",
  "methods": ["GET", "POST"],
  "route_pattern": "/orders/{order_id}/items/{item_id}",
  "workflow_id": "order-detail-workflow"
}
```

Path parameters are extracted and injected into the message metadata. Routes are matched by priority (descending) then specificity (segment count).

### Simple HTTP (Sync)

Simple HTTP channels are matched by channel name. Requests to `/api/v1/data/{channel-name}` are routed directly:

```json
{
  "name": "orders",
  "channel_type": "sync",
  "protocol": "http",
  "workflow_id": "order-processing"
}
```

### Kafka (Async)

Kafka channels consume from topics and process messages asynchronously:

```json
{
  "name": "kafka-orders",
  "channel_type": "async",
  "protocol": "kafka",
  "topic": "incoming-orders",
  "consumer_group": "orion-orders",
  "workflow_id": "order-processing"
}
```

DB-driven Kafka channels are automatically registered as consumers at startup and on engine reload. Add Kafka ingestion via the API without restarting Orion.


---
source: ./features/availability.md
---

# Availability

Orion supports zero-downtime engine reloads, percentage-based canary rollouts, full version lifecycle management, and response caching, enabling continuous delivery without service interruptions.

## Hot-Reload

The engine is held in memory as `Arc<RwLock<Arc<Engine>>>`. A reload swaps the inner `Arc<Engine>` while existing readers continue using the old one. Zero dropped requests.

**Trigger a reload:**

```bash
curl -s -X POST http://localhost:8080/api/v1/admin/engine/reload
```

A reload performs three operations atomically:

1. **Engine swap:** rebuilds the engine from all active workflows and channels in the database
2. **Channel registry rebuild:** reconstructs the route table, validation logic, rate limiters, backpressure semaphores, dedup stores, and response caches
3. **Kafka consumer restart:** if the topic set changed, the Kafka consumer is stopped and restarted with the new topics

Reloads are triggered automatically on status changes (activate/archive) and deletes. Draft creates and updates do not trigger reload.

In multi-instance deployments, reload only affects the instance that receives the request. Script the reload to broadcast to all instances:

```bash
for host in $INSTANCE_HOSTS; do
  curl -X POST "http://$host:8080/api/v1/admin/engine/reload" \
    -H "Authorization: Bearer $API_KEY"
done
```

## Canary Rollouts

Control traffic exposure for active workflows with rollout percentages:

```bash
# Activate a workflow at 10% rollout
curl -s -X PATCH http://localhost:8080/api/v1/admin/workflows/<id>/status \
  -H "Content-Type: application/json" \
  -d '{"status": "active", "rollout_percentage": 10}'

# Increase rollout to 50%
curl -s -X PATCH http://localhost:8080/api/v1/admin/workflows/<id>/rollout \
  -H "Content-Type: application/json" \
  -d '{"rollout_percentage": 50}'

# Full rollout
curl -s -X PATCH http://localhost:8080/api/v1/admin/workflows/<id>/rollout \
  -H "Content-Type: application/json" \
  -d '{"rollout_percentage": 100}'
```

The rollout percentage determines the probability that incoming requests are matched to this workflow. This enables:

- **Gradual migration:** slowly ramp traffic from 0% to 100%
- **A/B testing:** run two workflow versions at different percentages
- **Instant rollback:** set rollout to 0% or archive the workflow

## Versioning

Both workflows and channels follow a **draft → active → archived** lifecycle with automatic version tracking:

```bash
# Create (starts as draft, version 1)
curl -s -X POST http://localhost:8080/api/v1/admin/workflows \
  -H "Content-Type: application/json" \
  -d '{ "name": "Order Processor", ... }'

# Update (only drafts can be updated)
curl -s -X PUT http://localhost:8080/api/v1/admin/workflows/<id> \
  -H "Content-Type: application/json" -d '{ ... }'

# Activate (loads into engine)
curl -s -X PATCH http://localhost:8080/api/v1/admin/workflows/<id>/status \
  -H "Content-Type: application/json" -d '{"status": "active"}'

# Create new version (new draft from active)
curl -s -X POST http://localhost:8080/api/v1/admin/workflows/<id>/versions

# Archive (removes from engine)
curl -s -X PATCH http://localhost:8080/api/v1/admin/workflows/<id>/status \
  -H "Content-Type: application/json" -d '{"status": "archived"}'
```

All versions are stored with incrementing version numbers. List the version history:

```bash
curl -s http://localhost:8080/api/v1/admin/workflows/<id>/versions
```

**Import and export:** bulk operations for GitOps and migration:

```bash
# Export workflows (as JSON)
curl -s http://localhost:8080/api/v1/admin/workflows/export?status=active

# Import workflows (created as drafts)
curl -s -X POST http://localhost:8080/api/v1/admin/workflows/import \
  -H "Content-Type: application/json" -d @workflows.json
```

## Performance

**Response caching:** cache responses for identical requests to reduce redundant workflow execution:

```json
{
  "cache": {
    "enabled": true,
    "ttl_secs": 60,
    "cache_key_fields": ["data.user_id", "data.action"]
  }
}
```

Cache keys are computed from the specified fields. Cached responses are returned directly without executing the workflow. The cache backend is in-memory by default; Redis-backed caching is available via a cache connector.

**Request deduplication:** prevent duplicate processing using idempotency keys:

```json
{
  "deduplication": {
    "header": "Idempotency-Key",
    "retention_secs": 300
  }
}
```

When a request with the same idempotency key arrives within the retention window, it returns `409 Conflict` instead of re-processing.

**Connection pool caching:** external database and MongoDB connector pools are cached and reused across requests, with configurable pool sizes and idle timeouts:

```toml
[engine]
max_pool_cache_entries = 100
cache_cleanup_interval_secs = 60
```


---
source: ./features/maintainability.md
---

# Maintainability

Orion provides comprehensive admin APIs, CI/CD integration patterns, dry-run testing, and operational tools for managing services in production.

## Admin APIs

Full CRUD operations for all entities through a RESTful admin API:

| Resource | Endpoints |
|----------|-----------|
| **Workflows** | Create, read, update, delete, status management, versioning, rollout, dry-run test, import/export, validate |
| **Channels** | Create, read, update, delete, status management, versioning |
| **Connectors** | Create, read, update, delete, reload, circuit breaker inspection/reset |
| **Engine** | Status, hot-reload |
| **Audit logs** | List with filtering by action and resource type |
| **Backup** | Database export and restore |

**Version management:** both workflows and channels support the draft → active → archived lifecycle. Filter by status:

```bash
curl -s "http://localhost:8080/api/v1/admin/workflows?status=active"
curl -s "http://localhost:8080/api/v1/admin/channels?status=draft"
```

**Engine control:**

```bash
# Check engine status
curl -s http://localhost:8080/api/v1/admin/engine/status

# Hot-reload after changes
curl -s -X POST http://localhost:8080/api/v1/admin/engine/reload
```

**OpenAPI / Swagger UI:** interactive API documentation is always available at `/docs`, and the OpenAPI 3.0 spec at `/api/v1/openapi.json`.

## CI/CD Integration

Orion workflows are JSON files that version, diff, and review like any other config.

**Bulk import and export:**

```bash
# Export active workflows
curl -s "http://localhost:8080/api/v1/admin/workflows/export?status=active" -o workflows.json

# Import workflows (created as drafts)
curl -s -X POST http://localhost:8080/api/v1/admin/workflows/import \
  -H "Content-Type: application/json" -d @workflows.json
```

**Pre-deploy validation:** validate workflow structure without creating:

```bash
curl -s -X POST http://localhost:8080/api/v1/admin/workflows/validate \
  -H "Content-Type: application/json" -d @workflow.json
```

**GitOps pipeline:** a typical CI/CD flow:

```
AI generates workflow → commit as JSON → CI validates & dry-runs → review → import → activate
```

GitHub Actions example:

```yaml
name: Validate Workflows
on:
  pull_request:
    paths: ['workflows/**/*.json']

jobs:
  validate:
    runs-on: ubuntu-latest
    services:
      orion:
        image: ghcr.io/goplasmatic/orion:latest
        ports: ['8080:8080']
    steps:
      - uses: actions/checkout@v4
      - name: Import and test workflows
        run: |
          for file in workflows/**/*.json; do
            curl -sf -X POST http://localhost:8080/api/v1/admin/workflows \
              -H "Content-Type: application/json" -d @"$file"
          done
```

**Tag-based organization:** tag workflows for filtering:

```json
{ "tags": ["fraud", "high-priority", "v2"] }
```

```bash
curl -s "http://localhost:8080/api/v1/admin/workflows?tag=fraud"
```

## Testing

**Dry-run execution:** test a workflow against sample data without activating it:

```bash
curl -s -X POST http://localhost:8080/api/v1/admin/workflows/<id>/test \
  -H "Content-Type: application/json" \
  -d '{"data": {"amount": 50000, "currency": "USD"}}'
```

The response includes a full execution trace showing which tasks ran and which were skipped:

```json
{
  "matched": true,
  "trace": {
    "steps": [
      { "task_id": "parse", "result": "executed" },
      { "task_id": "high_risk", "result": "executed" },
      { "task_id": "normal_risk", "result": "skipped" }
    ]
  },
  "output": {
    "txn": { "amount": 50000, "risk_level": "high", "requires_review": true }
  }
}
```

**Workflow validation:** check that a workflow definition is structurally valid:

```bash
curl -s -X POST http://localhost:8080/api/v1/admin/workflows/validate \
  -H "Content-Type: application/json" -d @workflow.json
```

**Step-by-step traces:** async traces record the full execution path and can be retrieved for debugging:

```bash
# Submit async request
curl -s -X POST http://localhost:8080/api/v1/data/orders/async \
  -H "Content-Type: application/json" -d '{ "data": { "order_id": "ORD-123" } }'

# Get trace with execution details
curl -s http://localhost:8080/api/v1/data/traces/{trace-id}
```

## Operations

**Audit logging:** all admin actions are recorded for compliance and debugging:

```bash
curl -s http://localhost:8080/api/v1/admin/audit-logs
curl -s "http://localhost:8080/api/v1/admin/audit-logs?action=activate&resource_type=workflow"
```

Each entry captures: principal, action, resource type, resource ID, details, and timestamp.

**Database backup and restore:**

```bash
# Export backup
curl -s -X POST http://localhost:8080/api/v1/admin/backup -o backup.json

# Restore from backup
curl -s -X POST http://localhost:8080/api/v1/admin/restore \
  -H "Content-Type: application/json" -d @backup.json
```

**Config validation CLI:** validate your configuration without starting the server:

```bash
orion-server validate-config
orion-server validate-config -c config.toml
```

**Database migrations:** run or preview pending migrations:

```bash
orion-server migrate              # Run migrations
orion-server migrate --dry-run    # Preview pending migrations
```


---
source: ./api/admin.md
---

# Admin API

All admin endpoints are under `/api/v1/admin/`. When admin authentication is enabled, requests must include a valid bearer token or API key.

## Channels

| Method | Path | Description |
|--------|------|-------------|
| POST | `/api/v1/admin/channels` | Create channel (as draft) |
| GET | `/api/v1/admin/channels` | List channels. Filter with `?status=`, `?channel_type=`, `?protocol=` |
| GET | `/api/v1/admin/channels/{id}` | Get channel by ID |
| PUT | `/api/v1/admin/channels/{id}` | Update draft channel |
| DELETE | `/api/v1/admin/channels/{id}` | Delete channel (all versions) |
| PATCH | `/api/v1/admin/channels/{id}/status` | Change status (active/archived) |
| GET | `/api/v1/admin/channels/{id}/versions` | List channel version history |
| POST | `/api/v1/admin/channels/{id}/versions` | Create new draft version from active channel |
| POST | `/api/v1/admin/channels/import` | Bulk import channels (as drafts). `?dry_run=true` validates without writing |

## Workflows

| Method | Path | Description |
|--------|------|-------------|
| POST | `/api/v1/admin/workflows` | Create workflow (as draft; optional `id` field for custom IDs) |
| GET | `/api/v1/admin/workflows` | List workflows. Filter with `?tag=`, `?status=` |
| GET | `/api/v1/admin/workflows/{id}` | Get workflow by ID |
| PUT | `/api/v1/admin/workflows/{id}` | Update draft workflow |
| DELETE | `/api/v1/admin/workflows/{id}` | Delete workflow (all versions) |
| PATCH | `/api/v1/admin/workflows/{id}/status` | Change status (active/archived) |
| GET | `/api/v1/admin/workflows/{id}/versions` | List workflow version history |
| POST | `/api/v1/admin/workflows/{id}/versions` | Create new draft version from active workflow |
| PATCH | `/api/v1/admin/workflows/{id}/rollout` | Update rollout percentage |
| POST | `/api/v1/admin/workflows/{id}/test` | Dry-run on sample payload |
| POST | `/api/v1/admin/workflows/import` | Bulk import workflows (as drafts). `?dry_run=true` validates without writing |
| GET | `/api/v1/admin/workflows/export` | Export workflows. Filter with `?tag=`, `?status=` |
| POST | `/api/v1/admin/workflows/validate` | Validate workflow definition |

## Connectors

| Method | Path | Description |
|--------|------|-------------|
| POST | `/api/v1/admin/connectors` | Create connector. String fields may use `env://VAR_NAME` to pull values from the process environment |
| GET | `/api/v1/admin/connectors` | List connectors (secrets masked) |
| GET | `/api/v1/admin/connectors/{id}` | Get connector by ID (secrets masked) |
| PUT | `/api/v1/admin/connectors/{id}` | Update connector |
| DELETE | `/api/v1/admin/connectors/{id}` | Delete connector |
| POST | `/api/v1/admin/connectors/import` | Bulk import connectors. `?dry_run=true` validates without writing |
| POST | `/api/v1/admin/connectors/reload` | Reload all connectors from DB |
| GET | `/api/v1/admin/connectors/circuit-breakers` | List circuit breaker states |
| POST | `/api/v1/admin/connectors/circuit-breakers/{key}` | Reset a circuit breaker |

Connector types: `http`, `kafka`, `db` (PostgreSQL/MySQL/SQLite/MongoDB), `cache`, `storage`, `es` (Elasticsearch). `db` and `es` connector configs accept an optional `operations` block that en/disables operation types per connector (read / insert / update / delete / upsert / raw_write) — see [Operation Gates](../features/extensibility.md#operation-gates).

## Engine

| Method | Path | Description |
|--------|------|-------------|
| GET | `/api/v1/admin/engine/status` | Engine status (version, uptime, workflows count, channels) |
| POST | `/api/v1/admin/engine/reload` | Hot-reload channels and workflows |

## Functions

| Method | Path | Description |
|--------|------|-------------|
| GET | `/api/v1/admin/functions` | List every task function with its input-field schema (category, type, required flag, description). Used by CLI tools and IDEs for autocompletion and by workflow validators to give field-pathed errors |

## Audit Logs

| Method | Path | Description |
|--------|------|-------------|
| GET | `/api/v1/admin/audit-logs` | List audit log entries. Filter with `?action=`, `?resource_type=` |

## Backups

| Method | Path | Description |
|--------|------|-------------|
| POST | `/api/v1/admin/backups` | Create a database backup (SQLite only — `VACUUM INTO` a timestamped file in `storage.backup_dir`) |
| GET | `/api/v1/admin/backups` | List backup files currently in `storage.backup_dir` |

## Lifecycle

Both channels and workflows follow a **draft → active → archived** lifecycle:

1. **Create:** entities are created as `draft` (not loaded into the engine)
2. **Update:** only draft versions can be updated via `PUT`
3. **Activate:** `PATCH /status` with `{"status": "active"}` loads the entity into the engine
4. **New version:** `POST /versions` creates a new draft version from the active entity
5. **Archive:** `PATCH /status` with `{"status": "archived"}` removes from the engine

A channel links to a workflow via `workflow_id`. Activating a channel makes it available for data processing; activating a workflow makes its logic available to the engine.

## Authentication

Admin API endpoints support bearer token or API key authentication when enabled:

```bash
# Bearer token (default header: Authorization)
curl -H "Authorization: Bearer your-secret-key" \
  http://localhost:8080/api/v1/admin/workflows

# API key via custom header
curl -H "X-API-Key: your-secret-key" \
  http://localhost:8080/api/v1/admin/workflows
```

Configure via `[admin_auth]` in config or `ORION_ADMIN_AUTH__ENABLED=true` environment variable.

## Error Response Format

All error responses follow a consistent structure:

```json
{
  "error": {
    "code": "NOT_FOUND",
    "message": "Workflow with id '...' not found"
  }
}
```

| Code | HTTP Status | Description |
|------|-------------|-------------|
| `NOT_FOUND` | 404 | Resource not found |
| `BAD_REQUEST` | 400 | Invalid input |
| `UNAUTHORIZED` | 401 | Missing or invalid credentials |
| `FORBIDDEN` | 403 | Access denied |
| `CONFLICT` | 409 | Duplicate or conflicting state |
| `RATE_LIMITED` | 429 | Too many requests |
| `TIMEOUT` | 504 | Workflow execution exceeded timeout |
| `SERVICE_UNAVAILABLE` | 503 | Backpressure or circuit breaker open |
| `UNSUPPORTED_MEDIA_TYPE` | 415 | Invalid content type |
| `INTERNAL_ERROR` | 500 | Internal server error |

When a workflow, channel, or connector fails strict validation on create/update, the envelope is extended with a `details` array of field-pathed errors (kept omitted for single-message errors so v0.1 clients aren't broken):

```json
{
  "error": {
    "code": "BAD_REQUEST",
    "message": "Workflow validation failed",
    "details": [
      { "field": "tasks[0].function.input.connector", "message": "is required" },
      { "field": "tasks[2].function.input.method",    "message": "expected string, got number" }
    ]
  }
}
```

The `field` path mirrors the JSON structure the API received, so editors can jump straight to the failing key. The same envelope is returned by `POST /workflows/validate`, `POST /workflows/{id}/test`, and the `orion-server lint` / `dry-run` CLI subcommands.


---
source: ./api/data.md
---

# Data API

The data API handles runtime request processing: routing messages to channels, executing workflows, and returning results.

## Endpoints

| Method | Path | Description |
|--------|------|-------------|
| `POST` | `/api/v1/data/{channel}` | Process message synchronously (simple channel name) |
| `POST` | `/api/v1/data/{channel}/async` | Submit for async processing (returns trace ID) |
| `ANY` | `/api/v1/data/{path...}` | REST route matching: method + path matched against channel route patterns |
| `ANY` | `/api/v1/data/{path...}/async` | Async submission via REST route matching |
| `GET` | `/api/v1/data/traces` | List traces. Filter with `?status=`, `?channel=`, `?mode=` |
| `GET` | `/api/v1/data/traces/{id}` | Poll async trace result |

## Route Resolution

When a request arrives at `/api/v1/data/{path}`, Orion resolves the target channel in this order:

1. **Async check:** strip trailing `/async` suffix (switches to async mode)
2. **REST route table:** match HTTP method + path against channel `route_pattern` values (e.g., `GET /orders/{order_id}`)
3. **Channel name fallback:** direct lookup by single path segment (e.g., `/api/v1/data/orders` → channel named `orders`)

REST routes are matched by priority (descending) then specificity (segment count). Path parameters are extracted and injected into the message metadata.

## Synchronous Processing

Send a POST to the channel name or a matching REST route:

```bash
# By channel name
curl -s -X POST http://localhost:8080/api/v1/data/orders \
  -H "Content-Type: application/json" \
  -d '{ "data": { "order_id": "ORD-123", "total": 25000 } }'

# By REST route pattern
curl -s -X GET http://localhost:8080/api/v1/data/orders/ORD-123/items/ITEM-1
```

**Response:**

```json
{
  "status": "ok",
  "data": {
    "order": { "order_id": "ORD-123", "total": 25000, "flagged": true }
  },
  "errors": []
}
```

### Per-request profiling

Add `X-Orion-Profile: 1` (or `?profile=1`) to the request and the response gains a `_orion.profile` block that breaks the request down by phase. The header is opt-in so you only pay the cost on the requests you care about, and `tracing.profile_enabled` in config gates the surface entirely. The debug surface always sits under the `_orion` namespace so workflow-produced output keys can never collide with future debug fields.

```json
{
  "status": "ok",
  "data": { ... },
  "errors": [],
  "_orion": {
    "profile": {
      "version": 1,
      "engine_lock_ms": 0.04,
      "workflow_ms": 6.71,
      "tasks": [
        { "id": "parse",    "ms": 0.18 },
        { "id": "validate", "ms": 0.42 },
        { "id": "enrich",   "ms": 4.91 }
      ]
    }
  }
}
```

## Asynchronous Processing

Append `/async` to submit for background processing:

```bash
curl -s -X POST http://localhost:8080/api/v1/data/orders/async \
  -H "Content-Type: application/json" \
  -d '{ "data": { "order_id": "ORD-456" } }'
```

**Response:** returns immediately with a trace ID:

```json
{
  "trace_id": "550e8400-e29b-41d4-a716-446655440000",
  "status": "pending"
}
```

**Poll for the result:**

```bash
curl -s http://localhost:8080/api/v1/data/traces/550e8400-e29b-41d4-a716-446655440000
```

**Trace statuses:** `pending` → `completed` or `failed`.

## Trace Endpoints

List and filter traces:

```bash
# List all traces
curl -s http://localhost:8080/api/v1/data/traces

# Filter by channel and status
curl -s "http://localhost:8080/api/v1/data/traces?channel=orders&status=completed"

# Filter by mode
curl -s "http://localhost:8080/api/v1/data/traces?mode=async"
```

Get a specific trace:

```bash
curl -s http://localhost:8080/api/v1/data/traces/{trace-id}
```

## Operational Endpoints

| Method | Path | Description |
|--------|------|-------------|
| GET | `/health` | Health check (200 OK / 503 degraded). Checks DB, engine, uptime |
| GET | `/healthz` | Kubernetes liveness probe. Always returns 200 |
| GET | `/readyz` | Kubernetes readiness probe. 503 if DB, engine, or startup not ready |
| GET | `/metrics` | Prometheus metrics (when enabled) |
| GET | `/docs` | Swagger UI |
| GET | `/api/v1/openapi.json` | OpenAPI 3.0 specification |


---
source: ./reference/workflows.md
---

# Workflow Reference

A **workflow** is a versioned, JSON-defined pipeline of tasks. A
[channel](../architecture/overview.md) links to a workflow by `workflow_id`;
when a request arrives, Orion matches an active workflow, runs its tasks in
order, and returns the resulting data context.

This page is the authoritative reference for the workflow JSON shape, the data
context model, conditions, and the draft → active lifecycle. For the per-task
`function.input` schemas, see the [Function Reference](./functions.md).

## The workflow object

Send this shape to `POST /api/v1/admin/workflows` (and `PUT .../{id}` to update
the draft). Fields marked **server-managed** are set by Orion and returned in
responses — you don't send them on create.

| Field | Type | Required | Default | Notes |
|-------|------|:--------:|---------|-------|
| `workflow_id` | string | no | auto (UUID v4) | Stable identifier. ≤128 chars, alphanumeric plus `.`, `-`, `_`, must start alphanumeric |
| `name` | string | **yes** | — | Human-readable name. ≤255 chars, non-empty |
| `description` | string | no | — | ≤2048 chars |
| `priority` | integer | no | `0` | Match order — higher priority workflows are evaluated first (see [Matching](#matching)) |
| `condition` | JSONLogic | no | `true` | Whether the workflow matches a request (see [Conditions](#conditions)) |
| `tasks` | array | **yes** | — | Ordered, non-empty list of [task objects](#tasks) |
| `tags` | string[] | no | `[]` | Free-form labels for filtering |
| `continue_on_error` | bool | no | `false` | If `true`, a failing task does not halt the pipeline (see [Error handling](#error-handling)) |
| `version` | integer | server-managed | `1` | Increments per saved version of a `workflow_id` |
| `status` | string | server-managed | `draft` | `draft` \| `active` \| `archived` |
| `rollout_percentage` | integer | server-managed | `100` | Share of traffic when activated (see [Rollout](#rollout)) |
| `created_at` / `updated_at` | string | server-managed | — | RFC 3339 timestamps |

Responses wrap the resource in a `data` envelope:

```json
{ "data": { "workflow_id": "high-value-order", "version": 1, "status": "draft", "...": "..." } }
```

Validation failures return `400` with a structured error envelope — see the
[Admin API](../api/admin.md#error-response-format) for the `FieldError` format.

## Tasks

Each entry in `tasks` is a single step in the pipeline:

| Field | Type | Required | Notes |
|-------|------|:--------:|-------|
| `id` | string | **yes** | Unique within the workflow; used in tracing |
| `name` | string | **yes** | Human-readable label |
| `function` | object | **yes** | The function to run — see below |
| `condition` | JSONLogic | no | If present and falsy, this task is skipped |

The `function` object names a [built-in function](./functions.md) and supplies
its `input`:

| Field | Type | Required | Notes |
|-------|------|:--------:|-------|
| `name` | string | **yes** | One of the [16 built-in functions](./functions.md) |
| `input` | object | depends | Function-specific parameters. Connector functions are schema-validated on create |

```json
{
  "id": "flag",
  "name": "Flag high-value order",
  "condition": { ">": [{ "var": "data.order.total" }, 10000] },
  "function": {
    "name": "map",
    "input": { "mappings": [{ "path": "data.order.flagged", "logic": true }] }
  }
}
```

## The data context

Tasks share a single JSON document, the **data context**, with two top-level
areas your JSONLogic can read:

- `data` — the working document. The request body is parsed into it by
  `parse_json` (or `parse_xml`), tasks read and write `data.*`, and for a sync
  channel the final `data` object is what's returned to the caller.
- `metadata` — request context such as headers, query params, and path params,
  available to conditions and validation.

Most functions write their result to a **dotted output path** (e.g.
`response_path`, `output`, or a `map` mapping `path`), which is created inside
the context if it doesn't exist.

> **The parse-then-process pattern.** A workflow that reads request data should
> start with `parse_json`; otherwise conditions referencing `data.*` see an
> empty context.

```json
{
  "tasks": [
    { "id": "parse",   "name": "Parse",   "function": { "name": "parse_json", "input": { "source": "payload", "target": "order" } } },
    { "id": "process", "name": "Process", "condition": { ">": [{ "var": "data.order.total" }, 100] }, "function": { "name": "map", "input": { "mappings": [] } } }
  ]
}
```

## Conditions

Conditions are [JSONLogic](https://jsonlogic.com) expressions, evaluated by
[datalogic-rs](https://github.com/GoPlasmatic/datalogic-rs) and compiled once at
engine build time. They appear at two levels:

- **Workflow-level `condition`** — decides whether the whole workflow *matches* a
  request. Defaults to `true` (always matches). If multiple active workflows are
  bound to a channel, the first match wins (see [Matching](#matching)).
- **Task-level `condition`** — decides whether *that task* runs within a matched
  workflow. Use it for branching inside a pipeline.

Common operators (see the [JSONLogic spec](https://jsonlogic.com/operations.html)
for the full set):

| Operator | Example | Meaning |
|----------|---------|---------|
| `var` | `{ "var": "data.order.total" }` | Read a value from the context |
| `==` / `!=` | `{ "==": [{ "var": "data.type" }, "order"] }` | Loose equality |
| `>` `>=` `<` `<=` | `{ ">": [{ "var": "data.order.total" }, 10000] }` | Comparison |
| `and` / `or` / `!` | `{ "and": [a, b] }` | Boolean logic |
| `!!` | `{ "!!": [{ "var": "data.order.id" }] }` | Truthiness (e.g. "is present") |
| `if` | `{ "if": [cond, then, else] }` | Conditional value |
| `in` | `{ "in": [{ "var": "data.tier" }, ["vip", "premium"]] }` | Membership |
| `cat` | `{ "cat": ["Order #", { "var": "data.order.id" }] }` | String concatenation |
| `+` `-` `*` `/` `%` | `{ "*": [{ "var": "data.qty" }, 1.1] }` | Arithmetic |

## Error handling

By default the pipeline **halts** on the first task that errors, and the error is
returned to the caller. Set `continue_on_error: true` on the workflow to keep
running subsequent tasks and collect errors instead. The
[`filter`](./functions.md#filter) function offers finer control: `on_reject:
"halt"` stops the workflow, while `on_reject: "skip"` skips only the current task.

For async channels, a task failure routes the trace to the Dead Letter Queue for
automatic retry — see [Resilience](../features/resilience.md).

## Lifecycle and versioning

Each `workflow_id` has one or more **versions**, identified by the composite key
`(workflow_id, version)`. Status moves in one direction:

```orion-diagram
{
  "direction": "LR",
  "nodes": [
    { "id": "draft",    "label": "draft",    "type": "infra" },
    { "id": "active",   "label": "active",   "type": "channel" },
    { "id": "archived", "label": "archived", "type": "datastore", "shape": "rectangle" }
  ],
  "edges": [
    { "from": "draft",  "to": "active",   "label": "activate" },
    { "from": "active", "to": "archived", "label": "archive" }
  ]
}
```

- **draft** — editable; not served. Only **one draft per `workflow_id`** may exist
  at a time. Creating a workflow starts it as a draft.
- **active** — served; **immutable**. To change an active workflow, create a new
  draft version, edit it, and activate it.
- **archived** — retired; kept for history and instant rollback.

Endpoints (see the [Admin API](../api/admin.md#workflows) for full details):

| Action | Endpoint |
|--------|----------|
| Validate without saving | `POST /api/v1/admin/workflows/validate` |
| Create (as draft) | `POST /api/v1/admin/workflows` |
| New draft version of an existing id | `POST /api/v1/admin/workflows/{id}/versions` |
| Dry-run against sample data | `POST /api/v1/admin/workflows/{id}/test` |
| Change status | `PATCH /api/v1/admin/workflows/{id}/status` |
| Adjust rollout | `PATCH /api/v1/admin/workflows/{id}/rollout` |

### Matching

When a channel resolves to its workflows, Orion evaluates **active** workflows in
descending `priority`, then runs the first whose `condition` is truthy. Give a
catch-all workflow a low priority and specific ones a higher priority to layer
behavior.

### Rollout

`rollout_percentage` (1–100) enables canary releases across versions. Activating a
new version at, say, `25` directs ~25% of traffic to it and the remainder to the
previously active version; traffic is bucketed by a stable hash of the request so
a given caller is routed consistently. Promote by raising the percentage to `100`
(which archives the older active version), or roll back instantly by re-activating
a previous version.

```bash
# Activate a new version to 10% of traffic
curl -X PATCH http://localhost:8080/api/v1/admin/workflows/high-value-order/status \
  -H "Content-Type: application/json" -d '{ "status": "active", "rollout_percentage": 10 }'

# Ramp up later
curl -X PATCH http://localhost:8080/api/v1/admin/workflows/high-value-order/rollout \
  -H "Content-Type: application/json" -d '{ "rollout_percentage": 50 }'
```

## Complete example

```json
{
  "workflow_id": "high-value-order",
  "name": "High-Value Order",
  "description": "Flag orders over $10,000 for manual review",
  "priority": 10,
  "condition": { "==": [{ "var": "metadata.headers.x-source" }, "checkout"] },
  "tasks": [
    {
      "id": "parse",
      "name": "Parse payload",
      "function": { "name": "parse_json", "input": { "source": "payload", "target": "order" } }
    },
    {
      "id": "validate",
      "name": "Validate order",
      "function": {
        "name": "validation",
        "input": { "rules": [
          { "logic": { "!!": [{ "var": "data.order.id" }] }, "message": "order id is required" },
          { "logic": { ">": [{ "var": "data.order.total" }, 0] }, "message": "total must be positive" }
        ]}
      }
    },
    {
      "id": "flag",
      "name": "Flag for review",
      "condition": { ">": [{ "var": "data.order.total" }, 10000] },
      "function": {
        "name": "map",
        "input": { "mappings": [
          { "path": "data.order.flagged", "logic": true },
          { "path": "data.order.alert", "logic": { "cat": ["High-value order: $", { "var": "data.order.total" }] } }
        ]}
      }
    }
  ],
  "tags": ["orders", "risk"],
  "continue_on_error": false
}
```

See [Use Cases & Patterns](../tutorials/use-cases.md) for complete, tested
workflows, and the [Function Reference](./functions.md) for every function's
input schema.


---
source: ./reference/functions.md
---

# Function Reference

A workflow is an ordered list of **tasks**, and every task invokes one built-in
**function** with an `input` object:

```json
{
  "id": "enrich",
  "name": "Look up customer",
  "function": {
    "name": "http_call",
    "input": { "connector": "crm", "path": "/customers/42", "response_path": "data.customer" }
  }
}
```

Functions read from and write to the **data context** — the JSON document that
flows through the pipeline. By convention the request body is parsed into
`data.*` by `parse_json`, later tasks read `data.*` in their JSONLogic, and the
`data` object is what a sync channel returns. See the
[Workflow Reference](./workflows.md#the-data-context) for the context model and
how `condition` expressions are evaluated.

Orion ships **18 functions** (plus `validate`, an alias for `validation`). Eight
are contributed by the [dataflow-rs](https://github.com/GoPlasmatic/dataflow-rs)
engine; ten are Orion handlers that talk to [connectors](../features/extensibility.md#connectors)
or compose channels.

| Function | Category | Connector | Purpose |
|----------|----------|:---------:|---------|
| [`parse_json`](#parse_json) | Data | — | Parse the raw payload into the data context |
| [`parse_xml`](#parse_xml) | Data | — | Parse an XML payload into the data context |
| [`map`](#map) | Data | — | Transform/reshape data with JSONLogic |
| [`filter`](#filter) | Data | — | Gate the pipeline on a JSONLogic condition |
| [`validation`](#validation--validate) | Data | — | Collect validation errors from JSONLogic rules |
| [`log`](#log) | Data | — | Emit a structured log line |
| [`publish_json`](#publish_json) | Data | — | Serialize a context field to a JSON string |
| [`publish_xml`](#publish_xml) | Data | — | Serialize a context field to an XML string |
| [`http_call`](#http_call) | Connector | HTTP | Call an external API with retry + circuit breaker |
| [`data_query`](#data_query) | Connector | SQL / MongoDB / ES | Portable, backend-neutral query |
| [`data_write`](#data_write) | Connector | SQL / MongoDB / ES | Portable, backend-neutral insert/update/delete/upsert |
| [`db_read`](#db_read) | Connector | SQL | Run a raw `SELECT`, return rows as JSON |
| [`db_write`](#db_write) | Connector | SQL | Run raw `INSERT`/`UPDATE`/`DELETE`, return affected count |
| [`cache_read`](#cache_read) | Connector | Cache | Read a value from Redis or the in-memory cache |
| [`cache_write`](#cache_write) | Connector | Cache | Write a value to cache with optional TTL |
| [`mongo_read`](#mongo_read) | Connector | MongoDB | Run a raw `find()`, return documents as JSON |
| [`publish_kafka`](#publish_kafka) | Connector | Kafka | Publish a message to a Kafka topic |
| [`channel_call`](#channel_call) | Composition | — | Invoke another channel's workflow in-process |

> Wherever an input field is described as **JSONLogic**, you pass a JSONLogic
> expression that is evaluated against the data context. A plain JSON literal
> (string, number, object) is also valid JSONLogic and evaluates to itself.

---

## Data functions

These come from the dataflow-rs engine. Orion does **not** input-schema-validate
them at workflow-create time (unlike the connector functions below), so an
invalid `input` here surfaces at execution time rather than on create.

### `parse_json`

Reads a raw value (typically the request payload) and parses it as JSON into the
data context. Almost every workflow starts with this — without it, task
conditions that reference `data.*` see an empty context.

| Field | Type | Required | Default | Description |
|-------|------|:--------:|---------|-------------|
| `source` | string | yes | — | Where to read the raw value from, e.g. `"payload"` |
| `target` | string | yes | — | Field name under `data`; the parsed value is stored at `data.{target}` |

```json
{ "name": "parse_json", "input": { "source": "payload", "target": "order" } }
```

### `parse_xml`

Same input shape as `parse_json`, but parses an XML payload into a JSON
structure at `data.{target}`.

| Field | Type | Required | Default | Description |
|-------|------|:--------:|---------|-------------|
| `source` | string | yes | — | Where to read the raw XML from, e.g. `"payload"` |
| `target` | string | yes | — | Stored at `data.{target}` |

```json
{ "name": "parse_xml", "input": { "source": "payload", "target": "order" } }
```

### `map`

Applies an ordered list of JSONLogic expressions, writing each result to a
dotted path in the context. The primary tool for reshaping, computing, and
enriching data.

| Field | Type | Required | Default | Description |
|-------|------|:--------:|---------|-------------|
| `mappings` | array | yes | — | Ordered list of `{ "path", "logic" }` entries |
| `mappings[].path` | string | yes | — | Dotted target path, e.g. `"data.order.total"` |
| `mappings[].logic` | JSONLogic | yes | — | Expression whose result is written to `path` |

```json
{
  "name": "map",
  "input": {
    "mappings": [
      { "path": "data.order.flagged", "logic": true },
      { "path": "data.order.total_with_tax", "logic": { "*": [{ "var": "data.order.total" }, 1.1] } }
    ]
  }
}
```

### `filter`

Evaluates a JSONLogic condition. If it is truthy the pipeline continues;
otherwise the `on_reject` action is taken.

| Field | Type | Required | Default | Description |
|-------|------|:--------:|---------|-------------|
| `condition` | JSONLogic | yes | — | Evaluated against the data context |
| `on_reject` | string | no | `"halt"` | `"halt"` stops the whole workflow; `"skip"` skips only this task |

```json
{
  "name": "filter",
  "input": {
    "condition": { ">": [{ "var": "data.order.total" }, 0] },
    "on_reject": "halt"
  }
}
```

### `validation` / `validate`

Evaluates a list of rules. Each rule's `logic` must evaluate to exactly `true`;
any other result records the rule's `message` in the response's error list.
Validation is non-destructive — it never mutates the data context. `validate` is
an accepted alias for `validation`.

| Field | Type | Required | Default | Description |
|-------|------|:--------:|---------|-------------|
| `rules` | array | yes | — | List of `{ "logic", "message" }` rules |
| `rules[].logic` | JSONLogic | yes | — | Must evaluate to `true` to pass |
| `rules[].message` | string | yes | — | Error message recorded when the rule fails |

```json
{
  "name": "validation",
  "input": {
    "rules": [
      { "logic": { "!!": [{ "var": "data.order.customer_id" }] }, "message": "customer_id is required" },
      { "logic": { ">": [{ "var": "data.order.total" }, 0] },        "message": "total must be positive" }
    ]
  }
}
```

### `log`

Emits a structured log line. `message` is a JSONLogic expression (a plain string
is valid), and `fields` attaches additional JSONLogic-derived key/values.

| Field | Type | Required | Default | Description |
|-------|------|:--------:|---------|-------------|
| `message` | JSONLogic | yes | — | The log message (string literal or expression) |
| `level` | string | no | `"info"` | `trace` \| `debug` \| `info` \| `warn` \| `error` |
| `fields` | object | no | `{}` | Map of name → JSONLogic expression, logged as structured fields |

```json
{
  "name": "log",
  "input": {
    "level": "info",
    "message": "Order processed",
    "fields": { "order_id": { "var": "data.order.id" } }
  }
}
```

### `publish_json`

Serializes a field **inside** the data context to a JSON string and stores it at
another field. (It writes back into the context; it does not publish to an
external system.)

| Field | Type | Required | Default | Description |
|-------|------|:--------:|---------|-------------|
| `source` | string | yes | — | Field under `data` to serialize, e.g. `"order"` (reads `data.order`) |
| `target` | string | yes | — | Field under `data` to receive the serialized string |
| `pretty` | bool | no | `false` | Pretty-print the JSON output |

```json
{ "name": "publish_json", "input": { "source": "order", "target": "order_json", "pretty": true } }
```

### `publish_xml`

Like `publish_json`, but serializes to an XML string.

| Field | Type | Required | Default | Description |
|-------|------|:--------:|---------|-------------|
| `source` | string | yes | — | Field under `data` to serialize |
| `target` | string | yes | — | Field under `data` to receive the XML string |
| `root_element` | string | no | `"root"` | Name of the XML root element |

```json
{ "name": "publish_xml", "input": { "source": "order", "target": "order_xml", "root_element": "Order" } }
```

---

## Connector functions

These reference a [connector](../features/extensibility.md#connectors) by name —
credentials and endpoints live in the connector, not the workflow. Orion
validates their `input` at workflow create/update time and exposes the schema
via [`GET /api/v1/admin/functions`](../api/admin.md#functions). Every connector
call is wrapped in a [circuit breaker](../features/resilience.md).

### `http_call`

Makes an HTTP request through an HTTP connector, with retry and circuit-breaker
support. The connector supplies the base URL and auth.

| Field | Type | Required | Default | Description |
|-------|------|:--------:|---------|-------------|
| `connector` | string | yes | — | Name of the HTTP connector |
| `method` | string | no | `"GET"` | `GET` \| `POST` \| `PUT` \| `PATCH` \| `DELETE` |
| `path` | string | no | — | Path appended to the connector's base URL |
| `path_logic` | JSONLogic | no | — | Compute the path dynamically (use instead of `path`) |
| `headers` | object | no | `{}` | Extra request headers (string → string) |
| `body` | any | no | — | Static request body (serialized as JSON) |
| `body_logic` | JSONLogic | no | — | Compute the body dynamically (use instead of `body`) |
| `response_path` | string | no | — | Dotted path where the response body is written; omit to discard it |
| `timeout_ms` | number | no | `30000` | Per-request timeout in milliseconds |

```json
{
  "name": "http_call",
  "input": {
    "connector": "payment-api",
    "method": "POST",
    "path": "/charge",
    "body_logic": { "var": "data.payment" },
    "response_path": "data.charge_result",
    "timeout_ms": 5000
  }
}
```

### `data_query`

Runs one **backend-neutral query** against a SQL (PostgreSQL/MySQL/SQLite),
MongoDB, or Elasticsearch connector — the connector decides the rendering
(parameterized SQL via sea-query, a Mongo `find`, or an ES `_search` body).
The full envelope, operator vocabulary, schema registry, and relation support
are documented in the [Portable Data Dialect](./data-dialect.md) reference.

| Field | Type | Required | Default | Description |
|-------|------|:--------:|---------|-------------|
| `connector` | string | yes | — | Name of a `db` or `es` connector |
| `query` | object | yes | — | The query envelope: `source`, `filter`, `fields`, `sort`, `limit`, `skip`, `include` |
| `params` | object | no | `{}` | Named values referenced as `{ "param": "name" }` inside the filter; each value is JSONLogic resolved against the context |
| `schema` | object | no | identity | Inline entity schema: renames, types, allowlist, relations |
| `database` | string | MongoDB only | — | Database name (Mongo connectors) |
| `output` | string | no | `"data"` | Dotted path where the row array is written |

```json
{
  "name": "data_query",
  "input": {
    "connector": "orders-db",
    "query": {
      "source": "orders",
      "filter": { "and": [
        { "==": [{ "field": "customer_id" }, { "param": "cid" }] },
        { ">":  [{ "field": "total" }, 100] }
      ] },
      "sort": [{ "created_at": "desc" }],
      "limit": 20
    },
    "params": { "cid": { "var": "data.customer_id" } },
    "output": "data.orders"
  }
}
```

Page sizes are bounded by the `[query]` config section (`default_limit` /
`max_limit`); a query asking for more than the cap is rejected, never clamped.

### `data_write`

The write counterpart of `data_query`: one **backend-neutral mutation** —
`insert`, `update`, `delete`, or `upsert` — rendered natively for SQL, MongoDB,
or Elasticsearch. The `filter` of an update/delete is the query dialect's
filter, unchanged. See the [Portable Data Dialect](./data-dialect.md) reference
for the full envelope, backend mapping, and safety rules.

| Field | Type | Required | Default | Description |
|-------|------|:--------:|---------|-------------|
| `connector` | string | yes | — | Name of a `db` or `es` connector |
| `op` | string | yes | — | `insert` \| `update` \| `delete` \| `upsert` |
| `target` | string | yes | — | Logical entity → table / collection / index |
| `values` | object \| array | insert, upsert | — | Row object(s) to insert |
| `set` | object | update, upsert | — | Column → value/param assignments |
| `filter` | JSONLogic | update, delete | — | Row selection (same operators as `data_query`) |
| `on_conflict` | object | upsert | — | `{ "target": [cols], "action": "update" \| "nothing" }` |
| `returning` | array | no | — | Columns returned from mutated rows (PostgreSQL/SQLite only) |
| `all` | bool | no | `false` | Acknowledge an intentionally unfiltered update/delete |
| `params` / `schema` / `database` / `output` | | | | As in `data_query` |

```json
{
  "name": "data_write",
  "input": {
    "connector": "orders-db",
    "op": "update",
    "target": "orders",
    "set": { "status": "shipped" },
    "filter": { "==": [{ "field": "id" }, { "param": "id" }] },
    "params": { "id": { "var": "data.order_id" } },
    "output": "data.write_result"
  }
}
```

Safety guards: unfiltered mutations are rejected unless `"all": true` **and**
`write.allow_unfiltered` are both set; bulk inserts over `write.max_rows` are
rejected; and a connector's
[operation gates](./data-dialect.md#connector-operation-gates) can disable
individual ops entirely. Results are normalized per backend — SQL returns
`{ "rows_affected": n }` (plus `returning` / `last_insert_id` where supported);
MongoDB and Elasticsearch return doc-store counts (`inserted`/`ids`,
`matched`/`modified`, `deleted`).

### `db_read`

The raw-SQL escape hatch for reads — anything outside the portable dialect's
vocabulary (joins, aggregations, CTEs, database-specific SQL). Runs a `SELECT`
against a SQL connector and writes the result rows as a JSON array. Use
placeholders bound from `params` — `?` for SQLite/MySQL, `$1`, `$2`,
… for PostgreSQL.

| Field | Type | Required | Default | Description |
|-------|------|:--------:|---------|-------------|
| `connector` | string | yes | — | Name of the SQL connector |
| `query` | string | yes | — | `SELECT` statement with bind placeholders |
| `params` | array | no | — | Values bound to the placeholders, in order |
| `output` | string | no | `"data"` | Dotted path where the row array is written |

```json
{
  "name": "db_read",
  "input": {
    "connector": "primary-db",
    "query": "SELECT id, name, tier FROM customers WHERE id = ?",
    "params": [{ "var": "data.order.customer_id" }],
    "output": "data.customer"
  }
}
```

### `db_write`

The raw-SQL escape hatch for writes (multi-table statements, `UPDATE … FROM`,
SQL functions in `SET`, DDL). Runs an `INSERT`/`UPDATE`/`DELETE` against a SQL
connector and writes `{ "rows_affected": N }`. Note: the author writes
dialect-specific SQL, and a connector can disable this function entirely via
its [`raw_write` operation gate](./data-dialect.md#connector-operation-gates).

| Field | Type | Required | Default | Description |
|-------|------|:--------:|---------|-------------|
| `connector` | string | yes | — | Name of the SQL connector |
| `query` | string | yes | — | `INSERT`/`UPDATE`/`DELETE` statement with bind placeholders |
| `params` | array | no | — | Values bound to the placeholders, in order |
| `output` | string | no | `"data"` | Dotted path where `{ "rows_affected": N }` is written |

```json
{
  "name": "db_write",
  "input": {
    "connector": "primary-db",
    "query": "INSERT INTO orders (id, total) VALUES (?, ?)",
    "params": [{ "var": "data.order.id" }, { "var": "data.order.total" }],
    "output": "data.write_result"
  }
}
```

### `cache_read`

Reads a key from a cache connector (Redis or the built-in in-memory backend).
Missing keys yield `null`.

| Field | Type | Required | Default | Description |
|-------|------|:--------:|---------|-------------|
| `connector` | string | yes | — | Name of the cache connector |
| `key` | string | yes | — | Cache key to read |
| `output` | string | no | `"data"` | Dotted path where the value is written |

```json
{ "name": "cache_read", "input": { "connector": "redis", "key": "rate:42", "output": "data.cached" } }
```

### `cache_write`

Writes a key to a cache connector, optionally with a TTL.

| Field | Type | Required | Default | Description |
|-------|------|:--------:|---------|-------------|
| `connector` | string | yes | — | Name of the cache connector |
| `key` | string | yes | — | Cache key to set |
| `value` | any | yes | — | Value to store (non-strings are JSON-serialized) |
| `ttl_secs` | number | no | no expiry | Time-to-live in seconds |

```json
{ "name": "cache_write", "input": { "connector": "redis", "key": "rate:42", "value": 1, "ttl_secs": 60 } }
```

### `mongo_read`

The raw escape hatch for MongoDB reads: runs a `find()` with a hand-written
Mongo filter document and writes the matched documents as a JSON array. For
backend-portable queries, prefer [`data_query`](#data_query).

| Field | Type | Required | Default | Description |
|-------|------|:--------:|---------|-------------|
| `connector` | string | yes | — | Name of the MongoDB connector |
| `database` | string | yes | — | Database name |
| `collection` | string | yes | — | Collection name |
| `filter` | object | no | `{}` | MongoDB find filter document |
| `output` | string | no | `"data"` | Dotted path where matched documents are written |

```json
{
  "name": "mongo_read",
  "input": {
    "connector": "mongo",
    "database": "shop",
    "collection": "customers",
    "filter": { "tier": "vip" },
    "output": "data.vips"
  }
}
```

### `publish_kafka`

Publishes a message to a Kafka topic through a Kafka connector. Requires Kafka to
be enabled in config. If `value_logic` is omitted, the full data context is
published.

| Field | Type | Required | Default | Description |
|-------|------|:--------:|---------|-------------|
| `connector` | string | yes | — | Name of the Kafka connector |
| `topic` | string | yes | — | Target topic |
| `key_logic` | JSONLogic | no | — | Expression that derives the message key |
| `value_logic` | JSONLogic | no | full `data` | Expression that derives the message value |

```json
{
  "name": "publish_kafka",
  "input": {
    "connector": "events",
    "topic": "order.placed",
    "key_logic": { "var": "data.order.id" },
    "value_logic": { "var": "data.order" }
  }
}
```

---

## Composition functions

### `channel_call`

Invokes another channel's workflow **in-process** — no network hop. The called
channel keeps its own versioning and governance. Cycle detection and a max call
depth prevent runaway recursion. Provide exactly one of `channel`/`channel_logic`
and at most one of `data`/`data_logic`.

| Field | Type | Required | Default | Description |
|-------|------|:--------:|---------|-------------|
| `channel` | string | one of `channel`/`channel_logic` | — | Static target channel name |
| `channel_logic` | JSONLogic | one of `channel`/`channel_logic` | — | Expression that resolves to the target channel name |
| `data` | any | no | request payload | Static payload passed to the target channel |
| `data_logic` | JSONLogic | no | — | Expression that derives the payload |
| `response_path` | string | no | `"data"` | Dotted path where the called channel's response is stored |
| `timeout_ms` | number | no | from config | Per-call timeout in milliseconds |

```json
{
  "name": "channel_call",
  "input": {
    "channel": "customer-lookup",
    "data_logic": { "var": "data.order.customer_id" },
    "response_path": "data.customer"
  }
}
```

---

## Inspecting schemas at runtime

`GET /api/v1/admin/functions` returns the live input schema for the connector and
composition functions (the data functions are provided by dataflow-rs and are not
catalogued there). The [Orion CLI MCP server](../tutorials/mcp-setup.md) surfaces
the same schemas to AI assistants so generated workflows use correct field names.


---
source: ./reference/data-dialect.md
---

# Portable Data Dialect

The **portable data dialect** lets a workflow express a database query or
mutation once — in a backend-neutral, JSONLogic-shaped envelope — and run it
unchanged against **PostgreSQL, MySQL, SQLite, MongoDB, or Elasticsearch**.
Switching backend is a connector change, not a rewrite.

Two functions implement it:

- [`data_query`](./functions.md#data_query) — reads: filter, project, sort,
  paginate, and include related records.
- [`data_write`](./functions.md#data_write) — writes: insert, update, delete,
  and upsert.

The raw functions (`db_read`, `db_write`, `mongo_read`) remain the **escape
hatch** for anything outside the portable vocabulary — hand-written SQL, CTEs,
multi-table statements, aggregations.

```json
{
  "name": "data_query",
  "input": {
    "connector": "orders_db",
    "query": {
      "source": "users",
      "filter": { "and": [
        { ">":  [{ "field": "age" }, 25] },
        { "in": [{ "field": "status" }, ["active", "trial"]] }
      ] },
      "fields": ["id", "name", "age"],
      "sort": [{ "age": "asc" }],
      "limit": 50
    },
    "output": "data.users"
  }
}
```

Point `connector` at Postgres and this renders parameterized SQL via sea-query;
at MongoDB and it renders a `find` filter document; at Elasticsearch and it
renders a Query DSL search body. The results come back as the same JSON row
array either way.

## The token model

Three tokens appear inside envelopes, and they never collide:

| Token | Meaning | Valid in |
|-------|---------|----------|
| `{ "field": "name" }` | A column / document field | `filter` (and as keys in `values`/`set`) |
| `{ "param": "p" }` | A named value from the `params` map, resolved **before** rendering | `filter`, `values`, `set` |
| `{ "var": "data.x" }` | An ordinary datalogic context lookup | only inside the `params` map |

`params` is the single point where the message touches a query or mutation — it
produces literal values, never SQL or query text:

```json
{
  "filter": { "==": [{ "field": "id" }, { "param": "id" }] },
  "params": { "id": { "var": "data.req.id" } }
}
```

Every resolved value is a **bound parameter** (SQL) or a document value
(Mongo/ES). No user value is ever string-interpolated — the dialect is
injection-safe by construction.

## Query envelope (`data_query`)

| Field | Type | Description |
|-------|------|-------------|
| `source` | string | Logical entity → table / collection / index (schema-resolved) |
| `filter` | JSONLogic | Condition from the operator vocabulary below; omit for "match all" |
| `fields` | array | Projection; omit for all columns/fields |
| `sort` | array | `[{ "age": "asc" }, { "name": "desc" }]` — deterministic null ordering across backends |
| `limit` / `skip` | number | Pagination. Missing `limit` gets `query.default_limit`; above `query.max_limit` is **rejected, never clamped** |
| `include` | object | Relation name → `{ "fields": [..], "limit": n }`; nested related records, hydrated per relation (see [Relations](#relations-and-includes)) |

### Operator vocabulary

| Operator | Meaning |
|----------|---------|
| `and`, `or`, `!` | Boolean combinators |
| `==`, `!=`, `<`, `<=`, `>`, `>=` | Comparisons (`===`/`!==` are aliases). `{"==": [x, null]}` is a null test |
| `<`/`<=` (ternary) | Range: `{ "<=": [1, { "field": "x" }, 10] }` → BETWEEN |
| `in` | Membership (list haystack) or substring containment (string haystack) |
| `starts_with`, `ends_with` | Text anchors (rendered as `LIKE` / `$regex` / `prefix`+`wildcard`) |
| `missing` | Field(s) have no meaningful value |
| `some`, `all`, `none` | Quantifiers over a declared relation |

Anything outside this vocabulary is rejected with a **located error** naming
the operator and position — never silently ignored.

## Write envelope (`data_write`)

| Field | Used by | Description |
|-------|---------|-------------|
| `op` | all | `insert` \| `update` \| `delete` \| `upsert` |
| `target` | all | Logical entity → table / collection / index |
| `values` | insert, upsert | A row object, or an array of rows (bulk) |
| `set` | update, upsert | Column → value/param assignments |
| `filter` | update, delete | Row selection — **the query dialect's filter**, same operators, same rendering |
| `on_conflict` | upsert | `{ "target": ["email"], "action": "update" \| "nothing" }` |
| `returning` | all | Columns to return from mutated rows (capability-gated, see below) |
| `all` | update, delete | Explicit acknowledgement for an intentionally unfiltered mutation |

One task per operation:

```json
{ "name": "data_write", "input": {
    "connector": "orders_db", "op": "insert", "target": "users",
    "values": [ { "name": "Ada", "status": "active" }, { "name": "Grace", "status": "active" } ],
    "returning": ["id"], "output": "data.created" } }

{ "name": "data_write", "input": {
    "connector": "orders_db", "op": "update", "target": "users",
    "set": { "status": "inactive" },
    "filter": { "==": [{ "field": "id" }, { "param": "id" }] },
    "params": { "id": { "var": "data.req.id" } },
    "output": "data.updated" } }

{ "name": "data_write", "input": {
    "connector": "orders_db", "op": "upsert", "target": "users",
    "values": { "email": "ada@x.io", "name": "Ada" },
    "on_conflict": { "target": ["email"], "action": "update" },
    "output": "data.upserted" } }
```

### Safety guards

- **Unfiltered mutations are rejected.** An `update`/`delete` with no `filter`
  would rewrite or truncate the whole table. It fails unless the envelope
  carries `"all": true` **and** the server enables `write.allow_unfiltered`
  (default `false`) — a deliberate double opt-in.
- **Bulk inserts are capped.** A `values` array longer than `write.max_rows`
  (default 1000) is rejected — never silently truncated.
- **Connector operation gates.** A connector's config can disable operation
  types entirely (`operations: { "delete": false }`) — see
  [Connector operation gates](#connector-operation-gates).

## Backend mapping

| `op` | SQL (sea-query) | MongoDB | Elasticsearch |
|------|-----------------|---------|---------------|
| query | `SELECT … WHERE …` | `find(filter)` | `POST {index}/_search` |
| `insert` | `INSERT INTO … VALUES …` | `insertOne` / `insertMany` | `POST {index}/_bulk` |
| `update` | `UPDATE … SET … WHERE …` | `updateMany(filter, {$set})` | `POST {index}/_update_by_query` (painless script) |
| `delete` | `DELETE FROM … WHERE …` | `deleteMany(filter)` | `POST {index}/_delete_by_query` |
| `upsert` | `ON CONFLICT … DO UPDATE / DO NOTHING` (PG/SQLite); `ON DUPLICATE KEY UPDATE` (MySQL) | `updateOne(…, upsert: true)` | `POST {index}/_update/{id}` (`doc_as_upsert`); `op_type=create` for `nothing` |

The `filter` of an update/delete goes through the exact same rendering path as
a query's `WHERE` — including relation predicates (`some` → SQL `EXISTS`).

### Parity or error

The dialect's governing rule: **match the reference semantics where a backend
can; raise a precise, located capability error where it cannot; never
approximate silently.** The notable divergences:

| Feature | Behavior |
|---------|----------|
| `returning` | Native on PostgreSQL/SQLite. On MySQL it is rejected (`FeatureUnsupportedByTarget`); single-row inserts report `last_insert_id` instead. On MongoDB inserts report generated `ids`. On Elasticsearch it is rejected; inserts report `ids` |
| `all` over an ES relation | Rejected (not set-equivalent on nested documents) |
| Deep ES pagination | `skip + limit` beyond `max_result_window` (10k) is rejected, not truncated |
| Bulk upsert on Mongo/ES | Rejected — single-row upserts only |
| ES upsert conflict target | Must resolve to the document `_id` (declare a schema rename); anything else is rejected |

### Elasticsearch notes

- **`_id` is explicit.** ES keys documents on the metadata `_id`, which lives
  outside the document source. There is no implicit `id` → `_id` mapping —
  declare it in the schema (`{"columns": {"id": {"name": "_id"}}}`). A physical
  `_id` column is then lifted into the bulk action / URL path on insert and
  upsert; mutating `_id` in `set` is rejected.
- **Read-your-writes.** Every ES write requests a refresh (`wait_for`, or
  `true` on the by-query endpoints), so a `data_query` later in the same
  pipeline sees the write — parity with SQL/Mongo visibility, at a throughput
  cost.
- **`_bulk` is ordered but non-transactional.** Any item error fails the call,
  though earlier items may already be applied. Version conflicts on
  `_update_by_query`/`_delete_by_query` surface as errors (`conflicts=abort`).

## Result shapes

Written to the task's `output` path:

| Backend | Shape |
|---------|-------|
| Query (all) | JSON array of rows/documents |
| SQL write | `{ "rows_affected": n }`, plus `"returning": [..]` where supported and `"last_insert_id": n` on MySQL single-row inserts |
| MongoDB / ES write | Doc-store keys per op: `{ "inserted": n, "ids": [..] }`, `{ "matched": n, "modified": n }` (+ `"upserted_id"` when created), `{ "deleted": n }` |

## The schema registry

Both functions accept an optional inline `schema` — **privileged configuration
authored alongside the workflow, never built from request input**. Without one,
the dialect runs in *identity mode*: names pass through as-is.

```json
"schema": {
  "entities": {
    "users": {
      "table": "app_users",
      "columns": {
        "id":     { "name": "user_id", "type": "int" },
        "email":  { "type": "string" },
        "secret": { "queryable": false, "writable": false }
      },
      "relations": {
        "orders": { "to": "orders", "kind": "has_many", "local": "id", "foreign": "user_id" }
      }
    }
  },
  "unmapped": "reject"
}
```

- **Renames** — logical entity/column names map to physical tables/columns
  (`id` → `user_id`, or `id` → `_id` for Elasticsearch).
- **Types** — drive value coercion where a backend needs the hint.
- **Allowlist** — with `"unmapped": "reject"`, only declared entities/columns
  are usable; `queryable: false` hides a column from reads, `writable: false`
  protects it from writes (generated/identity columns).
- **Relations** — declare `has_one` / `has_many` / `many_to_many` (the latter
  via `through`) so `some`/`all`/`none` predicates and `include` work.

## Relations and includes

With relations declared, a filter can quantify over related records:

```json
{ "some": [{ "field": "orders" }, { ">": [{ "field": "total" }, 100] }] }
```

renders as a correlated `EXISTS` (SQL), `$elemMatch` over embedded documents
(Mongo), or a `nested`/`has_child` query (ES). `include` fetches the related records
themselves, hydrated with one child query per relation:

```json
"include": { "orders": { "fields": ["id", "total"], "limit": 10 } }
```

## Connector operation gates

A `db` or `es` connector's config can en/disable operation types, regardless
of what workflows ask for. Everything defaults to allowed:

```json
{
  "name": "orders-db-readonly",
  "connector_type": "db",
  "config": {
    "type": "db",
    "connection_string": "postgres://…",
    "operations": {
      "read": true,
      "insert": true,
      "update": false,
      "delete": false,
      "upsert": false,
      "raw_write": false
    }
  }
}
```

| Gate | Blocks |
|------|--------|
| `read` | `data_query`, `db_read`, `mongo_read` |
| `insert` / `update` / `delete` / `upsert` | the matching `data_write` op |
| `raw_write` | the raw-SQL `db_write` escape hatch |

A gated call fails with a validation error naming the operation and connector
(`operation 'delete' is disabled on connector 'orders-db-readonly'`). Because
raw SQL cannot be classified per-statement, `db_write` has its own `raw_write`
gate — to make a connector fully delete-proof, disable both `delete` and
`raw_write`.

## Configuration

```toml
[query]                    # Page-size bounds for data_query
# default_limit = 100      # Page size when a query omits `limit`
# max_limit = 1000         # Hard cap; a query asking for more is rejected

[write]                    # Safety bounds for data_write
# max_rows = 1000          # Cap on rows per bulk insert/upsert (over → rejected)
# allow_unfiltered = false # Permit unfiltered update/delete (still needs per-call "all": true)
```

Both sections are overridable via environment variables
(`ORION_QUERY__MAX_LIMIT`, `ORION_WRITE__MAX_ROWS`, …).


---
source: ./tutorials/use-cases.md
---

# Use Cases & Patterns

Real-world examples showing how AI generates Orion workflows from natural language. Every example follows the same pattern: **describe what you need → AI generates the workflow → create a channel → send data → get results**.

## E-Commerce Order Classification

Classify orders into tiers and compute discounts.

**AI prompt:**

```
Create a workflow for the "orders" channel that:
1. Parses the payload into "order"
2. Assigns tiers based on amount:
   - VIP: amount >= 500, discount 15%
   - Premium: amount 100-500, discount 5%
   - Standard: amount < 100, no discount
```

**Generated workflow:**

```json
{
  "name": "Order Classification",
  "condition": true,
  "tasks": [
    { "id": "parse", "name": "Parse Payload", "function": {
        "name": "parse_json", "input": { "source": "payload", "target": "order" }
    }},
    { "id": "vip_tier", "name": "Set VIP Tier",
      "condition": { ">=": [{ "var": "data.order.amount" }, 500] },
      "function": { "name": "map", "input": { "mappings": [
        { "path": "data.order.tier", "logic": "vip" },
        { "path": "data.order.discount_pct", "logic": 15 }
      ]}}
    },
    { "id": "premium_tier", "name": "Set Premium Tier",
      "condition": { "and": [
        { ">=": [{ "var": "data.order.amount" }, 100] },
        { "<": [{ "var": "data.order.amount" }, 500] }
      ]},
      "function": { "name": "map", "input": { "mappings": [
        { "path": "data.order.tier", "logic": "premium" },
        { "path": "data.order.discount_pct", "logic": 5 }
      ]}}
    },
    { "id": "standard_tier", "name": "Set Standard Tier",
      "condition": { "<": [{ "var": "data.order.amount" }, 100] },
      "function": { "name": "map", "input": { "mappings": [
        { "path": "data.order.tier", "logic": "standard" },
        { "path": "data.order.discount_pct", "logic": 0 }
      ]}}
    }
  ]
}
```

**Send data:**

```bash
curl -s -X POST http://localhost:8080/api/v1/data/orders \
  -H "Content-Type: application/json" \
  -d '{ "data": { "amount": 750, "product": "Diamond Ring" } }'
```

**Response:**

```json
{
  "status": "ok",
  "data": {
    "order": { "amount": 750, "product": "Diamond Ring", "tier": "vip", "discount_pct": 15 }
  },
  "errors": []
}
```

**Key patterns:** Task-level conditions, computed output fields.

## IoT Sensor Alert Classification

Classify sensor readings into severity levels using range-based conditions.

**AI prompt:**

```
Create a workflow for the "sensors" channel that classifies temperature readings:
- Critical: temperature > 90 or below 0, set alert flag
- Warning: temperature 70-90, set alert flag
- Normal: temperature 0-70, no alert
Parse the payload into "reading" and set severity and alert fields.
```

**Generated workflow:**

```json
{
  "name": "Sensor Alert Pipeline",
  "condition": true,
  "tasks": [
    { "id": "parse", "name": "Parse Payload", "function": {
        "name": "parse_json", "input": { "source": "payload", "target": "reading" }
    }},
    { "id": "critical", "name": "Mark Critical",
      "condition": { "or": [
        { ">": [{ "var": "data.reading.temperature" }, 90] },
        { "<": [{ "var": "data.reading.temperature" }, 0] }
      ]},
      "function": { "name": "map", "input": { "mappings": [
        { "path": "data.reading.severity", "logic": "critical" },
        { "path": "data.reading.alert", "logic": true }
      ]}}
    },
    { "id": "warning", "name": "Mark Warning",
      "condition": { "and": [
        { ">": [{ "var": "data.reading.temperature" }, 70] },
        { "<=": [{ "var": "data.reading.temperature" }, 90] }
      ]},
      "function": { "name": "map", "input": { "mappings": [
        { "path": "data.reading.severity", "logic": "warning" },
        { "path": "data.reading.alert", "logic": true }
      ]}}
    },
    { "id": "normal", "name": "Mark Normal",
      "condition": { "and": [
        { ">=": [{ "var": "data.reading.temperature" }, 0] },
        { "<=": [{ "var": "data.reading.temperature" }, 70] }
      ]},
      "function": { "name": "map", "input": { "mappings": [
        { "path": "data.reading.severity", "logic": "normal" },
        { "path": "data.reading.alert", "logic": false }
      ]}}
    }
  ]
}
```

**Send data:**

```bash
curl -s -X POST http://localhost:8080/api/v1/data/sensors \
  -H "Content-Type: application/json" \
  -d '{ "data": { "temperature": 80, "sensor_id": "SENSOR-42" } }'
```

**Response:**

```json
{
  "status": "ok",
  "data": {
    "reading": { "temperature": 80, "sensor_id": "SENSOR-42", "severity": "warning", "alert": true }
  },
  "errors": []
}
```

| Input | Severity | Alert |
|-------|----------|-------|
| `temperature: 45` | normal | false |
| `temperature: 80` | warning | true |
| `temperature: 95` | critical | true |
| `temperature: -5` | critical | true |

**Key patterns:** Range-based classification with `and`/`or` conditions, boolean flags.

## Webhook Payload Transformation

Normalize incoming webhook payloads from different providers into a consistent internal schema.

**AI prompt:**

```
Create a workflow for the "webhooks" channel that normalizes webhook payloads from any provider:
- Map "origin" to "source"
- Map "type" to "event_type"
- Map "body" to "payload"
- Add a "processed" flag set to true
Output should be under data.normalized.
```

**Generated workflow:**

```json
{
  "name": "Webhook Transform Pipeline",
  "condition": true,
  "tasks": [
    { "id": "parse", "name": "Parse Payload", "function": {
        "name": "parse_json", "input": { "source": "payload", "target": "event" }
    }},
    { "id": "normalize", "name": "Normalize Schema", "function": {
        "name": "map", "input": { "mappings": [
          { "path": "data.normalized.source", "logic": { "var": "data.event.origin" } },
          { "path": "data.normalized.event_type", "logic": { "var": "data.event.type" } },
          { "path": "data.normalized.payload", "logic": { "var": "data.event.body" } },
          { "path": "data.normalized.processed", "logic": true }
        ]}
    }}
  ]
}
```

**Send data:**

```bash
curl -s -X POST http://localhost:8080/api/v1/data/webhooks \
  -H "Content-Type: application/json" \
  -d '{ "data": { "origin": "github", "type": "push", "body": {"ref": "refs/heads/main"} } }'
```

**Response:**

```json
{
  "status": "ok",
  "data": {
    "normalized": { "source": "github", "event_type": "push", "payload": {"ref": "refs/heads/main"}, "processed": true }
  },
  "errors": []
}
```

Missing optional fields produce `null`, no errors. This makes the pipeline safe for partial payloads from different webhook providers.

**Key patterns:** Schema mapping with `var`, null-safe field access, static enrichment.

## Notification Routing

Route notifications to different delivery channels based on severity.

**AI prompt:**

```
Create a workflow for the "notifications" channel that routes by severity:
- Log all notifications
- Send email for anything except "low" severity
- Send SMS only for "high" and "critical" severity
Parse the payload into "notification".
```

**Generated workflow:**

```json
{
  "name": "Notification Router",
  "condition": true,
  "tasks": [
    { "id": "parse", "name": "Parse Payload", "function": {
        "name": "parse_json", "input": { "source": "payload", "target": "notification" }
    }},
    { "id": "log_all", "name": "Log All Notifications", "function": {
        "name": "map", "input": { "mappings": [
          { "path": "data.notification.logged", "logic": true }
        ]}
    }},
    { "id": "email", "name": "Send Email",
      "condition": { "!=": [{ "var": "data.notification.severity" }, "low"] },
      "function": { "name": "map", "input": { "mappings": [
        { "path": "data.notification.email_sent", "logic": true }
      ]}}
    },
    { "id": "sms", "name": "Send SMS for High/Critical",
      "condition": { "in": [{ "var": "data.notification.severity" }, ["high", "critical"]] },
      "function": { "name": "map", "input": { "mappings": [
        { "path": "data.notification.sms_sent", "logic": true }
      ]}}
    }
  ]
}
```

**Send data:**

```bash
curl -s -X POST http://localhost:8080/api/v1/data/notifications \
  -H "Content-Type: application/json" \
  -d '{ "data": { "message": "Disk usage at 92%", "severity": "high" } }'
```

**Response:**

```json
{
  "status": "ok",
  "data": {
    "notification": { "message": "Disk usage at 92%", "severity": "high", "logged": true, "email_sent": true, "sms_sent": true }
  },
  "errors": []
}
```

| Severity | Logged | Email | SMS |
|----------|--------|-------|-----|
| low | yes | no | no |
| medium | yes | yes | no |
| high | yes | yes | yes |
| critical | yes | yes | yes |

In production, replace the `map` tasks with `http_call` tasks pointing to your email and SMS connectors.

**Key patterns:** Task-level condition gating, `in` operator for set membership, progressive pipeline.

## Compliance Risk Classification

Classify transactions by risk level and use dry-run testing to verify workflows before activating them.

**AI prompt:**

```
Create a workflow for the "compliance" channel that classifies transaction risk:
- High risk: amount > 10000, requires manual review
- Normal risk: amount <= 10000, no review needed
Parse the payload into "txn".
```

**Generated workflow:**

```json
{
  "name": "Risk Classifier",
  "condition": true,
  "tasks": [
    { "id": "parse", "name": "Parse Payload", "function": {
        "name": "parse_json", "input": { "source": "payload", "target": "txn" }
    }},
    { "id": "high_risk", "name": "Flag High Risk",
      "condition": { ">": [{ "var": "data.txn.amount" }, 10000] },
      "function": { "name": "map", "input": { "mappings": [
        { "path": "data.txn.risk_level", "logic": "high" },
        { "path": "data.txn.requires_review", "logic": true }
      ]}}
    },
    { "id": "normal_risk", "name": "Normal Risk",
      "condition": { "<=": [{ "var": "data.txn.amount" }, 10000] },
      "function": { "name": "map", "input": { "mappings": [
        { "path": "data.txn.risk_level", "logic": "normal" },
        { "path": "data.txn.requires_review", "logic": false }
      ]}}
    }
  ]
}
```

**Dry-run before going live:**

```bash
curl -s -X POST http://localhost:8080/api/v1/admin/workflows/<workflow-id>/test \
  -H "Content-Type: application/json" \
  -d '{"data": {"amount": 50000, "currency": "USD"}}'
```

```json
{
  "matched": true,
  "trace": { "steps": [
    { "task_id": "parse", "result": "executed" },
    { "task_id": "high_risk", "result": "executed" },
    { "task_id": "normal_risk", "result": "skipped" }
  ]},
  "output": { "txn": { "amount": 50000, "currency": "USD", "risk_level": "high", "requires_review": true } }
}
```

The trace shows exactly which tasks ran and which were skipped. Verify the logic is correct before a single real transaction flows through.

**Key patterns:** Dry-run verification, execution trace inspection, regulatory workflow.

## AI Workflow & CI/CD

AI writes workflows, not services. Instead of generating microservices that need their own governance, LLMs generate Orion workflows: constrained JSON that the platform validates, versions, and monitors automatically.

### Prompt Templates

Structure your LLM prompts to produce valid Orion workflows. Here's a reusable system prompt:

```
You generate Orion workflows in JSON format. Workflows have:
- name, condition (JSONLogic or true), continue_on_error (optional boolean)
- tasks: array of { id, name, condition (optional, JSONLogic), function: { name, input } }
- Every workflow starts with a parse_json task: { "name": "parse_json", "input": { "source": "payload", "target": "<entity>" } }
- Use "map" function with "mappings" array for transforms. Each mapping has "path" (dot notation) and "logic" (value or JSONLogic).
- Use "http_call" with "connector" (by name) for external API calls. Do not embed URLs or credentials in workflows.
- Use "channel_call" with "channel" (by name) for in-process inter-channel invocation.
- Task conditions use { "var": "data.<entity>.<field>" } to reference parsed data.

Output ONLY the JSON workflow. No explanation.
```

### Validation Pipeline

Every AI-generated workflow should go through this pipeline before reaching production:

1. **Generate:** use your LLM with the prompt template above
2. **Validate:** `POST /api/v1/admin/workflows/validate` to check structure
3. **Create as draft:** `POST /api/v1/admin/workflows` (workflows are created as drafts by default, not loaded into the engine)
4. **Dry-run:** `POST /api/v1/admin/workflows/{id}/test` with representative test data
5. **Check the trace:** verify the right tasks ran, the right ones were skipped, and output matches expectations
6. **Activate:** `PATCH /api/v1/admin/workflows/{id}/status` with `"status": "active"`

### CI/CD Pipeline

Integrate AI workflow generation into your deployment pipeline. Workflows are JSON files that version, diff, and review like any other config.

```
AI generates workflow → commit as JSON → CI runs dry-run → review → import
```

**GitHub Actions example:**

```yaml
name: Validate AI Workflows
on:
  pull_request:
    paths: ['workflows/**/*.json']

jobs:
  validate:
    runs-on: ubuntu-latest
    services:
      orion:
        image: ghcr.io/goplasmatic/orion:latest
        ports: ['8080:8080']
    steps:
      - uses: actions/checkout@v4

      - name: Import workflows (as drafts)
        run: |
          for file in workflows/**/*.json; do
            curl -s -X POST http://localhost:8080/api/v1/admin/workflows \
              -H "Content-Type: application/json" \
              -d @"$file"
          done

      - name: Dry-run test cases
        run: |
          for test in workflows/tests/**/*.json; do
            WORKFLOW_ID=$(jq -r '.workflow_id' "$test")
            DATA=$(jq -c '.data' "$test")
            EXPECTED=$(jq -c '.expected_output' "$test")

            RESULT=$(curl -s -X POST \
              "http://localhost:8080/api/v1/admin/workflows/${WORKFLOW_ID}/test" \
              -H "Content-Type: application/json" \
              -d "$DATA")

            OUTPUT=$(echo "$RESULT" | jq -c '.output')
            if [ "$OUTPUT" != "$EXPECTED" ]; then
              echo "FAIL: $test"
              echo "Expected: $EXPECTED"
              echo "Got: $OUTPUT"
              exit 1
            fi
          done

      - name: Deploy to production
        if: github.event_name == 'push' && github.ref == 'refs/heads/main'
        run: |
          for file in workflows/**/*.json; do
            curl -s -X POST "${{ secrets.ORION_URL }}/api/v1/admin/workflows" \
              -H "Content-Type: application/json" \
              -d @"$file"
          done
```

**Test case format:** store test cases alongside workflows:

```
workflows/
  fraud-detection.json       # The workflow
  tests/
    fraud-high-risk.json     # Test case
    fraud-clear.json         # Test case
```

Each test case:

```json
{
  "workflow_id": "fraud-detection",
  "data": { "data": { "amount": 15000, "country": "US" } },
  "expected_output": { "order": { "amount": 15000, "risk": "high", "requires_review": true } }
}
```

### Safety Guardrails

AI-generated workflows get the same governance as hand-written ones:

- **Version history:** every workflow change is recorded. Roll back if an AI-generated workflow misbehaves.
- **Draft status:** workflows are created as `draft` by default and are not loaded into the engine until explicitly activated.
- **Dry-run before activate:** test with representative data and inspect the full execution trace.
- **Audit trail:** every workflow version is recorded in the `workflows` table with incrementing version numbers.
- **Connectors isolate secrets:** AI generates workflows that reference connector names, never credentials.

## Common Workflow Patterns

### The parse-then-process pattern

Every workflow that reads input data must start with `parse_json`. Without it, task conditions referencing `data.X` see empty context.

```json
{
  "tasks": [
    { "id": "parse", "function": { "name": "parse_json", "input": { "source": "payload", "target": "order" } } },
    { "id": "process", "condition": { ">": [{ "var": "data.order.total" }, 100] }, "function": { "..." : "..." } }
  ]
}
```

### Task-level vs workflow-level conditions

- **Workflow-level condition:** determines whether the entire workflow matches. Set to `true` for workflows that always run.
- **Task-level condition:** determines whether a specific task within a matched workflow executes. Use for branching logic within a pipeline.

```json
{
  "condition": true,
  "tasks": [
    { "id": "always", "function": { "..." : "..." } },
    { "id": "conditional", "condition": { ">": [{ "var": "data.amount" }, 500] }, "function": { "..." : "..." } }
  ]
}
```

### External API calls with connectors

Keep credentials in connectors, reference them by name in workflows:

```json
{
  "tasks": [
    { "id": "parse", "function": { "name": "parse_json", "input": { "source": "payload", "target": "event" } } },
    { "id": "notify", "function": { "name": "http_call", "input": {
        "connector": "slack-webhook",
        "method": "POST",
        "body_logic": { "var": "data.event" }
    }}}
  ]
}
```

### Inter-channel composition with channel_call

Invoke another channel's workflow in-process for service composition:

```json
{
  "tasks": [
    { "id": "parse", "function": { "name": "parse_json", "input": { "source": "payload", "target": "order" } } },
    { "id": "enrich", "function": { "name": "channel_call", "input": {
        "channel": "customer-lookup",
        "data_logic": { "var": "data.order.customer_id" },
        "response_path": "data.customer"
    }}},
    { "id": "process", "condition": { "==": [{ "var": "data.customer.tier" }, "vip"] }, "function": { "..." : "..." } }
  ]
}
```


---
source: ./configuration/reference.md
---

# Config Reference

All settings have sensible defaults. You can run Orion with no config file at all. `orion-server` just works.

## CLI Commands

```bash
orion-server                                       # Start the server (default)
orion-server -c config.toml                        # Start with a config file
orion-server validate-config                       # Validate config without starting
orion-server validate-config -c config.toml        # Validate a specific config file
orion-server migrate                               # Run database migrations
orion-server migrate --dry-run                     # Preview pending migrations
orion-server lint path/to/workflow.json            # Strict-validate a workflow JSON file
orion-server dry-run -w workflow.json -i input.json # Execute a workflow against a sample payload
orion-server test-connectivity                     # Probe DB (and Kafka if enabled)
```

All subcommands honour `${VAR}` / `${VAR:-default}` substitution in the loaded config file, so the same `config.toml` can be reused across environments.

## Database Backend

The database backend is selected at runtime from the `storage.url` scheme. No rebuild needed:

| Backend | URL Format | Example |
|---------|------------|---------|
| **SQLite** | `sqlite:` | `sqlite:orion.db` or `sqlite::memory:` |
| **PostgreSQL** | `postgres://` | `postgres://user:pass@host/db` |
| **MySQL** | `mysql://` | `mysql://user:pass@host/db` |

```bash
# SQLite (default)
orion-server

# PostgreSQL
ORION_STORAGE__URL="postgres://user:pass@localhost/orion" orion-server

# MySQL
ORION_STORAGE__URL="mysql://user:pass@localhost/orion" orion-server
```

Migrations for all backends are embedded in the binary and the correct set is selected automatically at startup.

## Complete Config File

```toml
[server]
host = "0.0.0.0"
port = 8080
# shutdown_drain_secs = 30     # HTTP connection drain timeout on shutdown

# [server.tls]
# enabled = false
# cert_path = "cert.pem"
# key_path = "key.pem"

[storage]
url = "sqlite:orion.db"       # Database URL (sqlite:, postgres://, mysql://)
# max_connections = 25          # Connection pool max connections
# min_connections = 5           # Connection pool min connections
# busy_timeout_ms = 5000        # SQLite busy timeout (ignored for other backends)
# acquire_timeout_secs = 5      # Connection pool acquire timeout
# idle_timeout_secs = 0         # Connection idle timeout (0 = no timeout)
# backup_dir = "./backups"      # Directory where POST /api/v1/admin/backups writes files (SQLite only)

[ingest]
max_payload_size = 1048576     # Maximum payload size in bytes (1 MB)

[query]                        # Page-size bounds for the portable `data_query` handler
# default_limit = 100          # Page size when a query omits `limit`
# max_limit = 1000             # Hard cap; a query asking for more is rejected (never clamped)

[write]                        # Safety bounds for the portable `data_write` handler
# max_rows = 1000              # Hard cap on rows per bulk insert/upsert (over this is rejected)
# allow_unfiltered = false     # Permit unfiltered update/delete (still needs per-call "all": true)

[engine]
# health_check_timeout_secs = 2   # Timeout for engine read lock in health checks
# reload_timeout_secs = 10        # Timeout for engine write lock during reload
# max_channel_call_depth = 10     # Max recursion depth for channel_call
# default_channel_call_timeout_ms = 30000  # Default timeout for channel_call
# global_http_timeout_secs = 30   # Global timeout for HTTP client
# max_pool_cache_entries = 100    # Max cached connection pools for external connectors
# cache_cleanup_interval_secs = 60  # Cleanup interval for idle connection pools

# [engine.circuit_breaker]
# enabled = false                 # Enable circuit breakers for connectors
# failure_threshold = 5           # Failures before tripping the breaker
# recovery_timeout_secs = 30      # Cooldown before half-open probe
# max_breakers = 10000            # Max circuit breaker instances (LRU eviction)

[queue]
workers = 4                    # Concurrent async trace workers
buffer_size = 1000             # Channel buffer for pending traces
# shutdown_timeout_secs = 30    # Timeout for in-flight jobs during shutdown
# trace_retention_hours = 72    # How long to keep completed traces (0 = disabled)
# trace_cleanup_interval_secs = 3600  # Cleanup check interval
# processing_timeout_ms = 60000  # Per-trace processing timeout
# max_result_size_bytes = 1048576  # Max size of trace result (1 MB)
# max_queue_memory_bytes = 104857600  # Max memory for queued traces (100 MB)
# dlq_retry_enabled = true       # Enable dead letter queue retry
# dlq_max_retries = 5            # Max retry attempts for DLQ entries
# dlq_poll_interval_secs = 30    # DLQ retry poll interval

[rate_limit]
# enabled = false               # Enable platform-level rate limiting
# default_rps = 100             # Default requests per second
# default_burst = 50            # Default burst allowance

# [rate_limit.endpoints]
# admin_rps = 50                # Rate limit for admin routes
# data_rps = 200                # Rate limit for data routes

[admin_auth]
# enabled = false                          # Require authentication for admin endpoints
# api_keys = ["key-1", "key-2"]            # Accepted bearer tokens / API keys (any matching key authorises a request)
# header = "Authorization"                 # "Authorization" = Bearer format, other = raw key

[cors]
# allowed_origins = ["*"]       # Global CORS allowed origins

[kafka]
enabled = false
brokers = ["localhost:9092"]
group_id = "orion"
# processing_timeout_ms = 60000  # Per-message processing timeout
# max_inflight = 100             # Max in-flight messages
# lag_poll_interval_secs = 30    # Consumer lag poll interval

[[kafka.topics]]               # Map Kafka topics to channels
topic = "incoming-orders"
channel = "orders"

[kafka.dlq]
enabled = false
topic = "orion-dlq"

[logging]
level = "info"                 # trace, debug, info, warn, error
format = "pretty"              # pretty or json

[metrics]
enabled = false

[tracing]
# enabled = false                   # Enable OpenTelemetry trace export
# otlp_endpoint = "http://localhost:4317"  # OTLP gRPC endpoint
# service_name = "orion"            # Service name in traces
# sample_rate = 1.0                 # 0.0 (none) to 1.0 (all)
# debug_profile_enabled = false     # Allow per-request `X-Orion-Profile: 1` to attach `_orion.profile` to responses

[tracing.storage]
# Persistence policy for engine traces (rows in the `traces` table, surfaced via /api/v1/data/traces).
# Unrelated to the OpenTelemetry export above. Per-channel `config.tracing` overrides this default.
#   sync   — write inline before responding (strongest durability)
#   async  — enqueue to a bounded background queue (one DB write per task)
#   batch  — bounded queue, workers commit batch_size rows in one TX (highest throughput)
#   off    — skip persistence entirely
# mode = "sync"
# sample_rate = 1.0                  # 0.0–1.0 fraction of traces persisted
# errors_only = false                # Only persist traces that ended in an error
# max_pending = 10000                # Queue depth (async + batch modes)

[channels]                          # Control which channels this instance loads
# include = ["orders.*", "payments.*"]   # Glob patterns to include (empty = all)
# exclude = ["analytics.*"]              # Glob patterns to exclude
```

### Environment Variable Substitution

Values in `config.toml` may reference process environment variables with `${VAR}` (required, fails fast if unset) or `${VAR:-default}` (optional with a fallback). Use `$$` to escape a literal `$`. The same substitution also runs against connector `config_json` blobs at startup, so secrets can stay out of the database. The complementary `env://VAR_NAME` resolver kicks in **after** JSON parsing on connector string fields — `${VAR}` rewrites text, `env://` rewrites parsed values.

## Environment Variable Overrides

Override any setting with environment variables using double-underscore nesting:

```bash
ORION_SERVER__PORT=9090
ORION_STORAGE__URL="postgres://user:pass@localhost/orion"
ORION_KAFKA__ENABLED=true
ORION_LOGGING__FORMAT=json
ORION_RATE_LIMIT__ENABLED=true
ORION_ADMIN_AUTH__ENABLED=true
ORION_ADMIN_AUTH__API_KEYS="key-1,key-2"   # comma-separated; merged into admin_auth.api_keys
ORION_TRACING__ENABLED=true
ORION_TRACING__OTLP_ENDPOINT="http://jaeger:4317"
ORION_METRICS__ENABLED=true
```

Environment variables take precedence over config file values.

## Built-in Capabilities

All capabilities are compiled into a single binary and controlled at runtime:

| Capability | Configuration | Default |
|-----------|--------------|---------|
| Database backend | `storage.url` scheme | SQLite |
| Kafka | `kafka.enabled` | Disabled |
| OpenTelemetry | `tracing.enabled` | Disabled |
| TLS/HTTPS | `server.tls.enabled` | Disabled |
| Swagger UI | Always at `/docs` | Enabled |
| SQL connectors | `db_read`/`db_write` functions | Always available |
| Redis cache | `cache_read`/`cache_write` with Redis backend | Always available |
| MongoDB connector | `mongo_read` function | Always available |
| Portable data dialect | `data_query`/`data_write` against SQL, MongoDB, or Elasticsearch | Always available |
| Elasticsearch connector | `es` connector type (used by `data_query`/`data_write`) | Always available |
| Rate limiting | `rate_limit.enabled` | Disabled |
| Metrics | `metrics.enabled` | Disabled |

## Production Checklist

- Mount a persistent volume for `orion.db` (SQLite) or configure `storage.url` for PostgreSQL/MySQL
- Enable admin API authentication: `ORION_ADMIN_AUTH__ENABLED=true`
- Set structured logging: `ORION_LOGGING__FORMAT=json`
- Enable Prometheus metrics: `ORION_METRICS__ENABLED=true`
- Configure rate limiting for traffic protection (platform-level and per-channel)
- Use per-crate filtering: `RUST_LOG=orion=info`
- Enable OpenTelemetry: `ORION_TRACING__ENABLED=true` for distributed tracing
- Enable TLS via the `server.tls` section
- Enable circuit breakers: `engine.circuit_breaker.enabled = true`
- Set trace retention: `queue.trace_retention_hours` for trace data lifecycle management