From 927cd33a8cbc2e42db492df49e2d8f390c1a48ab Mon Sep 17 00:00:00 2001 From: phil Date: Wed, 19 Aug 2026 20:06:51 -0400 Subject: [PATCH] 0: counts --- Cargo.lock | 17 +- Cargo.toml | 10 +- lexicons/de/atverkackt/icarus/getProfile.json | 46 +++ moots-0-counts/Cargo.toml | 31 +++ moots-0-counts/readme.md | 9 + {moots-1-simple => moots-0-counts}/src/lib.rs | 0 moots-0-counts/src/main.rs | 234 ++++++++++++++++ moots-0-counts/src/serve.rs | 117 ++++++++ moots-0-counts/src/store.rs | 45 +++ moots-0-counts/src/sync.rs | 219 +++++++++++++++ moots-1-mutuals/Cargo.toml | 31 +++ {moots-1-simple => moots-1-mutuals}/readme.md | 0 moots-1-mutuals/src/lib.rs | 9 + .../src/main.rs | 0 .../src/serve.rs | 0 moots-1-mutuals/src/store.rs | 89 ++++++ moots-1-mutuals/src/sync.rs | 261 ++++++++++++++++++ moots-1-simple/Cargo.toml | 30 -- moots-1-simple/src/store.rs | 20 -- moots-1-simple/src/sync.rs | 70 ----- moots-2-fast-reads/Cargo.toml | 7 + .../readme.md | 0 .../src/main.rs | 0 moots-2-read-optimized/Cargo.toml | 30 -- moots-lexicons/Cargo.toml | 3 +- moots-lexicons/src/de_atverkackt.rs | 7 + moots-lexicons/src/de_atverkackt/icarus.rs | 7 + .../src/de_atverkackt/icarus/get_profile.rs | 240 ++++++++++++++++ moots-lexicons/src/lib.rs | 6 +- readme.md | 2 +- 30 files changed, 1384 insertions(+), 156 deletions(-) create mode 100644 lexicons/de/atverkackt/icarus/getProfile.json create mode 100644 moots-0-counts/Cargo.toml create mode 100644 moots-0-counts/readme.md rename {moots-1-simple => moots-0-counts}/src/lib.rs (100%) create mode 100644 moots-0-counts/src/main.rs create mode 100644 moots-0-counts/src/serve.rs create mode 100644 moots-0-counts/src/store.rs create mode 100644 moots-0-counts/src/sync.rs create mode 100644 moots-1-mutuals/Cargo.toml rename {moots-1-simple => moots-1-mutuals}/readme.md (100%) create mode 100644 moots-1-mutuals/src/lib.rs rename {moots-1-simple => moots-1-mutuals}/src/main.rs (100%) rename {moots-1-simple => moots-1-mutuals}/src/serve.rs (100%) create mode 100644 moots-1-mutuals/src/store.rs create mode 100644 moots-1-mutuals/src/sync.rs delete mode 100644 moots-1-simple/Cargo.toml delete mode 100644 moots-1-simple/src/store.rs delete mode 100644 moots-1-simple/src/sync.rs create mode 100644 moots-2-fast-reads/Cargo.toml rename {moots-2-read-optimized => moots-2-fast-reads}/readme.md (100%) rename {moots-2-read-optimized => moots-2-fast-reads}/src/main.rs (100%) delete mode 100644 moots-2-read-optimized/Cargo.toml create mode 100644 moots-lexicons/src/de_atverkackt.rs create mode 100644 moots-lexicons/src/de_atverkackt/icarus.rs create mode 100644 moots-lexicons/src/de_atverkackt/icarus/get_profile.rs diff --git a/Cargo.lock b/Cargo.lock index 89c6b7b..6f6b7e7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2875,16 +2875,19 @@ dependencies = [ ] [[package]] -name = "moots-1-simple" +name = "moots-0-counts" version = "0.1.0" dependencies = [ "axum", "clap", "color-eyre", + "dasl", "hubble-sync", "hubble-sync-axum", "hubble-sync-rocksdb", + "jacquard-api", "jacquard-axum", + "jacquard-common", "metrics", "metrics-exporter-prometheus", "moots-lexicons", @@ -2903,16 +2906,19 @@ dependencies = [ ] [[package]] -name = "moots-2-read-optimized" +name = "moots-1-mutuals" version = "0.1.0" dependencies = [ "axum", "clap", "color-eyre", + "dasl", "hubble-sync", "hubble-sync-axum", "hubble-sync-rocksdb", + "jacquard-api", "jacquard-axum", + "jacquard-common", "metrics", "metrics-exporter-prometheus", "moots-lexicons", @@ -2930,6 +2936,13 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "moots-2-fast-reads" +version = "0.1.0" +dependencies = [ + "tikv-jemallocator", +] + [[package]] name = "moots-lexicons" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index d854ed9..1fcff47 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,12 @@ [workspace] resolver = "3" -members = ["moots-1-simple", "moots-2-read-optimized", "lexgen", "moots-lexicons"] +members = [ + "lexgen", + "moots-0-counts", + "moots-1-mutuals", + "moots-2-fast-reads", + "moots-lexicons" +] [workspace.package] version = "0.1.0" @@ -10,9 +16,11 @@ edition = "2024" axum = "0.8" clap = { version = "4.6", features = ["env", "derive"] } color-eyre = "0.6" +dasl = "0.2" hubble-sync = "0.0.4" hubble-sync-axum = "0.0.4" hubble-sync-rocksdb = "0.0.4" +jacquard-api = { version = "0.12", features = ["app_bsky"] } jacquard-axum = "0.12" jacquard-common = "0.12" jacquard-derive = "0.12" diff --git a/lexicons/de/atverkackt/icarus/getProfile.json b/lexicons/de/atverkackt/icarus/getProfile.json new file mode 100644 index 0000000..4221125 --- /dev/null +++ b/lexicons/de/atverkackt/icarus/getProfile.json @@ -0,0 +1,46 @@ +{ + "lexicon": 1, + "id": "de.atverkackt.icarus.getProfile", + "defs": { + "main": { + "type": "query", + "description": "Get followers/following counts for one account", + "parameters": { + "type": "params", + "required": ["actor"], + "properties": { + "actor": { + "type": "string", + "format": "at-identifier", + "description": "The account to list mutuals for." + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["did", "handle", "followersCount", "followsCount"], + "properties": { + "did": { + "type": "string", + "format": "did" + }, + "handle": { + "type": "string", + "format": "handle" + }, + "followersCount": { "type": "integer" }, + "followsCount": { "type": "integer" } + } + } + }, + "errors": [ + { + "name": "AccountNotFound", + "description": "The requested account is unknown to this service." + } + ] + } + } +} diff --git a/moots-0-counts/Cargo.toml b/moots-0-counts/Cargo.toml new file mode 100644 index 0000000..9e1081a --- /dev/null +++ b/moots-0-counts/Cargo.toml @@ -0,0 +1,31 @@ +[package] +name = "moots-0-counts" +version.workspace = true +edition.workspace = true + +[dependencies] +axum = { workspace = true } +clap = { workspace = true } +color-eyre = { workspace = true } +dasl = { workspace = true } +hubble-sync = { workspace = true } +hubble-sync-axum = { workspace = true } +hubble-sync-rocksdb = { workspace = true } +jacquard-api = { workspace = true } +jacquard-axum = { workspace = true } +jacquard-common = { workspace = true } +metrics = { workspace = true } +metrics-exporter-prometheus = { workspace = true } +moots-lexicons = { workspace = true } +rlimit = { workspace = true } +rocksdb = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +thiserror = { workspace = true } +tikv-jemallocator = { workspace = true } +tokio = { workspace = true } +tokio-util = { workspace = true } +tower-http = { workspace = true } +tracing = { workspace = true } +tracing-error = { workspace = true } +tracing-subscriber = { workspace = true } diff --git a/moots-0-counts/readme.md b/moots-0-counts/readme.md new file mode 100644 index 0000000..748641a --- /dev/null +++ b/moots-0-counts/readme.md @@ -0,0 +1,9 @@ +# mutuals index: tutorial part 0 + +TODO: link to tutorial + +maintain bluesky follower and following counts for all accounts on the network, using hubble-sync's built-in `counts` merge operator column family to aggregate. + +in this stage of the tutorial we don't de-duplicate follows (one account can create multiple follow records with the same subject), or consider moderation (blocks don't decrement counts if the follow records still exist). we *do* decrement counts when an account is deleted though, unlike bluesky. + +we use Eve's [`de.atverkackt.icarus.getProfile`](https://picon.atverkackt.de/) lexicon xrpc query, though we don't implement full `handle` functionality. only DIDs are accepted in the request, and handles in responses are *not* bidirectionally validated. diff --git a/moots-1-simple/src/lib.rs b/moots-0-counts/src/lib.rs similarity index 100% rename from moots-1-simple/src/lib.rs rename to moots-0-counts/src/lib.rs diff --git a/moots-0-counts/src/main.rs b/moots-0-counts/src/main.rs new file mode 100644 index 0000000..ef08e21 --- /dev/null +++ b/moots-0-counts/src/main.rs @@ -0,0 +1,234 @@ +use std::net::SocketAddr; +use std::num::NonZeroUsize; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; + +use clap::Parser; +use hubble_sync::{DeepCrawl, HubbleSync, SyncConfig, UpstreamKind}; +use hubble_sync_rocksdb::RocksEngine; +use metrics_exporter_prometheus::{Matcher, PrometheusBuilder}; +use tokio::task::JoinSet; +use tokio_util::sync::CancellationToken; +use tracing_subscriber::EnvFilter; + +use moots_0_counts::Moots; +use moots_0_counts::serve::{ServerState, route}; + +#[global_allocator] +static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc; + +#[derive(Debug, Parser)] +#[command(name = "hubble", version, about)] +struct Args { + /// base data directory for db and spill dirs + #[arg(long)] + data: PathBuf, + + /// http server listen + #[arg(long, default_value = "0.0.0.0:3000")] + bind: SocketAddr, + + /// prometheus metrics exporter listen + #[arg(long, default_value = "0.0.0.0:9000")] + metrics_bind: SocketAddr, + + /// hostname for com.atproto.sync.subscribeRepos, getRepoStatus, etc. + #[arg(long)] + upstream: String, + + /// for the user-agent in requests to upstream hosts (relays, PDSes, ...) + #[arg(long)] + contact: String, + + /// enable rocksdb statistics + /// + /// records cache hit ratios, write amp, etc to rocksdb LOG periodically + #[arg(long, action)] + rocksdb_statistics: bool, + + /// how many repo actors can be alive at once + #[arg(long, default_value_t = 16_384)] + max_repo_actors: usize, + + /// how many resyncs can be processed concurrently + /// + /// raise for faster backfill if your machine can handle it. + #[arg(long, default_value_t = 10)] + max_resync_concurrency: usize, + + /// how many "big" repos can be processed concurrently + /// + /// raise for faster backfill if you have enough memory for it. + #[arg(long, default_value_t = 2)] + max_big_repo_resync_concurrency: usize, + + /// enable the deep crawl: listHosts on the relay, then listRepos on each PDS + /// directly, to discover repos the relay doesn't list. + #[arg(long, action)] + deep_crawl: bool, +} + +#[tokio::main] +async fn main() -> color_eyre::Result<()> { + color_eyre::install()?; + tracing_subscriber::fmt() + .with_env_filter( + EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")), + ) + .init(); + + // rocksdb opens a lot of files, make sure the os doesn't kill us for that + if let Err(err) = rlimit::Resource::NOFILE.set(128 * 1024, 128 * 1024) { + tracing::warn!(%err, "failed to raise rlimit nofile"); + } + + let args = Args::parse(); + + start_metrics(args.metrics_bind); + + let cancel = install_signal_handler(); + + let db_dir = args.data.join("db"); + let resync_spill_dir = args.data.join("resync-spill"); + for dir in [&db_dir, &resync_spill_dir] { + tokio::fs::create_dir_all(dir).await?; + } + + let db = Arc::new(set_up_db(&db_dir, args.rocksdb_statistics)?); + let storage = RocksEngine::new(db.clone(), &Default::default())?; + + let moots = Arc::new(Moots { + storage: storage.clone(), + }); + + let mut config = SyncConfig { + max_repo_actors: args.max_repo_actors, + max_resync_concurrency: args.max_resync_concurrency, + big_repo_resync_limit: args.max_big_repo_resync_concurrency, + huge_repo_spill_dir: resync_spill_dir, + ..Default::default() + }; + config.upstream.hostname = args.upstream.clone(); + config.upstream.kind = UpstreamKind::Relay; + if args.deep_crawl { + config.deep_crawl = Some(DeepCrawl { + concurrency: NonZeroUsize::new(100).expect("nonzero"), + recrawl_interval: Duration::from_secs(7 * 86_400), + wait_for_upstream_crawl: true, + }); + } + + let info = format!("hubble-sync moots tutorial step 0: {}", args.contact); + let sync = HubbleSync::new(moots.clone(), storage.clone(), &info, &config); + + let server_state = ServerState { + sync_handle: sync.handle(), + }; + + let mut tasks: JoinSet> = JoinSet::new(); + + // http server + { + let cancel = cancel.clone(); + let bind = args.bind; + tasks.spawn(async move { + let app = route(Arc::new(server_state)); + let listener = tokio::net::TcpListener::bind(bind).await?; + axum::serve(listener, app) + .with_graceful_shutdown(async move { cancel.cancelled().await }) + .await?; + Ok(()) + }); + }; + + // atproto sync client + { + let cancel = cancel.clone(); + tasks.spawn(async move { + sync.with_graceful_shutdown(async move { cancel.cancelled().await }) + .run() + .await?; + Ok(()) + }) + }; + + let mut first_err = None; + + while let Some(joined) = tasks.join_next().await { + tracing::info!("task ended (shutdown)"); + + // if any task stops, fully shut down (supervisor restart) + cancel.cancel(); + + // flush + abort any compactions (don't wait, can take forever) + db.cancel_all_background_work(false); + + match joined { + Ok(Ok(())) => {} // clean shutdown + Ok(Err(err)) => { + // ended with error + if first_err.is_none() { + tracing::error!(%err, "task errored, shutting down"); + first_err = Some(err); + } else { + tracing::warn!(%err, "other task also/subsuquently errored"); + } + } + Err(join_err) => { + if first_err.is_none() { + tracing::error!(%join_err, is_panic = %join_err.is_panic(), "task exited, shutting down"); + first_err = Some(join_err.into()); + } else if join_err.is_cancelled() { + tracing::info!(err = %join_err, "task cancelled"); + } else { + tracing::warn!(err = %join_err, "other task also/subsequently panicked"); + } + } + } + } + + first_err.map(Err).unwrap_or(Ok(())) +} + +fn start_metrics(bind: SocketAddr) { + let mut builder = PrometheusBuilder::new().with_http_listener(bind); + for (name, buckets) in hubble_sync::histogram_buckets() { + builder = builder + .set_buckets_for_metric(Matcher::Full((*name).into()), buckets) + .expect("matcher for histogram"); + } + builder + .install() + .expect("failed to install recorder/exporter"); + hubble_sync::describe_metrics(); + hubble_sync_rocksdb::describe_rocksdb_metrics(); +} + +fn install_signal_handler() -> CancellationToken { + let cancel = CancellationToken::new(); + let signal_cancel = cancel.clone(); + tokio::spawn(async move { + match tokio::signal::ctrl_c().await { + Ok(()) => { + tracing::info!("ctrl-c received, shutting down"); + signal_cancel.cancel(); + } + Err(e) => tracing::error!(error = %e, "signal handler install failed"), + } + }); + cancel +} + +fn set_up_db(path: &Path, statistics: bool) -> Result { + let cache = rocksdb::Cache::new_lru_cache(512 * 2_usize.pow(20)); + let cfs = hubble_sync_rocksdb::recommended_cf_descriptors(&cache); + let mut db_opts = rocksdb::Options::default(); + db_opts.create_if_missing(true); + db_opts.create_missing_column_families(true); + db_opts.set_max_background_jobs(4); + if statistics { + db_opts.enable_statistics(); + } + rocksdb::DB::open_cf_descriptors(&db_opts, path, cfs) +} diff --git a/moots-0-counts/src/serve.rs b/moots-0-counts/src/serve.rs new file mode 100644 index 0000000..eb0979e --- /dev/null +++ b/moots-0-counts/src/serve.rs @@ -0,0 +1,117 @@ +use std::sync::Arc; + +use axum::{Json, Router, extract::State, http::StatusCode, routing::get}; +use hubble_sync::{Did, LoadError, StorageEngine, SyncHandle}; +use hubble_sync_rocksdb::RocksEngine; +use jacquard_axum::{ExtractXrpc, XrpcErrorResponse}; +use jacquard_common::xrpc::XrpcError; +use moots_lexicons::de_atverkackt::icarus::get_profile::{ + GetProfileError, GetProfileOutput, GetProfileRequest, +}; +use tokio::task::spawn_blocking; +use tower_http::cors::CorsLayer; + +use crate::store::{followers_count_key, following_count_key}; + +#[derive(Clone)] +pub struct ServerState { + pub sync_handle: SyncHandle, +} + +pub fn route(state: Arc) -> Router { + let endpoints_state = state.sync_handle.endpoints(); + Router::new() + .route("/", get(hello)) + .route("/xrpc/de.atverkackt.icarus.getProfile", get(get_profile)) + .with_state(state) + .merge(hubble_sync_axum::standard(endpoints_state.clone())) + .merge(hubble_sync_axum::meta(endpoints_state.clone())) + .layer(CorsLayer::permissive()) +} + +async fn hello() -> String { + "hello from step 1 of the hubble-sync moots tutorial".into() +} + +#[derive(Debug, thiserror::Error)] +enum LookupError { + #[error("repo load: {0}")] + Repo(#[from] LoadError), + #[error("record counter: {0}")] + Counter(#[from] hubble_sync_rocksdb::Error), +} + +/// `de.atverkackt.icarus.getProfile`: get follower and following counts +async fn get_profile( + State(state): State>, + ExtractXrpc(params): ExtractXrpc, +) -> Result, XrpcErrorResponse> { + let our_did = params.actor.as_str(); + if !our_did.starts_with("did:") { + return Err(nope("only did identifiers are supported at the moment")); + } + + let h = { + let did = our_did.to_string(); + let sync_handle = state.sync_handle.clone(); + + spawn_blocking(move || -> Result<_, LookupError> { + Ok(sync_handle + .get(&Did::raw(did.as_str()))? + .map(|view| view.context().unverified_handle().map(str::to_string))) + }) + .await + .expect("repo info lookup not to panic") + .map_err(|_| internal("failed to read repo state"))? + }; + + // outer none: the repo was not found + let maybe_handle = h.ok_or(nope("repo not found"))?; + let unverified_handle = maybe_handle.unwrap_or("handle.invalid".to_string()); + + let (follows_count, followers_count) = { + let did = our_did.to_string(); + let sync_handle = state.sync_handle.clone(); + spawn_blocking(move || -> Result<_, LookupError> { + let following = sync_handle + .storage() + .get_counter(&following_count_key(&did))?; + let followers = sync_handle + .storage() + .get_counter(&followers_count_key(&did))?; + + Ok((following, followers)) + }) + .await + .expect("repo info lookup not to panic") + .map_err(|err| { + tracing::error!(%our_did, %err, "getRepoInfo lookup failed"); + internal("failed to read repo state") + })? + }; + + Ok(Json(GetProfileOutput { + did: our_did.parse().unwrap(), // TOOD + handle: unverified_handle.into(), + follows_count, + followers_count, + extra_data: None, + })) +} + +/// shortcut to make an xrpc error response +/// +/// the lexicon only has AccountNotFound, so we shove stuff into the message +fn nope(message: &str) -> XrpcErrorResponse { + let e = GetProfileError::AccountNotFound(Some(message.into())); + XrpcErrorResponse::new(StatusCode::BAD_REQUEST, XrpcError::Xrpc(e)) +} + +/// shortcut for non-client-errors +fn internal(msg: &str) -> XrpcErrorResponse { + let e = GetProfileError::Other { + error: "InternalError".into(), + message: Some(msg.into()), + }; + XrpcErrorResponse::new(StatusCode::BAD_REQUEST, XrpcError::Xrpc(e)) +} diff --git a/moots-0-counts/src/store.rs b/moots-0-counts/src/store.rs new file mode 100644 index 0000000..0ea3f25 --- /dev/null +++ b/moots-0-counts/src/store.rs @@ -0,0 +1,45 @@ +//! our storage keys +//! +//! hubble-sync prefixes all its keys with `&[0x00]`: we are allowed to share +//! column families with it as long as we make sure to never write a key with +//! that prefix. +//! +//! we use the counts column family, with hubble-sync's recommended merge +//! operator. keys: +//! +//! - `m:following:` => [count int] +//! - `m:followers:` => [count int] +//! +//! and we share hubble-sync's point-optimized column family to record follow +//! subjects by rkey, so that we can decrement counts on unfollow: +//! +//! - `m:following-rkey:\0` => +//! +//! a prefix scan on this key enumerates all our follow subjects + +pub static FOLLOWING_COUNT_PREFIX: &[u8] = b"m:following:"; +pub static FOLLOWERS_COUNT_PREFIX: &[u8] = b"m:followers:"; + +pub static FOLLOWING_RKEY_PREFIX: &[u8] = b"m:following-rkey:"; + +pub fn following_count_key(our_did: &str) -> Vec { + [FOLLOWING_COUNT_PREFIX, our_did.as_bytes()].concat() +} + +pub fn followers_count_key(their_did: &str) -> Vec { + [FOLLOWERS_COUNT_PREFIX, their_did.as_bytes()].concat() +} + +pub fn following_by_rkey_key(our_did: &str, rkey: &str) -> Vec { + [ + FOLLOWING_RKEY_PREFIX, + our_did.as_bytes(), + &[0x00], + rkey.as_bytes(), + ] + .concat() +} + +pub fn our_follows_prefix(our_did: &str) -> Vec { + [FOLLOWING_RKEY_PREFIX, our_did.as_bytes(), &[0x00]].concat() +} diff --git a/moots-0-counts/src/sync.rs b/moots-0-counts/src/sync.rs new file mode 100644 index 0000000..09d31e4 --- /dev/null +++ b/moots-0-counts/src/sync.rs @@ -0,0 +1,219 @@ +use dasl::drisl; +use hubble_sync::{ + AccountStatus, AppResult, Commit, ConsumerAppError, OpKind, RepoContext, ResyncData, + SimpleSyncConsumer, StorageBatch, StorageEngine, +}; +use hubble_sync_rocksdb::RocksEngine; +use jacquard_api::app_bsky::graph::follow::Follow; + +use crate::{Moots, store}; + +const BSKY_FOLLOWING_COLLECTION: &str = "app.bsky.graph.follow"; + +/// following more than 100k: reject this account (~meaningless to graph?) +const FOLLOWING_LIMIT: usize = 100_000; + +/// helper functions +impl Moots { + /// increment counts for a new follow record and index by rkey + fn put_follow( + &self, + our_did: &str, + their_did: &str, + our_rkey: &str, + batch: &mut ::Batch, + ) { + batch.increment_counter(&store::following_count_key(our_did), 1); + batch.increment_counter(&store::followers_count_key(their_did), 1); + batch.put( + &store::following_by_rkey_key(our_did, our_rkey), + their_did.as_bytes(), + ); + } + + /// decrement counts for an unfollow + /// + /// just does the database batch, with already-resolved dids and rkeys. + fn delete_follow( + &self, + our_did: &str, + their_did: &str, + our_rkey: &str, + batch: &mut ::Batch, + ) { + batch.increment_counter(&store::following_count_key(our_did), -1); + batch.increment_counter(&store::followers_count_key(their_did), -1); + batch.delete(&store::following_by_rkey_key(our_did, our_rkey)); + } + + /// clear follow counts and rkey index entries for an account + fn clear_follows( + &self, + our_did: &str, + batch: &mut ::Batch, + ) -> AppResult { + for pair in self + .storage + .scan_from(&store::our_follows_prefix(our_did), &[]) + { + let (our_rkey, their_did) = pair.map_err(ConsumerAppError::Storage)?; + let our_rkey = str::from_utf8(&our_rkey).map_err(|_| ConsumerAppError::Fatal { + reason: "db corruption: failed to decode rkey as utf8".to_string(), + })?; + let their_did = str::from_utf8(&their_did).map_err(|_| ConsumerAppError::Fatal { + reason: "db corruption: failed to decode did as utf8".to_string(), + })?; + self.delete_follow(our_did, their_did, our_rkey, batch); + } + Ok(()) + } +} + +/// this is our *entire* sync1.1 integration!!!! +impl SimpleSyncConsumer for Moots { + type Engine = RocksEngine; + + /// handle a single commit that arrived from the firehose + /// + /// as long as we put all of our *writes* on the provided `batch`, our + /// state will be updated with sync1.1-level consistency guarantee. + fn apply_commit( + &self, + commit: Commit, + ctx: &RepoContext, + batch: &mut ::Batch, + ) -> AppResult { + let our_did = ctx.did().as_str(); + + for op in commit + .ops + .iter() + .filter(|op| op.collection() == BSKY_FOLLOWING_COLLECTION) + { + match op.kind { + OpKind::Create { cid } => { + // new follow! + let record_bytes = commit.block(cid).unwrap(); // guaranteed present + + let Ok(follow) = drisl::from_slice::(record_bytes) else { + // failed to decode: ignore + tracing::debug!("failed to decode follow record, ignoring it"); + continue; + }; + + let their_did = follow.subject.as_str(); + self.put_follow(our_did, their_did, op.rkey(), batch); + } + OpKind::Delete { .. } => { + // unfollow! + + // we only get the **key** of deleted records from the + // firehose deletes, so we check our rkey index to find the + // follow subject that needs to be undone. + let our_rkey = op.rkey(); + let key = store::following_by_rkey_key(our_did, our_rkey); + + // it's actually safe to do this _read_ here without a + // transaction, as long as we only look at "our own keys" -- + // any keys which no other DID will write. hubble-sync + // ensures that only one event per actor/DID is running at + // a time. + let maybe_bytes = self.storage.get(&key).map_err(ConsumerAppError::Storage)?; + + // it's possible to receive `None` from that `get`, if the + // original record failed to parse (see above) + let Some(val_bytes) = maybe_bytes else { + tracing::debug!("ignoring missing rkey index entry"); + continue; + }; + + let their_did = + str::from_utf8(&val_bytes).map_err(|_| ConsumerAppError::Fatal { + reason: "db corruption: failed to decode did as utf8".to_string(), + })?; + + self.delete_follow(our_did, their_did, op.rkey(), batch); + } + OpKind::Update { .. } => { + // updates to follow records changing a subject are trash! + // meaningless operation!! + // this cuts to a sort of values question in atproto: who is + // the source of truth? the pds or the app? + // apps are who define the meaning of data, so we take the + // view that an app is *allowed to reject* meaningless data + // operations, leaving them in a different-from-pds state. + // this does mean that re-doing backfill can lead to a + // different app state!! ew! + tracing::debug!("rejecting update to follow record (meaningless)"); + } + } + } + Ok(()) + } + + /// reset repo state + /// + /// new account, or we lost firehose sync + fn apply_resync( + &self, + data: &mut ResyncData, + ctx: &RepoContext, + batch: &mut ::Batch, + ) -> AppResult { + let our_did = ctx.did().as_str(); + + // first: just clear everything (staged in batch) + self.clear_follows(our_did, batch)?; + + // then: reinsert all our follows from scratch (in same batch) + let mut following_total = 0; + while let Some(chunk) = data.walk.next_chunk(4096)? { + for entry in chunk { + let Some((collection, rkey)) = entry.key.split_once('/') else { + tracing::debug!(%our_did, %entry.key, "skipping invalid record path"); + continue; + }; + if collection != BSKY_FOLLOWING_COLLECTION { + continue; + } + let Ok(follow) = drisl::from_slice::(&entry.data) else { + // failed to decode: ignore this record + tracing::debug!("failed to decode follow record, ignoring it"); + continue; + }; + following_total += 1; + if following_total > FOLLOWING_LIMIT { + tracing::warn!(%our_did, "follows over limit: keeping partial state"); + return Err(ConsumerAppError::Desynchronize { + reason: format!("more than {FOLLOWING_LIMIT} following"), + }); + // TODO: hubble-sync follows a retry schedule for desyncs, but we + // probably don't want to waste effort on this account. need a way + // to signal to hubble-sync to put it on a slowwww schedule. + } + + let their_did = follow.subject.as_str(); + self.put_follow(our_did, their_did, rkey, batch); + } + } + + Ok(()) + } + + fn apply_status( + &self, + prev: &AccountStatus, + ctx: &RepoContext, + batch: &mut ::Batch, + ) -> AppResult { + use AccountStatus as S; + + // clear data for account deletes + let deleting = (*prev != S::Deleted) && (ctx.status() == S::Deleted); + if deleting { + self.clear_follows(ctx.did().as_str(), batch)?; + } + + Ok(()) + } +} diff --git a/moots-1-mutuals/Cargo.toml b/moots-1-mutuals/Cargo.toml new file mode 100644 index 0000000..35d2da8 --- /dev/null +++ b/moots-1-mutuals/Cargo.toml @@ -0,0 +1,31 @@ +[package] +name = "moots-1-mutuals" +version.workspace = true +edition.workspace = true + +[dependencies] +axum = { workspace = true } +clap = { workspace = true } +color-eyre = { workspace = true } +dasl = { workspace = true } +hubble-sync = { workspace = true } +hubble-sync-axum = { workspace = true } +hubble-sync-rocksdb = { workspace = true } +jacquard-api = { workspace = true } +jacquard-axum = { workspace = true } +jacquard-common = { workspace = true } +metrics = { workspace = true } +metrics-exporter-prometheus = { workspace = true } +moots-lexicons = { workspace = true } +rlimit = { workspace = true } +rocksdb = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +thiserror = { workspace = true } +tikv-jemallocator = { workspace = true } +tokio = { workspace = true } +tokio-util = { workspace = true } +tower-http = { workspace = true } +tracing = { workspace = true } +tracing-error = { workspace = true } +tracing-subscriber = { workspace = true } diff --git a/moots-1-simple/readme.md b/moots-1-mutuals/readme.md similarity index 100% rename from moots-1-simple/readme.md rename to moots-1-mutuals/readme.md diff --git a/moots-1-mutuals/src/lib.rs b/moots-1-mutuals/src/lib.rs new file mode 100644 index 0000000..0683428 --- /dev/null +++ b/moots-1-mutuals/src/lib.rs @@ -0,0 +1,9 @@ +use hubble_sync_rocksdb::RocksEngine; + +pub mod serve; +pub mod store; +pub mod sync; + +pub struct Moots { + pub storage: RocksEngine, +} diff --git a/moots-1-simple/src/main.rs b/moots-1-mutuals/src/main.rs similarity index 100% rename from moots-1-simple/src/main.rs rename to moots-1-mutuals/src/main.rs diff --git a/moots-1-simple/src/serve.rs b/moots-1-mutuals/src/serve.rs similarity index 100% rename from moots-1-simple/src/serve.rs rename to moots-1-mutuals/src/serve.rs diff --git a/moots-1-mutuals/src/store.rs b/moots-1-mutuals/src/store.rs new file mode 100644 index 0000000..c1b1a12 --- /dev/null +++ b/moots-1-mutuals/src/store.rs @@ -0,0 +1,89 @@ +//! our app keys and where they live +//! +//! everything goes in the shared point column family +//! ([`hubble_sync_rocksdb::DEFAULT_POINT_CF`]), alongside hubble-sync's own +//! state. hubble-sync reserves keys starting with `0x00`; our ascii prefixes +//! stay clear of that. +//! +//! - `follows:\0\0` (no value): one key per follow record, +//! ordered for reads: enumerate who `us` follows, or check "does `us` follow +//! `them`" with a bounded prefix seek. +//! - `followsrk:\0` (value: `them`): sync-side index. a firehose +//! delete only tells us `(us, rkey)`, so this key recovers `them` in order +//! to delete the main key. + +pub static FOLLOWS_KEY_PREFIX: &[u8] = b"follows:"; +pub static FOLLOWS_RKEY_PREFIX: &[u8] = b"followsrk:"; + +pub fn follows_key(us: &str, them: &str, rkey: &str) -> Vec { + [ + FOLLOWS_KEY_PREFIX, + us.as_bytes(), + &[0x00], + them.as_bytes(), + &[0x00], + rkey.as_bytes(), + ] + .concat() +} + +pub fn our_follows_prefix(us: &str) -> Vec { + [FOLLOWS_KEY_PREFIX, us.as_bytes(), &[0x00]].concat() +} + +pub fn follows_rkey_key(us: &str, rkey: &str) -> Vec { + [FOLLOWS_RKEY_PREFIX, us.as_bytes(), &[0x00], rkey.as_bytes()].concat() +} + +pub fn our_follows_rkey_prefix(us: &str) -> Vec { + [FOLLOWS_RKEY_PREFIX, us.as_bytes(), &[0x00]].concat() +} + +/// the `\0` tail of a [`follows_key`] +/// +/// scans trim the prefix they were given, so this is the key shape a scan of +/// [`our_follows_prefix`] yields. `follows_suffix_layout` keeps it honest. +pub fn follows_suffix(them: &str, rkey: &str) -> Vec { + [them.as_bytes(), &[0x00], rkey.as_bytes()].concat() +} + +/// split a [`follows_suffix`] back into `(them, rkey)` +/// +/// `None` if it isn't the expected shape, which would mean corrupt storage +pub fn split_follows_suffix(suffix: &[u8]) -> Option<(&str, &str)> { + let mut parts = suffix.splitn(2, |b| *b == 0x00); + let them = str::from_utf8(parts.next()?).ok()?; + let rkey = str::from_utf8(parts.next()?).ok()?; + Some((them, rkey)) +} + +#[cfg(test)] +mod tests { + use super::*; + + const US: &str = "did:web:bad-example.com"; + const THEM: &str = "did:plc:hdhoaan3xa3jiuq4fg4mefid"; + const RKEY: &str = "3l5xyzabcde2k"; + + /// the suffix helpers duplicate the key layout: prove they agree + #[test] + fn follows_suffix_layout() { + let key = follows_key(US, THEM, RKEY); + let tail = key + .strip_prefix(our_follows_prefix(US).as_slice()) + .expect("key starts with our prefix"); + assert_eq!(tail, follows_suffix(THEM, RKEY)); + } + + #[test] + fn follows_suffix_round_trips() { + let suffix = follows_suffix(THEM, RKEY); + assert_eq!(split_follows_suffix(&suffix), Some((THEM, RKEY))); + } + + #[test] + fn splitting_a_malformed_suffix_fails() { + // no separator, so there's no rkey to find + assert_eq!(split_follows_suffix(THEM.as_bytes()), None); + } +} diff --git a/moots-1-mutuals/src/sync.rs b/moots-1-mutuals/src/sync.rs new file mode 100644 index 0000000..9e68a6f --- /dev/null +++ b/moots-1-mutuals/src/sync.rs @@ -0,0 +1,261 @@ +use std::collections::HashSet; + +use hubble_sync::{ + AccountStatus, AppResult, Commit, ConsumerAppError, OpKind, RepoContext, ResyncData, + SimpleSyncConsumer, StorageBatch, StorageEngine, +}; +use hubble_sync_rocksdb::RocksEngine; +use jacquard_api::app_bsky::graph::follow::{Follow, FollowRecord}; +use jacquard_common::types::collection::Collection; + +use crate::{Moots, store}; + +/// following more than 100k: stop resync, lossy. moots kinda meaningless +const FOLLOWS_LIMIT: usize = 100_000; + +/// resync repo-walk chunk +const RESYNC_CHUNK: usize = 4096; + +impl Moots { + /// record a follow (and its rkey index for deletion) + /// + /// `us` and `them`: DIDs + fn put_follow( + &self, + batch: &mut ::Batch, + us: &str, + them: &str, + rkey: &str, + ) { + batch.put(&store::follows_key(us, them, rkey), b""); + batch.put(&store::follows_rkey_key(us, rkey), them.as_bytes()); + } + + /// forget a follow (the inverse of [`Self::put_follow`]) + fn delete_follow( + &self, + batch: &mut ::Batch, + us: &str, + them: &str, + rkey: &str, + ) { + batch.delete(&store::follows_key(us, them, rkey)); + batch.delete(&store::follows_rkey_key(us, rkey)); + } + + /// every follow we have stored for `us`, as `\0` suffixes + /// + /// this reads committed state, not the pending batch. batches are atomic, + /// so it never sees a half-applied earlier resync. + fn stored_follows(&self, us: &str) -> AppResult>> { + self.storage + .scan_from(&store::our_follows_prefix(us), b"") + .map(|pair| { + pair.map(|(suffix, _empty)| suffix) + .map_err(ConsumerAppError::Storage) + }) + .collect() + } + + /// stage deletes for all of `us`'s follow state + fn clear_follows( + &self, + batch: &mut ::Batch, + us: &str, + ) -> AppResult { + for pair in self.storage.scan_from(&store::our_follows_prefix(us), b"") { + let (suffix, _empty) = pair.map_err(ConsumerAppError::Storage)?; + let (them, rkey) = split_stored(us, &suffix)?; + self.delete_follow(batch, us, them, rkey); + } + Ok(()) + } +} + +/// split a stored follows key, treating a malformed one as fatal +/// +/// unlike record data from the network, these bytes are ours: a bad one means +/// our own storage is corrupt, so stop rather than guess. +fn split_stored<'k>(us: &str, suffix: &'k [u8]) -> AppResult { + store::split_follows_suffix(suffix).ok_or_else(|| ConsumerAppError::Fatal { + reason: format!("corrupt follows key for {us}: {suffix:?}"), + }) +} + +/// decode a follow record's cbor, warning + `None` for junk +/// +/// skipping junk is safe: a resync would replay the same bytes (and +/// desynchronizing over it would loop), so log it and move on. +fn decode_follow(us: &str, path: &str, data: &[u8]) -> Option { + match dasl::drisl::from_slice(data) { + Ok(follow) => Some(follow), + Err(e) => { + tracing::warn!(%us, %path, "undecodable follow record: {e}"); + None + } + } +} + +impl SimpleSyncConsumer for Moots { + type Engine = RocksEngine; + + fn apply_commit( + &self, + commit: Commit, + ctx: &RepoContext, + batch: &mut ::Batch, + ) -> AppResult { + let us = ctx.did().as_str(); + for op in commit.ops { + if op.collection() != FollowRecord::NSID { + // ignore everything that's not a change to follows + continue; + } + match op.kind { + OpKind::Create { cid } => { + let block = commit + .block(cid) + .expect("create record block present (sync1.1-validated)"); + let Some(follow) = decode_follow(us, &op.path, block) else { + continue; + }; + self.put_follow(batch, us, follow.subject.as_str(), op.rkey()); + } + OpKind::Delete { .. } => { + // a delete op doesn't carry the record body (sync1.1 only + // includes blocks for creates/updates), so recover `them` + // from our rkey index. reading the engine mid-batch is + // consistent: commits apply serially per-repo, and a single + // commit can't create and delete the same path (its MST + // diff would cancel out). + let them = self + .storage + .get(&store::follows_rkey_key(us, op.rkey())) + .map_err(ConsumerAppError::Storage)?; + let Some(them) = them else { + // expected for accounts we dropped (over FOLLOWS_LIMIT) + // or whose records didn't decode + tracing::debug!(%us, path = %op.path, "unfollow for a follow we don't have"); + continue; + }; + let them = String::from_utf8(them).map_err(|_| ConsumerAppError::Fatal { + reason: format!("corrupt followsrk value for {us}"), + })?; + self.delete_follow(batch, us, &them, op.rkey()); + } + OpKind::Update { .. } => { + tracing::warn!("ignoring update to follows (not a valid operation)"); + } + } + } + Ok(()) + } + + fn apply_resync( + &self, + data: &mut ResyncData, + ctx: &RepoContext, + batch: &mut ::Batch, + ) -> AppResult { + let us = ctx.did().as_str(); + + // diff the snapshot against what we already have, so a resync usually + // stages a handful of keys instead of rewriting the whole follow set. + // everything shares the one batch, so it lands atomically with + // hubble-sync's bookkeeping. + // + // `stale` starts as everything we have stored and loses each follow the + // snapshot still contains; whatever's left went away while we weren't + // looking. it holds one ~50 byte key per follow, so it stays within a + // few MB for accounts under FOLLOWS_LIMIT. + // + // future work: repo-stream guarantees walk order is sorted, and the + // followsrk index is rkey-ordered too, so the two could be merge-joined + // for O(1) memory instead of buffering this set. + let mut stale = self.stored_follows(us)?; + + let mut count = 0_usize; + 'walk: while let Some(chunk) = data.walk.next_chunk(RESYNC_CHUNK)? { + for rec in chunk { + let Some((collection, rkey)) = rec.key.split_once('/') else { + tracing::warn!(%us, key = %rec.key, "skipping invalid record path"); + continue; + }; + if collection != FollowRecord::NSID { + continue; + } + let Some(follow) = decode_follow(us, &rec.key, &rec.data) else { + continue; + }; + count += 1; + if count > FOLLOWS_LIMIT { + // we're giving up before seeing the whole follow set, so + // anything left in `stale` might still be real: leave it + // alone rather than deleting live follows. we keep what's + // staged plus what's stored, which can mean holding more + // follows than the account actually has. + tracing::warn!(%us, "follows over limit: keeping partial state"); + stale.clear(); + break 'walk; + } + let them = follow.subject.as_str(); + if !stale.remove(&store::follows_suffix(them, rkey)) { + // not one we already had + self.put_follow(batch, us, them, rkey); + } + } + } + + for suffix in &stale { + let (them, rkey) = split_stored(us, suffix)?; + self.delete_follow(batch, us, them, rkey); + } + + Ok(()) + } + + fn apply_status( + &self, + prev: &AccountStatus, + ctx: &RepoContext, + batch: &mut ::Batch, + ) -> AppResult { + use AccountStatus as S; + + // only clear data for explicit deletes + // takedowns and deactivations reversible etc., filter on read side + let deleting = (*prev != S::Deleted) && (ctx.status() == S::Deleted); + if deleting { + self.clear_follows(batch, ctx.did().as_str())?; + } + + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const US: &str = "did:web:bad-example.com"; + + /// records on the wire carry `$type` -- make sure the decode tolerates it + #[test] + fn decodes_pds_style_follow_record() { + let record = serde_json::json!({ + "$type": "app.bsky.graph.follow", + "subject": "did:plc:hdhoaan3xa3jiuq4fg4mefid", + "createdAt": "2026-08-11T00:00:00.000Z", + }); + let cbor = dasl::drisl::to_vec(&record).expect("encodes"); + + let follow = + decode_follow(US, "app.bsky.graph.follow/3l5xyzabcde2k", &cbor).expect("decodes"); + assert_eq!(follow.subject.as_str(), "did:plc:hdhoaan3xa3jiuq4fg4mefid"); + } + + #[test] + fn junk_record_decodes_to_none() { + assert!(decode_follow(US, "app.bsky.graph.follow/3l5xyzabcde2k", b"nonsense").is_none()); + } +} diff --git a/moots-1-simple/Cargo.toml b/moots-1-simple/Cargo.toml deleted file mode 100644 index 73532ba..0000000 --- a/moots-1-simple/Cargo.toml +++ /dev/null @@ -1,30 +0,0 @@ -[package] -name = "moots-1-simple" -version.workspace = true -edition.workspace = true - -[dependencies] -axum.workspace = true -clap.workspace = true -color-eyre.workspace = true -hubble-sync.workspace = true -hubble-sync-axum.workspace = true -hubble-sync-rocksdb.workspace = true -jacquard-axum.workspace = true -metrics.workspace = true -metrics-exporter-prometheus.workspace = true -moots-lexicons.workspace = true -rlimit.workspace = true -rocksdb.workspace = true -serde.workspace = true -serde_json.workspace = true -thiserror.workspace = true -tokio.workspace = true -tokio-util.workspace = true -tower-http.workspace = true -tracing.workspace = true -tracing-error.workspace = true -tracing-subscriber.workspace = true - -[target.'cfg(not(target_env = "msvc"))'.dependencies] -tikv-jemallocator.workspace = true diff --git a/moots-1-simple/src/store.rs b/moots-1-simple/src/store.rs deleted file mode 100644 index 48b6ed1..0000000 --- a/moots-1-simple/src/store.rs +++ /dev/null @@ -1,20 +0,0 @@ -pub static FOLLOWS_KEY_PREFIX: &[u8] = b"follows:"; - -pub fn follows_key(us: &str, them: &str, rkey: &str) -> Vec { - [ - FOLLOWS_KEY_PREFIX, - us.as_bytes(), - &[0x00], - them.as_bytes(), - &[0x00], - rkey.as_bytes(), - ].concat() -} - -pub fn our_follows_prefix(us: &str) -> Vec { - [ - FOLLOWS_KEY_PREFIX, - us.as_bytes(), - &[0x00], - ].concat() -} diff --git a/moots-1-simple/src/sync.rs b/moots-1-simple/src/sync.rs deleted file mode 100644 index b90823d..0000000 --- a/moots-1-simple/src/sync.rs +++ /dev/null @@ -1,70 +0,0 @@ -use hubble_sync::{ - AccountStatus, AppResult, Commit, RepoContext, ResyncData, SimpleSyncConsumer, StorageEngine, - OpKind, -}; -use hubble_sync_rocksdb::RocksEngine; - -use crate::{Moots, store}; - -/// if someone follows more than 1M people, we just drop them. moots meaningless. -const FOLLOWS_LIMIT: usize = 1_000_000; - -impl SimpleSyncConsumer for Moots { - type Engine = RocksEngine; - - fn apply_commit( - &self, - commit: Commit, - ctx: &RepoContext, - batch: &mut ::Batch, - ) -> AppResult { - for op in commit.ops { - if !op.path.starts_with("app.bsky.graph.follow/") { - // ignore everything that's not a change to follows - continue; - } - match op.kind { - OpKind::Create { cid } => { - // 1. get the block, decode cbor (i guess we should grab struct from jacq) - // 2. extract the follow subject - // 3. generate the key with our did (ctx), the subject (cbor), and the rkey (commit.path) - // 4. add a put for the key to the batch (empty value) - }, - OpKind::Delete { prev } => { - // 1. generate the key, same way as create (i guess we can combine things?) - // 2. add a delete for the key to the batch - }, - OpKind::Update { .. } => { - tracing::warn!("ignoring update to follows (not a valid operation)"); - } - } - } - Ok(()) - } - - fn apply_resync( - &self, - data: &mut ResyncData, - ctx: &RepoContext, - batch: &mut ::Batch, - ) -> AppResult { - Ok(()) - } - - fn apply_status( - &self, - prev: &AccountStatus, - ctx: &RepoContext, - batch: &mut ::Batch, - ) -> AppResult { - use AccountStatus as S; - - let deleting = (*prev != S::Deleted) && (ctx.status() == S::Deleted); - if deleting { - // let records = self.db.cf_handle("records").expect("records cf"); - // batch.inner_mut().delete_range_cf(&records, from, to); - } - - Ok(()) - } -} diff --git a/moots-2-fast-reads/Cargo.toml b/moots-2-fast-reads/Cargo.toml new file mode 100644 index 0000000..6d2b9f8 --- /dev/null +++ b/moots-2-fast-reads/Cargo.toml @@ -0,0 +1,7 @@ +[package] +name = "moots-2-fast-reads" +version.workspace = true +edition.workspace = true + +[dependencies] +tikv-jemallocator = { workspace = true } diff --git a/moots-2-read-optimized/readme.md b/moots-2-fast-reads/readme.md similarity index 100% rename from moots-2-read-optimized/readme.md rename to moots-2-fast-reads/readme.md diff --git a/moots-2-read-optimized/src/main.rs b/moots-2-fast-reads/src/main.rs similarity index 100% rename from moots-2-read-optimized/src/main.rs rename to moots-2-fast-reads/src/main.rs diff --git a/moots-2-read-optimized/Cargo.toml b/moots-2-read-optimized/Cargo.toml deleted file mode 100644 index df53939..0000000 --- a/moots-2-read-optimized/Cargo.toml +++ /dev/null @@ -1,30 +0,0 @@ -[package] -name = "moots-2-read-optimized" -version.workspace = true -edition.workspace = true - -[dependencies] -axum.workspace = true -clap.workspace = true -color-eyre.workspace = true -hubble-sync.workspace = true -hubble-sync-axum.workspace = true -hubble-sync-rocksdb.workspace = true -jacquard-axum.workspace = true -metrics.workspace = true -metrics-exporter-prometheus.workspace = true -moots-lexicons.workspace = true -rlimit.workspace = true -rocksdb.workspace = true -serde.workspace = true -serde_json.workspace = true -thiserror.workspace = true -tokio.workspace = true -tokio-util.workspace = true -tower-http.workspace = true -tracing.workspace = true -tracing-error.workspace = true -tracing-subscriber.workspace = true - -[target.'cfg(not(target_env = "msvc"))'.dependencies] -tikv-jemallocator.workspace = true diff --git a/moots-lexicons/Cargo.toml b/moots-lexicons/Cargo.toml index 9417af0..42d4240 100644 --- a/moots-lexicons/Cargo.toml +++ b/moots-lexicons/Cargo.toml @@ -7,8 +7,9 @@ version.workspace = true edition.workspace = true [features] -default = ["com_bad_example"] +default = ["com_bad_example", "de_atverkackt"] com_bad_example = [] +de_atverkackt = [] [dependencies] jacquard-common.workspace = true diff --git a/moots-lexicons/src/de_atverkackt.rs b/moots-lexicons/src/de_atverkackt.rs new file mode 100644 index 0000000..6e65aee --- /dev/null +++ b/moots-lexicons/src/de_atverkackt.rs @@ -0,0 +1,7 @@ +// @generated by jacquard-lexicon. DO NOT EDIT. +// +// This file was automatically generated from Lexicon schemas. +// Any manual changes will be overwritten on the next regeneration. + +//! Generated bindings for the `de.atverkackt` Lexicon namespace/module. +pub mod icarus; \ No newline at end of file diff --git a/moots-lexicons/src/de_atverkackt/icarus.rs b/moots-lexicons/src/de_atverkackt/icarus.rs new file mode 100644 index 0000000..0003e64 --- /dev/null +++ b/moots-lexicons/src/de_atverkackt/icarus.rs @@ -0,0 +1,7 @@ +// @generated by jacquard-lexicon. DO NOT EDIT. +// +// This file was automatically generated from Lexicon schemas. +// Any manual changes will be overwritten on the next regeneration. + +//! Generated bindings for the `de.atverkackt.icarus` Lexicon namespace/module. +pub mod get_profile; \ No newline at end of file diff --git a/moots-lexicons/src/de_atverkackt/icarus/get_profile.rs b/moots-lexicons/src/de_atverkackt/icarus/get_profile.rs new file mode 100644 index 0000000..343448a --- /dev/null +++ b/moots-lexicons/src/de_atverkackt/icarus/get_profile.rs @@ -0,0 +1,240 @@ +// @generated by jacquard-lexicon. DO NOT EDIT. +// +// Lexicon: de.atverkackt.icarus.getProfile +// +// This file was automatically generated from Lexicon schemas. +// Any manual changes will be overwritten on the next regeneration. + +#[derive( + serde::Serialize, + serde::Deserialize, + Debug, + Clone, + PartialEq, + Eq, + jacquard_derive::IntoStatic +)] + +#[serde( + rename_all = "camelCase", + bound(deserialize = "S: serde::Deserialize<'de> + jacquard_common::BosStr") +)] +pub struct GetProfile { + pub actor: jacquard_common::types::ident::AtIdentifier, +} + + +#[derive( + serde::Serialize, + serde::Deserialize, + Debug, + Clone, + PartialEq, + Eq, + jacquard_derive::IntoStatic +)] + +#[serde( + rename_all = "camelCase", + bound(deserialize = "S: serde::Deserialize<'de> + jacquard_common::BosStr") +)] +pub struct GetProfileOutput { + pub did: jacquard_common::types::string::Did, + pub followers_count: i64, + pub follows_count: i64, + pub handle: jacquard_common::types::string::Handle, + #[serde(flatten, default, skip_serializing_if = "core::option::Option::is_none")] + pub extra_data: core::option::Option< + alloc::collections::BTreeMap< + jacquard_common::deps::smol_str::SmolStr, + jacquard_common::types::value::Data, + >, + >, +} + + +#[derive( + serde::Serialize, + serde::Deserialize, + Debug, + Clone, + PartialEq, + Eq, + thiserror::Error, + miette::Diagnostic +)] + +#[serde(tag = "error", content = "message")] +pub enum GetProfileError { + /// The requested account is unknown to this service. + #[serde(rename = "AccountNotFound")] + AccountNotFound(core::option::Option), + /// Catch-all for unknown error codes. + #[serde(untagged)] + Other { + error: jacquard_common::deps::smol_str::SmolStr, + message: Option, + }, +} + +impl core::fmt::Display for GetProfileError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + match self { + Self::AccountNotFound(msg) => { + write!(f, "AccountNotFound")?; + if let Some(msg) = msg { + write!(f, ": {}", msg)?; + } + Ok(()) + } + Self::Other { error, message } => { + write!(f, "{}", error)?; + if let Some(msg) = message { + write!(f, ": {}", msg)?; + } + Ok(()) + } + } + } +} + +/** Response marker for the `de.atverkackt.icarus.getProfile` query. + +Implements `jacquard_common::xrpc::XrpcResp`; successful bodies decode as `Self::Output`, which is `GetProfileOutput` for this endpoint.*/ +pub struct GetProfileResponse; +impl jacquard_common::xrpc::XrpcResp for GetProfileResponse { + const NSID: &'static str = "de.atverkackt.icarus.getProfile"; + const ENCODING: &'static str = "application/json"; + type Output = GetProfileOutput; + type Err = GetProfileError; +} + +impl jacquard_common::xrpc::XrpcRequest for GetProfile { + const NSID: &'static str = "de.atverkackt.icarus.getProfile"; + const METHOD: jacquard_common::xrpc::XrpcMethod = jacquard_common::xrpc::XrpcMethod::Query; + type Response = GetProfileResponse; +} + +/** Endpoint marker for the `de.atverkackt.icarus.getProfile` query. + +Path: `/xrpc/de.atverkackt.icarus.getProfile`. The request payload type is `GetProfile`; send that request with `jacquard::Client` or use this marker through lower-level `XrpcEndpoint` APIs.*/ +pub struct GetProfileRequest; +impl jacquard_common::xrpc::XrpcEndpoint for GetProfileRequest { + const PATH: &'static str = "/xrpc/de.atverkackt.icarus.getProfile"; + const METHOD: jacquard_common::xrpc::XrpcMethod = jacquard_common::xrpc::XrpcMethod::Query; + type Request = GetProfile; + type Response = GetProfileResponse; +} + +pub mod get_profile_state { + + pub use crate::builder_types::{Set, Unset, IsSet, IsUnset}; + #[allow(unused)] + use ::core::marker::PhantomData; + mod sealed { + pub trait Sealed {} + } + /// State trait tracking which required fields have been set + pub trait State: sealed::Sealed { + type Actor; + } + /// Empty state - all required fields are unset + pub struct Empty(()); + impl sealed::Sealed for Empty {} + impl State for Empty { + type Actor = Unset; + } + ///State transition - sets the `actor` field to Set + pub struct SetActor(PhantomData St>); + impl sealed::Sealed for SetActor {} + impl State for SetActor { + type Actor = Set; + } + /// Marker types for field names + #[allow(non_camel_case_types)] + pub mod members { + ///Marker type for the `actor` field + pub struct actor(()); + } +} + +/// Builder for constructing an instance of this type. +pub struct GetProfileBuilder< + St: get_profile_state::State, + S: jacquard_common::BosStr = jacquard_common::DefaultStr, +> { + _state: ::core::marker::PhantomData St>, + _fields: (core::option::Option>,), + _type: ::core::marker::PhantomData S>, +} + +impl GetProfile { + /// Create a new builder for this type, using the default string type (DefaultStr = SmolStr) if needed + pub fn new() -> GetProfileBuilder< + get_profile_state::Empty, + jacquard_common::DefaultStr, + > { + GetProfileBuilder::new() + } +} + +impl GetProfile { + /// Create a new builder for this type + pub fn builder() -> GetProfileBuilder { + GetProfileBuilder::builder() + } +} + +impl GetProfileBuilder { + /// Create a new builder with all fields unset, using the default string type, if needed + pub fn new() -> Self { + GetProfileBuilder { + _state: ::core::marker::PhantomData, + _fields: (None,), + _type: ::core::marker::PhantomData, + } + } +} + +impl GetProfileBuilder { + /// Create a new builder with all fields unset + pub fn builder() -> Self { + GetProfileBuilder { + _state: ::core::marker::PhantomData, + _fields: (None,), + _type: ::core::marker::PhantomData, + } + } +} + +impl GetProfileBuilder +where + St: get_profile_state::State, + St::Actor: get_profile_state::IsUnset, +{ + /// Set the `actor` field (required) + pub fn actor( + mut self, + value: impl Into>, + ) -> GetProfileBuilder, S> { + self._fields.0 = ::core::option::Option::Some(value.into()); + GetProfileBuilder { + _state: ::core::marker::PhantomData, + _fields: self._fields, + _type: ::core::marker::PhantomData, + } + } +} + +impl GetProfileBuilder +where + St: get_profile_state::State, + St::Actor: get_profile_state::IsSet, +{ + /// Build the final struct. + pub fn build(self) -> GetProfile { + GetProfile { + actor: self._fields.0.unwrap(), + } + } +} \ No newline at end of file diff --git a/moots-lexicons/src/lib.rs b/moots-lexicons/src/lib.rs index 4dc7404..2a80898 100644 --- a/moots-lexicons/src/lib.rs +++ b/moots-lexicons/src/lib.rs @@ -45,4 +45,8 @@ pub mod builder_types; #[cfg(feature = "com_bad_example")] -pub mod com_bad_example; \ No newline at end of file +pub mod com_bad_example; + + +#[cfg(feature = "de_atverkackt")] +pub mod de_atverkackt; \ No newline at end of file diff --git a/readme.md b/readme.md index 73a89e6..981a6c7 100644 --- a/readme.md +++ b/readme.md @@ -6,5 +6,5 @@ wip ## codegen for lexicons ```bash -cargo run -p moots-lexgen +cargo run -p lexgen ``` -- 2.51.2