arche

The opinionated backend foundation for Axum applications.

Cloud integrations, databases, auth, LLM inference, tool-calling agents, encryption, streaming JSON/CSV, WebSockets, and structured error handling — wired up and ready to go.

arche sits around Axum, not in place of it.

Getting Started · Modules · API Reference · Design Principles

Why arche?

Every backend service re-implements the same infrastructure plumbing — cloud SDK setup, database pools, auth primitives, error handling, config resolution. arche bundles these into a single, cohesive Rust crate built on well-established libraries so you can skip the boilerplate and focus on business logic.

Getting Started

Add arche to your Cargo.toml:

[dependencies]
arche = "4.15.0"

Modules

Module	What it does
`aws`	S3, SES, KMS, and CloudFront via official AWS SDKs
`gcp`	Generic GCP REST client; wrappers for Sheets, Drive, Cloud KMS, Cloud Storage, Cloud CDN, and Google OAuth login
`oidc`	OpenID Connect both ways — "Sign in with Google" client + build-your-own identity provider (authorization-code + PKCE, RS256)
`senno`	LLM client + tool-calling agents — the re-exported senno crate (Vertex AI: Gemini + Claude)
`database`	Postgres, Redis, and ClickHouse connection pooling with health checks
`queue`	Kafka producer/consumer with a raw message stream, explicit offset commits, and health checks
`jwt`	HS256 token generation, verification, and expiry helpers
`csv`	Async CSV read/write — batch, streaming, and from URL
`json`	Streaming JSON array parsing with metadata extraction
`crypto`	AES-128-CBC encryption with PBKDF2 key derivation
`sockets`	WebSocket connection registry with broadcast
`error`	Axum-compatible structured error responses (400–503)
`middleware`	Axum layers — rewrites extractor rejections into `AppError` JSON so clients see one error contract
`utils`	Alphanumeric nano IDs, timestamp validation, date/time conversions, pagination

Tip

Every service module exports a config builder to wire up credentials programmatically — or omit it entirely (pass None) and let arche resolve everything from environment variables.

// Pass None to resolve entirely from env vars
let pool = arche::database::pg::get_pg_pool(None).await?;

// Or configure explicitly
let config = arche::database::pg::PgConfigBuilder::default()
    .host(Some("localhost".into()))
    .port(Some(5432))
    .build();
let pool = arche::database::pg::get_pg_pool(config).await?;

All components are modular and explicit — nothing is hidden or magical.

API Reference

AWS

AWS SDK integrations built on official SDKs. Default region: ap-south-1.

S3

use arche::aws::s3::{get_s3_client, S3ConfigBuilder};

// From env vars
let client = get_s3_client(None).await?;

// Or with explicit config
let config = S3ConfigBuilder::default()
    .credential_source(Some("env".into()))
    .access_key_id(Some("AKIA...".into()))
    .secret_access_key(Some("secret".into()))
    .build();
let client = get_s3_client(config).await?;

Env Var	Description
`S3_CRED_SOURCE`	`"IAM"` (default) or `"env"`
`S3_ACCESS_KEY_ID`	Required when source is `"env"`
`S3_SECRET_ACCESS_KEY`	Required when source is `"env"`
`S3_REGION`	AWS region (default: `ap-south-1`)

KMS

use arche::aws::kms::KMSClient;

// Default region
let kms = KMSClient::new_with_region("ap-south-1").await;

// Encrypt / decrypt
let ciphertext = kms.encrypt("alias/my-key", b"sensitive data").await?;
let plaintext = kms.decrypt(&ciphertext).await?;

// Decrypt base64-encoded ciphertext directly
let plaintext = kms.decrypt_base64("base64string...").await?;

Env Var	Description
`AWS_REGION`	AWS region (default: `ap-south-1`)

SES

use arche::aws::ses::SESClient;

let ses = SESClient::new_with_region("ap-south-1").await;

// Plain email (with optional HTML body)
let message_id = ses.send_email(
    "from@example.com",
    "to@example.com",
    "Subject line",
    "Plain text body",
    Some("<h1>HTML body</h1>"),
).await?;

// Templated email
let message_id = ses.send_templated_email(
    "from@example.com",
    "to@example.com",
    "TemplateName",
    r#"{"name": "Alice"}"#,
).await?;

Env Var	Description
`AWS_REGION`	AWS region (default: `ap-south-1`)

CloudFront

use arche::aws::cloudfront::{get_cloudfront_client, CloudFrontClient, CloudFrontConfigBuilder};

let aws = get_cloudfront_client(None).await;
let cf = CloudFrontClient::new(aws, None);

Invalidate paths — submits a CloudFront invalidation and returns immediately with the invalidation ID and status (typically "InProgress").

let result = cf.invalidate_paths(
    Some("E1ABCXYZ"),
    vec!["/index.html".into(), "/assets/*".into()],
    None, // caller_reference: None auto-generates a nanoid (not retry-safe); pass a stable value for idempotent retries
).await?;
println!("{} -> {}", result.id, result.status);

Per CloudFront limits: paths must start with /, max 3000 paths per call, caller reference ≤ 128 chars.

Note

caller_reference: None auto-generates a fresh nanoid on every call, so a retried request creates a duplicate invalidation. Pass a stable value for idempotent retries.

Get invalidation status — fetch the current status of a previously created invalidation. Returns the same InvalidationResult shape; status transitions from "InProgress" to "Completed" (typically 5–15 minutes).

let status = cf.get_invalidation(Some("E1ABCXYZ"), &result.id).await?;
println!("{}", status.status);

Default distribution ID — set once on the client (or via CLOUDFRONT_DISTRIBUTION_ID env) so per-call distribution_id can be None:

let config = CloudFrontConfigBuilder::default()
    .distribution_id("E1ABCXYZ")
    .build();
let cf = CloudFrontClient::new(aws, config);

cf.invalidate_paths(None, vec!["/index.html".into()], None).await?;
cf.get_invalidation(None, "I2J3K4L5...").await?;

Env Var	Description
`AWS_REGION`	AWS region (default: `ap-south-1`)
`CLOUDFRONT_DISTRIBUTION_ID`	Optional default distribution ID

GCP

