amqp_api_seda — a reference-grade Rust AMQP/SEDA API template

amqp_api_seda is a reference-grade, staged event-driven (SEDA) API template in Rust. An event enters over HTTP and flows through a durable, decoupled pipeline: HTTP ingress → transactional outbox → RabbitMQ stream → router → topic exchange → per-domain quorum queues → workers. The messaging topology is config-driven (a single topology.toml), business logic is added through trait + registry handlers that auto-register with no central dispatch edit, and queues can be made useful with zero Rust by binding them to a shipped no-code passthrough handler. The result is durable, at-least-once delivery with idempotent processing, operable defaults, and a low-friction extension path.

For the full architecture (the stages, the outbox/relay, the at-least-once and idempotency model, and the design decisions behind each), see specs/sad.md.

Prerequisites

You only need two things:

Rust (stable toolchain) — install via rustup. Build and run with the standard cargo commands below.
Docker + Docker Compose — to run the local backbone (RabbitMQ with the stream plugin, MariaDB, Mailpit) defined in docker-compose.yml.

Copy the sample environment file before your first run; the documented defaults work for local development out of the box:

cp .env.example .env

Every variable in .env.example has a documented default in AppConfig::from_env (src/config.rs). For a local run you typically do not need to change anything — the defaults (DATABASE_URL, AMQP_URL, RABBITMQ_STREAM_URL) already point at the Compose services. The pipeline runs frictionless-local by default: plaintext, no TLS, and no credential setup beyond the Compose defaults (user/password app/app).

Quickstart: clone to a running pipeline

A reproducible walkthrough from a fresh clone to a processed event. Defaults are chosen so this just works locally.

1. Clone the repository

git clone <your-fork-or-repo-url> amqp_api_seda
cd amqp_api_seda

2. Create your environment file

cp .env.example .env

The defaults are sufficient for local development. If you want to override anything, the most commonly adjusted variables are:

Variable	Default	What it does
`INGRESS_MODE`	`stream`	Durable on-ramp: `stream` (router active) or `direct`.
`HTTP_HOST`	`0.0.0.0`	API bind host.
`HTTP_PORT`	`8080`	API bind port.
`DATABASE_URL`	`mysql://app:app@127.0.0.1:3306/amqp_api_seda`	Outbox + idempotency-inbox store (MariaDB).
`AMQP_URL`	`amqp://app:app@127.0.0.1:5672/%2f`	Broker AMQP 0-9-1 URL.
`RABBITMQ_STREAM_URL`	`rabbitmq-stream://app:app@127.0.0.1:5552/%2f`	RabbitMQ stream-protocol URL.
`TOPOLOGY_PATH`	`config/topology.toml`	The messaging topology loaded at startup.
`RUST_LOG`	`info`	`tracing-subscriber` log filter.

3. Bring up the local backbone

docker compose up -d

This starts the three services from docker-compose.yml:

rabbitmq (rabbitmq:4-management) — AMQP on 5672, the stream protocol on 5552, and the management UI on 15672 (login app/app). The stream plugin is enabled via the mounted rabbitmq/enabled_plugins.
mariadb (mariadb:11) — the SQL store on 3306 (database amqp_api_seda, user/password app/app).
mailpit (axllent/mailpit) — a dev SMTP catcher on 1025 with a web UI on 8025.

Check that the stack is healthy before continuing:

docker compose ps

4. Run the API

cargo run --bin api

On startup the API builds its database pool and applies migrations automatically (db::run_migrations, which embeds migrations/ via sqlx::migrate!) — so the outbox and processed_events tables are created for you. There is no separate migrate command to run. The API comes up even if the broker is down (the outbox absorbs writes; the relay publishes later), and serves ingest plus the ops endpoints on HTTP_PORT (default 8080).

5. Run the router and the worker (separate terminals)

cargo run --bin router    # terminal 2

cargo run --bin worker    # terminal 3

The router is active when INGRESS_MODE=stream (the default): it consumes the ingest stream and forwards each event to the topic exchange. In INGRESS_MODE=direct the relay publishes straight to the exchange and the router logs one line and exits 0. The worker consumes the per-domain quorum queues and dispatches each message to its configured handler. Both also run a headless ops listener on OPS_ADDR (default 0.0.0.0:9090).

6. Submit an event

The typed endpoint, POST /votes, takes a vote body and durably captures it under the fixed routing key votes.cast. bill_id must be >= 1 and value must be one of -1 (nay), 0 (abstain), 1 (yea). The optional Idempotency-Key header becomes the event's dedup token:

curl -i -X POST http://localhost:8080/votes \
  -H 'Content-Type: application/json' \
  -H 'Idempotency-Key: demo-vote-1' \
  -d '{"bill_id": 4271, "value": 1}'

You get back 202 Accepted with a JSON body {"event_token": "..."} once the event is durably committed.

You can also use the generic ingress, POST /events/{routing_key}, which accepts any JSON object under a validated routing key (lowercase dot-delimited segments). Use a key under the votes. namespace so it matches the votes queue's votes.# binding and is dispatched to the example_votes handler:

