cairn_mod/server/
mod.rs

1//! HTTP server: XRPC endpoints over axum.
2//!
3//! Currently exposes one route — `GET /xrpc/com.atproto.label.subscribeLabels`
4//! (§F4, WebSocket). Other endpoints land in their respective issues: the
5//! binary wire-up (main.rs serving a long-lived listener) is parked until
6//! #8 (queryLabels) so both public endpoints compose into one router.
7//!
8//! Design notes:
9//!
10//! - [`SubscribeConfig`] centralizes every runtime knob §F4 mentions: per-IP
11//!   cap, global cap, batch size, ping/pong cadence, retention window.
12//!   `Default` values match §F4 defaults so operators only override what
13//!   their deployment requires.
14//! - The [`router`] constructor takes ownership of a [`Pool<Sqlite>`] and a
15//!   [`WriterHandle`]; both clone cheaply. The limiter is created here and
16//!   wrapped in `Arc` so tests can inspect it with [`Limiter::snapshot`].
17//! - [`serve`] is a thin helper over `axum::serve`, mostly to keep
18//!   integration tests from reimplementing the listen/accept dance.
19
20use std::net::SocketAddr;
21use std::sync::Arc;
22use std::time::Duration;
23
24use axum::Extension;
25use axum::Router;
26use axum::http::Method;
27use axum::routing::get;
28use sqlx::{Pool, Sqlite};
29use tokio::net::TcpListener;
30use tower_http::cors::{Any, CorsLayer};
31
32use crate::error::Result;
33use crate::writer::WriterHandle;
34
35pub mod admin;
36mod create_report;
37pub mod did_document;
38pub mod health;
39pub mod limits;
40mod query;
41pub(crate) mod subscribe;
42pub mod wellknown;
43pub mod xrpc;
44
45pub use admin::{AdminConfig, admin_router};
46pub use create_report::{CreateReportConfig, create_report_router};
47pub use subscribe::current_retention_floor;
48
49/// Sweep-execution policy for the subscribeLabels retention sweep (§F4
50/// retention task). Distinct from [`SubscribeConfig::retention_days`] —
51/// that field is the cutoff source of truth (read-side floor + write-
52/// side cutoff); this struct holds *how* the sweep runs, not *when*
53/// the cutoff lies. Operators tune the two independently:
54///
55/// - `[subscribe].retention_days` (logical owner: [`SubscribeConfig`])
56///   sets the rolling window ("180 days back").
57/// - `[retention]` (logical owner: this struct) sets sweep schedule
58///   and batching ("run at 04:00 UTC, 1000 rows per transaction").
59///
60/// Defaults match §F4 prose: sweep enabled, runs daily at 04:00 UTC,
61/// 1000-row batches.
62#[derive(Debug, Clone)]
63pub struct RetentionConfig {
64    /// Master toggle for the scheduled sweep. When `false`, no
65    /// scheduled sweep runs; the manual CLI / admin-XRPC path still
66    /// works (operators retain explicit control). Default `true`.
67    pub sweep_enabled: bool,
68    /// UTC hour-of-day (0..=23) at which the scheduled sweep fires.
69    /// Default 4 (04:00 UTC — quiet hour for most operator regions,
70    /// matches §F4 example).
71    pub sweep_run_at_utc_hour: u8,
72    /// Rows per DELETE transaction. Larger batches are throughput-
73    /// efficient but hold the writer for longer; smaller batches let
74    /// normal label writes interleave with finer granularity. Default
75    /// 1000. Distinct from [`SubscribeConfig::batch_size`] (replay
76    /// batching is latency-sensitive, sweep batching is throughput-
77    /// sensitive — tune independently).
78    pub sweep_batch_size: i64,
79}
80
81impl Default for RetentionConfig {
82    fn default() -> Self {
83        Self {
84            sweep_enabled: true,
85            sweep_run_at_utc_hour: 4,
86            sweep_batch_size: 1000,
87        }
88    }
89}
90pub use did_document::did_document_router;
91pub use health::health_router;
92pub use limits::Limiter;
93pub use wellknown::wellknown_router;
94
95/// Tunables for the subscribeLabels endpoint. All defaults match §F4.
96#[derive(Debug, Clone)]
97pub struct SubscribeConfig {
98    /// Maximum concurrent subscribers from one IP (§F4 default 8).
99    pub per_ip_cap: usize,
100    /// Maximum concurrent subscribers overall (§F4 default 256).
101    pub global_cap: usize,
102    /// Replay batch size in rows (§F4 "up to 1000 sequence rows per batch").
103    pub batch_size: i64,
104    /// Ping cadence (§F4 "30s").
105    pub ping_interval: Duration,
106    /// Pong silence that closes the connection (§F4 "90s").
107    pub pong_timeout: Duration,
108    /// Rolling retention window. `None` disables the read-side floor
109    /// (useful in tests and when an operator wants unbounded replay).
110    /// Default per §F4: 180 days.
111    pub retention_days: Option<u32>,
112}
113
114impl Default for SubscribeConfig {
115    fn default() -> Self {
116        Self {
117            per_ip_cap: 8,
118            global_cap: 256,
119            batch_size: 1000,
120            ping_interval: Duration::from_secs(30),
121            pong_timeout: Duration::from_secs(90),
122            retention_days: Some(180),
123        }
124    }
125}
126
127/// Build a router exposing the public label endpoints. The caller owns
128/// the pool and writer; dropping all cloned `WriterHandle`s signals
129/// shutdown.
130///
131/// CORS is applied to `queryLabels` only (§F3 "accepts requests from any
132/// origin"). subscribeLabels is WebSocket — browsers don't apply CORS to
133/// WS connections the same way, and the WS handshake rejects cross-
134/// origin reads at the Sec-WebSocket-* layer. Credentials are never
135/// echoed (`allow_credentials` defaults to false on `CorsLayer`).
136pub fn router(pool: Pool<Sqlite>, writer: WriterHandle, config: SubscribeConfig) -> Router {
137    let limiter = Limiter::new(config.global_cap, config.per_ip_cap);
138    let state = subscribe::AppState {
139        pool,
140        writer,
141        limiter,
142        config: Arc::new(config),
143    };
144
145    let query_cors = CorsLayer::new()
146        .allow_origin(Any)
147        .allow_methods([Method::GET, Method::OPTIONS]);
148
149    Router::new()
150        .route(
151            "/xrpc/com.atproto.label.subscribeLabels",
152            get(subscribe::handler),
153        )
154        .route(
155            "/xrpc/com.atproto.label.queryLabels",
156            get(query::handler).layer(query_cors),
157        )
158        .layer(Extension(state))
159}
160
161/// Serve `router` on `addr`. Returns when the listener errors or the
162/// surrounding runtime shuts down.
163pub async fn serve(router: Router, addr: SocketAddr) -> Result<()> {
164    let listener = TcpListener::bind(addr).await?;
165    axum::serve(
166        listener,
167        router.into_make_service_with_connect_info::<SocketAddr>(),
168    )
169    .await?;
170    Ok(())
171}
cairn_mod/server/mod.rs

cairn_mod/server/
mod.rs
