+115

Cargo.lock

··· 97 97 ] 98 98 99 99 [[package]] 100 100 + name = "atproto-jetstream" 101 101 + version = "0.11.3" 102 102 + source = "registry+https://github.com/rust-lang/crates.io-index" 103 103 + checksum = "9e55d1140fcf5c6be0c03e7db773c6a9258986340118015d9bb536bf22689de2" 104 104 + dependencies = [ 105 105 + "anyhow", 106 106 + "async-trait", 107 107 + "atproto-identity", 108 108 + "futures", 109 109 + "http", 110 110 + "serde", 111 111 + "serde_json", 112 112 + "thiserror 2.0.16", 113 113 + "tokio", 114 114 + "tokio-util", 115 115 + "tokio-websockets", 116 116 + "tracing", 117 117 + "tracing-subscriber", 118 118 + "urlencoding", 119 119 + "zstd", 120 120 + ] 121 121 + 122 122 + [[package]] 100 123 name = "autocfg" 101 124 version = "1.5.0" 102 125 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 285 308 checksum = "5252b3d2648e5eedbc1a6f501e3c795e07025c1e93bbf8bbdd6eef7f447a6d54" 286 309 dependencies = [ 287 310 "find-msvc-tools", 311 311 + "jobserver", 312 312 + "libc", 288 313 "shlex", 289 314 ] 290 315 ··· 736 761 ] 737 762 738 763 [[package]] 764 764 + name = "futures" 765 765 + version = "0.3.31" 766 766 + source = "registry+https://github.com/rust-lang/crates.io-index" 767 767 + checksum = "65bc07b1a8bc7c85c5f2e110c476c7389b4554ba72af57d8445ea63a576b0876" 768 768 + dependencies = [ 769 769 + "futures-channel", 770 770 + "futures-core", 771 771 + "futures-executor", 772 772 + "futures-io", 773 773 + "futures-sink", 774 774 + "futures-task", 775 775 + "futures-util", 776 776 + ] 777 777 + 778 778 + [[package]] 739 779 name = "futures-channel" 740 780 version = "0.3.31" 741 781 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 808 848 source = "registry+https://github.com/rust-lang/crates.io-index" 809 849 checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81" 810 850 dependencies = [ 851 851 + "futures-channel", 811 852 "futures-core", 812 853 "futures-io", 813 854 "futures-macro", ··· 1326 1367 checksum = "4a5f13b858c8d314ee3e8f639011f7ccefe71f97f96e50151fb991f267928e2c" 1327 1368 1328 1369 [[package]] 1370 1370 + name = "jobserver" 1371 1371 + version = "0.1.34" 1372 1372 + source = "registry+https://github.com/rust-lang/crates.io-index" 1373 1373 + checksum = "9afb3de4395d6b3e67a780b6de64b51c978ecf11cb9a462c66be7d4ca9039d33" 1374 1374 + dependencies = [ 1375 1375 + "getrandom 0.3.3", 1376 1376 + "libc", 1377 1377 + ] 1378 1378 + 1379 1379 + [[package]] 1329 1380 name = "js-sys" 1330 1381 version = "0.3.78" 1331 1382 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 1874 1925 "anyhow", 1875 1926 "async-trait", 1876 1927 "atproto-identity", 1928 1928 + "atproto-jetstream", 1877 1929 "axum", 1878 1930 "bincode", 1879 1931 "cadence", ··· 2480 2532 "digest", 2481 2533 "rand_core 0.6.4", 2482 2534 ] 2535 2535 + 2536 2536 + [[package]] 2537 2537 + name = "simdutf8" 2538 2538 + version = "0.1.5" 2539 2539 + source = "registry+https://github.com/rust-lang/crates.io-index" 2540 2540 + checksum = "e3a9fe34e3e7a50316060351f37187a3f546bce95496156754b601a5fa71b76e" 2483 2541 2484 2542 [[package]] 2485 2543 name = "slab" ··· 2902 2960 "io-uring", 2903 2961 "libc", 2904 2962 "mio", 2963 2963 + "parking_lot", 2905 2964 "pin-project-lite", 2906 2965 "signal-hook-registry", 2907 2966 "slab", ··· 2964 3023 "futures-util", 2965 3024 "pin-project-lite", 2966 3025 "tokio", 3026 3026 + ] 3027 3027 + 3028 3028 + [[package]] 3029 3029 + name = "tokio-websockets" 3030 3030 + version = "0.11.4" 3031 3031 + source = "registry+https://github.com/rust-lang/crates.io-index" 3032 3032 + checksum = "9fcaf159b4e7a376b05b5bfd77bfd38f3324f5fce751b4213bfc7eaa47affb4e" 3033 3033 + dependencies = [ 3034 3034 + "base64", 3035 3035 + "bytes", 3036 3036 + "futures-core", 3037 3037 + "futures-sink", 3038 3038 + "http", 3039 3039 + "httparse", 3040 3040 + "rand 0.9.2", 3041 3041 + "ring", 3042 3042 + "rustls-native-certs", 3043 3043 + "rustls-pki-types", 3044 3044 + "simdutf8", 3045 3045 + "tokio", 3046 3046 + "tokio-rustls", 3047 3047 + "tokio-util", 2967 3048 ] 2968 3049 2969 3050 [[package]] ··· 3158 3239 "percent-encoding", 3159 3240 "serde", 3160 3241 ] 3242 3242 + 3243 3243 + [[package]] 3244 3244 + name = "urlencoding" 3245 3245 + version = "2.1.3" 3246 3246 + source = "registry+https://github.com/rust-lang/crates.io-index" 3247 3247 + checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da" 3161 3248 3162 3249 [[package]] 3163 3250 name = "utf8_iter" ··· 3816 3903 "quote", 3817 3904 "syn", 3818 3905 ] 3906 3906 + 3907 3907 + [[package]] 3908 3908 + name = "zstd" 3909 3909 + version = "0.13.3" 3910 3910 + source = "registry+https://github.com/rust-lang/crates.io-index" 3911 3911 + checksum = "e91ee311a569c327171651566e07972200e76fcfe2242a4fa446149a3881c08a" 3912 3912 + dependencies = [ 3913 3913 + "zstd-safe", 3914 3914 + ] 3915 3915 + 3916 3916 + [[package]] 3917 3917 + name = "zstd-safe" 3918 3918 + version = "7.2.4" 3919 3919 + source = "registry+https://github.com/rust-lang/crates.io-index" 3920 3920 + checksum = "8f49c4d5f0abb602a93fb8736af2a4f4dd9512e36f7f570d66e65ff867ed3b9d" 3921 3921 + dependencies = [ 3922 3922 + "zstd-sys", 3923 3923 + ] 3924 3924 + 3925 3925 + [[package]] 3926 3926 + name = "zstd-sys" 3927 3927 + version = "2.0.16+zstd.1.5.7" 3928 3928 + source = "registry+https://github.com/rust-lang/crates.io-index" 3929 3929 + checksum = "91e19ebc2adc8f83e43039e79776e3fda8ca919132d68a1fed6a5faca2683748" 3930 3930 + dependencies = [ 3931 3931 + "cc", 3932 3932 + "pkg-config", 3933 3933 + ]

