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.10.0"

Modules

Module	What it does
`aws`	S3, SES, KMS, and CloudFront via official AWS SDKs
`gcp`	Generic GCP REST client + Vertex AI (Gemini + Claude); 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)
`llm`	Canonical LLM types + `LlmProvider` trait — backend-agnostic
`agent`	Tool-calling agent engine, session state, SSE streaming
`database`	Postgres, Redis, and ClickHouse connection pooling with 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)
`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

VertexClient implements arche::llm::LlmProvider for Gemini and Anthropic Claude models on Google Cloud. The provider (Gemini or Anthropic) is captured at construction; the model is specified per-request.

use arche::gcp::vertex::{get_vertex_client, VertexConfig, VertexProvider};
use arche::gcp::ServiceAccountKey;
use arche::llm::{GenerateRequest, LlmProvider, Message, StreamChunk};

// Gemini via API key (resolved from VERTEX_API_KEY / GEMINI_API_KEY env)
let client = get_vertex_client(VertexProvider::Gemini, None).await?;

// Service-account auth (required for Anthropic, optional for Gemini)
let key = ServiceAccountKey::new(client_email, private_key);
let client = get_vertex_client(
    VertexProvider::Anthropic,
    Some(VertexConfig::default()
        .with_service_account_key(key)
        .with_project_id("my-project")
        .with_region("us-east5")),
).await?;

let request = GenerateRequest::new(
    "gemini-2.0-flash",
    vec![Message::user("Explain quantum computing in one sentence.")],
)
.with_system("You are a helpful assistant.")
.with_max_tokens(256)
.with_temperature(0.7);

// Non-streaming
let response = client.generate(&request).await?;
println!("{}", response.text().unwrap_or_default());

// Streaming
use futures::StreamExt;
let mut stream = client.stream_generate(&request).await?;
while let Some(chunk) = stream.next().await {
    match chunk? {
        StreamChunk::Text(text) => print!("{text}"),
        StreamChunk::ToolCall { name, arguments, .. } => { /* dispatch tool */ }
        StreamChunk::Done { finish_reason, usage } => {
            println!("\n[{finish_reason}] usage={usage:?}");
        }
    }
}

Function calling (typed schemas via arche::llm::ParameterSchema):

use arche::llm::{ParameterSchema, ToolDefinition};

let tools = vec![
    ToolDefinition::new("get_weather", "Get current weather for a city")
        .with_parameters(
            ParameterSchema::object()
                .with_property("city", ParameterSchema::string("City name"))
                .with_required(["city"]),
        ),
];

let request = GenerateRequest::new(
    "gemini-2.0-flash",
    vec![Message::user("What's the weather in Tokyo?")],
)
.with_tools(tools);

Authentication:

Method	When	Source
API Key	Gemini only	`VertexConfig::with_api_key(...)` or `VERTEX_API_KEY` / `GEMINI_API_KEY` env
Service Account	Gemini + Anthropic	`VertexConfig::with_service_account_key(ServiceAccountKey)` or `with_service_account_key_path("/path/to/sa.json")`

[!NOTE] If an API key is present it takes priority; Anthropic models require service-account auth. VERTEX_PROJECT_ID / VERTEX_REGION override config (default region asia-south1). Service-account creds must be passed via VertexConfig — arche does not auto-resolve GOOGLE_APPLICATION_CREDENTIALS.

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 four 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`

use arche::error::AppError;
use arche::oidc::server::{
    AccessTokenIssuer, ClientRegistration, ClientRegistry, CodeStore,
    IssuedAccessToken, PendingGrant, 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))
    }
}

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
        allowed_scopes: None,  // ["openid", "email", "profile"]
    },
    clients, key, OpaqueTokens, MemStore::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.

[!WARNING] Not supported, by design: refresh tokens, 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, four 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 four traits

LLM

Canonical, provider-agnostic types and the LlmProvider trait that every backend implements. Use it directly when you just want to call an LLM; build on top of it when you want tool-calling orchestration (see agent).

use arche::llm::{GenerateRequest, LlmProvider, Message, ParameterSchema, ToolDefinition};

// `client` is anything implementing `LlmProvider` —
// VertexClient, or your own OpenAi/Bedrock/Ollama/local impl.
let request = GenerateRequest::new(
    "gemini-2.0-flash",
    vec![Message::user("Hello!")],
)
.with_system("Be concise.")
.with_temperature(0.3);

