# noema-actix-webapi
Actix-web backend runtime on [Noema](../noema): modules, Postgres, unit of work, Swagger, WebSocket → `subscribe!`.
```rust
use actix_web::{App, HttpServer};
use noema::MapSource;
use noema_actix_webapi::prelude::*;
#[actix_web::main]
async fn main() -> Result<(), noema_actix_webapi::Error> {
let modules = Modules::new().add::<UsersModule>();
let app = start(
modules,
MapSource::new([("DATABASE_URL", "postgres://localhost/app")]),
ApplicationConfig {
http: HttpConfig { port: 8080, ..Default::default() },
..Default::default()
},
)
.await?
.health()
.ready();
HttpServer::new(move || {
App::new()
.wrap(request_context::<()>())
.wrap(cors())
.configure(|cfg| app.configure(cfg))
})
.bind(app.bind_addr())?
.run()
.await?;
Ok(())
}
```
`start` / `Application::start(modules, source, config)` reads required `DATABASE_URL` from the source, installs `ApplicationConfig` (singleton), Postgres, migrators, infra, logger. It does **not** run Actix. You own `HttpServer`. Swagger UI and `{scope}/docs/openapi.json` are always mounted but gated by `Presentation::authorize_docs` (default **deny**). `.health()` / `.ready()` are opt-in. CORS is a wrap on **your** `App`, not inside `configure`.
Mount WebSockets on the module scope (as many routes / transports as you want):
```rust
fn configure_scope(&self, scope: actix_web::Scope) -> actix_web::Scope {
scope
.route("/ws", web::get().to(connect::<ChatWs>))
.route("/live", web::get().to(connect::<LiveWs>))
}
```
`ChatWs` is **your** type: `EventDispatcherContext` (use `noema_actix_webapi::actix::dispatch_context()`), `subscribe!(ChatWs, …)`, `dependency!(singleton, ChatWs)`, and `WsConnection`.
Auth runs **once** on handshake. Event handlers do not re-check the token. `SessionContext<T>` / `RequestContext<T>` bind **one** `T` per task (the wrap / `connect::<T>()`). Mixing extras on the same task is unsupported.
```rust
struct PlayerSession { player_id: Uuid }
#[async_trait::async_trait(?Send)]
impl WsConnection for ChatWs {
type Ctx = PlayerSession;
async fn on_connect(&self, req: &HttpRequest) -> Result<Self::Ctx, WsError> {
let token = bearer(req).ok_or_else(|| WsError::unauthorized("missing token"))?;
Ok(self.auth.decode(token).map_err(|_| WsError::unauthorized("invalid"))?)
}
// Default already replies `{ "name": "error", "data": ErrorBody }`. Override to no-op:
// async fn on_dispatch_error(&self, _err: &(dyn std::error::Error + Send + Sync + 'static), _event_name: &str) {}
}
// In an EventListener — reply to this socket; read Ctx only if you need identity:
async fn handle(&self, event: Arc<ChatSend>) -> NoemaResult<()> {
let hub = resolve::<SessionHub>();
hub.join_current("lobby");
hub.reply(&ChatAck { ok: true });
let player_id = resolve::<dyn SessionContext<ChatWs> + Send + Sync>()
.ctx()
.player_id;
Ok(())
}
```
Incoming text is `{ "name", "data" }` → `dispatch`. Outgoing hub messages are `{ name: E::WIRE_NAME, data }`. Handler `Err`, unknown `name`, and invalid JSON go to `WsConnection::on_dispatch_error` (default: `{ "name": "error", "data": ErrorBody }` on **this** socket). Override the method to silence or remap.
Rooms on the hub are **local last-mile** (which sockets on this process get the frame). Membership for the product lives in your store. Multi-pod: `broadcast_event` sends locally then, if you registered a hook, publishes a `WsFanoutMessage` (`origin`, `room`, `body`). Other pods deserialize, skip `is_local()`, then `broadcast_raw` (does **not** publish again).
```rust
app.on_ws_publish(|room, envelope| {
// redis.publish(room, envelope) — spawn if your client is async
let _ = (room, envelope);
});
// subscriber BackgroundTask:
let msg: WsFanoutMessage = serde_json::from_str(&payload)?;
if msg.is_local() { return; }
resolve::<SessionHub>().broadcast_raw(&msg.room, msg.body);
```
`process_origin()` is a UUID v7 generated once in `start` (so a bus subscriber can skip this process). Not configurable.
## Layout
| Piece | Role |
|-------|------|
| `Application::start` / `start` | `DATABASE_URL` from source + `ApplicationConfig` → pool → migrators → infra → logger |
| `Application::configure` | scopes + **Swagger** (+ optional `/health` `/ready`) |
| `Module` / `Presentation` / `Infrastructure` | scopes, OpenAPI, `migrator()` |
| `PgPool` | **one** process pool from `DATABASE_URL` (default max 32, acquire 5s). A read replica / snapshot is an app type + its own sqlx pool, not a second `PgPool` |
| `UnitOfWork` / `with_transaction!` | opaque `Tx`; infra uses `db::postgres_tx` |
| `/swagger-ui` | always mounted; gated by `authorize_docs` (default deny); one tab per module |
| `ws::connect::<T>()` | per-route WS; `on_connect` then `{ name, data }` → `dispatch`; errors → `on_dispatch_error` |
| `SessionHub` | local rooms + `broadcast_event` / `broadcast_raw`; opt-in `on_ws_publish` |
| `Session::<T>::get()` | connect-time `Ctx` during dispatch |
| `SessionContext<T>` | injectable port; `resolve` reads the bound session |
| `RequestScope<T>` | opt-in wrap snapshot; `id` + idempotency + extra |
| `RequestContext<T>` | injectable port; `resolve` reads the bound scope |
| `Logger` | `resolve::<dyn Logger + Send + Sync>()` (default `TracingLogger`) — do not also `dependency_as!` `Logger` in the app |
| `Hasher` | `resolve::<dyn Hasher + Send + Sync>()` (default `Argon2Hasher`) — same: one binding in this crate; another algorithm is **your** type, not a second `Hasher` |
| `Clock` | `resolve::<dyn Clock + Send + Sync>()` (default `SystemClock`) |
| `HttpClient` | `resolve::<dyn HttpClient + Send + Sync>()` — **one** pooled `reqwest::Client` |
| `cors()` / `cors_from` | opt-in wrap from `CorsConfig`; not mounted in `configure` |
| `PageRequest` / `PageResult<T>` | pagination query + JSON (`ToSchema`) |
| `HttpResult` / `HttpError` | presentation routes: `send()?`; JSON `{ code, message, details? }` |
| `BackgroundTask` / `spawn_background_tasks` | process-lifetime loops (`init` then `run`); pass instances, no many-batch |
| `send` | mediator (prelude); route validates, then `send(input)` |
Application prelude: `noema_actix_webapi::prelude`. Do not import `db::postgres_tx` there.
## Unit of work
`DATABASE_URL` is required on the `ConfigSource` passed to `start` (`EnvSource`, `MapSource`, …). Pool size is `ApplicationConfig.database` (`max_connections` default 32, `acquire_timeout_ms` default 5000). Size the pool under Postgres `max_connections` and against Actix workers × in-flight queries. A full pool waits up to the acquire timeout then surfaces `PoolTimedOut` (`503`). `UnitOfWork` uses this pool only. For a read replica, open another `sqlx::PgPool` in the app and inject it as your own type. `resolve::<ApplicationConfig>()` after `start`.
```rust
with_transaction!(tx, {
users.save(tx, &user).await?;
Ok(())
});
let uow = resolve::<dyn UnitOfWork + Send + Sync>();
uow.transaction(Box::new(|tx| Box::pin(async move { Ok(()) }))).await?;
```
## Logger and hasher
`LogConfig` (on `ApplicationConfig`): `level` (default `info`), optional `file`, `stdout` (default `true`), `json` (default `false`).
This crate already `dependency_as!` `Logger` → `TracingLogger` and `Hasher` → `Argon2Hasher`. A second `dependency_as!(singleton, Hasher: …)` in the app **does not compile**. Use `resolve::<dyn Hasher + Send + Sync>()` for passwords, or your own port if you need a different algorithm.
```rust
let log = resolve::<dyn Logger + Send + Sync>();
log.info("send CreateUser");
let hasher = resolve::<dyn Hasher + Send + Sync>();
let hashed = hasher.hash(b"secret");
assert!(hasher.verify(b"secret", &hashed));
```
HTTP handlers stay thin: validate input, then `send(CreateUser { .. }).await`.
## Errors
`HttpResult` / `HttpError` are **presentation** (module routes), not domain. Handlers keep returning their own errors; `send` boxes them. Routes:
```rust
async fn create(body: web::Json<CreateUser>) -> HttpResult {
let page: PageRequest = /* query */;
page.validate()?;
let out = send(body.into_inner()).await?;
Ok(HttpResponse::Ok().json(out))
}
```
Map a module domain error in presentation: `impl From<UserError> for HttpError` using `MappedError::not_found` / `conflict` / … .
`map_error` understands `ValidationError` (`400` `validation.failed`) and `sqlx::Error`: `RowNotFound` → `404` `not_found` (typical of `fetch_one` with zero rows; `fetch_optional` is `Ok(None)` and the app maps that itself); pool timeout/closed/crashed → `503` `infrastructure.unavailable`; other sqlx → `502` `infrastructure.database`. sqlx `details.reason` (not 404) is included only when `ApplicationConfig.http.env` is `Environment::Development` (default `Production`). `/ready` ping failures follow the same rule. Anything else is `500` `internal`.
WebSocket: the connect loop calls `WsConnection::on_dispatch_error`. The default uses `error_ws_envelope` → `{ "name": "error", "data": ErrorBody }` on the socket that sent the frame. Invalid JSON uses `code` `bad_request`; a handler `MappedError` keeps its code (e.g. `bad_request`). Override `on_dispatch_error` to drop or customize the reply.
## Pagination and validation
`PageRequest` query params `page` (default 1) and `page_size` (default 20, max 100). `PageResult { items, page, page_size, total }` — both have utoipa schemas.
```rust
page.validate()?;
require_non_empty(&name, "name")?;
require_email(&email, "email")?;
```
## Clock
```rust
let clock = resolve::<dyn Clock + Send + Sync>();
let _now = clock.now();
```
## HTTP client
One process-wide `reqwest::Client` (connection pool **per host** + TLS reuse). Do not build a new client per call. Do not retry. If too many `send`s are in flight, the extra call fails immediately (`HttpClientError::is_busy`).
`HttpClientConfig` (on `ApplicationConfig`): `timeout_ms` (30000), `connect_timeout_ms` (10000), `pool_max_idle_per_host` (32), `pool_idle_timeout_secs` (optional), `user_agent` (`noema-actix-webapi`), `max_in_flight` (64; `0` = unlimited).
```rust
let http = resolve::<dyn HttpClient + Send + Sync>();
let resp = http.send(OutboundRequest::get("https://example.com/v1/x")).await?;
let created = http
.send(OutboundRequest::post("https://example.com/v1/x").json(&body)?)
.await?;
```
Unit tests inject a fake `HttpClient` (no reqwest).
## CORS
Opt-in wrap, same as `request_context`. Empty `origins` (`Vec`, the default) allows none. `"*"` in the list allows any origin (credentials are skipped). With `*`, Actix echoes the request `Origin` header (it is not the literal `*`).
`CorsConfig`: `origins`, `methods`, `headers` as `Vec<String>`; `credentials` (default `false`); `max_age` (3600).
Last `.wrap()` is outermost. Put `cors()` **after** `request_context` so a 401 from `from_request` still gets `Access-Control-Allow-Origin`. `connect()` also stamps the same policy on the handshake (101 and `WsError`) because `actix-cors` does not reliably cover WebSocket upgrades.
```rust
App::new()
.wrap(request_context::<Principal>())
.wrap(cors())
.configure(|cfg| app.configure(cfg))
```
## Swagger
Mounted by `configure`. Default `authorize_docs` is **deny** (`401`). The UI requires every module to allow; `{scope}/docs/openapi.json` uses that module only.
```rust
fn authorize_docs(&self, req: &HttpRequest) -> Result<(), HttpError> {
// local: Ok(())
// prod: check a header, cookie, or your auth
let _ = req;
Ok(())
}
```
## Background tasks
Pass the loops after `start` (does not run `HttpServer`). Each task `init`s then `run`s; if `init` fails, `run` is skipped.
```rust
let app = start(modules, source, ApplicationConfig::default()).await?;
app.spawn_background_tasks([
background_task(KafkaConsumer { .. }),
background_task(MetricsPoller { .. }),
]);
```
## Request context
Opt-in wrap. The crate fills `id` and `Idempotency-Key`; `T` is yours (`()` if you only need the frame). The wrap and `connect::<T>()` each bind **one** extra type per task.
```rust
App::new()
.wrap(request_context::<Principal>())
.configure(|cfg| app.configure(cfg))
// production handler field:
// ctx: Arc<dyn RequestContext<Principal> + Send + Sync>
let ctx = resolve::<dyn RequestContext<Principal> + Send + Sync>();
ctx.id();
ctx.idempotency_key();
ctx.extra();
```
Unit tests inject a double (no Actix):
```rust
let h = CreateUserHandler {
ctx: Arc::new(RequestScope::new(id, Some("k".into()), Principal::User { .. })),
};
```
Same for WebSocket: `Arc<dyn SessionContext<ChatWs> + Send + Sync>` in production via `resolve`, `Session::new(id, ctx)` in tests.