curl -i -X POST http://localhost:8080/events/votes.cast \
  -H 'Content-Type: application/json' \
  -H 'Idempotency-Key: demo-event-1' \
  -d '{"bill_id": 4271, "value": 1}'

7. Observe it being processed

The votes queue binds the topic pattern votes.#, so a votes.* event flows ingress → outbox → stream → router → exchange → the votes quorum queue → the worker. The generic POST /events/votes.cast call above carries the routing key votes.cast, which matches that binding — the worker dispatches it to the example_votes handler, which emits a structured log line. Watch the worker terminal for:

example_votes handled a vote event

Both entry points reach the same handler: the typed POST /votes maps every vote to the fixed key votes.cast, and the generic POST /events/votes.cast carries that key directly — both match the votes.# binding and are dispatched to example_votes. Use the typed endpoint when you want a validated, domain-specific request body, and the generic endpoint for arbitrary routing keys.

For operational visibility, every long-running component serves:

GET /health — liveness (always 200 while running).
GET /ready — readiness (200 when dependencies are reachable, else 503).
GET /metrics — Prometheus text exposition.

On the API these are on the main port (e.g. http://localhost:8080/metrics); on the router/worker they are on OPS_ADDR (e.g. http://localhost:9090/metrics):

curl http://localhost:8080/health
curl http://localhost:8080/ready
curl http://localhost:8080/metrics

Worked examples

The sample config/topology.toml wires both extension styles at once — typed handlers and the no-code path — so you can see both working on the shipped topology.

Typed handlers (Rust)

Two domains are bound to typed handlers that self-register with the worker:

Routing-key pattern	Queue	Handler	Source file
`votes.#`	`votes`	`example_votes`	`src/handlers/example_votes.rs`
`notify.#`	`notify`	`example_notify`	`src/handlers/example_notify.rs`

Submit a notification event and watch the worker terminal — example_notify processes it end to end:

curl -i -X POST http://localhost:8080/events/notify.welcome \
  -H 'Content-Type: application/json' \
  -H 'Idempotency-Key: demo-notify-1' \
  -d '{"recipient": "ada@example.com", "message": "Welcome aboard"}'

The worker logs:

example_notify handled a notification event

Both files are deliberately minimal — copy either one as the template when you add your own typed domain handler (see the recipe below).

The no-code path (zero Rust)

The audit queue is bound to the built-in logger handler purely by config — no Rust, no recompile. This is the demonstrated no-code proof: a queue becomes useful just by binding it to a shipped handler key in topology.toml.

[[queues]]
name = "audit"
routing_key = "audit.#"
handler = "logger"

Submit an audit event and watch the worker terminal — the built-in logger records it:

curl -i -X POST http://localhost:8080/events/audit.user_login \
  -H 'Content-Type: application/json' \
  -H 'Idempotency-Key: demo-audit-1' \
  -d '{"actor": "ada", "action": "login"}'

The worker logs (via src/handlers/logger.rs):

logger handled an event

Extension recipes

Three concise recipes covering the real extension mechanisms. Each maps to one specific way to grow the system.

Recipe 1 — Add a queue (config only)

Adding a queue is pure config: no recompile, as long as the queue binds to a handler key that already exists (a shipped typed handler, or a built-in no-code handler). Add a [[queues]] block to config/topology.toml.

This example uses the built-in webhook no-code handler, which POSTs each event's full envelope to a per-queue URL read from handler_config:

[[queues]]
name = "orders"
routing_key = "orders.#"
handler = "webhook"

  [queues.handler_config]
  # Point this at a receiver YOU control — e.g. a local HTTP sink you are running,
  # or a request-bin you own. Do NOT leave a dead/placeholder URL: a webhook queue
  # whose target never returns 2xx will fail, retry, and dead-letter every event.
  url = "http://127.0.0.1:9000/hook"
  timeout_ms = 10000

Restart the router and worker so the new queue is declared and consumed. The webhook handler is shown here as a recipe example, not an always-on golden-path queue — keep it pointed at a live receiver of your own.

Recipe 2 — Add a worker/handler (one file + one config binding)

To add typed business logic, implement the Handler trait in a new src/handlers/example_<domain>.rs, register it with one line, and bind a queue to it. Use src/handlers/example_notify.rs as the worked template.

Create the handler file src/handlers/example_orders.rs:

use async_trait::async_trait;

use crate::envelope::EventEnvelope;
use crate::errors::HandlerError;
use crate::handler::{Handler, HandlerCtx};
use crate::register_handler;

pub const HANDLER_KEY: &str = "orders";

pub struct ExampleOrdersHandler;

#[async_trait]
impl Handler for ExampleOrdersHandler {
    async fn handle(
        &self,
        envelope: &EventEnvelope,
        _ctx: &HandlerCtx,
    ) -> Result<(), HandlerError> {
        tracing::info!(
            handler = HANDLER_KEY,
            event_token = %envelope.event_token,
            routing_key = %envelope.routing_key,
            "example_orders handled an order event"
        );
        Ok(())
    }
}

// One-line auto-registration — no central dispatch list to edit.
register_handler!(HANDLER_KEY, ExampleOrdersHandler);

Wire the module in src/handlers/mod.rs. Add the module declaration and add its HANDLER_KEY to EXPECTED_HANDLER_KEYS. The EXPECTED_HANDLER_KEYS entry is required: without it the linker drops the handler's inventory::submit! static and the handler never resolves.

pub mod example_orders;

pub const EXPECTED_HANDLER_KEYS: &[&str] = &[
    example_notify::HANDLER_KEY,
    example_votes::HANDLER_KEY,
    example_orders::HANDLER_KEY,   // <-- add this line
    webhook::HANDLER_KEY,
    logger::HANDLER_KEY,
];

Bind a queue to it by config in config/topology.toml:

[[queues]]
name = "orders"
routing_key = "orders.#"
handler = "orders"

Rebuild and restart the worker; the new handler is auto-discovered and dispatched for its queue.

Recipe 3 — Add an endpoint

There are two ways, depending on whether you need typed validation.

Option A — by config (no-code)

For a named endpoint that durably ingests a JSON object under a fixed routing key, add an [[endpoints]] block to config/topology.toml and restart the api — no Rust, no recompile:

[[endpoints]]
path = "/orders"
routing_key = "orders.created"

POST /orders with any JSON object now durably captures it under orders.created (the same outbox path as the generic endpoint, just at a stable named URL). Point the routing_key at a queue's binding so it reaches a handler — if it matches no queue binding, the api logs a startup warning (the event is still durably stored, but is dropped after the relay publishes it, since nothing subscribes). Validated at startup: the path must start with /, be static (no {}/*), and not collide with a built-in route (/votes, /health, /ready, /metrics, or the /events/ prefix). This is fixed-routing-key, JSON-object ingress only — there is no per-field validation.

You don't even need a named endpoint to ingest a new domain: POST /events/{routing_key} already accepts any routing key. Config endpoints just give you a stable, documented URL.

Option B — typed (Rust)

When you need a validated, typed request body (like /votes' bill_id/value checks), add an Axum route in src/api/routes/mod.rs. The router builder is the routes() function:

pub fn routes(endpoints: &[EndpointCfg]) -> Router<AppState> {
    let mut router = Router::new()
        .route("/events/{routing_key}", post(events::ingest_event))
        .route("/votes", post(votes::cast_vote));
        // Add your typed route here, e.g.:
        // .route("/orders", post(orders::create_order))
    // ... config-driven [[endpoints]] are appended here ...
    router
}

Model your handler on the typed POST /votes: deserialize + validate a typed body, map it to a fixed routing key, then ingest through the shared events::ingest_object path — which keeps every endpoint durable and at-least-once by construction.

Production

The golden path above runs frictionless-local by default: plaintext transports, no TLS, and no credential setup beyond the Compose defaults. TLS is configurable and optional — enabled purely by configuration (cert/key paths and a broker CA path), with no code change; when those settings are absent the components serve plain HTTP and connect over plaintext.

For production hardening — TLS on every connection, least-privilege RabbitMQ credentials, scaling the workers/relay/router, and high-availability configuration — see the Production Hardening Guide.

Name		Name	Last commit message	Last commit date
.cargo		.cargo
.claude		.claude
.github		.github
config		config
deploy		deploy
docs		docs
migrations		migrations
rabbitmq		rabbitmq
specs		specs
src		src
tests		tests
.claudeignore		.claudeignore
.env		.env
.env.example		.env.example
.env.production.example		.env.production.example
.gitignore		.gitignore
AGENTS.md		AGENTS.md
CLAUDE.md		CLAUDE.md
Cargo.lock		Cargo.lock
Cargo.toml		Cargo.toml
New Text Document.txt		New Text Document.txt
README.md		README.md
docker-compose.yml		docker-compose.yml
project-instructions.md		project-instructions.md

Navigation Menu

Repository files navigation

amqp_api_seda — a reference-grade Rust AMQP/SEDA API template

Prerequisites

Quickstart: clone to a running pipeline

1. Clone the repository

2. Create your environment file

3. Bring up the local backbone

4. Run the API

5. Run the router and the worker (separate terminals)

6. Submit an event

7. Observe it being processed

Worked examples

Typed handlers (Rust)

The no-code path (zero Rust)

Extension recipes

Recipe 1 — Add a queue (config only)

Recipe 2 — Add a worker/handler (one file + one config binding)

Recipe 3 — Add an endpoint

Option A — by config (no-code)

Option B — typed (Rust)

Production

About

Releases

Packages

Contributors

Languages

Latest commit

History

Folders and files

Repository files navigation

amqp_api_seda — a reference-grade Rust AMQP/SEDA API template

Prerequisites

Quickstart: clone to a running pipeline

1. Clone the repository

2. Create your environment file

3. Bring up the local backbone

4. Run the API

5. Run the router and the worker (separate terminals)

6. Submit an event

7. Observe it being processed

Worked examples

Typed handlers (Rust)

The no-code path (zero Rust)

Extension recipes

Recipe 1 — Add a queue (config only)

Recipe 2 — Add a worker/handler (one file + one config binding)

Recipe 3 — Add an endpoint

Option A — by config (no-code)

Option B — typed (Rust)

Production

About

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages