noema-actix-webapi

Actix-web backend runtime on Noema: modules, Postgres, unit of work, Swagger, WebSocket → subscribe!.

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):

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.

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).

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

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.

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.

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:

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.

page.validate()?;
require_non_empty(&name, "name")?;
require_email(&email, "email")?;

Clock

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 sends 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).

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.

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.

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 inits then runs; if init fails, run is skipped.

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.

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):

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.