+1

Cargo.toml

··· 17 17 anyhow = "1.0" 18 18 async-trait = "0.1" 19 19 atproto-identity = { version = "0.11.3" } 20 20 + atproto-jetstream = { version = "0.11.3" } 20 21 axum = { version = "0.8" } 21 22 bincode = { version = "2.0.1", features = ["serde"] } 22 23 cadence = "1.6.0"

+106 -1

src/bin/quickdid.rs

··· 3 3 config::{CertificateBundles, DnsNameservers}, 4 4 resolve::HickoryDnsResolver, 5 5 }; 6 6 + use atproto_jetstream::{Consumer as JetstreamConsumer, ConsumerTaskConfig}; 6 7 use quickdid::{ 7 8 cache::create_redis_pool, 8 9 config::Config, ··· 13 14 }, 14 15 handle_resolver_task::{HandleResolverTaskConfig, create_handle_resolver_task_with_config}, 15 16 http::{AppContext, create_router}, 17 17 + jetstream_handler::QuickDidEventHandler, 16 18 metrics::create_metrics_publisher, 17 19 queue::{ 18 20 HandleResolutionWork, QueueAdapter, create_mpsc_queue_from_channel, create_noop_queue, ··· 151 153 ); 152 154 println!( 153 155 " PROACTIVE_REFRESH_THRESHOLD Threshold as percentage of TTL (0.0-1.0, default: 0.8)" 156 156 + ); 157 157 + println!(); 158 158 + println!(" JETSTREAM:"); 159 159 + println!( 160 160 + " JETSTREAM_ENABLED Enable Jetstream consumer (default: false)" 161 161 + ); 162 162 + println!( 163 163 + " JETSTREAM_HOSTNAME Jetstream hostname (default: jetstream.atproto.tools)" 154 164 ); 155 165 println!(); 156 166 println!( ··· 523 533 let app_context = AppContext::new( 524 534 handle_resolver.clone(), 525 535 handle_queue, 526 526 - metrics_publisher, 536 536 + metrics_publisher.clone(), 527 537 config.etag_seed.clone(), 528 538 config.cache_control_header.clone(), 529 539 config.static_files_dir.clone(), ··· 573 583 signal_token.cancel(); 574 584 tracing::info!("Signal handler task completed"); 575 585 }); 586 586 + } 587 587 + 588 588 + // Start Jetstream consumer if enabled 589 589 + if config.jetstream_enabled { 590 590 + let jetstream_resolver = handle_resolver.clone(); 591 591 + let jetstream_metrics = metrics_publisher.clone(); 592 592 + let jetstream_hostname = config.jetstream_hostname.clone(); 593 593 + let jetstream_user_agent = config.user_agent.clone(); 594 594 + 595 595 + spawn_cancellable_task( 596 596 + &tracker, 597 597 + token.clone(), 598 598 + "jetstream_consumer", 599 599 + move |cancel_token| async move { 600 600 + tracing::info!(hostname = %jetstream_hostname, "Starting Jetstream consumer"); 601 601 + 602 602 + // Create event handler 603 603 + let event_handler = Arc::new(QuickDidEventHandler::new( 604 604 + jetstream_resolver, 605 605 + jetstream_metrics.clone(), 606 606 + )); 607 607 + 608 608 + // Reconnection loop 609 609 + let mut reconnect_count = 0u32; 610 610 + let max_reconnects_per_minute = 5; 611 611 + let reconnect_window = std::time::Duration::from_secs(60); 612 612 + let mut last_disconnect = std::time::Instant::now() - reconnect_window; 613 613 + 614 614 + while !cancel_token.is_cancelled() { 615 615 + let now = std::time::Instant::now(); 616 616 + if now.duration_since(last_disconnect) < reconnect_window { 617 617 + reconnect_count += 1; 618 618 + if reconnect_count > max_reconnects_per_minute { 619 619 + tracing::warn!( 620 620 + count = reconnect_count, 621 621 + "Too many Jetstream reconnects, waiting 60 seconds" 622 622 + ); 623 623 + tokio::time::sleep(reconnect_window).await; 624 624 + reconnect_count = 0; 625 625 + last_disconnect = now; 626 626 + continue; 627 627 + } 628 628 + } else { 629 629 + reconnect_count = 0; 630 630 + } 631 631 + 632 632 + // Create consumer configuration 633 633 + let consumer_config = ConsumerTaskConfig { 634 634 + user_agent: jetstream_user_agent.clone(), 635 635 + compression: false, 636 636 + zstd_dictionary_location: String::new(), 637 637 + jetstream_hostname: jetstream_hostname.clone(), 638 638 + collections: vec![], // Listen to all collections 639 639 + dids: vec![], 640 640 + max_message_size_bytes: None, 641 641 + cursor: None, 642 642 + require_hello: true, 643 643 + }; 644 644 + 645 645 + let consumer = JetstreamConsumer::new(consumer_config); 646 646 + 647 647 + // Register event handler 648 648 + if let Err(e) = consumer.register_handler(event_handler.clone()).await { 649 649 + tracing::error!(error = ?e, "Failed to register Jetstream event handler"); 650 650 + continue; 651 651 + } 652 652 + 653 653 + // Run consumer with cancellation support 654 654 + match consumer.run_background(cancel_token.clone()).await { 655 655 + Ok(()) => { 656 656 + tracing::info!("Jetstream consumer stopped normally"); 657 657 + if cancel_token.is_cancelled() { 658 658 + break; 659 659 + } 660 660 + last_disconnect = std::time::Instant::now(); 661 661 + tokio::time::sleep(std::time::Duration::from_secs(5)).await; 662 662 + } 663 663 + Err(e) => { 664 664 + tracing::error!(error = ?e, "Jetstream consumer connection failed, will reconnect"); 665 665 + jetstream_metrics.incr("jetstream.connection.error").await; 666 666 + last_disconnect = std::time::Instant::now(); 667 667 + 668 668 + if !cancel_token.is_cancelled() { 669 669 + tokio::time::sleep(std::time::Duration::from_secs(5)).await; 670 670 + } 671 671 + } 672 672 + } 673 673 + } 674 674 + 675 675 + tracing::info!("Jetstream consumer task shutting down"); 676 676 + Ok(()) 677 677 + }, 678 678 + ); 679 679 + } else { 680 680 + tracing::info!("Jetstream consumer disabled"); 576 681 } 577 682 578 683 // Start HTTP server with cancellation support

