feature: experimental jetstream update

crates/atproto-jetstream/src/consumer.rs

···
 //!
 //! WebSocket event consumption with background processing and
 //! customizable event handler dispatch.
+//!
+//! ## Memory Efficiency
+//!
+//! This module is optimized for high-throughput event processing with minimal allocations:
+//!
+//! - **Arc-based event sharing**: Events are wrapped in `Arc` and shared across all handlers,
+//!   avoiding expensive clones of event data structures.
+//! - **Zero-copy handler IDs**: Handler identifiers use string slices to avoid allocations
+//!   during registration and dispatch.
+//! - **Optimized query building**: WebSocket query strings are built with pre-allocated
+//!   capacity to minimize reallocations.
+//!
+//! ## Usage
+//!
+//! Implement the `EventHandler` trait to process events:
+//!
+//! ```rust
+//! use atproto_jetstream::{EventHandler, JetstreamEvent};
+//! use async_trait::async_trait;
+//! use std::sync::Arc;
+//! use anyhow::Result;
+//!
+//! struct MyHandler;
+//!
+//! #[async_trait]
+//! impl EventHandler for MyHandler {
+//!     async fn handle_event(&self, event: Arc<JetstreamEvent>) -> Result<()> {
+//!         // Process event without cloning
+//!         Ok(())
+//!     }
+//!
+//!     fn handler_id(&self) -> &str {
+//!         "my-handler"
+//!     }
+//! }
+//! ```
 
 use crate::errors::ConsumerError;
 use anyhow::Result;
···
 #[async_trait]
 pub trait EventHandler: Send + Sync {
     /// Handle a received event
-    async fn handle_event(&self, event: JetstreamEvent) -> Result<()>;
+    ///
+    /// Events are wrapped in Arc to enable efficient sharing across multiple handlers
+    /// without cloning the entire event data structure.
+    async fn handle_event(&self, event: Arc<JetstreamEvent>) -> Result<()>;
 
     /// Get the handler's identifier
-    fn handler_id(&self) -> String;
+    ///
+    /// Returns a string slice to avoid unnecessary allocations.
+    fn handler_id(&self) -> &str;
 }
 
 #[cfg_attr(debug_assertions, derive(Debug))]
···
 pub struct Consumer {
     config: ConsumerTaskConfig,
     handlers: Arc<RwLock<HashMap<String, Arc<dyn EventHandler>>>>,
-    event_sender: Arc<RwLock<Option<broadcast::Sender<JetstreamEvent>>>>,
+    event_sender: Arc<RwLock<Option<broadcast::Sender<Arc<JetstreamEvent>>>>>,
 }
 
 impl Consumer {
···
         let handler_id = handler.handler_id();
         let mut handlers = self.handlers.write().await;
 
-        if handlers.contains_key(&handler_id) {
+        if handlers.contains_key(handler_id) {
             return Err(ConsumerError::HandlerRegistrationFailed(format!(
                 "Handler with ID '{}' already registered",
                 handler_id
···
             .into());
         }
 
-        handlers.insert(handler_id.clone(), handler);
+        handlers.insert(handler_id.to_string(), handler);
         Ok(())
     }
 
···
     }
 
     /// Get a broadcast receiver for events
-    pub async fn get_event_receiver(&self) -> Result<broadcast::Receiver<JetstreamEvent>> {
+    ///
+    /// Events are wrapped in Arc to enable efficient sharing without cloning.
+    pub async fn get_event_receiver(&self) -> Result<broadcast::Receiver<Arc<JetstreamEvent>>> {
         let sender_guard = self.event_sender.read().await;
         match sender_guard.as_ref() {
             Some(sender) => Ok(sender.subscribe()),
···
         tracing::info!("Starting Jetstream consumer");
 
         // Build WebSocket URL with query parameters
-        let mut query_params = vec![];
+        // Pre-allocate capacity to avoid reallocations during string building
+        let capacity = 50 // Base parameters
+            + self.config.collections.len() * 30 // Estimate per collection
+            + self.config.dids.len() * 60; // Estimate per DID
+        let mut query_string = String::with_capacity(capacity);
 
         // Add compression parameter
-        query_params.push(format!("compress={}", self.config.compression));
+        query_string.push_str("compress=");
+        query_string.push_str(if self.config.compression { "true" } else { "false" });
 
         // Add requireHello parameter
-        query_params.push(format!("requireHello={}", self.config.require_hello));
+        query_string.push_str("&requireHello=");
+        query_string.push_str(if self.config.require_hello { "true" } else { "false" });
 
         // Add wantedCollections if specified (each collection as a separate query parameter)
         if !self.config.collections.is_empty() && !self.config.require_hello {
             for collection in &self.config.collections {
-                query_params.push(format!(
-                    "wantedCollections={}",
-                    urlencoding::encode(collection)
-                ));
+                query_string.push_str("&wantedCollections=");
+                query_string.push_str(&urlencoding::encode(collection));
             }
         }
 
         // Add wantedDids if specified (each DID as a separate query parameter)
         if !self.config.dids.is_empty() && !self.config.require_hello {
             for did in &self.config.dids {
-                query_params.push(format!("wantedDids={}", urlencoding::encode(did)));
+                query_string.push_str("&wantedDids=");
+                query_string.push_str(&urlencoding::encode(did));
             }
         }
 
         // Add maxMessageSizeBytes if specified
         if let Some(max_size) = self.config.max_message_size_bytes {
-            query_params.push(format!("maxMessageSizeBytes={}", max_size));
+            use std::fmt::Write;
+            write!(&mut query_string, "&maxMessageSizeBytes={}", max_size).unwrap();
         }
 
         // Add cursor if specified
         if let Some(cursor) = self.config.cursor {
-            query_params.push(format!("cursor={}", cursor));
+            use std::fmt::Write;
+            write!(&mut query_string, "&cursor={}", cursor).unwrap();
         }
-
-        let query_string = query_params.join("&");
         let ws_url = Uri::from_str(&format!(
             "wss://{}/subscribe?{}",
             self.config.jetstream_hostname, query_string
···
     }
 
     /// Dispatch event to all registered handlers
+    ///
+    /// Wraps the event in Arc once and shares it across all handlers,
+    /// avoiding expensive clones of the event data structure.
     async fn dispatch_to_handlers(&self, event: JetstreamEvent) -> Result<()> {
         let handlers = self.handlers.read().await;
+        let event = Arc::new(event);
 
         for (handler_id, handler) in handlers.iter() {
             let handler_span = tracing::debug_span!("handler_dispatch", handler_id = %handler_id);
+            let event_ref = Arc::clone(&event);
             async {
-                if let Err(err) = handler.handle_event(event.clone()).await {
+                if let Err(err) = handler.handle_event(event_ref).await {
                     tracing::error!(
                         error = ?err,
                         handler_id = %handler_id,
···
 
 #[async_trait]
 impl EventHandler for LoggingHandler {
-    async fn handle_event(&self, _event: JetstreamEvent) -> Result<()> {
+    async fn handle_event(&self, _event: Arc<JetstreamEvent>) -> Result<()> {
         Ok(())
     }
 
-    fn handler_id(&self) -> String {
-        self.id.clone()
+    fn handler_id(&self) -> &str {
+        &self.id
     }
 }