diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -64,6 +64,20 @@ events are de-duplicated using the `time` field. todo: decide what to do on relay-side account takedowns or if relays set the `time` field. +#### direct PDS connections + +a firehose source can also be a direct connection to a PDS rather than a relay. +prefix the URL with `pds::` to mark it as such: + +``` +HYDRANT_RELAY_HOSTS=wss://bsky.network,pds::wss://pds.example.com +``` + +only when a source is marked as a direct PDS (`is_pds: true`), hydrant enforces +host authority. relays (`is_pds: false`, the default) are exempt from this check, +since they forward commits from many PDSes by design. this means you will trust +the relay on this though. + ### crawler sources [<- back to toc](#table-of-contents) @@ -104,7 +118,7 @@ | `DATABASE_PATH` | `./hydrant.db` | path to the database folder. | | `RUST_LOG` | `info` | log filter directives (e.g., `debug`, `hydrant=trace`). [`tracing` env-filter syntax](https://docs.rs/tracing-subscriber/latest/tracing_subscriber/filter/struct.EnvFilter.html). | | `RELAY_HOST` | `wss://relay.fire.hose.cam/` | URL of the relay (firehose only). | -| `RELAY_HOSTS` | | comma-separated list of relay URLs (firehose only). if unset, falls back to `RELAY_HOST`. | +| `RELAY_HOSTS` | | comma-separated list of firehose sources (firehose only). if unset, falls back to `RELAY_HOST`. prefix a URL with `pds::` to mark it as a direct PDS connection (e.g. `pds::wss://pds.example.com`). bare URLs are treated as relays. | | `CRAWLER_URLS` | relay hosts in full-network mode, `https://lightrail.microcosm.blue` in filter mode | comma-separated list of `[mode::]url` crawler sources. mode is `relay` or `by_collection`; bare URLs use the default mode. set to empty string to disable crawling. | | `PLC_URL` | `https://plc.wtf`, `https://plc.directory` if full network | base URL(s) of the PLC directory (comma-separated for multiple). | | `EPHEMERAL` | `false` | if enabled, no records are stored. events are deleted after a certain duration (`EPHEMERAL_TTL`). | @@ -233,15 +247,16 @@ [<- back to toc](#table-of-contents) -- `GET /firehose/sources`: list all currently active firehose relay sources. - - returns a JSON array of `{ "url": string, "persisted": bool }`. +- `GET /firehose/sources`: list all currently active firehose sources. + - returns a JSON array of `{ "url": string, "persisted": bool, "is_pds": bool }`. - `persisted: true` means the source was added via the API and is stored in the database, it will survive a restart. `persisted: false` means the source came from `RELAY_HOSTS` and is not written to the database. -- `POST /firehose/sources`: add a firehose relay at runtime. - - body: `{ "url": string }`. + - `is_pds: true` means the source is a direct PDS connection with host authority enforcement enabled. +- `POST /firehose/sources`: add a firehose source at runtime. + - body: `{ "url": string, "is_pds": bool }`. `is_pds` defaults to `false`. - the source is persisted to the database before the ingestor task is started. - - if a relay with the same URL already exists, it is replaced: the running + - if a source with the same URL already exists, it is replaced: the running task is stopped and a new one is started. any existing cursor state for that URL is preserved. - returns `201 Created` on success. diff --git a/src/config.rs b/src/config.rs --- a/src/config.rs +++ b/src/config.rs @@ -122,6 +122,31 @@ pub mode: CrawlerMode, } +/// a single firehose source: a URL and whether it is a direct PDS connection. +/// +/// set via `HYDRANT_RELAY_HOSTS` as a comma-separated list of `[pds::]url` entries. +/// e.g. `wss://bsky.network,pds::wss://pds.example.com`. +/// a bare URL (no `pds::` prefix) is treated as an aggregating relay (`is_pds = false`). +#[derive(Debug, Clone)] +pub struct FirehoseSource { + pub url: Url, + /// true when this is a direct PDS connection; enables host authority enforcement. + pub is_pds: bool, +} + +impl FirehoseSource { + /// parse `[pds::]url`. the `pds::` prefix marks the source as a direct PDS connection. + pub fn parse(s: &str) -> Option { + if let Some(url_str) = s.strip_prefix("pds::") { + let url = Url::parse(url_str).ok()?; + Some(Self { url, is_pds: true }) + } else { + let url = Url::parse(s).ok()?; + Some(Self { url, is_pds: false }) + } + } +} + impl CrawlerSource { /// parse `[mode::]url`. mode prefix is optional, falls back to `default_mode`. fn parse(s: &str, default_mode: CrawlerMode) -> Option { @@ -214,9 +239,10 @@ /// set via `HYDRANT_EPHEMERAL_TTL` (humantime duration, e.g. `60min`). pub ephemeral_ttl: Duration, - /// relay URLs used for firehose ingestion. set via `HYDRANT_RELAY_HOST` (single) + /// firehose sources for ingestion. set via `HYDRANT_RELAY_HOST` (single) /// or `HYDRANT_RELAY_HOSTS` (comma-separated; takes precedence). - pub relays: Vec, + /// prefix a URL with `pds::` to mark it as a direct PDS connection. + pub relays: Vec, /// base URL(s) of the PLC directory (comma-separated for multiple). /// defaults to `https://plc.wtf`, or `https://plc.directory` in full-network mode. /// set via `HYDRANT_PLC_URL`. @@ -337,7 +363,10 @@ full_network: false, ephemeral: false, ephemeral_ttl: Duration::from_secs(3600), - relays: vec![Url::parse("wss://relay.fire.hose.cam/").unwrap()], + relays: vec![FirehoseSource { + url: Url::parse("wss://relay.fire.hose.cam/").unwrap(), + is_pds: false, + }], plc_urls: vec![Url::parse("https://plc.wtf").unwrap()], enable_firehose: true, firehose_workers: 8, @@ -412,17 +441,23 @@ let s = s.trim(); (!s.is_empty()) .then(|| { - Url::parse(s) - .inspect_err(|e| tracing::warn!("invalid relay host URL: {e}")) - .ok() + FirehoseSource::parse(s).or_else(|| { + tracing::warn!("invalid relay host URL: {s}"); + None + }) }) .flatten() }) .collect(), // HYDRANT_RELAY_HOSTS explicitly set to "" Ok(_) => vec![], - // not set at all, fall back to RELAY_HOST - Err(_) => vec![cfg!("RELAY_HOST", defaults.relays[0].clone())], + // not set at all, fall back to RELAY_HOST (bare URL, no pds:: prefix support here) + Err(_) => match std::env::var("HYDRANT_RELAY_HOST") { + Ok(s) if !s.trim().is_empty() => { + FirehoseSource::parse(s.trim()).into_iter().collect() + } + _ => defaults.relays.clone(), + }, }; let plc_urls: Vec = std::env::var("HYDRANT_PLC_URL") @@ -525,8 +560,8 @@ Err(_) => match default_mode { CrawlerMode::ListRepos => relay_hosts .iter() - .map(|url| CrawlerSource { - url: url.clone(), + .map(|source| CrawlerSource { + url: source.url.clone(), mode: CrawlerMode::ListRepos, }) .collect(), @@ -582,7 +617,21 @@ const LABEL_WIDTH: usize = 27; writeln!(f, "hydrant configuration:")?; - config_line!(f, "relay hosts", format_args!("{:?}", self.relays))?; + config_line!( + f, + "relay hosts", + format_args!( + "{:?}", + self.relays + .iter() + .map(|s| if s.is_pds { + format!("pds::{}", s.url) + } else { + s.url.to_string() + }) + .collect::>() + ) + )?; config_line!(f, "plc urls", format_args!("{:?}", self.plc_urls))?; config_line!(f, "full network indexing", self.full_network)?; config_line!(f, "verify signatures", self.verify_signatures)?; diff --git a/tests/common.nu b/tests/common.nu --- a/tests/common.nu +++ b/tests/common.nu @@ -56,7 +56,7 @@ } export def resolve-pds [did: string] { - let doc = (http get $"https://plc.wtf/($did)" | from json) + let doc = (http get $"https://plc.gaze.systems/($did)" | from json) ($doc.service | where type == "AtprotoPersonalDataServer" | first).serviceEndpoint } diff --git a/src/api/firehose.rs b/src/api/firehose.rs --- a/src/api/firehose.rs +++ b/src/api/firehose.rs @@ -24,6 +24,10 @@ #[derive(Deserialize)] pub struct AddSourceRequest { pub url: Url, + /// true to treat this as a direct PDS connection; enables host authority enforcement. + /// defaults to false (aggregating relay). + #[serde(default)] + pub is_pds: bool, } pub async fn add_source( @@ -32,7 +36,7 @@ ) -> Result { hydrant .firehose - .add_source(body.url) + .add_source(body.url, body.is_pds) .await .map(|_| StatusCode::CREATED) .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string())) diff --git a/src/control/firehose.rs b/src/control/firehose.rs --- a/src/control/firehose.rs +++ b/src/control/firehose.rs @@ -12,6 +12,7 @@ pub(super) struct FirehoseIngestorHandle { abort: tokio::task::AbortHandle, + pub(super) is_pds: bool, } impl Drop for FirehoseIngestorHandle { @@ -31,10 +32,13 @@ pub url: Url, /// true if added via the API and persisted to the database; false for `RELAY_HOSTS` sources. pub persisted: bool, + /// true when this is a direct PDS connection; enables host authority enforcement. + pub is_pds: bool, } pub(super) async fn spawn_firehose_ingestor( relay_url: &Url, + is_pds: bool, state: &Arc, shared: &FirehoseShared, enabled: watch::Receiver, @@ -48,12 +52,13 @@ .insert_async(relay_url.clone(), AtomicI64::new(start.unwrap_or(0))) .await; - info!(relay = %relay_url, cursor = ?start, "starting firehose ingestor"); + info!(relay = %relay_url, is_pds, cursor = ?start, "starting firehose ingestor"); let ingestor = FirehoseIngestor::new( state.clone(), shared.buffer_tx.clone(), relay_url.clone(), + is_pds, state.filter.clone(), enabled, shared.verify_signatures, @@ -67,7 +72,7 @@ }) .abort_handle(); - Ok(FirehoseIngestorHandle { abort }) + Ok(FirehoseIngestorHandle { abort, is_pds }) } /// runtime control over the firehose ingestor component. @@ -120,10 +125,11 @@ pub async fn list_sources(&self) -> Vec { let mut sources = Vec::new(); self.tasks - .iter_async(|url, _| { + .iter_async(|url, handle| { sources.push(FirehoseSourceInfo { url: url.clone(), persisted: self.persisted.contains_sync(url), + is_pds: handle.is_pds, }); true }) @@ -138,19 +144,21 @@ /// is started. any cursor state for that URL is preserved. /// /// returns an error if called before [`Hydrant::run`]. - pub async fn add_source(&self, url: Url) -> Result<()> { + pub async fn add_source(&self, url: Url, is_pds: bool) -> Result<()> { let Some(shared) = self.shared.get() else { miette::bail!("firehose not yet started: call Hydrant::run() first"); }; let db = self.state.db.clone(); let key = keys::firehose_source_key(url.as_str()); - tokio::task::spawn_blocking(move || db.crawler.insert(key, b"").into_diagnostic()) + let value = rmp_serde::to_vec(&crate::db::FirehoseSourceMeta { is_pds }) + .map_err(|e| miette::miette!("failed to serialize firehose source meta: {e}"))?; + tokio::task::spawn_blocking(move || db.crawler.insert(key, value).into_diagnostic()) .await .into_diagnostic()??; let enabled_rx = self.state.firehose_enabled.subscribe(); - let handle = spawn_firehose_ingestor(&url, &self.state, shared, enabled_rx).await?; + let handle = spawn_firehose_ingestor(&url, is_pds, &self.state, shared, enabled_rx).await?; let _ = self.persisted.insert_async(url.clone()).await; match self.tasks.entry_async(url).await { diff --git a/src/control/mod.rs b/src/control/mod.rs --- a/src/control/mod.rs +++ b/src/control/mod.rs @@ -350,35 +350,53 @@ relay_count = relay_hosts.len(), hosts = relay_hosts .iter() - .map(|h| h.as_str()) + .map(|h| h.url.as_str()) .collect::>() .join(", "), "starting firehose ingestor(s)" ); - for relay_url in &relay_hosts { + for source in &relay_hosts { let enabled_rx = state.firehose_enabled.subscribe(); - let handle = - spawn_firehose_ingestor(relay_url, &state, fire_shared, enabled_rx).await?; - let _ = firehose.tasks.insert_async(relay_url.clone(), handle).await; + let handle = spawn_firehose_ingestor( + &source.url, + source.is_pds, + &state, + fire_shared, + enabled_rx, + ) + .await?; + let _ = firehose + .tasks + .insert_async(source.url.clone(), handle) + .await; } } - let persisted_relay_urls = tokio::task::spawn_blocking({ + let persisted_sources = tokio::task::spawn_blocking({ let state = state.clone(); move || load_persisted_firehose_sources(&state.db) }) .await .into_diagnostic()??; - for relay_url in &persisted_relay_urls { - let _ = firehose.persisted.insert_async(relay_url.clone()).await; - if firehose.tasks.contains_async(relay_url).await { + for source in &persisted_sources { + let _ = firehose.persisted.insert_async(source.url.clone()).await; + if firehose.tasks.contains_async(&source.url).await { continue; } let enabled_rx = state.firehose_enabled.subscribe(); - let handle = - spawn_firehose_ingestor(relay_url, &state, fire_shared, enabled_rx).await?; - let _ = firehose.tasks.insert_async(relay_url.clone(), handle).await; + let handle = spawn_firehose_ingestor( + &source.url, + source.is_pds, + &state, + fire_shared, + enabled_rx, + ) + .await?; + let _ = firehose + .tasks + .insert_async(source.url.clone(), handle) + .await; } // 11. spawn crawler infrastructure (always, to support dynamic source management) diff --git a/src/db/mod.rs b/src/db/mod.rs --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -861,18 +861,31 @@ Ok(count.unwrap_or(0)) } -pub fn load_persisted_firehose_sources(db: &crate::db::Db) -> Result> { +#[derive(serde::Serialize, serde::Deserialize, Default)] +pub(crate) struct FirehoseSourceMeta { + #[serde(default)] + pub(crate) is_pds: bool, +} + +pub fn load_persisted_firehose_sources( + db: &crate::db::Db, +) -> Result> { use crate::db::keys::FIREHOSE_SOURCE_PREFIX; - let mut urls = Vec::new(); + let mut sources = Vec::new(); for entry in db.crawler.prefix(FIREHOSE_SOURCE_PREFIX) { - let (key, _) = entry.into_inner().into_diagnostic()?; + let (key, val) = entry.into_inner().into_diagnostic()?; let url_bytes = &key[FIREHOSE_SOURCE_PREFIX.len()..]; let url_str = std::str::from_utf8(url_bytes).into_diagnostic()?; let url = Url::parse(url_str).into_diagnostic()?; - urls.push(url); + let meta: FirehoseSourceMeta = rmp_serde::from_slice(&val) + .map_err(|e| miette::miette!("failed to deserialize firehose source meta: {e}"))?; + sources.push(crate::config::FirehoseSource { + url, + is_pds: meta.is_pds, + }); } - Ok(urls) + Ok(sources) } pub fn load_persisted_crawler_sources( diff --git a/src/ingest/firehose.rs b/src/ingest/firehose.rs --- a/src/ingest/firehose.rs +++ b/src/ingest/firehose.rs @@ -18,6 +18,7 @@ state: Arc, buffer_tx: BufferTx, relay_host: Url, + is_pds: bool, filter: FilterHandle, enabled: watch::Receiver, _verify_signatures: bool, @@ -28,6 +29,7 @@ state: Arc, buffer_tx: BufferTx, relay_host: Url, + is_pds: bool, filter: FilterHandle, enabled: watch::Receiver, verify_signatures: bool, @@ -36,6 +38,7 @@ state, buffer_tx, relay_host, + is_pds, filter, enabled, _verify_signatures: verify_signatures, @@ -130,6 +133,7 @@ if let Err(e) = self.buffer_tx.send(IngestMessage::Firehose { relay: self.relay_host.clone(), + is_pds: self.is_pds, msg: msg.into_static(), }) { error!(err = %e, "failed to send message to buffer processor"); diff --git a/src/ingest/mod.rs b/src/ingest/mod.rs --- a/src/ingest/mod.rs +++ b/src/ingest/mod.rs @@ -14,6 +14,9 @@ pub enum IngestMessage { Firehose { relay: Url, + /// true when `relay` is a direct PDS connection (not an aggregating relay). + /// enables host authority enforcement in the worker. + is_pds: bool, msg: SubscribeReposMessage<'static>, }, BackfillFinished(Did<'static>), diff --git a/src/ingest/worker.rs b/src/ingest/worker.rs --- a/src/ingest/worker.rs +++ b/src/ingest/worker.rs @@ -52,6 +52,15 @@ } } +enum HostAuthorityOutcome { + /// stored pds matched the source host immediately. + Authorized, + /// pds migrated: doc now points to this host, but our stored state was stale. trigger backfill. + Migration, + /// host did not match even after doc resolution. reject the message. + WrongHost, +} + // gate returned by check_repo_state, tells the shard loop what to do with the message enum ProcessGate<'s, 'c> { // did not exist in db, newly queued for backfill, drop @@ -249,8 +258,10 @@ } } } - IngestMessage::Firehose { relay, msg } => { + IngestMessage::Firehose { relay, is_pds, msg } => { let _span = tracing::info_span!("firehose", relay = %relay).entered(); + // only enforce host authority when the source is a direct PDS connection + let source_host = is_pds.then(|| relay.host_str()).flatten(); let (did, seq) = match &msg { SubscribeReposMessage::Commit(c) => (&c.repo, c.seq), SubscribeReposMessage::Identity(i) => (&i.did, i.seq), @@ -330,8 +341,14 @@ } } - match Self::process_message(&mut ctx, &msg, did, repo_state, pre_status) - { + match Self::process_message( + &mut ctx, + &msg, + did, + repo_state, + pre_status, + source_host, + ) { Ok(RepoProcessResult::Ok(_)) => {} Ok(RepoProcessResult::Deleted) => { state.db.update_count("repos", -1); @@ -411,15 +428,16 @@ did: &Did, repo_state: RepoState<'s>, pre_status: RepoStatus, + source_host: Option<&str>, ) -> Result, IngestError> { match msg { SubscribeReposMessage::Commit(commit) => { trace!(did = %did, "processing commit"); - Self::handle_commit(ctx, did, repo_state, commit) + Self::handle_commit(ctx, did, repo_state, commit, source_host) } SubscribeReposMessage::Sync(sync) => { debug!(did = %did, "processing sync"); - Self::handle_sync(ctx, did, repo_state, sync) + Self::handle_sync(ctx, did, repo_state, sync, source_host) } SubscribeReposMessage::Identity(identity) => { debug!(did = %did, "processing identity"); @@ -441,10 +459,38 @@ did: &Did, mut repo_state: RepoState<'s>, commit: &'c Commit<'c>, + source_host: Option<&str>, ) -> Result, IngestError> { repo_state.advance_message_time(commit.time.0.timestamp_millis()); - // TODO phase 2: host authority check (source_host not available in indexer mode) + if let Some(host) = source_host { + match Self::check_host_authority(ctx, did, &mut repo_state, host)? { + HostAuthorityOutcome::Authorized => {} + HostAuthorityOutcome::Migration => { + // pds migrated: our data may be stale, backfill from the new host + warn!(did = %did, source_host = host, "pds migration detected, triggering backfill"); + let mut batch = ctx.state.db.inner.batch(); + let _repo_state = ops::update_repo_status( + &mut batch, + &ctx.state.db, + did, + repo_state, + RepoStatus::Backfilling, + )?; + batch.commit().into_diagnostic()?; + ctx.state + .db + .update_gauge_diff(&GaugeState::Synced, &GaugeState::Pending); + ctx.state.notify_backfill(); + return Ok(RepoProcessResult::NeedsBackfill(Some(commit))); + } + // todo: ideally ban pds + HostAuthorityOutcome::WrongHost => { + warn!(did = %did, source_host = host, pds = ?repo_state.pds, "commit rejected: wrong host"); + return Ok(RepoProcessResult::Ok(repo_state)); + } + } + } // validate the commit: stale rev, size limits, future rev, CAR parse, field // consistency, signature, and chain-break detection @@ -536,10 +582,22 @@ did: &Did, mut repo_state: RepoState<'s>, sync: &'c Sync<'c>, + source_host: Option<&str>, ) -> Result, IngestError> { repo_state.advance_message_time(sync.time.0.timestamp_millis()); - // TODO phase 2: host authority check + if let Some(host) = source_host { + match Self::check_host_authority(ctx, did, &mut repo_state, host)? { + HostAuthorityOutcome::Authorized | HostAuthorityOutcome::Migration => { + // migration is fine here — sync already triggers a backfill below + } + // todo: ideally ban pds + HostAuthorityOutcome::WrongHost => { + warn!(did = %did, source_host = host, pds = ?repo_state.pds, "sync rejected: wrong host"); + return Ok(RepoProcessResult::Ok(repo_state)); + } + } + } // validate: size limit, CAR parse, field consistency, signature let signing_key = Self::fetch_key(ctx, did)?; @@ -864,7 +922,8 @@ let (key, value) = guard.into_inner().into_diagnostic()?; let commit: Commit = rmp_serde::from_slice(&value).into_diagnostic()?; - let res = Self::handle_commit(ctx, did, repo_state, &commit); + // buffered commits have already been source-checked on arrival; skip host check + let res = Self::handle_commit(ctx, did, repo_state, &commit, None); let res = match res { Ok(r) => r, Err(e) => { @@ -891,6 +950,43 @@ } Ok(RepoProcessResult::Ok(repo_state)) + } + + /// check that `source_host` is the authoritative PDS for `did`. + /// + /// - `Authorized`: stored pds matched immediately (fast path). + /// - `Migration`: stored pds was wrong but doc resolved to this host; caller should backfill. + /// - `WrongHost`: host did not match even after doc resolution; caller should reject. + fn check_host_authority( + ctx: &mut WorkerContext, + did: &Did, + repo_state: &mut RepoState, + source_host: &str, + ) -> Result { + let pds_host = repo_state + .pds + .as_deref() + .and_then(|pds| url::Url::parse(pds).ok()) + .and_then(|u| u.host_str().map(str::to_owned)); + + if pds_host.as_deref() == Some(source_host) { + return Ok(HostAuthorityOutcome::Authorized); + } + + // unknown pds or host mismatch — resolve doc to verify or detect a migration + Self::refresh_doc(ctx, repo_state, did)?; + + let updated_host = repo_state + .pds + .as_deref() + .and_then(|pds| url::Url::parse(pds).ok()) + .and_then(|u| u.host_str().map(str::to_owned)); + + if updated_host.as_deref() == Some(source_host) { + Ok(HostAuthorityOutcome::Migration) + } else { + Ok(HostAuthorityOutcome::WrongHost) + } } // refreshes the handle, pds url and signing key of a did diff --git a/src/db/migration/mod.rs b/src/db/migration/mod.rs --- a/src/db/migration/mod.rs +++ b/src/db/migration/mod.rs @@ -6,6 +6,7 @@ mod v1; mod v2; +mod v3; type MigrationFn = fn(&Db, &mut OwnedWriteBatch) -> Result<()>; @@ -13,6 +14,7 @@ const MIGRATIONS: &[(&str, MigrationFn)] = &[ ("stable_firehose_cursors", v1::stable_firehose_cursors), ("repo_state_root_commit", v2::repo_state_root_commit), + ("firehose_source_is_pds", v3::firehose_source_is_pds), ]; fn read_version(db: &Db) -> Result { diff --git a/src/db/migration/v3.rs b/src/db/migration/v3.rs new file mode 100644 --- /dev/null +++ b/src/db/migration/v3.rs @@ -0,0 +1,63 @@ +use fjall::OwnedWriteBatch; +use miette::{IntoDiagnostic, Result}; + +use crate::db::{Db, FirehoseSourceMeta, keys}; + +// for this migration, we default to everything being a PDS unless it matches +// one of the hosts mentioned in this list. this is best effort but its +// better than defaulting to false and entirely disabling validation, +// the user can manage the firehose sources when they see the error anyway. +// (and i like to be correct :P) +const KNOWN_RELAY_HOSTS: &[&str] = &[ + "atproto.africa", + "bsky.network", + "relay1.us-east.bsky.network", + "relay1.us-west.bsky.network", + "relay.fire.hose.cam", + "relay3.fr.hose.cam", + "relay.upcloud.world", + "relay.hayescmd.net", + "relay.xero.systems", + "relay.feeds.blue", + "zlay.waow.tech", + "asia.firehose.network", + "europe.firehose.network", + "northamerica.firehose.network", + "relay.bas.sh", + "relay.t4tlabs.net", + "relay.waow.tech", +]; + +fn is_known_relay(url_bytes: &[u8]) -> bool { + let Ok(url_str) = std::str::from_utf8(url_bytes) else { + return false; + }; + let Ok(url) = url::Url::parse(url_str) else { + return false; + }; + url.host_str() + .is_some_and(|h| KNOWN_RELAY_HOSTS.contains(&h)) +} + +pub(super) fn firehose_source_is_pds(db: &Db, batch: &mut OwnedWriteBatch) -> Result<()> { + let relay_bytes = rmp_serde::to_vec(&FirehoseSourceMeta { is_pds: false }) + .map_err(|e| miette::miette!("failed to serialize meta: {e}"))?; + let pds_bytes = rmp_serde::to_vec(&FirehoseSourceMeta { is_pds: true }) + .map_err(|e| miette::miette!("failed to serialize meta: {e}"))?; + + for item in db.crawler.prefix(keys::FIREHOSE_SOURCE_PREFIX) { + let (key, val) = item.into_inner().into_diagnostic()?; + if !val.is_empty() { + continue; + } + let url_bytes = &key[keys::FIREHOSE_SOURCE_PREFIX.len()..]; + let value = if is_known_relay(url_bytes) { + &relay_bytes + } else { + &pds_bytes + }; + batch.insert(&db.crawler, key, value); + } + + Ok(()) +}