ptr.pet
at protocol indexer with flexible filtering, xrpc queries, and a cursor-backed event stream, built on fjall
at-protocol
atproto
indexer
rust
fjall
[resolver] support multiple PLC URLs
+1
-1
src/bin/mst_dump.rs
+1
-1
src/bin/mst_dump.rs
+12
-4
src/config.rs
+12
-4
src/config.rs
···
39
pub struct Config {
40
pub database_path: PathBuf,
41
pub relay_host: Url,
42
-
pub plc_url: Url,
43
pub full_network: bool,
44
pub cursor_save_interval: Duration,
45
pub repo_fetch_timeout: Duration,
···
84
"RELAY_HOST",
85
Url::parse("wss://relay.fire.hose.cam").unwrap()
86
);
87
-
let plc_url = cfg!("PLC_URL", Url::parse("https://plc.wtf").unwrap());
88
89
let full_network = cfg!("FULL_NETWORK", false);
90
let backfill_concurrency_limit = cfg!("BACKFILL_CONCURRENCY_LIMIT", 32usize);
···
107
Ok(Self {
108
database_path,
109
relay_host,
110
-
plc_url,
111
full_network,
112
cursor_save_interval,
113
repo_fetch_timeout,
···
132
writeln!(f, "hydrant configuration:")?;
133
writeln!(f, " log level: {}", self.log_level)?;
134
writeln!(f, " relay host: {}", self.relay_host)?;
135
-
writeln!(f, " plc url: {}", self.plc_url)?;
136
writeln!(f, " full network indexing: {}", self.full_network)?;
137
writeln!(f, " verify signatures: {}", self.verify_signatures)?;
138
writeln!(
···
39
pub struct Config {
40
pub database_path: PathBuf,
41
pub relay_host: Url,
42
+
pub plc_urls: Vec<Url>,
43
pub full_network: bool,
44
pub cursor_save_interval: Duration,
45
pub repo_fetch_timeout: Duration,
···
84
"RELAY_HOST",
85
Url::parse("wss://relay.fire.hose.cam").unwrap()
86
);
87
+
let plc_urls: Vec<Url> = std::env::var("HYDRANT_PLC_URL")
88
+
.ok()
89
+
.map(|s| {
90
+
s.split(',')
91
+
.map(|s| Url::parse(s.trim()))
92
+
.collect::<Result<Vec<_>, _>>()
93
+
.map_err(|e| miette::miette!("invalid PLC URL: {}", e))
94
+
})
95
+
.unwrap_or_else(|| Ok(vec![Url::parse("https://plc.wtf").unwrap()]))?;
96
97
let full_network = cfg!("FULL_NETWORK", false);
98
let backfill_concurrency_limit = cfg!("BACKFILL_CONCURRENCY_LIMIT", 32usize);
···
115
Ok(Self {
116
database_path,
117
relay_host,
118
+
plc_urls,
119
full_network,
120
cursor_save_interval,
121
repo_fetch_timeout,
···
140
writeln!(f, "hydrant configuration:")?;
141
writeln!(f, " log level: {}", self.log_level)?;
142
writeln!(f, " relay host: {}", self.relay_host)?;
143
+
writeln!(f, " plc urls: {:?}", self.plc_urls)?;
144
writeln!(f, " full network indexing: {}", self.full_network)?;
145
writeln!(f, " verify signatures: {}", self.verify_signatures)?;
146
writeln!(
+27
-11
src/resolver.rs
+27
-11
src/resolver.rs
···
1
use std::ops::Not;
2
use std::sync::Arc;
3
use std::time::Duration;
4
5
use jacquard::IntoStatic;
···
46
}
47
48
struct ResolverInner {
49
-
jacquard: JacquardResolver,
50
key_cache: HashCache<Did<'static>, PublicKey<'static>>,
51
}
52
···
56
}
57
58
impl Resolver {
59
-
pub fn new(plc_url: Url, identity_cache_size: u64) -> Self {
60
let http = reqwest::Client::new();
61
-
let mut opts = ResolverOptions::default();
62
-
opts.plc_source = PlcSource::PlcDirectory { base: plc_url };
63
-
opts.request_timeout = Some(Duration::from_secs(3));
64
65
-
// no jacquard cache - we manage our own
66
-
let jacquard = JacquardResolver::new(http, opts);
67
68
Self {
69
inner: Arc::new(ResolverInner {
70
-
jacquard,
71
key_cache: HashCache::with_capacity(
72
std::cmp::min(1000, (identity_cache_size / 100) as usize),
73
identity_cache_size as usize,
···
76
}
77
}
78
79
pub async fn resolve_did(
80
&self,
81
identifier: &AtIdentifier<'_>,
···
83
match identifier {
84
AtIdentifier::Did(did) => Ok(did.clone().into_static()),
85
AtIdentifier::Handle(handle) => {
86
-
let did = self.inner.jacquard.resolve_handle(handle).await?;
87
Ok(did.into_static())
88
}
89
}
···
93
&self,
94
did: &Did<'_>,
95
) -> Result<(Url, Option<Handle<'_>>), ResolverError> {
96
-
let doc_resp = self.inner.jacquard.resolve_did_doc(did).await?;
97
let doc = doc_resp.parse()?;
98
99
let pds = doc
···
115
return Ok(entry.get().clone());
116
}
117
118
-
let doc_resp = self.inner.jacquard.resolve_did_doc(&did).await?;
119
let doc = doc_resp.parse()?;
120
121
let key = doc
···
1
use std::ops::Not;
2
use std::sync::Arc;
3
+
use std::sync::atomic::{AtomicUsize, Ordering};
4
use std::time::Duration;
5
6
use jacquard::IntoStatic;
···
47
}
48
49
struct ResolverInner {
50
+
jacquards: Vec<JacquardResolver>,
51
+
next_idx: AtomicUsize,
52
key_cache: HashCache<Did<'static>, PublicKey<'static>>,
53
}
54
···
58
}
59
60
impl Resolver {
61
+
pub fn new(plc_urls: Vec<Url>, identity_cache_size: u64) -> Self {
62
let http = reqwest::Client::new();
63
+
let mut jacquards = Vec::with_capacity(plc_urls.len());
64
65
+
for url in plc_urls {
66
+
let mut opts = ResolverOptions::default();
67
+
opts.plc_source = PlcSource::PlcDirectory { base: url };
68
+
opts.request_timeout = Some(Duration::from_secs(3));
69
+
70
+
// no jacquard cache - we manage our own
71
+
jacquards.push(JacquardResolver::new(http.clone(), opts));
72
+
}
73
+
74
+
if jacquards.is_empty() {
75
+
panic!("at least one PLC URL must be provided");
76
+
}
77
78
Self {
79
inner: Arc::new(ResolverInner {
80
+
jacquards,
81
+
next_idx: AtomicUsize::new(0),
82
key_cache: HashCache::with_capacity(
83
std::cmp::min(1000, (identity_cache_size / 100) as usize),
84
identity_cache_size as usize,
···
87
}
88
}
89
90
+
fn get_jacquard(&self) -> &JacquardResolver {
91
+
let idx = self.inner.next_idx.fetch_add(1, Ordering::Relaxed) % self.inner.jacquards.len();
92
+
&self.inner.jacquards[idx]
93
+
}
94
+
95
pub async fn resolve_did(
96
&self,
97
identifier: &AtIdentifier<'_>,
···
99
match identifier {
100
AtIdentifier::Did(did) => Ok(did.clone().into_static()),
101
AtIdentifier::Handle(handle) => {
102
+
let did = self.get_jacquard().resolve_handle(handle).await?;
103
Ok(did.into_static())
104
}
105
}
···
109
&self,
110
did: &Did<'_>,
111
) -> Result<(Url, Option<Handle<'_>>), ResolverError> {
112
+
let doc_resp = self.get_jacquard().resolve_did_doc(did).await?;
113
let doc = doc_resp.parse()?;
114
115
let pds = doc
···
131
return Ok(entry.get().clone());
132
}
133
134
+
let doc_resp = self.get_jacquard().resolve_did_doc(&did).await?;
135
let doc = doc_resp.parse()?;
136
137
let key = doc
+1
-1
src/state.rs
+1
-1
src/state.rs