let response = client.generate(&request).await?;

Types you'll use:

Type	Purpose
`LlmProvider` (trait)	`generate()` + `stream_generate()` on a canonical `GenerateRequest`. Implement this to add a backend.
`GenerateRequest` / `GenerateResponse`	Canonical request/response, provider-neutral
`Message`, `Role`, `ContentPart`	Conversation turns — text, tool calls, tool results
`StreamChunk`	`Text(String)` \| `ToolCall { id, name, arguments }` \| `Done { finish_reason, usage }`
`ToolDefinition` + `ParameterSchema`	Strictly-typed tool descriptions; serializes to valid JSON Schema
`Usage`	Token accounting (input/output/total)

Writing a custom backend:

use arche::llm::{GenerateRequest, GenerateResponse, LlmProvider, LlmStream};
use arche::error::AppError;
use std::future::Future;
use std::pin::Pin;

pub struct OpenAiClient { /* http client, api key */ }

impl LlmProvider for OpenAiClient {
    fn generate<'a>(&'a self, request: &'a GenerateRequest)
        -> Pin<Box<dyn Future<Output = Result<GenerateResponse, AppError>> + Send + 'a>>
    { Box::pin(async move { /* POST, convert */ todo!() }) }

    fn stream_generate<'a>(&'a self, request: &'a GenerateRequest)
        -> Pin<Box<dyn Future<Output = Result<LlmStream, AppError>> + Send + 'a>>
    { Box::pin(async move { /* POST stream, convert SSE */ todo!() }) }
}

Drops into arche::agent::get_agent_engine(my_client, config) with no other changes.

Agent

Tool-calling agent engine: orchestrates LLM rounds, invokes your tools, streams SSE events to the client, manages session history (with optional compaction).

use arche::agent::{get_agent_engine, AgentConfig, AgentFlow, AgentSession, ToolOutput, to_sse_event};
use arche::gcp::vertex::{get_vertex_client, VertexProvider};
use arche::llm::{ParameterSchema, ToolDefinition};

struct ShoppingFlow;

impl AgentFlow for ShoppingFlow {
    fn system_prompt(&self) -> String {
        "You help shoppers find products.".into()
    }

    fn tool_definitions(&self) -> Vec<ToolDefinition> {
        vec![
            ToolDefinition::new("search_catalog", "Search products by query")
                .with_parameters(
                    ParameterSchema::object()
                        .with_property("query", ParameterSchema::string("Query"))
                        .with_required(["query"]),
                ),
        ]
    }

    fn execute_tool<'a>(
        &'a self,
        name: &'a str,
        args: &'a serde_json::Value,
        _session: &'a AgentSession,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<ToolOutput, arche::error::AppError>> + Send + 'a>> {
        Box::pin(async move {
            // Run your business logic, return text for the LLM + optional data for the client
            Ok(ToolOutput::text("Found 3 matches")
                .data("product_list", serde_json::json!([/* ... */])))
        })
    }
}

// Wire it up
let client = get_vertex_client(VertexProvider::Gemini, None).await?;
let config = AgentConfig::builder("gemini-2.0-flash").build()?;
let engine = get_agent_engine(client, config)
    .with_default_summarizer("gemini-2.0-flash-lite"); // optional, cheap summarization

// Per request
let mut session = AgentSession::new("sess-1", "shopping");
let stream = engine.run(&ShoppingFlow, &mut session, "find red shoes");
// Map each SseEvent via `to_sse_event(..)` to an axum SSE Event.

What arche provides vs. what you write:

Arche provides	You write
Orchestration loop, streaming, SSE event types, session mutation, tool-calling loop, history compaction	System prompt, tool schemas, tool executors (`impl AgentFlow`), HTTP handler, session persistence

Extension points:

Need	Plug point
Different LLM backend	`impl LlmProvider for YourClient`
Custom history compaction (vector recall, server-side memory)	`impl HistoryCompactor`
Custom UI events from tools	`ToolOutput::text(..).data(type, payload)` → reaches client via `SseEvent::Data`

Deeper reading:

docs/agent/architecture.md — module layering, component diagram with hover tooltips
docs/agent/sequence.md — request lifecycle, error paths, SSE wire format
docs/agent/extending.md — step-by-step guides for each plug point

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`

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"] }

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

arche 4.11.1