From 4f7db3e42819449ef641d237612e4fce0300cc73 Mon Sep 17 00:00:00 2001 From: npub13n66s06epmqf2kc3v373ez8hj65cuzyvxzjf93vwpervxqn2u7jq2qd9je <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Date: Sat, 1 Aug 2026 10:15:11 -0400 Subject: [PATCH] refactor(relay): establish single-node backend seams Signed-off-by: npub13n66s06epmqf2kc3v373ez8hj65cuzyvxzjf93vwpervxqn2u7jq2qd9je <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> --- crates/buzz-media/src/storage.rs | 56 ++++++-- crates/buzz-pubsub/src/lib.rs | 191 ++++++++++++++++++++++++- crates/buzz-pubsub/src/rate_limiter.rs | 87 +++++++++++ crates/buzz-relay/src/config.rs | 34 +++++ crates/buzz-relay/src/local_mode.rs | 4 +- crates/buzz-relay/src/main.rs | 26 +++- crates/buzz-relay/src/router.rs | 7 +- crates/buzz-relay/src/state.rs | 94 +++++++++--- crates/buzz-search/src/error.rs | 3 + crates/buzz-search/src/lib.rs | 30 +++- 10 files changed, 477 insertions(+), 55 deletions(-) diff --git a/crates/buzz-media/src/storage.rs b/crates/buzz-media/src/storage.rs index cbf980201f..ffed5ef07a 100644 --- a/crates/buzz-media/src/storage.rs +++ b/crates/buzz-media/src/storage.rs @@ -17,6 +17,14 @@ pub type ByteStream = Pin, } @@ -66,7 +74,15 @@ impl MediaStorage { S3AddressingStyle::Path => bucket.with_path_style(), S3AddressingStyle::Virtual => bucket, }; - Ok(Self { bucket }) + Ok(Self { + backend: MediaBackend::S3(S3MediaStorage { bucket }), + }) + } + + fn s3(&self) -> &S3MediaStorage { + match &self.backend { + MediaBackend::S3(storage) => storage, + } } /// Store an object from a byte slice. @@ -74,7 +90,8 @@ impl MediaStorage { /// Used for images, sidecars, and thumbnails. For large video files use /// [`put_file`] to avoid loading the entire blob into RAM. pub async fn put(&self, key: &str, bytes: &[u8], content_type: &str) -> Result<(), MediaError> { - self.bucket + self.s3() + .bucket .put_object_with_content_type(key, bytes, content_type) .await?; Ok(()) @@ -98,7 +115,8 @@ impl MediaStorage { .map_err(|e| MediaError::Io(e.to_string()))?; let mut reader = tokio::io::BufReader::with_capacity(BUF, file); - self.bucket + self.s3() + .bucket .put_object_stream_with_content_type(&mut reader, key, content_type) .await?; Ok(()) @@ -106,7 +124,7 @@ impl MediaStorage { /// Retrieve an object's bytes. pub async fn get(&self, key: &str) -> Result, MediaError> { - match self.bucket.get_object(key).await { + match self.s3().bucket.get_object(key).await { Ok(response) => Ok(response.to_vec()), Err(s3::error::S3Error::HttpFailWithBody(404, _)) => Err(MediaError::NotFound), Err(e) => Err(MediaError::StorageError(e.to_string())), @@ -119,7 +137,12 @@ impl MediaStorage { /// is transferred from S3 — the full object is never loaded into RAM. /// Intended for HTTP 206 range responses on large video blobs. pub async fn get_range(&self, key: &str, start: u64, end: u64) -> Result, MediaError> { - match self.bucket.get_object_range(key, start, Some(end)).await { + match self + .s3() + .bucket + .get_object_range(key, start, Some(end)) + .await + { Ok(response) => Ok(response.to_vec()), Err(s3::error::S3Error::HttpFailWithBody(404, _)) => Err(MediaError::NotFound), Err(e) => Err(MediaError::StorageError(e.to_string())), @@ -133,6 +156,7 @@ impl MediaStorage { /// blobs (video) directly into HTTP responses via `Body::from_stream()`. pub async fn get_stream(&self, key: &str) -> Result { let response = self + .s3() .bucket .get_object_stream(key) .await @@ -150,7 +174,7 @@ impl MediaStorage { /// Check if an object exists. Returns false on 404. pub async fn head(&self, key: &str) -> Result { - match self.bucket.head_object(key).await { + match self.s3().bucket.head_object(key).await { Ok(_) => Ok(true), Err(s3::error::S3Error::HttpFailWithBody(404, _)) => Ok(false), Err(e) => Err(MediaError::StorageError(e.to_string())), @@ -159,7 +183,8 @@ impl MediaStorage { /// Delete an object. Returns an error on failure — callers decide whether to propagate. pub async fn delete(&self, key: &str) -> Result<(), MediaError> { - self.bucket + self.s3() + .bucket .delete_object(key) .await .map_err(|e| MediaError::StorageError(e.to_string()))?; @@ -168,7 +193,7 @@ impl MediaStorage { /// HEAD with metadata — returns Content-Length (size). pub async fn head_with_metadata(&self, key: &str) -> Result, MediaError> { - match self.bucket.head_object(key).await { + match self.s3().bucket.head_object(key).await { Ok((result, _)) => Ok(Some(BlobHeadMeta { size: result.content_length.unwrap_or(0) as u64, })), @@ -199,7 +224,7 @@ impl MediaStorage { sha256: &str, ) -> Result { let key = Self::ctx_sidecar_key(ctx, sha256); - let resp = self.bucket.get_object(&key).await?; + let resp = self.s3().bucket.get_object(&key).await?; let meta: BlobMeta = serde_json::from_slice(&resp.to_vec())?; Ok(meta) } @@ -248,6 +273,7 @@ impl MediaStorage { max_keys: usize, ) -> Result { let (result, _status) = self + .s3() .bucket .list_page( String::new(), @@ -307,8 +333,8 @@ mod tests { fn static_keys_build_client_with_configured_region() { let storage = MediaStorage::new(&storage_config("buzz_dev", "buzz_dev_secret")) .expect("static creds should build a client"); - match storage.bucket.region { - Region::Custom { ref region, .. } => assert_eq!(region, "us-west-2"), + match &storage.s3().bucket.region { + Region::Custom { region, .. } => assert_eq!(region, "us-west-2"), other => panic!("expected Custom region, got {other:?}"), } } @@ -317,15 +343,15 @@ mod tests { fn client_constructor_applies_both_addressing_styles() { let path = MediaStorage::new(&storage_config("buzz_dev", "buzz_dev_secret")) .expect("path-style client"); - assert!(path.bucket.is_path_style()); - assert_eq!(path.bucket.url(), "http://localhost:9000/buzz-media"); + assert!(path.s3().bucket.is_path_style()); + assert_eq!(path.s3().bucket.url(), "http://localhost:9000/buzz-media"); let mut virtual_config = storage_config("buzz_dev", "buzz_dev_secret"); virtual_config.s3_addressing_style = S3AddressingStyle::Virtual; let virtual_hosted = MediaStorage::new(&virtual_config).expect("virtual-hosted client"); - assert!(virtual_hosted.bucket.is_subdomain_style()); + assert!(virtual_hosted.s3().bucket.is_subdomain_style()); assert_eq!( - virtual_hosted.bucket.url(), + virtual_hosted.s3().bucket.url(), "http://buzz-media.localhost:9000" ); } diff --git a/crates/buzz-pubsub/src/lib.rs b/crates/buzz-pubsub/src/lib.rs index 4f1690beef..b0dd9510fb 100644 --- a/crates/buzz-pubsub/src/lib.rs +++ b/crates/buzz-pubsub/src/lib.rs @@ -96,8 +96,190 @@ impl PubSubConfig { } } -/// Central pub/sub manager for a Buzz relay instance. +/// Backend-neutral pub/sub handle selected once at relay startup. +/// +/// Request handlers use this stable surface; backend selection never occurs in +/// the event, presence, or subscription paths. pub struct PubSubManager { + backend: PubSubBackend, +} + +enum PubSubBackend { + Redis(Arc), +} + +impl PubSubManager { + /// Creates the production Redis pub/sub backend. + pub async fn new(redis_url: &str, pool: deadpool_redis::Pool) -> Result { + Self::with_config(PubSubConfig::new(redis_url), pool).await + } + + /// Creates the production Redis backend with explicit configuration. + pub async fn with_config( + config: PubSubConfig, + pool: deadpool_redis::Pool, + ) -> Result { + Ok(Self { + backend: PubSubBackend::Redis(Arc::new( + RedisPubSubManager::with_config(config, pool).await?, + )), + }) + } + + /// Runs the backend event subscriber. Process-local backends may implement + /// this as a pending task because publication already reaches local receivers. + pub async fn run_subscriber(self: Arc) { + match &self.backend { + PubSubBackend::Redis(backend) => Arc::clone(backend).run_subscriber().await, + } + } + + /// Runs the backend cache-invalidation subscriber. + pub async fn run_cache_invalidation_subscriber(self: Arc) { + match &self.backend { + PubSubBackend::Redis(backend) => { + Arc::clone(backend) + .run_cache_invalidation_subscriber() + .await + } + } + } + + /// Runs the backend connection-control subscriber. + pub async fn run_conn_control_subscriber(self: Arc) { + match &self.backend { + PubSubBackend::Redis(backend) => { + Arc::clone(backend).run_conn_control_subscriber().await + } + } + } + + /// Returns a receiver for locally visible channel events. + pub fn subscribe_local(&self) -> broadcast::Receiver { + match &self.backend { + PubSubBackend::Redis(backend) => backend.subscribe_local(), + } + } + + /// Retains interest in a scoped event topic. + pub async fn retain_topic(&self, ctx: &TenantContext, topic: EventTopic) { + match &self.backend { + PubSubBackend::Redis(backend) => backend.retain_topic(ctx, topic).await, + } + } + + /// Releases interest in a scoped event topic. + pub async fn release_topic(&self, ctx: &TenantContext, topic: EventTopic) { + match &self.backend { + PubSubBackend::Redis(backend) => backend.release_topic(ctx, topic).await, + } + } + + /// Returns the current topic retain count. + pub async fn topic_refcount(&self, ctx: &TenantContext, topic: EventTopic) -> usize { + match &self.backend { + PubSubBackend::Redis(backend) => backend.topic_refcount(ctx, topic).await, + } + } + + /// Returns a receiver for cache invalidations. + pub fn subscribe_cache_invalidations(&self) -> broadcast::Receiver { + match &self.backend { + PubSubBackend::Redis(backend) => backend.subscribe_cache_invalidations(), + } + } + + /// Returns a receiver for connection-control commands. + pub fn subscribe_conn_control(&self) -> broadcast::Receiver { + match &self.backend { + PubSubBackend::Redis(backend) => backend.subscribe_conn_control(), + } + } + + /// Publishes a cache invalidation. + pub async fn publish_cache_invalidation( + &self, + ctx: &TenantContext, + invalidation: &CacheInvalidation, + ) -> Result { + match &self.backend { + PubSubBackend::Redis(backend) => { + backend.publish_cache_invalidation(ctx, invalidation).await + } + } + } + + /// Publishes a connection-control command. + pub async fn publish_conn_control( + &self, + ctx: &TenantContext, + command: &ConnControl, + ) -> Result { + match &self.backend { + PubSubBackend::Redis(backend) => backend.publish_conn_control(ctx, command).await, + } + } + + /// Publishes an event to the selected backend. + pub async fn publish_event( + &self, + ctx: &TenantContext, + topic: EventTopic, + event: &nostr::Event, + ) -> Result { + match &self.backend { + PubSubBackend::Redis(backend) => backend.publish_event(ctx, topic, event).await, + } + } + + /// Sets presence for a principal. + pub async fn set_presence( + &self, + ctx: &TenantContext, + pubkey: &PublicKey, + status: &str, + ) -> Result<(), PubSubError> { + match &self.backend { + PubSubBackend::Redis(backend) => backend.set_presence(ctx, pubkey, status).await, + } + } + + /// Clears presence for a principal. + pub async fn clear_presence( + &self, + ctx: &TenantContext, + pubkey: &PublicKey, + ) -> Result<(), PubSubError> { + match &self.backend { + PubSubBackend::Redis(backend) => backend.clear_presence(ctx, pubkey).await, + } + } + + /// Gets presence for one principal. + pub async fn get_presence( + &self, + ctx: &TenantContext, + pubkey: &PublicKey, + ) -> Result, PubSubError> { + match &self.backend { + PubSubBackend::Redis(backend) => backend.get_presence(ctx, pubkey).await, + } + } + + /// Gets presence for multiple principals. + pub async fn get_presence_bulk( + &self, + ctx: &TenantContext, + pubkeys: &[PublicKey], + ) -> Result, PubSubError> { + match &self.backend { + PubSubBackend::Redis(backend) => backend.get_presence_bulk(ctx, pubkeys).await, + } + } +} + +/// Redis implementation behind [`PubSubManager`]. +struct RedisPubSubManager { pool: deadpool_redis::Pool, /// Redis URL used by the reconnect loop to re-establish pub/sub connections. redis_url: String, @@ -112,12 +294,7 @@ pub struct PubSubManager { conn_control_tx: broadcast::Sender, } -impl PubSubManager { - /// Creates a new `PubSubManager` connected to the given Redis URL. - pub async fn new(redis_url: &str, pool: deadpool_redis::Pool) -> Result { - Self::with_config(PubSubConfig::new(redis_url), pool).await - } - +impl RedisPubSubManager { /// Creates a new `PubSubManager` using explicit pub/sub configuration. pub async fn with_config( config: PubSubConfig, diff --git a/crates/buzz-pubsub/src/rate_limiter.rs b/crates/buzz-pubsub/src/rate_limiter.rs index 8d5494f1d5..2678573c52 100644 --- a/crates/buzz-pubsub/src/rate_limiter.rs +++ b/crates/buzz-pubsub/src/rate_limiter.rs @@ -78,6 +78,60 @@ async fn run_rate_limit( } } +/// Backend-neutral admission limiter selected once at relay startup. +pub enum AdmissionRateLimiter { + /// Production Redis fixed-window counters. + Redis(RedisRateLimiter), + /// Process-local profile policy until bounded in-memory counters are installed. + /// + /// This is explicit rather than a fallback: callers must select it at startup. + Permissive, +} + +impl AdmissionRateLimiter { + /// Creates the production Redis limiter. + pub fn redis(pool: deadpool_redis::Pool) -> Self { + Self::Redis(RedisRateLimiter::new(pool)) + } + + /// Creates the explicit permissive policy for a process-local relay. + pub fn permissive() -> Self { + Self::Permissive + } +} + +impl RateLimiter for AdmissionRateLimiter { + async fn check_and_increment( + &self, + ctx: &TenantContext, + pubkey: &PublicKey, + limit_type: LimitType, + window_secs: u64, + limit: u64, + ) -> Result { + match self { + Self::Redis(limiter) => { + limiter + .check_and_increment(ctx, pubkey, limit_type, window_secs, limit) + .await + } + Self::Permissive => Ok(RateLimitResult::allowed(1, limit, window_secs)), + } + } + + async fn check_ip_connection( + &self, + ip: &IpAddr, + window_secs: u64, + limit: u64, + ) -> Result { + match self { + Self::Redis(limiter) => limiter.check_ip_connection(ip, window_secs, limit).await, + Self::Permissive => Ok(RateLimitResult::allowed(1, limit, window_secs)), + } + } +} + /// Redis-backed rate limiter using fixed-window counters. /// /// Pubkey keys are community-scoped via `&TenantContext`: @@ -119,3 +173,36 @@ impl RateLimiter for RedisRateLimiter { run_rate_limit(&self.pool, &key, window_secs, limit).await } } + +#[cfg(test)] +mod admission_backend_tests { + use super::*; + use buzz_core::CommunityId; + use nostr::Keys; + use std::net::{IpAddr, Ipv4Addr}; + use uuid::Uuid; + + #[tokio::test] + async fn permissive_backend_allows_principal_and_ip_without_redis() { + let limiter = AdmissionRateLimiter::permissive(); + let tenant = + TenantContext::resolved(CommunityId::from_uuid(Uuid::from_u128(1)), "local.invalid"); + let principal = limiter + .check_and_increment( + &tenant, + &Keys::generate().public_key(), + LimitType::Messages, + 60, + 10, + ) + .await + .expect("permissive principal policy"); + let ip = limiter + .check_ip_connection(&IpAddr::V4(Ipv4Addr::LOCALHOST), 60, 10) + .await + .expect("permissive IP policy"); + + assert!(principal.allowed); + assert!(ip.allowed); + } +} diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index 037c6b1dd3..6e75e6e782 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -50,9 +50,41 @@ pub struct JoinPolicyConfig { /// WebSocket close-frame delivery after the final delayed cancellation. pub const MAX_DRAIN_JITTER_MS: u64 = 20_000; +/// Relay runtime profile. The default preserves the production service graph; +/// `single-node` selects only process-local or filesystem-backed services. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RelayProfile { + /// Postgres, Redis, S3, and Postgres FTS. + Production, + /// One-process local relay with no external service dependencies. + SingleNode, +} + +impl RelayProfile { + /// Whether this profile must avoid every external service connection. + pub fn is_single_node(self) -> bool { + matches!(self, Self::SingleNode) + } + + fn from_env() -> Result { + match std::env::var("BUZZ_PROFILE") { + Err(_) => Ok(Self::Production), + Ok(value) if value.trim().is_empty() || value.eq_ignore_ascii_case("production") => { + Ok(Self::Production) + } + Ok(value) if value.eq_ignore_ascii_case("single-node") => Ok(Self::SingleNode), + Ok(value) => Err(ConfigError::InvalidValue(format!( + "BUZZ_PROFILE must be 'production' or 'single-node', got {value:?}" + ))), + } + } +} + /// Relay runtime configuration, loaded from environment variables. #[derive(Debug, Clone)] pub struct Config { + /// Service topology selected by `BUZZ_PROFILE`. + pub profile: RelayProfile, /// Address the relay HTTP/WebSocket server binds to. pub bind_addr: SocketAddr, /// Postgres database connection URL. @@ -459,6 +491,7 @@ fn inert_env_vars<'a>(names: &[&'a str], lookup: impl Fn(&str) -> Option impl Config { /// Loads configuration from environment variables, falling back to development defaults. pub fn from_env() -> Result { + let profile = RelayProfile::from_env()?; let bind_addr_raw = std::env::var("BUZZ_BIND_ADDR").unwrap_or_else(|_| "0.0.0.0:3000".to_string()); let bind_addr = parse_bind_addr(&bind_addr_raw)?; @@ -987,6 +1020,7 @@ impl Config { } Ok(Self { + profile, bind_addr, database_url, read_database_url, diff --git a/crates/buzz-relay/src/local_mode.rs b/crates/buzz-relay/src/local_mode.rs index 39b7f2d5f6..14af524149 100644 --- a/crates/buzz-relay/src/local_mode.rs +++ b/crates/buzz-relay/src/local_mode.rs @@ -32,10 +32,10 @@ pub(crate) struct Config { impl Config { pub(crate) fn from_env() -> anyhow::Result> { - let enabled = std::env::var("BUZZ_LOCAL_MODE") + let legacy_spike_enabled = std::env::var("BUZZ_LOCAL_MODE") .ok() .is_some_and(|value| matches!(value.trim(), "1" | "true" | "on")); - if !enabled { + if !legacy_spike_enabled { return Ok(None); } let bind_addr = std::env::var("BUZZ_BIND_ADDR") diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index d701c388db..fb55175326 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -159,6 +159,15 @@ async fn main() -> anyhow::Result<()> { "Config loaded" ); + // Phase 1 constructor fence: never fall through to production's eager + // Postgres/Redis/S3 graph for a single-node profile. The local backend + // constructor replaces this explicit denial as its implementations land. + if config.profile.is_single_node() { + return Err(anyhow::anyhow!( + "BUZZ_PROFILE=single-node local backend bundle is not installed yet" + )); + } + let usage_interval_secs = usage_metrics_interval_secs(); let usage_idle_timeout_secs = usage_metrics_idle_timeout_secs(usage_interval_secs); relay_metrics::install(config.metrics_port, usage_idle_timeout_secs); @@ -474,7 +483,10 @@ async fn main() -> anyhow::Result<()> { // operator who asked for the mesh gets it or gets told why not. if let Some(handle) = buzz_relay::mesh_boot::boot_mesh( &state.config, - state.redis_pool.clone(), + state + .redis_pool + .clone() + .expect("production profile has Redis"), &state.relay_keypair, Arc::clone(&state.shutting_down), ) @@ -1019,11 +1031,13 @@ async fn main() -> anyhow::Result<()> { } } - let rs = pool_state.redis_pool.status(); - metrics::gauge!("buzz_redis_pool_available").set(rs.available as f64); - metrics::gauge!("buzz_redis_pool_size").set(rs.size as f64); - metrics::gauge!("buzz_redis_pool_max").set(rs.max_size as f64); - metrics::gauge!("buzz_redis_pool_waiting").set(rs.waiting as f64); + if let Some(redis_pool) = &pool_state.redis_pool { + let rs = redis_pool.status(); + metrics::gauge!("buzz_redis_pool_available").set(rs.available as f64); + metrics::gauge!("buzz_redis_pool_size").set(rs.size as f64); + metrics::gauge!("buzz_redis_pool_max").set(rs.max_size as f64); + metrics::gauge!("buzz_redis_pool_waiting").set(rs.waiting as f64); + } } }); } diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index 400ed1dfe3..379f0ead17 100644 --- a/crates/buzz-relay/src/router.rs +++ b/crates/buzz-relay/src/router.rs @@ -377,8 +377,11 @@ async fn readiness_handler(State(state): State>) -> impl IntoRespo let check = async { let (pg_ok, redis_ok) = tokio::join!(state.db.ping(), async { - state.redis_pool.get().await.is_ok() - },); + match &state.redis_pool { + Some(pool) => pool.get().await.is_ok(), + None => true, + } + }); (pg_ok, redis_ok) }; diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 14a50df7b7..9fc62f5ac4 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -24,7 +24,7 @@ use buzz_db::Db; use buzz_media::MediaStorage; use buzz_pubsub::cache_invalidation::CacheInvalidation; use buzz_pubsub::conn_control::ConnControl; -use buzz_pubsub::rate_limiter::RedisRateLimiter; +use buzz_pubsub::rate_limiter::AdmissionRateLimiter; use buzz_pubsub::{PubSubManager, RedisNip98ReplayGuard}; use buzz_search::SearchService; use buzz_workflow::WorkflowEngine; @@ -575,6 +575,29 @@ impl Default for ConnectionManager { } } +/// Runtime backends injected at the application-state boundary. +/// +/// Production construction remains in `main.rs`; the single-node profile supplies +/// alternate implementations without adding backend checks to request handlers. +pub struct AppBackends { + /// Database backend. + pub db: Db, + /// Redis health/command pool; absent for process-local profiles. + pub redis_pool: Option, + /// Event and control-plane pub/sub. + pub pubsub: Arc, + /// Full-text search backend. + pub search: SearchService, + /// Media blob backend. + pub media_storage: MediaStorage, + /// Git object backend. + pub git_store: crate::api::git::store::GitStore, + /// NIP-98 replay fence. + pub nip98_replay: Arc, + /// Admission rate limiter. + pub admission_rate_limiter: Arc, +} + /// Shared application state, cloned cheaply via inner `Arc` fields. #[derive(Clone)] pub struct AppState { @@ -582,8 +605,8 @@ pub struct AppState { pub config: Arc, /// Database connection pool. pub db: Db, - /// Redis pool for readiness health checks. - pub redis_pool: deadpool_redis::Pool, + /// Redis pool for readiness health checks; absent in single-node mode. + pub redis_pool: Option, /// Audit event service, absent when audit logging is disabled. pub audit: Option>, /// Pub/sub manager for broadcasting events to subscribers. @@ -673,7 +696,7 @@ pub struct AppState { /// cross-pod routing. pub nip98_replay: Arc, /// Shared Redis-backed admission limits for ordinary HTTP and WebSocket work. - pub admission_rate_limiter: Arc, + pub admission_rate_limiter: Arc, /// Per-agent sliding-window rate limiter for observer frames (kind 24200). /// Key: (community_id, agent pubkey bytes). Value: (count, window_start). @@ -738,6 +761,57 @@ impl AppState { relay_keypair: nostr::Keys, media_storage: MediaStorage, ) -> (Self, AuditShutdownHandle) { + let git_store = crate::api::git::store::GitStore::new( + &config.media.s3_endpoint, + &config.media.s3_access_key, + &config.media.s3_secret_key, + &config.media.s3_bucket, + &config.media.s3_region, + config.media.s3_addressing_style, + ) + .expect("media storage was already constructed with this S3 config"); + let nip98_replay: Arc = + Arc::new(RedisNip98ReplayGuard::new(redis_pool.clone())); + let admission_rate_limiter = Arc::new(AdmissionRateLimiter::redis(redis_pool.clone())); + Self::new_with_backends( + config, + AppBackends { + db, + redis_pool: Some(redis_pool), + pubsub, + search, + media_storage, + git_store, + nip98_replay, + admission_rate_limiter, + }, + audit, + auth, + workflow_engine, + relay_keypair, + ) + } + + /// Constructs state from an explicit backend bundle. + #[allow(clippy::too_many_arguments)] + pub fn new_with_backends( + config: Config, + backends: AppBackends, + audit: impl Into>, + auth: AuthService, + workflow_engine: Arc, + relay_keypair: nostr::Keys, + ) -> (Self, AuditShutdownHandle) { + let AppBackends { + db, + redis_pool, + pubsub, + search, + media_storage, + git_store, + nip98_replay, + admission_rate_limiter, + } = backends; let max_connections = config.max_connections; let max_concurrent_handlers = config.max_concurrent_handlers; let search_arc = Arc::new(search); @@ -783,15 +857,6 @@ impl AppState { let git_max_concurrent_ops = config.git_max_concurrent_ops; let media_max_concurrent_uploads = config.media_max_concurrent_uploads; - let git_store = crate::api::git::store::GitStore::new( - &config.media.s3_endpoint, - &config.media.s3_access_key, - &config.media.s3_secret_key, - &config.media.s3_bucket, - &config.media.s3_region, - config.media.s3_addressing_style, - ) - .expect("media storage was already constructed with this S3 config"); let git_pack_cache = Arc::new( crate::api::git::pack_cache::GitPackCache::new( &config.git_pack_cache_path, @@ -800,9 +865,6 @@ impl AppState { ) .expect("git pack cache path must be available"), ); - let nip98_replay: Arc = - Arc::new(RedisNip98ReplayGuard::new(redis_pool.clone())); - let admission_rate_limiter = Arc::new(RedisRateLimiter::new(redis_pool.clone())); let audit_enabled = audit_arc.is_some(); let state = Self { config: Arc::new(config), diff --git a/crates/buzz-search/src/error.rs b/crates/buzz-search/src/error.rs index 4a71a01daa..dd0e91ffbb 100644 --- a/crates/buzz-search/src/error.rs +++ b/crates/buzz-search/src/error.rs @@ -3,6 +3,9 @@ use thiserror::Error; /// Errors produced by the FTS service. #[derive(Debug, Error)] pub enum SearchError { + /// Search is intentionally unavailable in the selected runtime profile. + #[error("search unsupported by runtime profile")] + Unsupported, /// A database error from sqlx. #[error("database error: {0}")] Db(#[from] sqlx::Error), diff --git a/crates/buzz-search/src/lib.rs b/crates/buzz-search/src/lib.rs index 3c206327b6..d24fde143b 100644 --- a/crates/buzz-search/src/lib.rs +++ b/crates/buzz-search/src/lib.rs @@ -32,23 +32,39 @@ pub use query::{search, ChannelScope, SearchHit, SearchMode, SearchQuery, Search use sqlx::PgPool; -/// Thin handle around a `PgPool` for community-scoped FTS. -/// -/// Holds nothing the pool itself doesn't already own. The whole purpose of -/// this type is a stable injection point for the relay's `AppState`. +/// Search backend selected at startup. +#[derive(Debug, Clone)] +enum SearchBackend { + Postgres(PgPool), + Unsupported, +} + +/// Stable search-service injection point. #[derive(Debug, Clone)] pub struct SearchService { - pool: PgPool, + backend: SearchBackend, } impl SearchService { /// Build a search service over an existing Postgres pool. pub fn new(pool: PgPool) -> Self { - Self { pool } + Self { + backend: SearchBackend::Postgres(pool), + } + } + + /// Build a backend that explicitly reports search as unsupported. + pub fn unsupported() -> Self { + Self { + backend: SearchBackend::Unsupported, + } } /// Execute a community-scoped FTS query. pub async fn search(&self, query: &SearchQuery) -> Result { - query::search(&self.pool, query).await + match &self.backend { + SearchBackend::Postgres(pool) => query::search(pool, query).await, + SearchBackend::Unsupported => Err(SearchError::Unsupported), + } } }