+13

src/config.rs

··· 245 245 /// When set, the root handler will serve files from this directory. 246 246 /// Default: "www" (relative to working directory) 247 247 pub static_files_dir: String, 248 248 + 249 249 + /// Enable Jetstream consumer for AT Protocol events. 250 250 + /// When enabled, the service will consume Account and Identity events 251 251 + /// to maintain cache consistency. 252 252 + /// Default: false 253 253 + pub jetstream_enabled: bool, 254 254 + 255 255 + /// Jetstream WebSocket hostname for consuming AT Protocol events. 256 256 + /// Example: "jetstream.atproto.tools" or "jetstream1.us-west.bsky.network" 257 257 + /// Default: "jetstream.atproto.tools" 258 258 + pub jetstream_hostname: String, 248 259 } 249 260 250 261 impl Config { ··· 327 338 proactive_refresh_enabled: parse_env("PROACTIVE_REFRESH_ENABLED", false)?, 328 339 proactive_refresh_threshold: parse_env("PROACTIVE_REFRESH_THRESHOLD", 0.8)?, 329 340 static_files_dir: get_env_or_default("STATIC_FILES_DIR", Some("www")).unwrap(), 341 341 + jetstream_enabled: parse_env("JETSTREAM_ENABLED", false)?, 342 342 + jetstream_hostname: get_env_or_default("JETSTREAM_HOSTNAME", Some("jetstream.atproto.tools")).unwrap(), 330 343 }; 331 344 332 345 // Calculate the Cache-Control header value if enabled

