smokesignal.events
QuickDID is a high-performance AT Protocol identity resolution service written in Rust. It provides handle-to-DID resolution with Redis-backed caching and queue processing.
refactor: code cleanup
+40
-62
src/bin/quickdid.rs
+40
-62
src/bin/quickdid.rs
···
1
use anyhow::Result;
2
-
use async_trait::async_trait;
3
use atproto_identity::{
4
config::{CertificateBundles, DnsNameservers},
5
-
key::{KeyData, KeyProvider, identify_key, to_public},
6
resolve::HickoryDnsResolver,
7
};
8
use clap::Parser;
···
21
task_manager::spawn_cancellable_task,
22
};
23
use serde_json::json;
24
-
use std::{collections::HashMap, sync::Arc};
25
use tokio::signal;
26
use tokio_util::{sync::CancellationToken, task::TaskTracker};
27
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
28
29
-
#[derive(Clone)]
30
-
pub struct SimpleKeyProvider {
31
-
keys: HashMap<String, KeyData>,
32
-
}
33
-
34
-
impl SimpleKeyProvider {
35
-
pub fn new() -> Self {
36
-
Self {
37
-
keys: HashMap::new(),
38
}
39
}
40
}
41
42
-
#[async_trait]
43
-
impl KeyProvider for SimpleKeyProvider {
44
-
async fn get_private_key_by_id(&self, key_id: &str) -> anyhow::Result<Option<KeyData>> {
45
-
Ok(self.keys.get(key_id).cloned())
46
-
}
47
-
}
48
-
49
#[tokio::main]
50
async fn main() -> Result<()> {
51
// Initialize tracing
···
124
let base_handle_resolver = create_base_resolver(dns_resolver_arc.clone(), http_client.clone());
125
126
// Create Redis pool if configured
127
-
let redis_pool = if let Some(redis_url) = &config.redis_url {
128
-
match create_redis_pool(redis_url) {
129
-
Ok(pool) => {
130
-
tracing::info!("Redis pool created for handle resolver cache");
131
-
Some(pool)
132
-
}
133
-
Err(e) => {
134
-
tracing::warn!("Failed to create Redis pool for handle resolver: {}", e);
135
-
None
136
-
}
137
-
}
138
-
} else {
139
-
None
140
-
};
141
142
// Create handle resolver with Redis caching if available, otherwise use in-memory caching
143
let handle_resolver: Arc<dyn quickdid::handle_resolver::HandleResolver> =
···
173
.as_ref()
174
.or(config.redis_url.as_ref());
175
176
-
match queue_redis_url {
177
-
Some(url) => match create_redis_pool(url) {
178
-
Ok(pool) => {
179
-
tracing::info!(
180
-
"Creating Redis queue adapter with prefix: {}",
181
-
config.queue_redis_prefix
182
);
183
-
create_redis_queue::<HandleResolutionWork>(
184
-
pool,
185
-
config.queue_worker_id.clone(),
186
-
config.queue_redis_prefix.clone(),
187
-
config.queue_redis_timeout, // Configurable timeout for blocking operations
188
-
)
189
-
}
190
-
Err(e) => {
191
-
tracing::error!("Failed to create Redis pool for queue adapter: {}", e);
192
-
tracing::warn!("Falling back to MPSC queue adapter");
193
-
// Fall back to MPSC if Redis fails
194
-
let (handle_sender, handle_receiver) =
195
-
tokio::sync::mpsc::channel::<HandleResolutionWork>(
196
-
config.queue_buffer_size,
197
-
);
198
-
create_mpsc_queue_from_channel(handle_sender, handle_receiver)
199
-
}
200
-
},
201
-
None => {
202
-
tracing::warn!(
203
-
"Redis queue adapter requested but no Redis URL configured, using no-op adapter"
204
-
);
205
-
create_noop_queue::<HandleResolutionWork>()
206
}
207
}
208
}
209
"mpsc" => {
···
1
use anyhow::Result;
2
use atproto_identity::{
3
config::{CertificateBundles, DnsNameservers},
4
+
key::{identify_key, to_public},
5
resolve::HickoryDnsResolver,
6
};
7
use clap::Parser;
···
20
task_manager::spawn_cancellable_task,
21
};
22
use serde_json::json;
23
+
use std::sync::Arc;
24
use tokio::signal;
25
use tokio_util::{sync::CancellationToken, task::TaskTracker};
26
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
27
28
+
/// Helper function to create a Redis pool with consistent error handling
29
+
fn try_create_redis_pool(redis_url: &str, purpose: &str) -> Option<deadpool_redis::Pool> {
30
+
match create_redis_pool(redis_url) {
31
+
Ok(pool) => {
32
+
tracing::info!("Redis pool created for {}", purpose);
33
+
Some(pool)
34
+
}
35
+
Err(e) => {
36
+
tracing::warn!("Failed to create Redis pool for {}: {}", purpose, e);
37
+
None
38
}
39
}
40
}
41
42
#[tokio::main]
43
async fn main() -> Result<()> {
44
// Initialize tracing
···
117
let base_handle_resolver = create_base_resolver(dns_resolver_arc.clone(), http_client.clone());
118
119
// Create Redis pool if configured
120
+
let redis_pool = config
121
+
.redis_url
122
+
.as_ref()
123
+
.and_then(|url| try_create_redis_pool(url, "handle resolver cache"));
124
125
// Create handle resolver with Redis caching if available, otherwise use in-memory caching
126
let handle_resolver: Arc<dyn quickdid::handle_resolver::HandleResolver> =
···
156
.as_ref()
157
.or(config.redis_url.as_ref());
158
159
+
if let Some(url) = queue_redis_url {
160
+
if let Some(pool) = try_create_redis_pool(url, "queue adapter") {
161
+
tracing::info!(
162
+
"Creating Redis queue adapter with prefix: {}",
163
+
config.queue_redis_prefix
164
+
);
165
+
create_redis_queue::<HandleResolutionWork>(
166
+
pool,
167
+
config.queue_worker_id.clone(),
168
+
config.queue_redis_prefix.clone(),
169
+
config.queue_redis_timeout,
170
+
)
171
+
} else {
172
+
tracing::warn!("Falling back to MPSC queue adapter");
173
+
// Fall back to MPSC if Redis fails
174
+
let (handle_sender, handle_receiver) =
175
+
tokio::sync::mpsc::channel::<HandleResolutionWork>(
176
+
config.queue_buffer_size,
177
);
178
+
create_mpsc_queue_from_channel(handle_sender, handle_receiver)
179
}
180
+
} else {
181
+
tracing::warn!(
182
+
"Redis queue adapter requested but no Redis URL configured, using no-op adapter"
183
+
);
184
+
create_noop_queue::<HandleResolutionWork>()
185
}
186
}
187
"mpsc" => {
+6
-31
src/handle_resolver/redis.rs
+6
-31
src/handle_resolver/redis.rs
···
264
#[cfg(test)]
265
mod tests {
266
use super::*;
267
-
use crate::cache::create_redis_pool;
268
269
// Mock handle resolver for testing
270
#[derive(Clone)]
···
286
287
#[tokio::test]
288
async fn test_redis_handle_resolver_cache_hit() {
289
-
// This test requires Redis to be running
290
-
// Set TEST_REDIS_URL environment variable to run this test
291
-
let redis_url = match std::env::var("TEST_REDIS_URL") {
292
-
Ok(url) => url,
293
-
Err(_) => {
294
-
eprintln!("Skipping test: Set TEST_REDIS_URL to run Redis tests");
295
-
return;
296
-
}
297
-
};
298
-
299
-
let pool = match create_redis_pool(&redis_url) {
300
-
Ok(p) => p,
301
-
Err(e) => {
302
-
eprintln!("Failed to create Redis pool: {}", e);
303
-
return;
304
-
}
305
};
306
307
// Create mock resolver
···
341
342
#[tokio::test]
343
async fn test_redis_handle_resolver_cache_error() {
344
-
let redis_url = match std::env::var("TEST_REDIS_URL") {
345
-
Ok(url) => url,
346
-
Err(_) => {
347
-
eprintln!("Skipping test: Set TEST_REDIS_URL to run Redis tests");
348
-
return;
349
-
}
350
-
};
351
-
352
-
let pool = match create_redis_pool(&redis_url) {
353
-
Ok(p) => p,
354
-
Err(e) => {
355
-
eprintln!("Failed to create Redis pool: {}", e);
356
-
return;
357
-
}
358
};
359
360
// Create mock resolver that fails
···
264
#[cfg(test)]
265
mod tests {
266
use super::*;
267
268
// Mock handle resolver for testing
269
#[derive(Clone)]
···
285
286
#[tokio::test]
287
async fn test_redis_handle_resolver_cache_hit() {
288
+
let pool = match crate::test_helpers::get_test_redis_pool() {
289
+
Some(p) => p,
290
+
None => return,
291
};
292
293
// Create mock resolver
···
327
328
#[tokio::test]
329
async fn test_redis_handle_resolver_cache_error() {
330
+
let pool = match crate::test_helpers::get_test_redis_pool() {
331
+
Some(p) => p,
332
+
None => return,
333
};
334
335
// Create mock resolver that fails
-6
src/handle_resolver_task.rs
-6
src/handle_resolver_task.rs
···
18
pub(crate) enum HandleResolverError {
19
#[error("Queue adapter health check failed: adapter is not healthy")]
20
QueueAdapterUnhealthy,
21
-
22
-
#[error("Handle resolution failed: {0}")]
23
-
ResolutionFailed(String),
24
-
25
-
#[error("Resolution timeout: exceeded {timeout_ms}ms")]
26
-
ResolutionTimeout { timeout_ms: u64 },
27
}
28
29
/// Configuration for the handle resolver task processor
+3
-3
src/http/handle_xrpc_resolve_handle.rs
+3
-3
src/http/handle_xrpc_resolve_handle.rs
···
100
match handle_resolver.resolve(&handle).await {
101
Ok(did) => {
102
tracing::debug!("Found cached DID for handle {}: {}", handle, did);
103
-
return Ok(Json(ResolveHandleResponse { did }).into_response());
104
}
105
Err(_) => {
106
// {"error":"InvalidRequest","message":"Unable to resolve handle"}
107
-
return Err((
108
StatusCode::BAD_REQUEST,
109
Json(ErrorResponse {
110
error: "InvalidRequest".to_string(),
111
message: "Unable to resolve handle".to_string(),
112
}),
113
-
));
114
}
115
}
116
}
···
100
match handle_resolver.resolve(&handle).await {
101
Ok(did) => {
102
tracing::debug!("Found cached DID for handle {}: {}", handle, did);
103
+
Ok(Json(ResolveHandleResponse { did }).into_response())
104
}
105
Err(_) => {
106
// {"error":"InvalidRequest","message":"Unable to resolve handle"}
107
+
Err((
108
StatusCode::BAD_REQUEST,
109
Json(ErrorResponse {
110
error: "InvalidRequest".to_string(),
111
message: "Unable to resolve handle".to_string(),
112
}),
113
+
))
114
}
115
}
116
}
+4
src/lib.rs
+4
src/lib.rs
+12
-74
src/queue_adapter.rs
+12
-74
src/queue_adapter.rs
···
176
}
177
}
178
179
-
/// Generic work type for different kinds of background tasks
180
-
#[derive(Debug, Clone, Serialize, Deserialize)]
181
-
pub(crate) enum WorkItem {
182
-
/// Handle resolution work
183
-
HandleResolution(HandleResolutionWork),
184
-
// Future work types can be added here
185
-
}
186
-
187
/// Redis-backed queue adapter implementation.
188
///
189
/// This adapter uses Redis lists with a reliable queue pattern:
···
554
555
#[tokio::test]
556
async fn test_redis_queue_adapter_push_pull() {
557
-
// This test requires Redis to be running
558
-
let redis_url = match std::env::var("TEST_REDIS_URL") {
559
-
Ok(url) => url,
560
-
Err(_) => {
561
-
eprintln!("Skipping test: Set TEST_REDIS_URL to run Redis tests");
562
-
return;
563
-
}
564
-
};
565
-
566
-
// Import create_redis_pool
567
-
use crate::cache::create_redis_pool;
568
-
569
-
let pool = match create_redis_pool(&redis_url) {
570
-
Ok(p) => p,
571
-
Err(e) => {
572
-
eprintln!("Failed to create Redis pool: {}", e);
573
-
return;
574
-
}
575
};
576
577
// Create adapter with unique prefix for testing
···
603
604
#[tokio::test]
605
async fn test_redis_queue_adapter_reliable_queue() {
606
-
let redis_url = match std::env::var("TEST_REDIS_URL") {
607
-
Ok(url) => url,
608
-
Err(_) => {
609
-
eprintln!("Skipping test: Set TEST_REDIS_URL to run Redis tests");
610
-
return;
611
-
}
612
-
};
613
-
614
-
use crate::cache::create_redis_pool;
615
-
616
-
let pool = match create_redis_pool(&redis_url) {
617
-
Ok(p) => p,
618
-
Err(e) => {
619
-
eprintln!("Failed to create Redis pool: {}", e);
620
-
return;
621
-
}
622
};
623
624
let test_prefix = format!("test:queue:{}:", uuid::Uuid::new_v4());
···
658
659
#[tokio::test]
660
async fn test_redis_queue_adapter_depth() {
661
-
let redis_url = match std::env::var("TEST_REDIS_URL") {
662
-
Ok(url) => url,
663
-
Err(_) => {
664
-
eprintln!("Skipping test: Set TEST_REDIS_URL to run Redis tests");
665
-
return;
666
-
}
667
-
};
668
-
669
-
use crate::cache::create_redis_pool;
670
-
671
-
let pool = match create_redis_pool(&redis_url) {
672
-
Ok(p) => p,
673
-
Err(e) => {
674
-
eprintln!("Failed to create Redis pool: {}", e);
675
-
return;
676
-
}
677
};
678
679
let test_prefix = format!("test:queue:{}:", uuid::Uuid::new_v4());
···
705
706
#[tokio::test]
707
async fn test_redis_queue_adapter_health() {
708
-
let redis_url = match std::env::var("TEST_REDIS_URL") {
709
-
Ok(url) => url,
710
-
Err(_) => {
711
-
eprintln!("Skipping test: Set TEST_REDIS_URL to run Redis tests");
712
-
return;
713
-
}
714
-
};
715
-
716
-
use crate::cache::create_redis_pool;
717
-
718
-
let pool = match create_redis_pool(&redis_url) {
719
-
Ok(p) => p,
720
-
Err(e) => {
721
-
eprintln!("Failed to create Redis pool: {}", e);
722
-
return;
723
-
}
724
};
725
726
let adapter = Arc::new(RedisQueueAdapter::<String>::with_config(
···
176
}
177
}
178
179
/// Redis-backed queue adapter implementation.
180
///
181
/// This adapter uses Redis lists with a reliable queue pattern:
···
546
547
#[tokio::test]
548
async fn test_redis_queue_adapter_push_pull() {
549
+
let pool = match crate::test_helpers::get_test_redis_pool() {
550
+
Some(p) => p,
551
+
None => return,
552
};
553
554
// Create adapter with unique prefix for testing
···
580
581
#[tokio::test]
582
async fn test_redis_queue_adapter_reliable_queue() {
583
+
let pool = match crate::test_helpers::get_test_redis_pool() {
584
+
Some(p) => p,
585
+
None => return,
586
};
587
588
let test_prefix = format!("test:queue:{}:", uuid::Uuid::new_v4());
···
622
623
#[tokio::test]
624
async fn test_redis_queue_adapter_depth() {
625
+
let pool = match crate::test_helpers::get_test_redis_pool() {
626
+
Some(p) => p,
627
+
None => return,
628
};
629
630
let test_prefix = format!("test:queue:{}:", uuid::Uuid::new_v4());
···
656
657
#[tokio::test]
658
async fn test_redis_queue_adapter_health() {
659
+
let pool = match crate::test_helpers::get_test_redis_pool() {
660
+
Some(p) => p,
661
+
None => return,
662
};
663
664
let adapter = Arc::new(RedisQueueAdapter::<String>::with_config(
+36
src/test_helpers.rs
+36
src/test_helpers.rs
···
···
1
+
//! Test helper utilities for QuickDID tests
2
+
#![cfg(test)]
3
+
4
+
use crate::cache::create_redis_pool;
5
+
use deadpool_redis::Pool;
6
+
7
+
/// Helper function to get a Redis pool for testing.
8
+
///
9
+
/// Returns None if TEST_REDIS_URL is not set, logging a skip message.
10
+
/// This consolidates the repeated Redis test setup code.
11
+
pub(crate) fn get_test_redis_pool() -> Option<Pool> {
12
+
match std::env::var("TEST_REDIS_URL") {
13
+
Ok(url) => match create_redis_pool(&url) {
14
+
Ok(pool) => Some(pool),
15
+
Err(e) => {
16
+
eprintln!("Failed to create Redis pool: {}", e);
17
+
None
18
+
}
19
+
},
20
+
Err(_) => {
21
+
eprintln!("Skipping test: Set TEST_REDIS_URL to run Redis tests");
22
+
None
23
+
}
24
+
}
25
+
}
26
+
27
+
/// Macro to skip Redis tests if pool is not available
28
+
#[macro_export]
29
+
macro_rules! require_redis {
30
+
() => {
31
+
match $crate::test_helpers::get_test_redis_pool() {
32
+
Some(pool) => pool,
33
+
None => return,
34
+
}
35
+
};
36
+
}