diff --git a/.gitignore b/.gitignore index d75d36a..3de7c1f 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ /target .env .ceres_data +lightrail diff --git a/Cargo.lock b/Cargo.lock index f5f5f50..9c91981 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -405,16 +405,22 @@ dependencies = [ "dotenvy", "env_logger", "fjall", + "futures", "jacquard", "jacquard-api", "jacquard-axum", + "jacquard-common", "jacquard-identity", "log", + "postcard", "reqwest", "serde", "serde_json", + "thiserror 2.0.18", "tokio", + "tokio-util", "tower-http", + "url", ] [[package]] @@ -1141,6 +1147,21 @@ dependencies = [ "new_debug_unreachable", ] +[[package]] +name = "futures" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + [[package]] name = "futures-buffered" version = "0.2.13" @@ -1161,6 +1182,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d" dependencies = [ "futures-core", + "futures-sink", ] [[package]] @@ -1169,6 +1191,17 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" +[[package]] +name = "futures-executor" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "baf29c38818342a3b26b5b923639e7b1f4a61fc5e76102d4b1981c6dc7a7579d" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + [[package]] name = "futures-io" version = "0.3.32" @@ -1217,6 +1250,7 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" dependencies = [ + "futures-channel", "futures-core", "futures-io", "futures-macro", @@ -1921,6 +1955,7 @@ dependencies = [ "ciborium-io", "cid", "fluent-uri", + "futures", "getrandom 0.2.17", "getrandom 0.3.4", "hashbrown 0.15.5", @@ -1931,6 +1966,7 @@ dependencies = [ "miette", "multibase", "multihash", + "n0-future", "ouroboros", "oxilangtag", "p256", @@ -1952,6 +1988,7 @@ dependencies = [ "spin 0.10.0", "thiserror 2.0.18", "tokio", + "tokio-tungstenite-wasm", "tokio-util", "trait-variant", "unicode-segmentation", @@ -3389,6 +3426,18 @@ dependencies = [ "zeroize", ] +[[package]] +name = "rustls-native-certs" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "612460d5f7bea540c490b2b6395d8e34a953e52b491accd6c86c8164c5932a63" +dependencies = [ + "openssl-probe", + "rustls-pki-types", + "schannel", + "security-framework", +] + [[package]] name = "rustls-pki-types" version = "1.14.0" @@ -3662,6 +3711,17 @@ dependencies = [ "xxhash-rust", ] +[[package]] +name = "sha1" +version = "0.10.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + [[package]] name = "sha1_smol" version = "1.0.1" @@ -4049,6 +4109,7 @@ dependencies = [ "bytes", "libc", "mio", + "parking_lot", "pin-project-lite", "signal-hook-registry", "socket2 0.6.3", @@ -4087,6 +4148,41 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-tungstenite" +version = "0.24.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "edc5f74e248dc973e0dbb7b74c7e0d6fcc301c694ff50049504004ef4d0cdcd9" +dependencies = [ + "futures-util", + "log", + "rustls", + "rustls-native-certs", + "rustls-pki-types", + "tokio", + "tokio-rustls", + "tungstenite", +] + +[[package]] +name = "tokio-tungstenite-wasm" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e21a5c399399c3db9f08d8297ac12b500e86bca82e930253fdc62eaf9c0de6ae" +dependencies = [ + "futures-channel", + "futures-util", + "http", + "httparse", + "js-sys", + "rustls", + "thiserror 1.0.69", + "tokio", + "tokio-tungstenite", + "wasm-bindgen", + "web-sys", +] + [[package]] name = "tokio-util" version = "0.7.18" @@ -4238,6 +4334,26 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "tungstenite" +version = "0.24.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "18e5b8366ee7a95b16d32197d0b2604b43a0be89dc5fac9f8e96ccafbaedda8a" +dependencies = [ + "byteorder", + "bytes", + "data-encoding", + "http", + "httparse", + "log", + "rand 0.8.5", + "rustls", + "rustls-pki-types", + "sha1", + "thiserror 1.0.69", + "utf-8", +] + [[package]] name = "twoway" version = "0.1.8" diff --git a/Cargo.toml b/Cargo.toml index 93a5296..d0def80 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,17 +9,24 @@ axum = "0.8.9" dotenvy = "0.15.7" env_logger = "0.11.10" fjall = "3.1.4" +futures = "0.3" # jacquard = "0.11.0" jacquard = { path = "../jacquard/crates/jacquard" } # jacquard-api = { version = "0.11.1", features = ["default", "app_bsky"] } -jacquard-api = { path = "../jacquard/crates/jacquard-api", features = ["default", "app_bsky"] } +jacquard-api = { path = "../jacquard/crates/jacquard-api", features = ["default", "app_bsky", "streaming"] } # jacquard-axum = "0.11.0" jacquard-axum = { path = "../jacquard/crates/jacquard-axum" } +# jacquard-common = "0.11.0" +jacquard-common = { path = "../jacquard/crates/jacquard-common", features = ["websocket", "streaming", "reqwest-client"] } # jacquard-identity = { version = "0.11.0", features = ["cache", "dns"] } jacquard-identity = { path = "../jacquard/crates/jacquard-identity", features = ["cache", "dns"] } log = "0.4.29" +postcard = { version = "1", features = ["alloc"] } reqwest = { version = "0.12.23", features = ["stream", "json"] } serde = { version = "1.0.228", features = ["derive"] } serde_json = "1.0.149" -tokio = { version = "1.52.1", features = ["macros", "rt-multi-thread", "net", "signal"] } +thiserror = "2" +tokio = { version = "1.52.1", features = ["full"] } +tokio-util = { version = "0.7", features = ["rt"] } tower-http = { version = "0.6.8", features = ["cors"] } +url = "2" diff --git a/README.md b/README.md index df25201..782e8f2 100644 --- a/README.md +++ b/README.md @@ -4,5 +4,5 @@ Ceres is a Bluesky AppView that aims to be easy on resources, does most of what # Work done currently - `app.bsky.actor.getProfile` - Just started, and let me tell you. It got hands -- `app.bsky.actor.getPreferences` - Works, but not as expected. It has it's own internal preferences. Not the ones from your PDS +- `app.bsky.actor.getPreferences` - Works, but not as expected. It has it's own internal preferences. Not the ones from your PDS. best to just use a social app that clears the atproto-proxy to get from your own PDS - `app.bsky.actor.putPreferences` - [lol](https://github.com/bluesky-social/atproto/issues/4193). Best to just use a bsky-social app fork like https://blacksky.community for running this AppView diff --git a/src/error.rs b/src/error.rs new file mode 100644 index 0000000..fb76988 --- /dev/null +++ b/src/error.rs @@ -0,0 +1,16 @@ +use crate::storage::StorageError; +use thiserror::Error; + +#[derive(Debug, Error)] +pub enum Error { + #[error("storage: {0}")] + Storage(#[from] StorageError), + #[error("I/O: {0}")] + Io(#[from] std::io::Error), + #[error("task: {0}")] + Join(#[from] tokio::task::JoinError), + #[error("{0}")] + Other(String), +} + +pub type Result = std::result::Result; diff --git a/src/handlers/mod.rs b/src/handlers/mod.rs deleted file mode 100644 index d49e1b5..0000000 --- a/src/handlers/mod.rs +++ /dev/null @@ -1,2 +0,0 @@ -pub mod well_known; -pub mod xrpc; diff --git a/src/main.rs b/src/main.rs index c52a67e..9bab421 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,38 +1,44 @@ -use axum::body::Bytes; -use axum::{ - Json, Router, body, - extract::{Request, State}, - http::{Method, StatusCode}, - response::{IntoResponse, Response}, - routing::get, -}; +use std::env; +use std::net::SocketAddr; +use std::path::Path; + use env_logger::Env; -use fjall::Database; -use handlers::{well_known::did_document, xrpc::app_bsky_actor}; use jacquard::{identity::resolver::ResolverOptions, prelude::JacquardResolver, types::did::Did}; use jacquard_axum::service_auth::ServiceAuthConfig; -use log::info; -use serde_json::{Value, json}; -use state::AppState; -use std::env; -use tokio::net::TcpListener; -use tower_http::cors::CorsLayer; +use log::{info, warn}; +use tokio::task::JoinSet; +use tokio_util::sync::CancellationToken; -mod handlers; +use crate::error::{Error, Result}; +use crate::state::AppState; + +mod error; +mod server; mod state; +mod storage; +mod sync; -static APP_USER_AGENT: &str = concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION"),); +static APP_USER_AGENT: &str = concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION")); #[tokio::main] async fn main() -> anyhow::Result<()> { dotenvy::dotenv().ok(); env_logger::Builder::from_env(Env::default().default_filter_or("info")).init(); - let bind_addr = env::var("BIND_ADDR").unwrap_or_else(|_| "127.0.0.1:3000".to_string()); + run().await?; + Ok(()) +} + +async fn run() -> Result<()> { + let bind_addr: SocketAddr = env::var("BIND_ADDR") + .unwrap_or_else(|_| "127.0.0.1:3000".to_string()) + .parse() + .map_err(|e: std::net::AddrParseError| Error::Other(format!("BIND_ADDR: {e}")))?; let reqwest_client = reqwest::Client::builder() .user_agent(APP_USER_AGENT) - .build()?; + .build() + .map_err(|e| Error::Other(format!("reqwest build: {e}")))?; let resolver = JacquardResolver::new(reqwest_client.clone(), ResolverOptions::default()); @@ -41,107 +47,56 @@ async fn main() -> anyhow::Result<()> { .expect("APP_VIEW_DOMAIN produced an invalid did:web"); let service_auth = ServiceAuthConfig::new(service_did, &["bsky_appview"], resolver.clone()); - //fjall db setup - let data_directory = - env::var("CERES_DATA_DIRECTORY").unwrap_or_else(|_| ".ceres_data".to_string()); - let db = Database::builder(data_directory).open()?; - let persistent_ks = db.keyspace("persistent", fjall::KeyspaceCreateOptions::default)?; - let cache_ks = db.keyspace("cache", fjall::KeyspaceCreateOptions::default)?; + let data_dir = env::var("CERES_DATA_DIRECTORY").unwrap_or_else(|_| ".ceres_data".to_string()); + let db = storage::open(Path::new(&data_dir))?; let state = AppState { service_auth, reqwest_client, resolver, forwarded_app_view: env::var("FORWARDED_APP_VIEW").ok(), - database: db, - persistent_ks, - cache_ks, + db: db.clone(), }; - let app = Router::new() - .route("/.well-known/did.json", get(did_document)) - .merge(app_bsky_actor::routes()) - .fallback(log_request) - .layer(CorsLayer::permissive()) - .with_state(state); - - let listener = TcpListener::bind(&bind_addr) - .await - .expect("failed to bind listener"); - info!("listening on https://{bind_addr}"); - - axum::serve(listener, app).await.expect("server error"); - Ok(()) -} - -async fn log_request(State(state): State, req: Request) -> Response { - let (parts, body) = req.into_parts(); - let method = parts.method.clone(); - let path = parts.uri.path().to_string(); - let query = parts - .uri - .query() - .map(|q| format!("?{}", q)) - .unwrap_or_default(); - - info!("{method} {path}{query}"); - - if method == Method::GET { - match parts.uri.query() { - Some(q) => info!(" query: {q}"), - None => info!(" query: "), - } + let token = CancellationToken::new(); + let mut tasks: JoinSet> = JoinSet::new(); + + // Firehose — optional, gated on CERES_FIREHOSE_HOST + if let Ok(fh_host) = env::var("CERES_FIREHOSE_HOST") { + let db = db.clone(); + let token = token.clone(); + tasks.spawn(async move { + sync::firehose::Subscriber::new(fh_host, db).run(token).await + }); + } else { + info!("firehose: CERES_FIREHOSE_HOST unset; not spawning subscriber"); } - let bytes = match body::to_bytes(body, usize::MAX).await { - Ok(b) => b, - Err(e) => { - info!(" body: "); - if state.forwarded_app_view.is_none() { - return (StatusCode::OK, Json(json!({}))).into_response(); - } - // If we are proxying, we might want to try anyway or fail. - // Let's just use empty bytes if it failed to read. - Bytes::new() - } - }; + // Backfill dispatcher + { + let db = db.clone(); + let token = token.clone(); + tasks.spawn(async move { sync::backfill::dispatcher::run(db, token).await }); + } - if !bytes.is_empty() { - match serde_json::from_slice::(&bytes) { - Ok(value) => { - let pretty = - serde_json::to_string_pretty(&value).unwrap_or_else(|_| value.to_string()); - info!(" body (json):\n{pretty}"); - } - Err(_) => { - info!(" body: ", bytes.len()); - } - } - } else if method != Method::GET { - info!(" body: "); + // HTTP server + { + let state = state.clone(); + let token = token.clone(); + tasks.spawn(async move { server::serve(bind_addr, state, token).await }); } - if let Some(ref base_url) = state.forwarded_app_view { - let target_url = format!("{base_url}{path}{query}"); - let proxy_res = state - .reqwest_client - .request(method, &target_url) - .body(bytes) - .send() - .await; - - match proxy_res { - Ok(res) => { - let status = res.status(); - let body_bytes = res.bytes().await.unwrap_or_default(); - return (status, body_bytes).into_response(); - } - Err(e) => { - log::error!("Proxy request failed: {e}"); - return (StatusCode::INTERNAL_SERVER_ERROR, "Proxy Error").into_response(); - } + tokio::select! { + _ = tokio::signal::ctrl_c() => info!("ctrl-c received; shutting down"), + r = tasks.join_next() => warn!("task exited early: {r:?}"), + } + token.cancel(); + while let Some(r) = tasks.join_next().await { + match r { + Ok(Ok(())) => {} + Ok(Err(e)) => warn!("task error: {e}"), + Err(e) => warn!("task join: {e}"), } } - - (StatusCode::OK, Json(json!({}))).into_response() + Ok(()) } diff --git a/src/server/fallback.rs b/src/server/fallback.rs new file mode 100644 index 0000000..34cee53 --- /dev/null +++ b/src/server/fallback.rs @@ -0,0 +1,81 @@ +use axum::{ + Json, + body::{self, Bytes}, + extract::{Request, State}, + http::{Method, StatusCode}, + response::{IntoResponse, Response}, +}; +use log::info; +use serde_json::{Value, json}; + +use crate::state::AppState; + +pub async fn log_request(State(state): State, req: Request) -> Response { + let (parts, body) = req.into_parts(); + let method = parts.method.clone(); + let path = parts.uri.path().to_string(); + let query = parts + .uri + .query() + .map(|q| format!("?{}", q)) + .unwrap_or_default(); + + info!("{method} {path}{query}"); + + if method == Method::GET { + match parts.uri.query() { + Some(q) => info!(" query: {q}"), + None => info!(" query: "), + } + } + + let bytes = match body::to_bytes(body, usize::MAX).await { + Ok(b) => b, + Err(e) => { + info!(" body: "); + if state.forwarded_app_view.is_none() { + return (StatusCode::OK, Json(json!({}))).into_response(); + } + Bytes::new() + } + }; + + if !bytes.is_empty() { + match serde_json::from_slice::(&bytes) { + Ok(value) => { + let pretty = + serde_json::to_string_pretty(&value).unwrap_or_else(|_| value.to_string()); + info!(" body (json):\n{pretty}"); + } + Err(_) => { + info!(" body: ", bytes.len()); + } + } + } else if method != Method::GET { + info!(" body: "); + } + + if let Some(ref base_url) = state.forwarded_app_view { + let target_url = format!("{base_url}{path}{query}"); + let proxy_res = state + .reqwest_client + .request(method, &target_url) + .body(bytes) + .send() + .await; + + match proxy_res { + Ok(res) => { + let status = res.status(); + let body_bytes = res.bytes().await.unwrap_or_default(); + return (status, body_bytes).into_response(); + } + Err(e) => { + log::error!("Proxy request failed: {e}"); + return (StatusCode::INTERNAL_SERVER_ERROR, "Proxy Error").into_response(); + } + } + } + + (StatusCode::OK, Json(json!({}))).into_response() +} diff --git a/src/server/mod.rs b/src/server/mod.rs new file mode 100644 index 0000000..bb7dc4a --- /dev/null +++ b/src/server/mod.rs @@ -0,0 +1,34 @@ +pub mod fallback; +pub mod well_known; +pub mod xrpc; + +use std::net::SocketAddr; + +use axum::{Router, routing::get}; +use log::info; +use tokio::net::TcpListener; +use tokio_util::sync::CancellationToken; +use tower_http::cors::CorsLayer; + +use crate::error::Result; +use crate::state::AppState; + +pub async fn serve( + addr: SocketAddr, + state: AppState, + token: CancellationToken, +) -> Result<()> { + let app: Router = Router::new() + .route("/.well-known/did.json", get(well_known::did_document)) + .merge(xrpc::app_bsky_actor::routes()) + .fallback(fallback::log_request) + .layer(CorsLayer::permissive()) + .with_state(state); + + let listener = TcpListener::bind(addr).await?; + info!("listening on http://{addr}"); + axum::serve(listener, app) + .with_graceful_shutdown(token.cancelled_owned()) + .await?; + Ok(()) +} diff --git a/src/handlers/well_known.rs b/src/server/well_known.rs similarity index 77% rename from src/handlers/well_known.rs rename to src/server/well_known.rs index 00faf61..8d3e2ef 100644 --- a/src/handlers/well_known.rs +++ b/src/server/well_known.rs @@ -50,18 +50,11 @@ pub async fn did_document() -> Json { "z42tusbpVy6BCE8tqcQAkvvRi9W3M7LEqLWNdTxtW2KicMNJ".to_string(), ), }], - service: vec![ - // Service { - // id: "#bsky_notif".to_string(), - // type_: "BskyNotificationService".to_string(), - // service_endpoint: endpoint.clone(), - // }, - Service { - id: "#bsky_appview".to_string(), - type_: "BskyAppView".to_string(), - service_endpoint: endpoint, - }, - ], + service: vec![Service { + id: "#bsky_appview".to_string(), + type_: "BskyAppView".to_string(), + service_endpoint: endpoint, + }], id: did_id, }) } diff --git a/src/handlers/xrpc/app_bsky_actor.rs b/src/server/xrpc/app_bsky_actor.rs similarity index 90% rename from src/handlers/xrpc/app_bsky_actor.rs rename to src/server/xrpc/app_bsky_actor.rs index 75aba18..e88fb0e 100644 --- a/src/handlers/xrpc/app_bsky_actor.rs +++ b/src/server/xrpc/app_bsky_actor.rs @@ -1,15 +1,11 @@ -use crate::handlers::xrpc::XrpcErrorResponse; +use crate::server::xrpc::XrpcErrorResponse; use crate::state::AppState; -use axum::{ - Json, Router, - extract::State, - response::{IntoResponse, Response}, -}; +use crate::storage; +use axum::{Json, Router, extract::State}; use jacquard::{ IntoStatic, prelude::IdentityResolver, types::{datetime::Datetime, did::Did, uri::UriValue}, - xrpc::XrpcEndpoint, }; use jacquard_api::app_bsky::actor::{ PreferencesItem, ProfileViewDetailed, @@ -101,10 +97,6 @@ pub async fn get_profile( Ok(profile.into()) } -fn pref_key(did: &Did<'_>) -> Vec { - format!("preferences:{}", did.as_ref()).into_bytes() -} - pub async fn get_preferences( State(state): State, ExtractOptionalServiceAuth(auth): ExtractOptionalServiceAuth, @@ -112,7 +104,7 @@ pub async fn get_preferences( let auth = auth.ok_or_else(XrpcErrorResponse::auth_missing)?; let did = auth.did(); - let stored = state.persistent_ks.get(pref_key(did)).map_err(|err| { + let stored = storage::preferences::get(&state.db, did).map_err(|err| { log::error!("fjall get preferences: {err}"); XrpcErrorResponse::internal_server_error() })?; @@ -146,13 +138,10 @@ pub async fn put_preferences( XrpcErrorResponse::internal_server_error() })?; - state - .persistent_ks - .insert(pref_key(did), bytes) - .map_err(|err| { - log::error!("fjall insert preferences: {err}"); - XrpcErrorResponse::internal_server_error() - })?; + storage::preferences::put(&state.db, did, &bytes).map_err(|err| { + log::error!("fjall insert preferences: {err}"); + XrpcErrorResponse::internal_server_error() + })?; Ok(Json(())) } diff --git a/src/handlers/xrpc/mod.rs b/src/server/xrpc/mod.rs similarity index 100% rename from src/handlers/xrpc/mod.rs rename to src/server/xrpc/mod.rs diff --git a/src/state.rs b/src/state.rs index 8856717..694361a 100644 --- a/src/state.rs +++ b/src/state.rs @@ -1,15 +1,15 @@ use jacquard::{prelude::JacquardResolver, types::did::Did}; use jacquard_axum::service_auth::{ServiceAuth, ServiceAuthConfig}; +use crate::storage::DbRef; + #[derive(Clone)] pub struct AppState { pub service_auth: ServiceAuthConfig, pub reqwest_client: reqwest::Client, pub resolver: JacquardResolver, pub forwarded_app_view: Option, - pub database: fjall::Database, - pub persistent_ks: fjall::Keyspace, - pub cache_ks: fjall::Keyspace, + pub db: DbRef, } impl ServiceAuth for AppState { diff --git a/src/storage/backfill_queue.rs b/src/storage/backfill_queue.rs new file mode 100644 index 0000000..dc3b876 --- /dev/null +++ b/src/storage/backfill_queue.rs @@ -0,0 +1,113 @@ +//! Timestamp-ordered backfill queue backed by fjall. +//! +//! Keys: `"bfq"\0` — big-endian timestamp gives FIFO ordering. +//! Values: postcard-encoded [`BackfillJob`]. + +use std::collections::HashSet; +use std::time::{SystemTime, UNIX_EPOCH}; + +use serde::{Deserialize, Serialize}; + +use crate::storage::{DbRef, PREFIX_BACKFILL, StorageError, StorageResult}; + +const NUL: u8 = b'\0'; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct BackfillJob { + pub did: String, + pub pds_host: String, + pub cursor: Option, + pub retry_count: u16, + pub reason: String, +} + +fn to_millis(t: SystemTime) -> u64 { + t.duration_since(UNIX_EPOCH) + .expect("system clock before Unix epoch") + .as_millis() as u64 +} + +fn key(ts: u64, did: &str) -> Vec { + let mut k = Vec::with_capacity(PREFIX_BACKFILL.len() + 8 + 1 + did.len()); + k.extend_from_slice(&PREFIX_BACKFILL); + k.extend_from_slice(&ts.to_be_bytes()); + k.push(NUL); + k.extend_from_slice(did.as_bytes()); + k +} + +fn parse_did(raw: &[u8]) -> StorageResult { + let rest = raw + .strip_prefix(PREFIX_BACKFILL.as_slice()) + .ok_or(StorageError::Corrupt { + key: String::from_utf8_lossy(raw).into_owned(), + reason: "wrong prefix", + })?; + if rest.len() < 9 { + return Err(StorageError::Corrupt { + key: String::from_utf8_lossy(raw).into_owned(), + reason: "key too short", + }); + } + let did_bytes = rest[9..].to_vec(); + String::from_utf8(did_bytes).map_err(|_| StorageError::Corrupt { + key: String::from_utf8_lossy(raw).into_owned(), + reason: "did not utf-8", + }) +} + +/// Enqueue a job to be claimed when wall-clock time reaches `when`. +pub fn enqueue(db: &DbRef, when: SystemTime, job: &BackfillJob) -> StorageResult<()> { + let ts = to_millis(when); + let k = key(ts, &job.did); + let v = postcard::to_allocvec(job)?; + db.persistent.insert(k, v)?; + Ok(()) +} + +/// Claim the next ready job whose timestamp is `<= now` and whose DID is not already in flight. +/// +/// Atomically removes the key before returning. `since` is an opaque cursor from a prior +/// `claim` call — callers can hold it in memory to skip over tombstones left by prior claims. +pub fn claim( + db: &DbRef, + now: SystemTime, + since: Option>, + busy: &HashSet, +) -> StorageResult)>> { + let now_ms = to_millis(now); + + let lower = match since { + Some(suffix) => { + let mut k = PREFIX_BACKFILL.to_vec(); + k.extend_from_slice(&suffix); + k + } + None => PREFIX_BACKFILL.to_vec(), + }; + // Exclusive upper bound at "now": only claim jobs whose ts < now_ms. + let mut upper = PREFIX_BACKFILL.to_vec(); + upper.extend_from_slice(&now_ms.to_be_bytes()); + upper.push(NUL); + + for guard in db.persistent.range(lower..upper) { + let (key_slice, val_slice) = guard.into_inner()?; + let key_bytes = key_slice.as_ref(); + let did = parse_did(key_bytes)?; + + if busy.contains(&did) { + continue; + } + + let job: BackfillJob = postcard::from_bytes(val_slice.as_ref())?; + let next_since = key_bytes[PREFIX_BACKFILL.len()..].to_vec(); + + let mut batch = db.database.batch(); + batch.remove_weak(&db.persistent, key_bytes); + batch.commit()?; + + return Ok(Some((job, next_since))); + } + + Ok(None) +} diff --git a/src/storage/error.rs b/src/storage/error.rs new file mode 100644 index 0000000..04b914c --- /dev/null +++ b/src/storage/error.rs @@ -0,0 +1,16 @@ +use thiserror::Error; + +#[derive(Debug, Error)] +pub enum StorageError { + #[error("fjall: {0}")] + Fjall(#[from] fjall::Error), + #[error("postcard: {0}")] + Postcard(#[from] postcard::Error), + #[error("corrupt key {key}: {reason}")] + Corrupt { + key: String, + reason: &'static str, + }, +} + +pub type StorageResult = Result; diff --git a/src/storage/firehose_cursor.rs b/src/storage/firehose_cursor.rs new file mode 100644 index 0000000..c5d4614 --- /dev/null +++ b/src/storage/firehose_cursor.rs @@ -0,0 +1,33 @@ +use crate::storage::{DbRef, PREFIX_FH_CURSOR, StorageError, StorageResult}; + +const NUL: u8 = b'\0'; +const CURSOR_SUFFIX: &[u8] = b"cursor"; + +fn key(host: &str) -> Vec { + let mut k = + Vec::with_capacity(PREFIX_FH_CURSOR.len() + host.len() + 1 + CURSOR_SUFFIX.len()); + k.extend_from_slice(&PREFIX_FH_CURSOR); + k.extend_from_slice(host.as_bytes()); + k.push(NUL); + k.extend_from_slice(CURSOR_SUFFIX); + k +} + +pub fn get(db: &DbRef, host: &str) -> StorageResult> { + let k = key(host); + match db.persistent.get(&k)? { + None => Ok(None), + Some(v) => { + let bytes: [u8; 8] = v.as_ref().try_into().map_err(|_| StorageError::Corrupt { + key: String::from_utf8_lossy(&k).into_owned(), + reason: "cursor value is not 8 bytes", + })?; + Ok(Some(u64::from_be_bytes(bytes))) + } + } +} + +pub fn set(db: &DbRef, host: &str, cursor: u64) -> StorageResult<()> { + db.persistent.insert(key(host), cursor.to_be_bytes())?; + Ok(()) +} diff --git a/src/storage/mod.rs b/src/storage/mod.rs new file mode 100644 index 0000000..ab30649 --- /dev/null +++ b/src/storage/mod.rs @@ -0,0 +1,40 @@ +pub mod backfill_queue; +pub mod error; +pub mod firehose_cursor; +pub mod preferences; + +pub use error::{StorageError, StorageResult}; + +use std::path::Path; +use std::sync::Arc; + +/// Fixed-length 3-byte key prefix per data type. +pub(crate) type KeyPrefix = [u8; 3]; + +pub(crate) const PREFIX_PREFS: KeyPrefix = *b"prf"; +pub(crate) const PREFIX_FH_CURSOR: KeyPrefix = *b"fhc"; +pub(crate) const PREFIX_BACKFILL: KeyPrefix = *b"bfq"; + +/// Shared handle to the fjall database and its keyspaces. +pub struct Db { + pub(crate) database: fjall::Database, + /// Durable state: preferences, queue jobs, firehose cursor. + pub(crate) persistent: fjall::Keyspace, + /// Ephemeral cache — currently unused; kept as a home for future read-through caches. + #[allow(dead_code)] + pub(crate) cache: fjall::Keyspace, +} + +pub type DbRef = Arc; + +/// Open (or create) the fjall database at `path` and return a shared handle. +pub fn open(path: &Path) -> StorageResult { + let database = fjall::Database::builder(path).open()?; + let persistent = database.keyspace("persistent", fjall::KeyspaceCreateOptions::default)?; + let cache = database.keyspace("cache", fjall::KeyspaceCreateOptions::default)?; + Ok(Arc::new(Db { + database, + persistent, + cache, + })) +} diff --git a/src/storage/preferences.rs b/src/storage/preferences.rs new file mode 100644 index 0000000..5dc1a15 --- /dev/null +++ b/src/storage/preferences.rs @@ -0,0 +1,20 @@ +use jacquard::types::did::Did; + +use crate::storage::{DbRef, PREFIX_PREFS, StorageResult}; + +fn key(did: &Did<'_>) -> Vec { + let d = did.as_str(); + let mut k = Vec::with_capacity(PREFIX_PREFS.len() + d.len()); + k.extend_from_slice(&PREFIX_PREFS); + k.extend_from_slice(d.as_bytes()); + k +} + +pub fn get(db: &DbRef, did: &Did<'_>) -> StorageResult>> { + Ok(db.persistent.get(key(did))?.map(|v| v.as_ref().to_vec())) +} + +pub fn put(db: &DbRef, did: &Did<'_>, bytes: &[u8]) -> StorageResult<()> { + db.persistent.insert(key(did), bytes)?; + Ok(()) +} diff --git a/src/sync/backfill/dispatcher.rs b/src/sync/backfill/dispatcher.rs new file mode 100644 index 0000000..0f5bd82 --- /dev/null +++ b/src/sync/backfill/dispatcher.rs @@ -0,0 +1,79 @@ +//! Backfill job dispatcher. +//! +//! Claims [`BackfillJob`]s from the on-disk queue and runs each through a stub worker. +//! The worker currently just logs the job — real repo-walking logic lands later. + +use std::collections::HashSet; +use std::time::{Duration, SystemTime}; + +use log::{debug, info}; +use tokio::task::JoinSet; +use tokio_util::sync::CancellationToken; + +use crate::error::Result; +use crate::storage::{DbRef, backfill_queue::{self, BackfillJob}}; + +const MAX_WORKERS: usize = 4; +const IDLE_POLL: Duration = Duration::from_millis(500); + +pub async fn run(db: DbRef, token: CancellationToken) -> Result<()> { + info!("backfill dispatcher started (max_workers={MAX_WORKERS})"); + + let mut busy: HashSet = HashSet::new(); + let mut since: Option> = None; + let mut workers: JoinSet = JoinSet::new(); + + loop { + // Fill worker slots from the queue. + while workers.len() < MAX_WORKERS { + let claim_db = db.clone(); + let claim_since = since.clone(); + let claim_busy = busy.clone(); + let claimed = tokio::task::spawn_blocking(move || { + backfill_queue::claim(&claim_db, SystemTime::now(), claim_since, &claim_busy) + }) + .await??; + + match claimed { + Some((job, next_since)) => { + since = Some(next_since); + busy.insert(job.did.clone()); + workers.spawn(run_worker(job)); + } + None => { + since = None; + break; + } + } + } + + tokio::select! { + _ = token.cancelled() => { + info!("backfill dispatcher: cancelled; waiting for {} workers to drain", workers.len()); + while let Some(r) = workers.join_next().await { + if let Err(e) = r { + log::warn!("worker join: {e}"); + } + } + return Ok(()); + } + Some(res) = workers.join_next(), if !workers.is_empty() => { + match res { + Ok(did) => { busy.remove(&did); } + Err(e) => log::warn!("worker panicked: {e}"), + } + } + _ = tokio::time::sleep(IDLE_POLL), if workers.is_empty() => { + debug!("backfill dispatcher: idle"); + } + } + } +} + +async fn run_worker(job: BackfillJob) -> String { + info!( + "backfill stub: did={} pds={} cursor={:?} retry={} reason={}", + job.did, job.pds_host, job.cursor, job.retry_count, job.reason + ); + job.did +} diff --git a/src/sync/backfill/mod.rs b/src/sync/backfill/mod.rs new file mode 100644 index 0000000..1d7ced1 --- /dev/null +++ b/src/sync/backfill/mod.rs @@ -0,0 +1,3 @@ +pub mod dispatcher; + +pub use crate::storage::backfill_queue::BackfillJob; diff --git a/src/sync/firehose/mod.rs b/src/sync/firehose/mod.rs new file mode 100644 index 0000000..049bb10 --- /dev/null +++ b/src/sync/firehose/mod.rs @@ -0,0 +1,183 @@ +//! Firehose WebSocket subscriber. +//! +//! Connects to an ATProto relay's `com.atproto.sync.subscribeRepos` endpoint, +//! reconnects with exponential backoff on failure, and persists the latest +//! sequence number as a cursor. +//! +//! Events are **dropped** — real commit/account/identity handling lands later. + +use std::time::{Duration, Instant}; + +use futures::StreamExt; +use jacquard_api::com_atproto::sync::subscribe_repos::{SubscribeRepos, SubscribeReposMessage}; +use jacquard_common::deps::fluent_uri::Uri; +use jacquard_common::xrpc::SubscriptionExt; +use jacquard_common::{StreamErrorKind, TungsteniteClient}; +use log::{debug, info, trace, warn}; +use tokio_util::sync::CancellationToken; + +use crate::error::{Error, Result}; +use crate::storage::{self, DbRef}; + +const MAX_BACKOFF_SECS: u64 = 64; +const CURSOR_SAVE_EVERY: u64 = 1000; + +pub struct Subscriber { + host: String, + db: DbRef, +} + +impl Subscriber { + pub fn new(host: String, db: DbRef) -> Self { + Self { host, db } + } + + pub async fn run(&mut self, token: CancellationToken) -> Result<()> { + let base = Uri::parse(format!("wss://{}", self.host)) + .map_err(|e| Error::Other(format!("invalid firehose host: {e:?}")))? + .to_owned(); + + let client = TungsteniteClient::new(); + let mut backoff_secs: u64 = 1; + + loop { + if token.is_cancelled() { + return Ok(()); + } + + let connect_cursor = { + let db = self.db.clone(); + let host = self.host.clone(); + tokio::task::spawn_blocking(move || storage::firehose_cursor::get(&db, &host)) + .await?? + }; + + info!( + "firehose: connecting host={} cursor={:?}", + self.host, connect_cursor + ); + + let params = SubscribeRepos { + cursor: connect_cursor.map(|c| c as i64), + }; + + let stream = tokio::select! { + _ = token.cancelled() => return Ok(()), + r = client.subscription(base.clone()).subscribe(¶ms) => r, + }; + + let stream = match stream { + Ok(s) => s, + Err(e) => { + warn!( + "firehose: connect failed host={} err={e} backoff_secs={backoff_secs}", + self.host + ); + if !sleep_cancellable(&token, Duration::from_secs(backoff_secs)).await { + return Ok(()); + } + backoff_secs = (backoff_secs * 2).min(MAX_BACKOFF_SECS); + continue; + } + }; + + info!("firehose: connected host={}", self.host); + backoff_secs = 1; + + let (_sink, mut messages) = stream.into_stream(); + let mut processed: u64 = 0; + let mut last_seq: Option = None; + let mut last_message_at = Instant::now(); + + loop { + tokio::select! { + _ = token.cancelled() => { + persist_cursor(&self.db, &self.host, last_seq).await; + return Ok(()); + } + next = messages.next() => { + match next { + None => { + info!("firehose: stream ended host={}; reconnecting", self.host); + persist_cursor(&self.db, &self.host, last_seq).await; + break; + } + Some(Err(e)) => { + last_message_at = Instant::now(); + match e.kind() { + StreamErrorKind::Decode + | StreamErrorKind::WrongMessageFormat => { + warn!("firehose: decode error host={} err={e}", self.host); + } + StreamErrorKind::Closed => { + info!("firehose: closed host={}; reconnecting", self.host); + persist_cursor(&self.db, &self.host, last_seq).await; + break; + } + _ => { + warn!( + "firehose: stream error host={} err={e} backoff={backoff_secs}s", + self.host + ); + persist_cursor(&self.db, &self.host, last_seq).await; + if !sleep_cancellable(&token, Duration::from_secs(backoff_secs)).await { + return Ok(()); + } + backoff_secs = (backoff_secs * 2).min(MAX_BACKOFF_SECS); + break; + } + } + } + Some(Ok(msg)) => { + last_message_at = Instant::now(); + processed += 1; + if let Some(seq) = seq_of(&msg) { + trace!("firehose: msg seq={seq}"); + last_seq = Some(seq); + } + if processed % CURSOR_SAVE_EVERY == 0 { + persist_cursor(&self.db, &self.host, last_seq).await; + } + // Events are dropped — real handling lands later. + let _ = msg; + let _ = last_message_at; + } + } + } + } + } + } + } +} + +fn seq_of(msg: &SubscribeReposMessage<'_>) -> Option { + match msg { + SubscribeReposMessage::Commit(c) => Some(c.seq), + SubscribeReposMessage::Sync(s) => Some(s.seq), + SubscribeReposMessage::Identity(i) => Some(i.seq), + SubscribeReposMessage::Account(a) => Some(a.seq), + _ => None, + } +} + +async fn sleep_cancellable(token: &CancellationToken, d: Duration) -> bool { + tokio::select! { + _ = token.cancelled() => false, + _ = tokio::time::sleep(d) => true, + } +} + +async fn persist_cursor(db: &DbRef, host: &str, seq: Option) { + let Some(seq) = seq else { return }; + let db = db.clone(); + let host = host.to_string(); + let res = tokio::task::spawn_blocking(move || { + storage::firehose_cursor::set(&db, &host, seq as u64) + }) + .await; + match res { + Ok(Ok(())) => debug!("firehose: cursor saved seq={seq}"), + Ok(Err(e)) => warn!("firehose: cursor save failed: {e}"), + Err(e) => warn!("firehose: cursor save task panicked: {e}"), + } +} diff --git a/src/sync/mod.rs b/src/sync/mod.rs new file mode 100644 index 0000000..098c31e --- /dev/null +++ b/src/sync/mod.rs @@ -0,0 +1,2 @@ +pub mod backfill; +pub mod firehose;