+354

src/jetstream_handler.rs

··· 1 1 + //! Jetstream event handler for QuickDID 2 2 + //! 3 3 + //! This module provides the event handler for processing AT Protocol Jetstream events, 4 4 + //! specifically handling Account and Identity events to maintain cache consistency. 5 5 + 6 6 + use crate::handle_resolver::HandleResolver; 7 7 + use crate::metrics::MetricsPublisher; 8 8 + use anyhow::Result; 9 9 + use atproto_jetstream::{EventHandler, JetstreamEvent}; 10 10 + use std::sync::Arc; 11 11 + use tracing::{debug, info, warn}; 12 12 + 13 13 + /// Jetstream event handler for QuickDID 14 14 + /// 15 15 + /// This handler processes AT Protocol events from the Jetstream firehose to keep 16 16 + /// the handle resolver cache in sync with the network state. 17 17 + /// 18 18 + /// # Event Processing 19 19 + /// 20 20 + /// ## Account Events 21 21 + /// - When an account is marked as "deleted" or "deactivated", the DID is purged from the cache 22 22 + /// - Metrics are tracked for successful and failed purge operations 23 23 + /// 24 24 + /// ## Identity Events 25 25 + /// - When an identity event contains a handle, the handle-to-DID mapping is updated 26 26 + /// - When an identity event lacks a handle (indicating removal), the DID is purged 27 27 + /// - Metrics are tracked for successful and failed update/purge operations 28 28 + /// 29 29 + /// # Example 30 30 + /// 31 31 + /// ```no_run 32 32 + /// use quickdid::jetstream_handler::QuickDidEventHandler; 33 33 + /// use quickdid::handle_resolver::HandleResolver; 34 34 + /// use quickdid::metrics::MetricsPublisher; 35 35 + /// use std::sync::Arc; 36 36 + /// 37 37 + /// # async fn example(resolver: Arc<dyn HandleResolver>, metrics: Arc<dyn MetricsPublisher>) { 38 38 + /// let handler = QuickDidEventHandler::new(resolver, metrics); 39 39 + /// // Register with a JetstreamConsumer 40 40 + /// # } 41 41 + /// ``` 42 42 + pub struct QuickDidEventHandler { 43 43 + resolver: Arc<dyn HandleResolver>, 44 44 + metrics: Arc<dyn MetricsPublisher>, 45 45 + } 46 46 + 47 47 + impl QuickDidEventHandler { 48 48 + /// Create a new Jetstream event handler 49 49 + /// 50 50 + /// # Arguments 51 51 + /// 52 52 + /// * `resolver` - The handle resolver to use for cache operations 53 53 + /// * `metrics` - The metrics publisher for tracking event processing 54 54 + pub fn new(resolver: Arc<dyn HandleResolver>, metrics: Arc<dyn MetricsPublisher>) -> Self { 55 55 + Self { resolver, metrics } 56 56 + } 57 57 + } 58 58 + 59 59 + #[async_trait::async_trait] 60 60 + impl EventHandler for QuickDidEventHandler { 61 61 + fn handler_id(&self) -> String { 62 62 + "quickdid_handler".to_string() 63 63 + } 64 64 + 65 65 + async fn handle_event(&self, event: JetstreamEvent) -> Result<()> { 66 66 + match event { 67 67 + JetstreamEvent::Account { did, kind, .. } => { 68 68 + // If account kind is "deleted" or "deactivated", purge the DID 69 69 + if kind == "deleted" || kind == "deactivated" { 70 70 + info!(did = %did, kind = %kind, "Purging account"); 71 71 + match self.resolver.purge(&did).await { 72 72 + Ok(()) => { 73 73 + self.metrics.incr("jetstream.account.purged").await; 74 74 + } 75 75 + Err(e) => { 76 76 + warn!(did = %did, error = ?e, "Failed to purge DID"); 77 77 + self.metrics.incr("jetstream.account.purge_error").await; 78 78 + } 79 79 + } 80 80 + } 81 81 + self.metrics.incr("jetstream.account.processed").await; 82 82 + } 83 83 + JetstreamEvent::Identity { did, identity, .. } => { 84 84 + // Extract handle from identity JSON if available 85 85 + if !identity.is_null() { 86 86 + if let Some(handle_value) = identity.get("handle") { 87 87 + if let Some(handle) = handle_value.as_str() { 88 88 + info!(handle = %handle, did = %did, "Updating identity mapping"); 89 89 + match self.resolver.set(handle, &did).await { 90 90 + Ok(()) => { 91 91 + self.metrics.incr("jetstream.identity.updated").await; 92 92 + } 93 93 + Err(e) => { 94 94 + warn!(handle = %handle, did = %did, error = ?e, "Failed to update mapping"); 95 95 + self.metrics.incr("jetstream.identity.update_error").await; 96 96 + } 97 97 + } 98 98 + } else { 99 99 + // No handle or invalid handle, purge the DID 100 100 + info!(did = %did, "Purging identity without valid handle"); 101 101 + match self.resolver.purge(&did).await { 102 102 + Ok(()) => { 103 103 + self.metrics.incr("jetstream.identity.purged").await; 104 104 + } 105 105 + Err(e) => { 106 106 + warn!(did = %did, error = ?e, "Failed to purge DID"); 107 107 + self.metrics.incr("jetstream.identity.purge_error").await; 108 108 + } 109 109 + } 110 110 + } 111 111 + } else { 112 112 + // No handle field, purge the DID 113 113 + info!(did = %did, "Purging identity without handle field"); 114 114 + match self.resolver.purge(&did).await { 115 115 + Ok(()) => { 116 116 + self.metrics.incr("jetstream.identity.purged").await; 117 117 + } 118 118 + Err(e) => { 119 119 + warn!(did = %did, error = ?e, "Failed to purge DID"); 120 120 + self.metrics.incr("jetstream.identity.purge_error").await; 121 121 + } 122 122 + } 123 123 + } 124 124 + } else { 125 125 + // Null identity means removed, purge the DID 126 126 + info!(did = %did, "Purging identity with null info"); 127 127 + match self.resolver.purge(&did).await { 128 128 + Ok(()) => { 129 129 + self.metrics.incr("jetstream.identity.purged").await; 130 130 + } 131 131 + Err(e) => { 132 132 + warn!(did = %did, error = ?e, "Failed to purge DID"); 133 133 + self.metrics.incr("jetstream.identity.purge_error").await; 134 134 + } 135 135 + } 136 136 + } 137 137 + self.metrics.incr("jetstream.identity.processed").await; 138 138 + } 139 139 + _ => { 140 140 + // Other event types we don't care about 141 141 + debug!("Ignoring unhandled Jetstream event type"); 142 142 + } 143 143 + } 144 144 + Ok(()) 145 145 + } 146 146 + } 147 147 + 148 148 + #[cfg(test)] 149 149 + mod tests { 150 150 + use super::*; 151 151 + use crate::handle_resolver::HandleResolverError; 152 152 + use crate::metrics::NoOpMetricsPublisher; 153 153 + use async_trait::async_trait; 154 154 + use serde_json::json; 155 155 + 156 156 + /// Mock resolver for testing 157 157 + struct MockResolver { 158 158 + purge_called: std::sync::Arc<std::sync::Mutex<Vec<String>>>, 159 159 + set_called: std::sync::Arc<std::sync::Mutex<Vec<(String, String)>>>, 160 160 + } 161 161 + 162 162 + impl MockResolver { 163 163 + fn new() -> Self { 164 164 + Self { 165 165 + purge_called: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())), 166 166 + set_called: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())), 167 167 + } 168 168 + } 169 169 + 170 170 + fn get_purge_calls(&self) -> Vec<String> { 171 171 + self.purge_called.lock().unwrap().clone() 172 172 + } 173 173 + 174 174 + fn get_set_calls(&self) -> Vec<(String, String)> { 175 175 + self.set_called.lock().unwrap().clone() 176 176 + } 177 177 + } 178 178 + 179 179 + #[async_trait] 180 180 + impl HandleResolver for MockResolver { 181 181 + async fn resolve(&self, _handle: &str) -> Result<(String, u64), HandleResolverError> { 182 182 + unimplemented!("Not needed for tests") 183 183 + } 184 184 + 185 185 + async fn purge(&self, subject: &str) -> Result<(), HandleResolverError> { 186 186 + self.purge_called.lock().unwrap().push(subject.to_string()); 187 187 + Ok(()) 188 188 + } 189 189 + 190 190 + async fn set(&self, handle: &str, did: &str) -> Result<(), HandleResolverError> { 191 191 + self.set_called 192 192 + .lock() 193 193 + .unwrap() 194 194 + .push((handle.to_string(), did.to_string())); 195 195 + Ok(()) 196 196 + } 197 197 + } 198 198 + 199 199 + #[tokio::test] 200 200 + async fn test_account_deleted_event() { 201 201 + let resolver = Arc::new(MockResolver::new()); 202 202 + let metrics = Arc::new(NoOpMetricsPublisher::new()); 203 203 + let handler = QuickDidEventHandler::new(resolver.clone(), metrics); 204 204 + 205 205 + // Create a deleted account event 206 206 + let event = JetstreamEvent::Account { 207 207 + did: "did:plc:test123".to_string(), 208 208 + kind: "deleted".to_string(), 209 209 + time_us: 0, 210 210 + identity: json!(null), 211 211 + }; 212 212 + 213 213 + handler.handle_event(event).await.unwrap(); 214 214 + 215 215 + // Verify the DID was purged 216 216 + let purge_calls = resolver.get_purge_calls(); 217 217 + assert_eq!(purge_calls.len(), 1); 218 218 + assert_eq!(purge_calls[0], "did:plc:test123"); 219 219 + } 220 220 + 221 221 + #[tokio::test] 222 222 + async fn test_account_deactivated_event() { 223 223 + let resolver = Arc::new(MockResolver::new()); 224 224 + let metrics = Arc::new(NoOpMetricsPublisher::new()); 225 225 + let handler = QuickDidEventHandler::new(resolver.clone(), metrics); 226 226 + 227 227 + // Create a deactivated account event 228 228 + let event = JetstreamEvent::Account { 229 229 + did: "did:plc:test456".to_string(), 230 230 + kind: "deactivated".to_string(), 231 231 + time_us: 0, 232 232 + identity: json!(null), 233 233 + }; 234 234 + 235 235 + handler.handle_event(event).await.unwrap(); 236 236 + 237 237 + // Verify the DID was purged 238 238 + let purge_calls = resolver.get_purge_calls(); 239 239 + assert_eq!(purge_calls.len(), 1); 240 240 + assert_eq!(purge_calls[0], "did:plc:test456"); 241 241 + } 242 242 + 243 243 + #[tokio::test] 244 244 + async fn test_account_active_event() { 245 245 + let resolver = Arc::new(MockResolver::new()); 246 246 + let metrics = Arc::new(NoOpMetricsPublisher::new()); 247 247 + let handler = QuickDidEventHandler::new(resolver.clone(), metrics); 248 248 + 249 249 + // Create an active account event (should not purge) 250 250 + let event = JetstreamEvent::Account { 251 251 + did: "did:plc:test789".to_string(), 252 252 + kind: "active".to_string(), 253 253 + time_us: 0, 254 254 + identity: json!(null), 255 255 + }; 256 256 + 257 257 + handler.handle_event(event).await.unwrap(); 258 258 + 259 259 + // Verify the DID was NOT purged 260 260 + let purge_calls = resolver.get_purge_calls(); 261 261 + assert_eq!(purge_calls.len(), 0); 262 262 + } 263 263 + 264 264 + #[tokio::test] 265 265 + async fn test_identity_with_handle_event() { 266 266 + let resolver = Arc::new(MockResolver::new()); 267 267 + let metrics = Arc::new(NoOpMetricsPublisher::new()); 268 268 + let handler = QuickDidEventHandler::new(resolver.clone(), metrics); 269 269 + 270 270 + // Create an identity event with a handle 271 271 + let event = JetstreamEvent::Identity { 272 272 + did: "did:plc:testuser".to_string(), 273 273 + kind: "update".to_string(), 274 274 + time_us: 0, 275 275 + identity: json!({ 276 276 + "handle": "alice.bsky.social" 277 277 + }), 278 278 + }; 279 279 + 280 280 + handler.handle_event(event).await.unwrap(); 281 281 + 282 282 + // Verify the set method was called 283 283 + let set_calls = resolver.get_set_calls(); 284 284 + assert_eq!(set_calls.len(), 1); 285 285 + assert_eq!(set_calls[0], ("alice.bsky.social".to_string(), "did:plc:testuser".to_string())); 286 286 + 287 287 + // Verify no purge was called 288 288 + let purge_calls = resolver.get_purge_calls(); 289 289 + assert_eq!(purge_calls.len(), 0); 290 290 + } 291 291 + 292 292 + #[tokio::test] 293 293 + async fn test_identity_without_handle_event() { 294 294 + let resolver = Arc::new(MockResolver::new()); 295 295 + let metrics = Arc::new(NoOpMetricsPublisher::new()); 296 296 + let handler = QuickDidEventHandler::new(resolver.clone(), metrics); 297 297 + 298 298 + // Create an identity event without a handle field 299 299 + let event = JetstreamEvent::Identity { 300 300 + did: "did:plc:nohandle".to_string(), 301 301 + kind: "update".to_string(), 302 302 + time_us: 0, 303 303 + identity: json!({ 304 304 + "other_field": "value" 305 305 + }), 306 306 + }; 307 307 + 308 308 + handler.handle_event(event).await.unwrap(); 309 309 + 310 310 + // Verify the DID was purged 311 311 + let purge_calls = resolver.get_purge_calls(); 312 312 + assert_eq!(purge_calls.len(), 1); 313 313 + assert_eq!(purge_calls[0], "did:plc:nohandle"); 314 314 + 315 315 + // Verify set was not called 316 316 + let set_calls = resolver.get_set_calls(); 317 317 + assert_eq!(set_calls.len(), 0); 318 318 + } 319 319 + 320 320 + #[tokio::test] 321 321 + async fn test_identity_with_null_identity() { 322 322 + let resolver = Arc::new(MockResolver::new()); 323 323 + let metrics = Arc::new(NoOpMetricsPublisher::new()); 324 324 + let handler = QuickDidEventHandler::new(resolver.clone(), metrics); 325 325 + 326 326 + // Create an identity event with null identity 327 327 + let event = JetstreamEvent::Identity { 328 328 + did: "did:plc:nullidentity".to_string(), 329 329 + kind: "delete".to_string(), 330 330 + time_us: 0, 331 331 + identity: json!(null), 332 332 + }; 333 333 + 334 334 + handler.handle_event(event).await.unwrap(); 335 335 + 336 336 + // Verify the DID was purged 337 337 + let purge_calls = resolver.get_purge_calls(); 338 338 + assert_eq!(purge_calls.len(), 1); 339 339 + assert_eq!(purge_calls[0], "did:plc:nullidentity"); 340 340 + 341 341 + // Verify set was not called 342 342 + let set_calls = resolver.get_set_calls(); 343 343 + assert_eq!(set_calls.len(), 0); 344 344 + } 345 345 + 346 346 + #[tokio::test] 347 347 + async fn test_handler_id() { 348 348 + let resolver = Arc::new(MockResolver::new()); 349 349 + let metrics = Arc::new(NoOpMetricsPublisher::new()); 350 350 + let handler = QuickDidEventHandler::new(resolver, metrics); 351 351 + 352 352 + assert_eq!(handler.handler_id(), "quickdid_handler"); 353 353 + } 354 354 + }

+1

src/lib.rs

··· 2 2 pub mod config; // Config and Args needed by binary 3 3 pub mod handle_resolver; // Only traits and factory functions exposed 4 4 pub mod http; // Only create_router exposed 5 5 + pub mod jetstream_handler; // Jetstream event handler for AT Protocol events 5 6 6 7 // Semi-public modules - needed by binary but with limited exposure 7 8 pub mod cache; // Only create_redis_pool exposed