Service-account-authenticated REST client for any Google Cloud API, plus ergonomic wrappers for Sheets, Drive, and Vertex AI. Built on reqwest — honors HTTPS_PROXY / NO_PROXY like everything else.

Service account credentials

ServiceAccountKey is the canonical credential type. Two ways to construct:

use arche::gcp::ServiceAccountKey;

// From individual fields (e.g. separate env vars or a secrets manager)
let key = ServiceAccountKey::new(client_email, private_key);

// Or from a GCP service-account JSON file on disk
let key = ServiceAccountKey::from_path("/etc/secrets/sa.json").await?;

\n literals from .env-style storage are normalized to real newlines automatically. The private_key is never readable back from the struct and is masked in Debug output.

Authentication modes

Every GcpClient constructor accepts credentials in three forms, tried in order:

Explicit ServiceAccountKey — GcpClient::new(Some(key), None, scopes).
Path to a service-account JSON file — GcpClient::new(None, Some(path), scopes).
Neither → GKE / GCE metadata server (Workload Identity). When both are None, arche falls back to ${GCP_METADATA_URL:-http://metadata.google.internal} and exchanges the pod's bound service account for an OAuth token. This is the standard no-secrets-on-disk flow for pods running on GKE, Cloud Run, or GCE.

// Workload Identity — no creds passed in, no env vars required on GKE.
let kms = arche::gcp::kms::get_kms_client(None, None, None).await?;

Tokens are cached with a 60 s safety margin, single-flighted per scope set, and retried once on transient failures — uniformly across all three modes.

Warning

Caveats for the metadata-server path:

Scopes are ignored — the endpoint returns whatever scopes the pod's bound service account has. Need narrower scopes? Use an explicit ServiceAccountKey.
GCP_METADATA_URL is read once, at construction. Changing it later has no effect — set it before the first GcpClient::new(None, None, …) call (handy for pointing tests at a mock).
Signed URLs are unsupported — V4 signing (GcsClient::sign_*) needs the SA's private key, which the metadata server never exposes. Those calls return a clear error here.

Sheets

let sheets = arche::gcp::sheets::client(Some(key), None).await?;

let resp = sheets
    .get(format!(
        "https://sheets.googleapis.com/v4/spreadsheets/{spreadsheet_id}/values/{range}",
    ))
    .await?
    .send()
    .await?;

Note

Pass either Some(key) or Some(path) — never both. Scope is preset to https://www.googleapis.com/auth/spreadsheets.

Drive

let drive = arche::gcp::drive::client(None, Some("/etc/secrets/sa.json".into())).await?;

let bytes = drive
    .get(format!("https://www.googleapis.com/drive/v3/files/{file_id}?alt=media"))
    .await?
    .send().await?
    .bytes().await?;

Scope is preset to https://www.googleapis.com/auth/drive.

Cloud KMS

Encrypt and decrypt against Google Cloud KMS using the same service-account JWT auth as the rest of the GCP family — token caching, retries, and concurrent-fetch deduplication come for free via GcpClient.

use arche::gcp::kms::{get_kms_client, GcpKmsConfig, GcpKmsKey};

// Build the client. Any unset field falls back to its GCP_KMS_* env var.
let kms = get_kms_client(
    Some(key),
    None,
    GcpKmsConfig::builder().project_id("my-project").build(),
).await?;

// Or fully env-driven (project_id required; location defaults to "global"):
let kms = get_kms_client(Some(key), None, None).await?;

// Key identifier — passed per call so one client can target multiple keys
let kms_key = GcpKmsKey::new("my-keyring", "my-key");
// or: let kms_key = GcpKmsKey::from_env()?;  // GCP_KMS_KEY_RING + GCP_KMS_KEY_NAME

// Encrypt — returns ciphertext + the key version that wrapped it
let out = kms.encrypt(&kms_key, b"sensitive data").await?;
// out.ciphertext: Vec<u8>
// out.key_version: e.g. "projects/.../cryptoKeys/my-key/cryptoKeyVersions/3"
//   Persist this alongside the ciphertext for key-rotation auditing.

let plaintext = kms.decrypt(&kms_key, &out.ciphertext).await?;

// If the ciphertext is already base64 (e.g. read from a DB column):
let plaintext = kms.decrypt_base64(&kms_key, &b64_string).await?;

Env Var	Description
`GCP_KMS_PROJECT_ID`	GCP project hosting the KMS key (required)
`GCP_KMS_LOCATION`	KMS location (default: `global`)
`GCP_KMS_BASE_URL`	Override the Cloud KMS endpoint (testing / VPC-SC)
`GCP_KMS_KEY_RING`	Used by `GcpKmsKey::from_env()`
`GCP_KMS_KEY_NAME`	Used by `GcpKmsKey::from_env()`

Already have a GcpClient configured for other services? Reuse it via the exported scope — keeps a single token cache across Sheets / Drive / KMS:

let kms_gcp = my_gcp_client.with_scopes([arche::gcp::kms::KMS_SCOPE]);
let kms = arche::gcp::kms::GcpKmsClient::new(
    kms_gcp,
    "my-project".into(),
    "global".into(),
    None,
);

Cloud Storage (GCS)

Object upload / download / delete / list / head, plus V4-signed GET URLs. Bucket is per-call so one client can target many buckets.

Warning

Uploads and downloads buffer the full object in memory — fine for reports and assets, but route large media through a streaming path instead.

use arche::gcp::gcs::{get_gcs_client, GcsConfig};
use std::collections::HashMap;
use std::time::Duration;

// All fields optional — gcs_base_url and signed-URL expiry default fine.
let gcs = get_gcs_client(Some(key), None, None).await?;

// Upload with optional user metadata (e.g. `modified_by`)
let mut meta = HashMap::new();
meta.insert("modified_by".into(), "alice".into());
let object = gcs.upload(
    "my-bucket",
    "reports/q4.pdf",
    pdf_bytes,
    "application/pdf",
    meta,
).await?;
// object.generation: Option<i64>  — round-trippable into download/head/delete

// Read it back
let bytes = gcs.download("my-bucket", "reports/q4.pdf", None).await?;

// Or a specific historical version (requires Object Versioning on the bucket)
let old = gcs.download("my-bucket", "reports/q4.pdf", Some(1_700_000_123_456_789)).await?;

// Metadata-only fetch
let meta = gcs.head("my-bucket", "reports/q4.pdf", None).await?;

// Merge-patch user metadata (existing keys overwritten, others untouched)
let mut update = HashMap::new();
update.insert("reviewed_by".into(), "bob".into());
gcs.patch_metadata("my-bucket", "reports/q4.pdf", update).await?;

// List a prefix; pass `versions: true` to include non-current generations
let page = gcs.list("my-bucket", Some("reports/"), None, false).await?;

// V4-signed GET URL (defaults to client's expiry, max 7 days). Always
// points at `storage.googleapis.com` regardless of GCS_BASE_URL.
let url = gcs.signed_get_url("my-bucket", "reports/q4.pdf", Some(Duration::from_secs(600)))?;

Env Var	Description
`GCS_BASE_URL`	Override the storage endpoint (testing / VPC-SC; ignored for signed URLs)
`GCS_SIGNED_URL_EXPIRY_SECS`	Default expiry for `signed_get_url` (default: 900, max: 604800)

upload switches automatically to multipart when called — user metadata travels with the bytes in one request, no separate PATCH needed. Object names containing / are URL-encoded for the JSON-API path but left literal in V4-signed paths, matching GCS spec.

Cloud CDN

Cache invalidation against a global URL map, plus operation polling.

use arche::gcp::cdn::{get_cdn_client, GcpCdnConfig};

let cdn = get_cdn_client(
    Some(key),
    None,
    GcpCdnConfig::builder()
        .project_id("my-project")
        .url_map("my-lb-url-map") // optional default; pass per-call to override
        .build(),
).await?;

// Invalidate. `path` must start with `/` and may use `*` as a suffix wildcard.
// `host` scopes the invalidation to a hostname routed by the URL map.
let op = cdn.invalidate(None, "/static/*", Some("cdn.example.com")).await?;
// op.name — the operation name; pass it to `invalidation_status` to poll
// op.status — "PENDING" / "RUNNING" / "DONE"
// op.progress — Option<i32>, 0..=100

// Poll until done
let status = cdn.invalidation_status(&op.name).await?;
assert_eq!(status.status, "DONE");

Env Var	Description
`GCP_CDN_PROJECT_ID`	GCP project hosting the URL map (required)
`GCP_CDN_URL_MAP`	Optional default URL map name
`GCP_CDN_BASE_URL`	Override the Compute API endpoint

Scope: global URL maps only — regional URL maps (/regions/{region}/urlMaps/...) are not supported and will return 404.

Any other GCP REST API

GcpClient works for any Google API that accepts Authorization: Bearer …:

use arche::gcp::GcpClient;

let pubsub = GcpClient::new(
    Some(key),
    None,
    ["https://www.googleapis.com/auth/pubsub"],
).await?;

pubsub
    .post(format!("https://pubsub.googleapis.com/v1/projects/{p}/topics/{t}:publish"))
    .await?
    .json(&payload)
    .send().await?;

For full HTTP control (custom timeouts, TLS config, connection pool), bring your own reqwest::Client:

let http = reqwest::Client::builder()
    .connect_timeout(std::time::Duration::from_secs(5))
    .build()?;

let storage = GcpClient::with_http(
    http,
    Some(key),
    None,
    ["https://www.googleapis.com/auth/devstorage.read_only"],
).await?;

One service account, multiple APIs — share a single token cache:

let drive   = arche::gcp::drive::client(Some(key), None).await?;
let sheets  = drive.with_scopes(["https://www.googleapis.com/auth/spreadsheets"]);
let storage = drive.with_scopes(["https://www.googleapis.com/auth/devstorage.read_only"]);

Vertex AI

Moved to senno as of arche 5.0 and re-exported here: arche::senno::providers::vertex (Gemini + Anthropic Claude, API-key or service-account auth). See LLM & Agents.

Token cache — every GCP REST call goes through a process-local token cache: JWT-bearer flow against oauth2.googleapis.com/token, signed RS256 with the service-account key, retried once on transient failures, refreshed 60 s before expiry, single-flighted per (client_email, scopes) pair.

OIDC

Both halves of OpenID Connect — speak to a provider, or be one.

	`arche::oidc` (client)	`arche::oidc::server`
Role	Relying party — "Sign in with Google"	Identity provider — "Sign in with your service"
You get	authorize URL → code → verified ID token	validate → single-use code → signed ID token
You bring	provider metadata + your config	four small trait impls (registry · signer · tokens · store)

Authorization-code flow with mandatory PKCE (S256), RS256 ID tokens, and a JWKS cache that handles key rotation transparently. The two halves never share code — an interop test drives the client against the server over real HTTP to prove they speak the same dialect.

Tip

Full walkthrough, both sequence diagrams, and a single hover-tooltip canvas live in docs/oidc/.

Client — "Sign in with Google" (or any OIDC provider)

Provider endpoints come from a ProviderMetadata: the shipped google() preset, a struct literal, or ProviderMetadata::discover(...) (fetches /.well-known/openid-configuration).

use arche::gcp::oauth::google;
use arche::oidc::{OidcClient, OidcConfig, ProviderMetadata, Verifier};

// Build once, hold on state — both are Clone.
let client = OidcClient::new(
    google(),                                        // or ProviderMetadata::discover(...).await?
    OidcConfig {
        client_id: "123.apps.googleusercontent.com".into(),
        client_secret: "secret".into(),
        redirect_uri: "https://app.example/auth/google/callback".into(),
        scopes: None,                                // defaults to "openid email profile"
    },
)?;
let verifier = Verifier::new(client.provider())?;

// 1 ─ send the user to the provider's consent screen
let url = client.auth_url(&state, &pkce_challenge);

// 2 ─ on callback, trade the code for tokens
let tokens = client.exchange_code(&code, &pkce_verifier).await?;

// 3 ─ verify the ID token into YOUR claims type
#[derive(serde::Deserialize)]
struct Claims { sub: String, email: String, email_verified: bool }

let claims: Claims = verifier
    .verify_id_token(&tokens.id_token, &[client.client_id()])
    .await?;

Note

verify_id_token::<C> validates RS256, the signing kid (against a rotation-aware JWKS cache), the signature, and iss / aud / exp / iat / nbf (60 s skew) — then deserializes into your C. A bad token is AppError::Unauthorized (the real reason is logged at debug, never leaked); an unreachable provider is AppError::DependencyFailed; a valid token that doesn't fit C is an internal error, not a 401. Nonce is yours to check — put it in C and compare.

Server — be your own identity provider

arche runs the protocol logic; it does not serve HTTP. You wire four routes and call four methods. The server is generic over five capabilities — arche ships a built-in only where correctness is math, never policy:

Capability	Trait	Built-in	Bring your own when…
Client resolution	`ClientRegistry`	— always yours	static list · DB · config service (override `verify_secret` for hashed secrets)
Token signing	`TokenSigner`	`SigningKey` (local RSA)	keys live in KMS / HSM, or you rotate
Access tokens	`AccessTokenIssuer`	— always yours	opaque random · your own JWT · a stored token
Code storage	`CodeStore`	— always yours	in-memory (single node) · Redis `GETDEL` · PG `DELETE…RETURNING`
Refresh tokens	`RefreshTokenStore`	— always yours	any store with atomic delete-on-read — same shape as `CodeStore`

1 · Implement your five seams — click to expand

use arche::error::AppError;
use arche::oidc::server::{
    AccessTokenIssuer, ClientRegistration, ClientRegistry, CodeStore, IssuedAccessToken,
    PendingGrant, RefreshTokenStore, SigningKey,
};
use std::collections::HashMap;
use std::time::{Duration, Instant};
use tokio::sync::Mutex;

// Signer: a local RSA key is the one built-in. Persist it — a fresh key
// invalidates every issued ID token.  openssl genpkey -algorithm RSA -pkeyopt rsa_keygen_bits:2048
let key = SigningKey::from_pem("2026-07-key", &std::fs::read_to_string("key.pem")?)?;

// Registry: resolve partner apps. `verify_secret` is defaulted (constant-time
// plaintext); override it when you store hashed secrets.
struct StaticClients(Vec<ClientRegistration>);
impl ClientRegistry for StaticClients {
    async fn find(&self, id: &str) -> Result<Option<ClientRegistration>, AppError> {
        Ok(self.0.iter().find(|c| c.client_id == id).cloned())
    }
}

// Access token: opaque noise is fine when only the ID token is consumed;
// mint a real JWT when something must verify it.
struct OpaqueTokens;
impl AccessTokenIssuer for OpaqueTokens {
    async fn issue(&self, _g: &PendingGrant) -> Result<IssuedAccessToken, AppError> {
        Ok(IssuedAccessToken { token: arche::utils::nano_id_of(43), expires_in: 3600 })
    }
}

// Code store: `take` MUST be atomic delete-on-read (the single-use guarantee).
// In-memory works on one node; behind a load balancer use Redis GETDEL / PG.
#[derive(Default)]
struct MemStore(Mutex<HashMap<String, (PendingGrant, Instant)>>);
impl CodeStore for MemStore {
    async fn put(&self, code: String, g: PendingGrant, ttl: Duration) -> Result<(), AppError> {
        let exp = Instant::now().checked_add(ttl).unwrap_or_else(Instant::now);
        self.0.lock().await.insert(code, (g, exp));
        Ok(())
    }
    async fn take(&self, code: &str) -> Result<Option<PendingGrant>, AppError> {
        Ok(self.0.lock().await.remove(code)
            .filter(|(_, exp)| *exp > Instant::now())
            .map(|(g, _)| g))
    }
}

// Refresh store: identical shape — `take` is delete-on-read, giving rotation +
// reuse-invalidation for free. Only touched when a client is granted `offline_access`.
#[derive(Default)]
struct MemRefreshStore(Mutex<HashMap<String, (PendingGrant, Instant)>>);
impl RefreshTokenStore for MemRefreshStore {
    async fn put(&self, token: String, g: PendingGrant, ttl: Duration) -> Result<(), AppError> {
        let exp = Instant::now().checked_add(ttl).unwrap_or_else(Instant::now);
        self.0.lock().await.insert(token, (g, exp));
        Ok(())
    }
    async fn take(&self, token: &str) -> Result<Option<PendingGrant>, AppError> {
        Ok(self.0.lock().await.remove(token)
            .filter(|(_, exp)| *exp > Instant::now())
            .map(|(g, _)| g))
    }
}

2 · Build the server — once, at startup:

use arche::oidc::server::{OidcServer, OidcServerConfig};

let server = OidcServer::new(
    OidcServerConfig {
        issuer: "https://id.example.com".into(),     // https; becomes `iss` + endpoint prefix
        code_ttl: None,           // 5 min
        id_token_ttl: None,       // 1 h
        refresh_token_ttl: None,  // 30 days
        // Add "offline_access" here to enable refresh tokens for clients that request it:
        allowed_scopes: None,     // ["openid", "email", "profile"]
    },
    clients, key, OpaqueTokens, MemStore::default(), MemRefreshStore::default(),
)?;

3 · Wire four routes — each is one method call:

use arche::oidc::server::{DiscoveryDocument, OidcServerError, TokenRequest};

// GET /.well-known/openid-configuration
let mut doc = DiscoveryDocument::standard(server.issuer());   // all fields pub — override to taste
Json(doc)

// GET /jwks   (send Cache-Control: public, max-age=3600)
Json(server.jwks_document())

// GET /authorize
let validated = match server.validate_authorize(&params).await {
    Ok(v) => v,
    Err(e) if e.redirectable() =>
        return redirect(e.redirect_url(&params.redirect_uri, params.state.as_deref())?),
    Err(e) => return bad_request(e.to_string()),   // unknown client / bad redirect — NEVER redirect
};
match session_user(&headers) {
    Some(u) => redirect(server.issue_code(validated, u.id, u.claims()).await?),
    None    => redirect("/login"),                 // stash `validated`, resume after login
}

// POST /token   (send Cache-Control: no-store on success)
let req = TokenRequest { /* form fields */,
    basic_auth: auth_header.and_then(TokenRequest::parse_basic_authorization) };
match server.exchange(req).await {
    Ok(payload)                        => json_no_store(payload),
    Err(OidcServerError::Internal(_))  => server_error(),     // 5xx — infra, retryable
    Err(e)                             => token_error(e.error_code()),   // 400 / 401
}

Important

One instance, cloned per request — never one-per-request. Build the OidcServer once at startup and keep it on state; .clone() is a single refcount bump. This isn't just efficiency: the CodeStore spans two requests — /authorize calls put(code), and /token (a different request, seconds later) calls take(code). A server rebuilt per request hands the second call an empty store, and every login fails with invalid_grant.

Tip

Keep it isolated: hold the server in its own service state, reached through the State extractor — no shared god-object, no middleware required. Give each service its own state and compose routers; reserve shared state for truly-global primitives (DB pools, secrets).

Claims are yours; protocol claims are arche's. issue_code(request, subject, claims) takes the sub (1–255 ASCII) and any Serialize claims and mints them verbatim — except the reserved set (iss, sub, aud, iat, exp, nbf, nonce, jti, azp, at_hash, c_hash), which arche strips from your input and stamps itself. auth_time is deliberately not reserved — you assert it (and must, to honor a max_age request). To resume after a login page, round-trip the serializable ValidatedAuthorizeRequest back into issue_code; it re-checks client_id / redirect_uri against the registry, so a tampered stash is rejected, not used.

Tip

Refresh tokens are opt-in per client via the offline_access scope. Add offline_access to allowed_scopes; when a client requests it and you grant it, exchange returns a refresh_token alongside the ID token. A subsequent grant_type=refresh_token call rotates it — arche re-mints the ID/access tokens (dropping the one-time nonce, keeping your original claims) and issues a fresh refresh token, invalidating the old one via the store's atomic take. A replayed old token gets invalid_grant. Claims are snapshotted at authorization time — arche has no user model to re-fetch them.

Warning

Not supported, by design: the client-credentials grant and /userinfo (claims ride in the ID token). Key rotation is a TokenSigner choice, not a limitation.

Deeper reading:

docs/oidc/README.md — index for both halves
docs/oidc/architecture.md — the two halves, component diagram, five seams, where each defense lives
docs/oidc/sequence.md — login flow both directions, what state / PKCE defend, error + wire tables
docs/oidc/extending.md — client & server quickstarts, code for each of the five traits

LLM & Agents (senno)

As of arche 5.0, the entire LLM stack — canonical types, the LlmProvider trait, one_shot, the tool-calling agent engine with history compaction, and the Vertex AI provider (Gemini + Anthropic Claude) — lives in its own crate: senno (crates.io · docs.rs).

arche depends on senno with the vertex and axum features enabled and re-exports it, so no extra dependency is needed:

use arche::senno::providers::vertex::{get_vertex_client, VertexProvider};
use arche::senno::{GenerateRequest, one_shot};

let client = get_vertex_client(VertexProvider::Gemini, None).await?;
let resp = one_shot(
    &client,
    &GenerateRequest::one_shot("gemini-2.5-flash", "Be concise.", "Hello!"),
)
.await?;

Error bridging — senno returns its own senno::Error; arche provides impl From<senno::Error> for AppError, so ? works unchanged in handlers: provider failures map to DependencyFailed (retryability preserved), config/tool failures map to InternalError.

Migrating from arche 4.x:

arche 4.x	arche 5.0
`arche::llm::*`	`arche::senno::*`
`arche::agent::*`	`arche::senno::agent::*`
`arche::gcp::vertex::*`	`arche::senno::providers::vertex::*`
tool errors as `AppError`	tool errors as `senno::Error`

Full documentation lives in the senno repo (README + docs/agent/).

Database

Postgres

Connection pooling with sqlx, configurable credentials, and health checks.

use arche::database::pg::{get_pg_pool, test_pg, PgConfigBuilder};

let pool = get_pg_pool(None).await?;
let is_healthy = test_pg(pool.clone()).await?;

Env Var	Description
`PG_HOST`	Database host
`PG_PORT`	Database port
`PG_DATABASE`	Database name
`PG_MAX_CONN`	Maximum pool connections
`PG_USERNAME`	Username
`PG_PASSWORD`	Password
`PG_CREDENTIALS`	JSON string `{"username":"...","password":"..."}` (alternative to separate vars)

Redis

Connection pooling with bb8, optional password auth, and health checks.

use arche::database::redis::{get_redis_pool, test_redis, RedisConfigBuilder};

let pool = get_redis_pool(None).await?;
let is_healthy = test_redis(pool.clone()).await?;

Env Var	Description
`REDIS_HOST`	Redis host
`REDIS_PORT`	Redis port
`REDIS_MAX_CONN`	Maximum pool connections
`REDIS_PASSWORD`	Optional password

ClickHouse

Read-only connection pooling with bb8 (round-robin across replicas) and a typed row API. SQL templates are &'static str — a compile-time check that prevents user input from being concatenated into a query.

use arche::database::clickhouse::{
    get_clickhouse_pool, ClickHousePoolExt, Row, Deserialize,
};

let pool = get_clickhouse_pool(None).await?;
let conn = pool.get_conn().await?;

#[derive(Row, Deserialize)]
struct EventCount { event: String, n: u64 }

let counts: Vec<EventCount> = conn
    .query("SELECT event, count() AS n FROM events WHERE day = ? GROUP BY event")
    .bind("2026-05-25")
    .fetch_all().await?;

Notes:

Bare SELECT * / SELECT t.* are blocked. Call .allow_select_star() on a query, set .allow_select_star(true) on the config, or set CLICKHOUSE_ALLOW_SELECT_STAR=true to bypass.
Writes go through Kafka → Kafka Connect ClickHouse Sink, not this connector.

Warning

query / execute take &'static str on purpose — user input can't be concatenated into the SQL. Runtime-built SQL is still possible via query_dynamic(String) / execute_dynamic(String), but those shift injection-safety onto you.

Env Var	Description	Default
`CLICKHOUSE_HOSTS`	Comma-separated replica hostnames	— (required)
`CLICKHOUSE_HOST`	Single-host fallback if `CLICKHOUSE_HOSTS` is unset	—
`CLICKHOUSE_PORT`	Server port	8443 (secure) / 8123 (plain)
`CLICKHOUSE_DATABASE`	Default database	`default`
`CLICKHOUSE_USERNAME`	Username	`default`
`CLICKHOUSE_PASSWORD`	Password	(empty)
`CLICKHOUSE_SECURE`	HTTPS toggle	`true`
`CLICKHOUSE_MAX_POOL_SIZE`	Max pool connections	`32`
`CLICKHOUSE_CONNECTION_TIMEOUT_MS`	Pool-acquire timeout	`5000`
`CLICKHOUSE_REQUEST_TIMEOUT_MS`	Per-request `max_execution_time`	`30000`
`CLICKHOUSE_COMPRESSION`	`lz4` or `none`	`none`
`CLICKHOUSE_ALLOW_SELECT_STAR`	Global `SELECT *` escape hatch	`false`

Queue

Kafka

Enabled with the kafka cargo feature (arche = { version = "...", features = ["kafka"] }), since it compiles librdkafka and OpenSSL from source and most services don't need it. A thin wrapper around rdkafka providing a convenience producer, a JSON-decoded consumer stream (plus a raw-bytes stream), and a broker health check. Unlike a typical high-level consumer wrapper, batching and commit timing are not owned by arche — callers decide that policy entirely, for example using tokio_stream::StreamExt::chunks_timeout.

Kafka configuration is always explicit — no field is ever resolved from the process environment (read your own env vars and pass them in). Configuration is split into three types: KafkaConnectionConfig (fields shared by every role: brokers, socket_timeout_ms, security/SASL/SSL settings, extra_options), KafkaProducerConfig (embeds KafkaConnectionConfig plus topic/message_timeout_ms), and KafkaConsumerConfig (embeds KafkaConnectionConfig plus topics/group_id/etc.). Build the connection once and hand it to each role builder via .connection(..) (call it first — it replaces the embedded connection); the role builders also expose .broker(), .brokers(), .socket_timeout_ms() and .extra_option(s)() directly for the simple plaintext case.

use arche::queue::kafka::{
    get_kafka_producer, get_kafka_consumer, test_kafka,
    KafkaConnectionConfig, KafkaProducerConfig, KafkaConsumerConfig,
    SecurityProtocol, SaslMechanism, CommitMode, KafkaConsumeError,
};
use arche::tokio_stream::StreamExt;
use std::time::Duration;

// Shared connection settings — here for a SASL/SCRAM-over-TLS cluster
// (AWS MSK, Confluent Cloud, Aiven, ...). For a local plaintext broker
// just `.broker("localhost:9092")` is enough.
let connection = KafkaConnectionConfig::builder()
    .brokers(["b-1.example:9096", "b-2.example:9096"])
    .security_protocol(SecurityProtocol::SaslSsl)
    .sasl_mechanism(SaslMechanism::ScramSha512)
    .sasl_username("svc-user")
    .sasl_password("secret")
    // .ssl_ca_location("/etc/ssl/certs/ca.pem") // only for private CAs
    .build();

let producer_config = KafkaProducerConfig::builder()
    .connection(connection.clone())
    .topic("orders")
    .build();

let consumer_config = KafkaConsumerConfig::builder()
    .connection(connection.clone())
    .topic("orders") // or .topics(["orders", "returns"]) for multiple
    .group_id("orders-service")
    .build();

// Health check — same connection config, so SSL/SASL is honoured here too.
let is_healthy = test_kafka(connection).await?;

// Producer
let producer = get_kafka_producer(producer_config).await?;
producer.send_json("order-123", &serde_json::json!({ "event": "order_placed" })).await?;

// Consumer — caller owns batching policy; here, up to 100 messages or every 5s
let consumer = get_kafka_consumer(consumer_config).await?;
let mut batches = consumer.messages().chunks_timeout(100, Duration::from_secs(5));
tokio::pin!(batches);
while let Some(batch) = batches.next().await {
    let mut last_ok = None;
    for result in batch {
        match result {
            Ok(msg) => {
                println!("key={:?}, message={}", msg.key, msg.value);
                last_ok = Some(msg);
            }
            Err(KafkaConsumeError::Decode { reason, message }) => {
                // poison message: dead-letter it and commit past it so the
                // group doesn't get stuck re-reading it forever
                tracing::warn!(%reason, topic = message.topic(), offset = message.offset(), "undecodable message");
                consumer.commit_raw(&message, CommitMode::Async)?;
            }
            Err(KafkaConsumeError::Consumer(e)) => tracing::warn!(?e, "consumer error"),
        }
    }

    if let Some(msg) = last_ok {
        consumer.commit(&msg, CommitMode::Async)?;
    }
}

Notes:

KafkaProducerConfig/KafkaConsumerConfig are deliberately separate types (each embedding a shared KafkaConnectionConfig), rather than one struct with fields only some roles read — this avoids the ambiguity of a producer silently needing to pick just one topic out of a list meant for a consumer's subscriptions, and makes it a compile error to pass a consumer's config where a producer's is expected (or vice versa).
Security. security_protocol (Plaintext default, Ssl, SaslPlaintext, SaslSsl), sasl_mechanism (Plain, ScramSha256, ScramSha512), sasl_username/sasl_password, and ssl_ca_location/ssl_certificate_location/ssl_key_location/ ssl_key_password map 1:1 to the librdkafka settings. When the protocol is SaslPlaintext/SaslSsl, mechanism + username + password are validated as required at client creation. rdkafka is built with the ssl-vendored and zstd features, so TLS, SASL PLAIN/SCRAM and zstd-compressed topics work out of the box with no system dependencies beyond a C toolchain; GSSAPI/Kerberos and OAUTHBEARER/OIDC are not enabled. Because OpenSSL is statically linked, librdkafka probes the standard CA bundle paths at runtime (/etc/ssl/certs/ca-certificates.crt, /etc/ssl/cert.pem, ...) — make sure the runtime image has ca-certificates installed (distroless/scratch images don't), or point ssl_ca_location at a bundle/private CA explicitly.
Any other librdkafka setting (client.id, compression.type, ...) can be passed via .extra_option(...) / .extra_options([...]). They are applied last, so they override the typed fields if both are set.
Missing/empty required config (brokers, topic, topics, group_id, SASL fields) returns AppError::InternalError with a Config error [kafka/<field>]: ... message, consistent with the other connectors; broker/network failures return AppError::DependencyFailed.
socket_timeout_ms (default 5000) applies to all three roles (producer, consumer, and test_kafka) via KafkaConnectionConfig. Note this is a behavior change from librdkafka's own bare default of 60000ms — producer/consumer requests will now time out after 5s by default instead of 60s; override via .socket_timeout_ms(60_000) if you need the longer window.
Offsets are committed manually by default (enable.auto.commit=false, auto.offset.reset=earliest by default) only when the caller explicitly calls commit, giving full control over delivery semantics and an at-least-once guarantee (a message is never marked done until your code says so). CommitMode::Async returns immediately; CommitMode::Sync blocks the calling thread until the broker acknowledges — in async code prefer Async, or wrap a Sync commit in tokio::task::block_in_place.
Optionally, KafkaConsumerConfigBuilder::auto_commit(true) re-enables librdkafka's periodic background auto-commit (auto_commit_interval_ms controls the frequency, default 5000). Tradeoff: with auto-commit enabled, an offset becomes eligible for commit as soon as the message is handed to your code via messages() — not when your processing logic finishes — so a crash between those two points can silently skip a message on redelivery (at-most-once), unlike the default manual-commit flow (at-least-once). KafkaConsumer::commit() remains callable regardless, if you want to force an earlier commit alongside the background timer.
KafkaConsumer::is_rebalancing() reports whether a partition rebalance is currently in progress; this is informational only and not enforced — callers may check it if they want to defer committing during reassignment.
Empty keys and null/empty values are rejected on the producer side (send_json, send_json_to_topic, send_bytes, and send_bytes_to_topic return AppError::BadRequest for an empty key, a null/empty-string JSON value, or an empty byte payload — nothing is sent to the broker).
The consumer never drops messages. messages() yields every message from the subscription: successfully decoded ones as KafkaMessage (key: Option<String> — None for Kafka null keys; value — a null payload/tombstone decodes to Value::Null), and undecodable ones (non-UTF-8 key, non-JSON payload) as KafkaConsumeError::Decode { reason, message } carrying the raw message, so the caller can dead-letter it and/or commit_raw(&message, ..) to move past it. Broker/consumer errors are KafkaConsumeError::Consumer(AppError). KafkaConsumeError converts into AppError via From for ? in handlers.
Raw messages. KafkaConsumer::raw_messages() yields every message untouched as KafkaRawMessage (key()/payload() as Option<&[u8]>, plus topic()/partition()/offset() and into_inner() for the underlying rdkafka message). Use it for Avro/protobuf/bytes topics or with send_bytes, and commit with commit_raw(&msg, mode).

JWT

Token generation and verification using HS256.

Note

This is symmetric HS256 for your app's own access / refresh tokens. For asymmetric RS256 ID tokens in an OIDC flow (signing or verifying "Sign in with…" tokens), see oidc — different keys, different purpose.

use arche::jwt::{generate_tokens, verify_token, generate_expiry_time};
use serde::{Serialize, Deserialize};

#[derive(Serialize, Deserialize)]
struct Claims {
    sub: String,
    exp: usize,
}

// Generate an access + refresh token pair
let tokens = generate_tokens(
    Claims { sub: "user_123".into(), exp: generate_expiry_time(3600) },
    Claims { sub: "user_123".into(), exp: generate_expiry_time(86400) },
    &access_secret,
    &refresh_secret,
)?;

// Verify a token
let data = verify_token::<Claims>(&tokens.access_token, &access_secret, None)?;

CSV

Async CSV processing powered by csv-async. Supports reading from bytes, files, and URLs — with both batch and streaming modes.

use arche::csv::CsvClient;

// Default config (comma-delimited, with headers)
let csv = CsvClient::new();

// Or customize
let csv = CsvClient::new()
    .delimiter(b';')
    .has_headers(true)
    .flexible(true);

Reading

use serde::Deserialize;

#[derive(Deserialize)]
struct Record { name: String, age: u32, city: String }

// From bytes
let records: Vec<Record> = csv.read().from_bytes(data).deserialize().await?;

// From file
let records: Vec<Record> = csv.read().from_file("data.csv").deserialize().await?;

// From URL
let records: Vec<Record> = csv.read().from_url("https://example.com/data.csv")
    .deserialize().await?;

// Batch processing (memory-efficient for large files)
csv.read().from_file("large.csv")
    .deserialize_batched(1000, |batch: Vec<Record>| async move {
        // Process 1000 records at a time
        Ok(())
    }).await?;

Writing

use serde::Serialize;

#[derive(Serialize)]
struct Output { name: String, score: f64 }

let records = vec![
    Output { name: "Alice".into(), score: 95.5 },
    Output { name: "Bob".into(),   score: 87.0 },
];

// To bytes
let bytes: Vec<u8> = csv.write_all(&records).await?;

// To file
csv.write_file("output.csv", &records).await?;

Streaming

// Record-by-record reading
let mut stream = csv.read().from_file("large.csv").stream().await?;
while let Some(record) = stream.next_deserialized::<Record>().await {
    let record = record?;
}

// Record-by-record writing
let mut writer = csv.writer_to_file("output.csv").await?;
writer.serialize(&Output { name: "Alice".into(), score: 95.5 }).await?;
writer.finish().await?;

JSON

Streaming JSON array parsing optimized for large payloads. Extracts metadata fields before the target array and streams array elements one-by-one or in batches — without loading the full document into memory.

use arche::json::JsonClient;
use serde::Deserialize;

#[derive(Deserialize)]
struct Item { id: u64, name: String }

let json = JsonClient::new();

// Stream a root-level JSON array from bytes
let source = json.from_bytes(data);
let mut stream = source.stream_root_array();

while let Some(item) = stream.next::<Item>().await {
    let item = item?;
}

// Stream a nested array with metadata capture
// Given: {"total": 1000, "items": [{...}, {...}, ...]}
let json = JsonClient::new();
let source = json.from_bytes(data);
let mut stream = source.stream_array("items").await;

while let Some(item) = stream.next::<Item>().await {
    let item = item?;
}
let total: u64 = stream.field("total")?;

// Batch iteration
let batch = stream.next_batch::<Item>(100).await;

// Stream directly from S3
let source = JsonClient::new().from_s3(&s3_client, "my-bucket", "data.json").await?;
let mut stream = source.stream_array("results").await;

Crypto

AES-128-CBC encryption with PBKDF2-HMAC-SHA1 key derivation (65,536 iterations).

Note

The salt must be ≥ 16 bytes. This is a fixed CBC + PBKDF2-SHA1 scheme; when you own both ends and want authenticated encryption, an AEAD cipher (AES-GCM) is the stronger default.

use arche::crypto::{encrypt_cbc, decrypt_cbc};

let secret = "my-secret-key";
let salt = "my-salt-value-16"; // minimum 16 bytes

// Encrypt — returns raw ciphertext bytes
let ciphertext = encrypt_cbc(secret, salt, "sensitive data")?;

// Decrypt — expects base64-encoded ciphertext input
let plaintext = decrypt_cbc(secret, salt, &base64_ciphertext)?;

Sockets

WebSocket connection registry with broadcast support. Manages a thread-safe map of active connections for fan-out messaging.

use arche::sockets::SocketConnectionManager;

let manager = SocketConnectionManager::new();

// Register a connection (typically in a WebSocket upgrade handler)
manager.add(&connection_id, sender)?;

// Broadcast to all connected clients
manager.broadcast("Hello, everyone!".into())?;

// List active connections
let ids = manager.get_connections()?;

// Remove a connection on disconnect
manager.remove(connection_id)?;

Error

Axum-compatible structured error handling. Every variant converts to a JSON response with the appropriate HTTP status code.

use arche::error::AppError;

async fn handler() -> Result<impl axum::response::IntoResponse, AppError> {
    Err(AppError::Unauthorized)
}

Variants:

Variant	Status	Constructor
`BadRequest`	400	`AppError::bad_request(errors, message, description)`
`Unauthorized`	401	Direct construction
`Forbidden`	403	Direct construction
`NotFound`	404	`AppError::not_found("resource")`
`Conflict`	409	`AppError::conflict("message")`
`UnprocessableEntity`	422	`AppError::unprocessable_entity(errors, message, description)`
`DependencyFailed`	424	`AppError::dependency_failed("upstream", "detail")`
`InternalError`	500	`AppError::internal_error(error, message)`
`Unavailable`	503	Direct construction

InternalError responses are sanitized by default — no leaked SQL or infra details.

Warning

The verbose-errors feature echoes raw error details into responses. Enable it in dev / staging only — never in production.

arche = { version = "4.10.0", features = ["verbose-errors"] }

Middleware

Axum layers that keep the wire format consistent with AppError.

extractor_rejection — by default, when an axum extractor fails (malformed JSON body, missing Content-Type, wrong field type, oversized payload), axum replies with a plain-text 400/413/415/422 before your handler runs — bypassing the AppError shape entirely. This layer intercepts those rejection responses and rewrites them as AppError::bad_request JSON, so clients see one error contract everywhere. Handler-authored responses (already JSON) pass through untouched.

use axum::{middleware::from_fn, Router};

let app: Router = Router::new()
    // ...routes...
    .layer(from_fn(arche::middleware::extractor_rejection));

// POST /plugin with body `{}` (missing field) now returns 400:
{
  "error_values": null,
  "message": "Failed to deserialize the JSON body into the target type: missing field `plugin_id` at line 1 column 2",
  "description": null
}

Add the layer before any logging/tracing layers (inner-most position) so observability middleware sees the rewritten response.

Utils

ID generation, date/time conversion traits, and pagination helpers.

Nano IDs

URL-safe, strictly alphanumeric unique IDs (0-9 a-z A-Z — no - or _), so they're safe in subdomains, file names, and anywhere symbol characters cause friction:

use arche::utils::{nano_id, nano_id_of};

// 21 characters — same collision resistance class as a standard nanoid
let id = nano_id();          // e.g. "V1StGXR8Z5jdHi6BmyT9k"

// Custom length
let short = nano_id_of(8);   // e.g. "fX3kQ9aZ"

Date/time & pagination

use arche::utils::{validate_timestamp, FromOffsetDateTime, PaginationParams};
use time::OffsetDateTime;

// Check if a timestamp is in the future
let is_valid = validate_timestamp(timestamp, false)?;

// Convert OffsetDateTime to ISO string
let iso = offset_dt.to_iso_string()?;

// Pagination query params (for Axum extractors)
let params = PaginationParams { page_number: Some(1), page_size: Some(20) };

Re-exported Dependencies

arche re-exports these crates so you don't need to add them separately:

axum · tokio · serde · serde_json · sqlx · time · tracing · tracing-subscriber · reqwest · jsonwebtoken · nanoid · thiserror · base64 · bb8 · bb8-redis · clickhouse (as ch_client) · rdkafka · csv-async · futures · tokio-stream · dotenv · aws-config · aws-sdk-s3 · aws-sdk-sesv2 · aws-sdk-kms · aws-sdk-cloudfront

Design Principles

Explicit over implicit — no hidden global state or magic
Composition over inheritance — thin wrappers you combine as needed
Production-first defaults — sane defaults, sanitized errors, pooled connections
Async-native — built on Tokio from the ground up

What arche is not

A framework that replaces Axum
A code generator or project template
A monolithic abstraction over third-party libraries

License

MIT

Name		Name	Last commit message	Last commit date
.cargo-husky/hooks		.cargo-husky/hooks
.github/workflows		.github/workflows
docs/oidc		docs/oidc
src		src
tests		tests
.gitignore		.gitignore
Cargo.lock		Cargo.lock
Cargo.toml		Cargo.toml
README.md		README.md
rust-toolchain.toml		rust-toolchain.toml

Latest commit

History

Folders and files

Repository files navigation

arche

Why arche?

Getting Started

Modules

API Reference

AWS

S3

KMS

SES

CloudFront

GCP

Service account credentials

Authentication modes

Sheets

Drive

Cloud KMS

Cloud Storage (GCS)

Cloud CDN

Any other GCP REST API

Vertex AI

OIDC

Client — "Sign in with Google" (or any OIDC provider)

Server — be your own identity provider

LLM & Agents (senno)

Database

Postgres

Redis

ClickHouse

Queue

Kafka

JWT

CSV

Reading

Writing

Streaming

JSON

Crypto

Sockets

Error

Middleware

Utils

Nano IDs

Date/time & pagination

Re-exported Dependencies

Design Principles

What arche is not

License

About

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages