use std::time::Duration;
use jacquard_common::types::string::Did;
use serde::Deserialize;
use smol_str::{SmolStr, ToSmolStr};
use tracing::warn;
use url::Url;
use crate::ingest::firehose_stats::{RelayMessageKind, ValidationStatsOutcome};
use crate::ingest::stream::{AccountStatus, SubscribeReposMessage};
#[derive(Deserialize)]
pub(crate) struct GetRepoStatusOutput<'a> {
pub active: bool,
#[serde(borrow)]
pub status: Option<AccountStatus<'a>>,
}
use crate::ingest::validation::{
CommitValidationError, SyncValidationError, ValidatedCommit, ValidatedSync,
};
use crate::net::{PublicEndpoint, PublicHttpClient};
use crate::types::{RepoState, RepoStatus};
pub mod context;
pub mod handlers;
pub(crate) mod lookup;
#[cfg(feature = "indexer")]
pub(crate) mod overlay;
pub(crate) mod parked;
#[cfg(test)]
mod pipeline_tests;
pub(crate) mod sink;
pub mod worker;
pub(crate) use context::WorkerContext;
pub use worker::RelayWorker;
pub(crate) use worker::WorkerMessage;
pub(crate) const WRONG_HOST_AUTHORITY_RECHECK_INTERVAL: Duration = Duration::from_secs(60);
pub(crate) const WRONG_HOST_AUTHORITY_CACHE_PRUNE_AT: usize = 8192;
pub(crate) fn map_repo_status_probe(
output: Option<GetRepoStatusOutput<'_>>,
) -> Option<RepoState<'static>> {
let output = output?;
let mut repo_state = RepoState::backfilling();
repo_state.active = output.active;
repo_state.status = match output.status {
Some(AccountStatus::Takendown) => RepoStatus::Takendown,
Some(AccountStatus::Suspended) => RepoStatus::Suspended,
Some(AccountStatus::Deactivated) => RepoStatus::Deactivated,
Some(AccountStatus::Deleted) => RepoStatus::Deleted,
Some(AccountStatus::Desynchronized) => RepoStatus::Desynchronized,
Some(AccountStatus::Throttled) => RepoStatus::Throttled,
Some(AccountStatus::Other(s)) => RepoStatus::Error(s.to_smolstr()),
None => output
.active
.then_some(RepoStatus::Synced)
.unwrap_or_else(|| RepoStatus::Error("unknown".into())),
};
Some(repo_state)
}
/// what a pds said when asked about one of its repos
pub(crate) enum StatusProbe {
/// `None` when it said nothing we can use, which is no reason to distrust the host
Answered(Option<RepoState<'static>>),
Unreachable,
}
pub(crate) async fn probe_repo_status(
http: &PublicHttpClient,
did: &Did,
pds: &Url,
) -> StatusProbe {
let Ok(mut url) = pds.join("xrpc/com.atproto.sync.getRepoStatus") else {
return StatusProbe::Answered(None);
};
url.query_pairs_mut().append_pair("did", did.as_str());
let Ok(endpoint) = PublicEndpoint::parse_http(&url) else {
warn!(%did, pds = %pds, "refusing repo status request to unsafe PDS endpoint");
return StatusProbe::Answered(None);
};
let Ok(request) = http::Request::builder()
.method(http::Method::GET)
.uri(endpoint.url().as_str())
.header(http::header::ACCEPT, "application/json")
.body(Vec::new())
else {
return StatusProbe::Answered(None);
};
let Ok(resp) = http.send_http(request).await else {
return StatusProbe::Unreachable;
};
let status = resp.status();
if status.is_success() {
StatusProbe::Answered(
serde_json::from_slice::<GetRepoStatusOutput>(resp.body())
.ok()
.and_then(|output| map_repo_status_probe(Some(output))),
)
} else if matches!(status.as_u16(), 400 | 401) {
StatusProbe::Answered(None)
} else {
StatusProbe::Unreachable
}
}
pub(crate) fn relay_message_kind(msg: &SubscribeReposMessage<'_>) -> Option<RelayMessageKind> {
match msg {
SubscribeReposMessage::Commit(_) => Some(RelayMessageKind::Commit),
SubscribeReposMessage::Sync(_) => Some(RelayMessageKind::Sync),
SubscribeReposMessage::Identity(_) => Some(RelayMessageKind::Identity),
SubscribeReposMessage::Account(_) => Some(RelayMessageKind::Account),
SubscribeReposMessage::Info(_) => None,
}
}
pub(crate) fn commit_validation_outcome(
res: &std::result::Result<ValidatedCommit<'_>, CommitValidationError>,
) -> ValidationStatsOutcome {
match res {
Ok(_) => ValidationStatsOutcome::Accepted,
Err(CommitValidationError::StaleRev) => ValidationStatsOutcome::Stale,
Err(CommitValidationError::SigFailure) => ValidationStatsOutcome::SigFailure,
Err(_) => ValidationStatsOutcome::Rejected,
}
}
pub(crate) fn sync_validation_outcome(
res: &std::result::Result<ValidatedSync, SyncValidationError>,
) -> ValidationStatsOutcome {
match res {
Ok(_) => ValidationStatsOutcome::Accepted,
Err(SyncValidationError::SigFailure) => ValidationStatsOutcome::SigFailure,
Err(_) => ValidationStatsOutcome::Rejected,
}
}
/// outcome of a host authority check.
pub(crate) enum AuthorityOutcome {
/// stored pds matched the source host immediately.
Authorized,
/// pds migrated: doc now points to this host, but our stored state was stale.
/// `identity_moved` says the re-resolve also rewrote handle, pds or key, so
/// consumers holding a cached identity need to be told.
WasStale {
#[cfg_attr(
not(any(feature = "indexer_stream", feature = "relay")),
allow(dead_code)
)]
identity_moved: bool,
},
/// host did not match even after doc resolution.
WrongHost { expected: SmolStr },
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn missing_repo_status_probe_falls_back_to_live_discovery() {
assert!(map_repo_status_probe(None).is_none());
}
#[test]
fn active_repo_status_probe_maps_to_synced_repo_state() {
let repo_state = map_repo_status_probe(Some(GetRepoStatusOutput {
active: true,
status: None,
}))
.expect("probe should map");
assert!(repo_state.active);
assert_eq!(repo_state.status, RepoStatus::Synced);
}
#[test]
fn throttled_repo_status_probe_preserves_host_status() {
let repo_state = map_repo_status_probe(Some(GetRepoStatusOutput {
active: true,
status: Some(AccountStatus::Throttled),
}))
.expect("probe should map");
assert!(repo_state.active);
assert_eq!(repo_state.status, RepoStatus::Throttled);
}
}