use std::collections::HashMap;
use std::sync::Arc;
use chrono::{DateTime, Utc};
use jacquard_common::cowstr::ToCowStr;
use jacquard_common::types::cid::{Cid, IpldCid};
use jacquard_common::types::ident::AtIdentifier;
use jacquard_common::types::nsid::Nsid;
use jacquard_common::types::string::{Did, Handle, Rkey};
use jacquard_common::types::tid::Tid;
use jacquard_common::{CowStr, Data, IntoStatic};
use miette::{Context, IntoDiagnostic, Result, WrapErr};
use smol_str::ToSmolStr;
use url::Url;
use crate::db::types::{DbRkey, DidKey, TrimmedDid};
use crate::db::{self, Db, keys};
#[cfg(feature = "indexer")]
use crate::ingest::indexer::repos::{Queued, queue_backfill};
use crate::state::AppState;
#[cfg(feature = "indexer")]
use crate::types::GaugeState;
use crate::types::{RepoMetadata, RepoState, RepoStatus};
use crate::util::invalid_handle;
#[cfg(feature = "indexer")]
mod indexer;
#[cfg(feature = "indexer")]
pub use indexer::*;
/// information about a repository known to hydrant. returned by [`ReposControl`] methods.
#[derive(Debug, Clone, serde::Serialize)]
pub struct RepoInfo {
/// the DID of the repository.
pub did: Did,
/// the status of the repository.
#[serde(serialize_with = "crate::util::repo_status_serialize_str")]
pub status: RepoStatus,
/// whether this repository is tracked or not.
/// untracked repositories are not updated and they stay frozen.
///
/// this will always be `true` in relay mode.
pub tracked: bool,
/// the revision of the root commit of this repository.
#[serde(skip_serializing_if = "Option::is_none")]
pub rev: Option<Tid>,
/// the CID of the MST root of this repository.
#[serde(serialize_with = "crate::util::opt_cid_serialize_str")]
#[serde(skip_serializing_if = "Option::is_none")]
pub data: Option<IpldCid>,
/// the handle for the DID of this repository.
///
/// note that this handle is not bi-directionally verified.
#[serde(skip_serializing_if = "Option::is_none")]
pub handle: Option<Handle>,
/// the URL for the PDS in which this repository is hosted on.
#[serde(skip_serializing_if = "Option::is_none")]
pub pds: Option<Url>,
/// ATProto signing key of this repository.
#[serde(serialize_with = "crate::util::opt_did_key_serialize_str")]
#[serde(skip_serializing_if = "Option::is_none")]
pub signing_key: Option<DidKey<'static>>,
/// when this repository was last touched (status update, commit ingested, etc.).
#[serde(skip_serializing_if = "Option::is_none")]
pub last_updated_at: Option<DateTime<Utc>>,
/// the time of the last message gotten from the firehose for this repository.
/// this is equal to the `time` field.
#[serde(skip_serializing_if = "Option::is_none")]
pub last_message_at: Option<DateTime<Utc>>,
/// whether a backfill of this repository is queued or running, retries
/// included.
#[cfg(feature = "indexer")]
pub pending: bool,
/// why the repository is parked in the resync queue, if it is.
#[cfg(feature = "indexer")]
#[serde(skip_serializing_if = "Option::is_none")]
pub resync: Option<ResyncInfo>,
}
/// a repository waiting in the resync queue, see [`RepoInfo::resync`].
#[cfg(feature = "indexer")]
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
#[serde(tag = "state", rename_all = "snake_case")]
pub enum ResyncInfo {
/// the last backfill failed and will be retried at `next_retry`.
Error {
kind: FailureKind,
retry_count: u32,
next_retry: DateTime<Utc>,
/// transport failures in a row the repository's pds has had since
/// hydrant started, 0 once it answers again. a `ratelimited` repo on
/// a pds that keeps failing is waiting for a dead host, not for a
/// rate limit to pass.
host_failures: usize,
},
/// the account is inactive, so it gets backfilled again once it comes back.
Gone {
#[serde(serialize_with = "crate::util::repo_status_serialize_str")]
status: RepoStatus,
},
}
/// why the last backfill of a repository failed, see [`ResyncInfo::Error`].
#[cfg(feature = "indexer")]
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
#[serde(rename_all = "snake_case")]
pub enum FailureKind {
/// the pds asked to slow down
Ratelimited,
/// the pds couldn't be reached
Transport,
/// anything else
Generic,
}
#[cfg(feature = "indexer")]
impl ResyncInfo {
fn new(state: crate::types::ResyncState, host_failures: usize) -> Result<Self> {
use crate::types::{ResyncErrorKind, ResyncState};
Ok(match state {
ResyncState::Error {
kind,
retry_count,
next_retry,
} => ResyncInfo::Error {
kind: match kind {
ResyncErrorKind::Ratelimited => FailureKind::Ratelimited,
ResyncErrorKind::Transport => FailureKind::Transport,
ResyncErrorKind::Generic => FailureKind::Generic,
},
retry_count,
next_retry: DateTime::from_timestamp_secs(next_retry).ok_or_else(|| {
miette::miette!("resync entry retries at an impossible time: {next_retry}")
})?,
host_failures,
},
ResyncState::Gone { status } => ResyncInfo::Gone { status },
})
}
}
/// fills in [`RepoInfo::pending`] and [`RepoInfo::resync`], which live in the
/// indexer's queues rather than in the repo state.
#[cfg(feature = "indexer")]
fn read_queues(state: &AppState, info: &mut RepoInfo, index_id: Option<u64>) -> Result<()> {
if let Some(index_id) = index_id {
info.pending = state
.db
.indexer
.is_pending(&keys::pending_key(index_id))
.into_diagnostic()?;
}
let Some(bytes) = state
.db
.indexer
.resync
.get(keys::repo_key(&info.did))
.into_diagnostic()?
else {
return Ok(());
};
// a repo that never backfilled has no pds in its state yet, but resolving
// it for the failed attempt left one in the identity cache
let pds = info
.pds
.clone()
.or_else(|| state.resolver.cached_pds(&info.did));
let host_failures = pds.map_or(0, |pds| state.throttler.snapshot(&pds).consecutive_failures);
// one unreadable entry would fail every listing page it lands on, so the cursor
// could never get past it. the row lists without it and the log says why
match rmp_serde::from_slice(&bytes)
.into_diagnostic()
.and_then(|resync| ResyncInfo::new(resync, host_failures))
{
Ok(resync) => info.resync = Some(resync),
Err(e) => {
// it stays unreadable and a poller lists the queue every few seconds, so each one
// warns once per process
static WARNED: std::sync::LazyLock<scc::HashSet<Did>> =
std::sync::LazyLock::new(scc::HashSet::new);
if WARNED.insert_sync(info.did.clone()).is_ok() {
tracing::warn!(did = %info.did, err = %e, "skipping an unreadable resync entry");
}
}
}
Ok(())
}
/// control over repositories and access to their state.
///
/// in indexer mode, repositories can be explicitly tracked or untracked:
/// - in `filter` mode, a repo is only indexed if it matches a signal or is
/// explicitly tracked via [`ReposControl::track`]. in `full` mode all repos
/// are indexed and tracking is implicit.
/// - tracking a DID that hydrant has never seen enqueues an immediate backfill.
/// - tracking a DID that hydrant already knows about (but has marked untracked)
/// re-enqueues it for backfill.
///
/// in relay mode, all observed repositories are passively indexed. explicit
/// tracking and backfill do not apply; [`RepoInfo::tracked`] is always `true`.
#[derive(Clone)]
pub struct ReposControl(pub(super) Arc<AppState>);
#[cfg(all(test, feature = "indexer"))]
impl ReposControl {
pub(crate) fn new(state: Arc<AppState>) -> Self {
Self(state)
}
}
impl ReposControl {
pub(crate) fn iter_states(
&self,
cursor: Option<&Did>,
) -> impl Iterator<Item = Result<(Did, RepoState<'static>)>> {
let start_bound = if let Some(cursor) = cursor {
let did_key = keys::repo_key(cursor);
std::ops::Bound::Excluded(did_key)
} else {
std::ops::Bound::Unbounded
};
self.0
.db
.repos
.range((start_bound, std::ops::Bound::Unbounded))
.map(|g| {
let (k, v) = g.into_inner().into_diagnostic()?;
let repo_state = crate::db::deser_repo_state(&v)?.into_static();
let did = TrimmedDid::try_from(k.as_ref())?.to_did();
Ok((did, repo_state))
})
}
/// iterates through all repositories, returning their state.
pub fn iter(&self, cursor: Option<&Did>) -> impl Iterator<Item = Result<RepoInfo>> {
let state = self.0.clone();
self.iter_states(cursor)
.map(move |r| r.and_then(|(did, s)| repo_row(&state, did, s)))
}
/// gets a handle for a repository to read from it.
pub fn get(&self, did: &Did) -> RepoHandle {
RepoHandle {
state: self.0.clone(),
did: did.clone(),
}
}
/// same as [`ReposControl::get`] but allows you to pass in an identifier that can be
/// either a handle or a DID.
pub async fn resolve(&self, repo: &AtIdentifier) -> Result<RepoHandle> {
let did = self.0.resolver.resolve_did(repo).await?;
Ok(RepoHandle {
state: self.0.clone(),
did,
})
}
/// fetch the current state of a repository.
/// returns `None` if hydrant has never seen this repository.
pub async fn info(&self, did: &Did) -> Result<Option<RepoInfo>> {
self.get(did).info().await
}
}
/// a repo as `/repos` lists it, with whether it's tracked and what the indexer
/// still has queued for it
#[cfg_attr(not(feature = "indexer"), allow(unused_variables))]
fn repo_row(state: &AppState, did: Did, s: RepoState<'_>) -> Result<RepoInfo> {
#[cfg(feature = "indexer")]
let metadata = state
.db
.repo_metadata
.get(keys::repo_metadata_key(&did))
.into_diagnostic()?
.map(|b| crate::db::deser_repo_meta(&b))
.transpose()?;
#[cfg(feature = "indexer")]
let tracked = metadata.as_ref().is_none_or(|m| m.tracked);
#[cfg(not(feature = "indexer"))]
let tracked = true;
#[allow(unused_mut)]
let mut info = repo_state_to_info(did, s, tracked);
#[cfg(feature = "indexer")]
read_queues(state, &mut info, metadata.map(|m| m.index_id))?;
Ok(info)
}
pub(crate) fn repo_state_to_info(did: Did, s: RepoState<'_>, tracked: bool) -> RepoInfo {
let (rev, data) = s
.root
.map(|c| (Some(c.rev.to_tid()), Some(c.data)))
.unwrap_or_default();
RepoInfo {
did,
status: s.status,
tracked,
rev,
data,
handle: s.handle.map(|h| h.into_static()),
pds: s.pds.and_then(|p| p.parse().ok()),
signing_key: s.signing_key.map(|k| k.into_static()),
last_updated_at: DateTime::from_timestamp_secs(s.last_updated_at),
last_message_at: s
.last_message_time
.and_then(DateTime::from_timestamp_millis),
#[cfg(feature = "indexer")]
pending: false,
#[cfg(feature = "indexer")]
resync: None,
}
}
pub struct Record {
pub did: Did,
pub cid: Cid,
pub value: Data,
}
#[cfg(test)]
mod tests {
use super::*;
use miette::IntoDiagnostic;
#[test]
fn identity_reconcile_trigger_matrix_is_narrow() -> miette::Result<()> {
let mut complete = RepoState::synced();
complete.pds = Some("https://pds.example".into());
complete.handle = Some(Handle::new_static("alice.test").into_diagnostic()?);
complete.signing_key = Some(DidKey::from_did_key(
"did:key:zQ3shokFTS3brHcDQrn82RUDfCZESWL1ZdCEJwekUDPQiYBme",
)?);
let same_handle = complete.handle.as_ref();
assert!(!needs_identity_reconcile(&complete, None, false));
assert!(!needs_identity_reconcile(&complete, same_handle, false));
assert!(needs_identity_reconcile(
&complete,
Some(&Handle::new_static("carol.test").into_diagnostic()?),
false,
));
for missing in [
{
let mut state = complete.clone();
state.pds = None;
state
},
{
let mut state = complete.clone();
state.signing_key = None;
state
},
] {
assert!(needs_identity_reconcile(&missing, None, false));
}
let mut deactivated = complete.clone();
deactivated.status = RepoStatus::Deactivated;
assert!(!needs_identity_reconcile(&deactivated, None, false));
let mut inactive_synced = complete.clone();
inactive_synced.active = false;
assert!(!needs_identity_reconcile(&inactive_synced, None, false));
assert!(needs_identity_reconcile(&complete, None, true));
let mut deleted = complete;
deleted.status = RepoStatus::Deleted;
assert!(!needs_identity_reconcile(&deleted, None, true));
deleted.pds = None;
deleted.signing_key = None;
assert!(!needs_identity_reconcile(
&deleted,
Some(&Handle::new_static("carol.test").into_diagnostic()?),
true,
));
Ok(())
}
#[test]
fn repo_info_renders_message_time_from_millis() -> miette::Result<()> {
let mut state = RepoState::backfilling();
state.last_updated_at = 1_783_785_600;
state.last_message_time = Some(1_783_785_600_123);
let info = repo_state_to_info(
Did::new_static("did:plc:testrepo").into_diagnostic()?,
state,
true,
);
assert_eq!(
info.last_updated_at.map(|t| t.timestamp()),
Some(1_783_785_600)
);
assert_eq!(
info.last_message_at
.map(|t| (t.timestamp(), t.timestamp_subsec_millis())),
Some((1_783_785_600, 123))
);
Ok(())
}
#[test]
fn repo_info_json_contract() {
let cid = crate::car::compute_cid(b"root");
let signing_key = DidKey(std::borrow::Cow::Borrowed(&[0xed, 1, 2, 3]));
let signing_key_encoded = signing_key.encode();
let info = RepoInfo {
did: Did::new_static("did:web:example.com").unwrap(),
status: RepoStatus::Throttled,
tracked: false,
rev: Some(Tid::new("3jzfcijpj2z2a").unwrap()),
data: Some(cid),
handle: Some(Handle::new_static("alice.test").unwrap()),
pds: Some(Url::parse("https://pds.example").unwrap()),
signing_key: Some(signing_key),
last_updated_at: DateTime::from_timestamp_secs(1_783_785_600),
last_message_at: DateTime::from_timestamp_millis(1_783_785_600_123),
#[cfg(feature = "indexer")]
pending: false,
#[cfg(feature = "indexer")]
resync: None,
};
let encoded = serde_json::to_value(info).unwrap();
assert_eq!(encoded["did"], "did:web:example.com");
assert_eq!(encoded["status"], "throttled");
assert_eq!(encoded["tracked"], false);
assert_eq!(encoded["rev"], "3jzfcijpj2z2a");
assert_eq!(encoded["data"], cid.to_string());
assert_eq!(encoded["handle"], "alice.test");
assert_eq!(encoded["pds"], "https://pds.example/");
assert_eq!(encoded["signing_key"], signing_key_encoded);
assert_eq!(encoded["last_updated_at"], "2026-07-11T16:00:00Z");
assert_eq!(encoded["last_message_at"], "2026-07-11T16:00:00.123Z");
}
#[cfg(feature = "indexer")]
#[test]
fn a_retry_time_out_of_range_is_an_error_rather_than_1970() {
use crate::types::{ResyncErrorKind, ResyncState};
let state = ResyncState::Error {
kind: ResyncErrorKind::Generic,
retry_count: 0,
next_retry: i64::MAX,
};
assert!(ResyncInfo::new(state, 0).is_err());
}
#[cfg(feature = "indexer")]
#[tokio::test]
async fn an_unreadable_resync_entry_does_not_stop_the_listing() -> miette::Result<()> {
use crate::types::{ResyncErrorKind, ResyncState};
let tmp = tempfile::tempdir().into_diagnostic()?;
let state = Arc::new(AppState::new(&crate::config::Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
})?);
let bad = Did::new_static("did:web:bad.example").into_diagnostic()?;
let garbled = Did::new_static("did:web:garbled.example").into_diagnostic()?;
let fine = Did::new_static("did:web:fine.example").into_diagnostic()?;
let mut batch = state.db.inner.batch();
for did in [&bad, &garbled, &fine] {
batch.insert(
&state.db.repos,
keys::repo_key(did),
db::ser_repo_state(&RepoState::backfilling())?,
);
}
let impossible = ResyncState::Error {
kind: ResyncErrorKind::Generic,
retry_count: 1,
next_retry: i64::MAX,
};
batch.insert(
&state.db.indexer.resync,
keys::repo_key(&bad),
rmp_serde::to_vec(&impossible).into_diagnostic()?,
);
batch.insert(
&state.db.indexer.resync,
keys::repo_key(&garbled),
b"not msgpack",
);
batch.commit().into_diagnostic()?;
#[derive(Clone, Default)]
struct Logs(Arc<std::sync::Mutex<Vec<u8>>>);
impl std::io::Write for Logs {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
self.0.lock().unwrap().extend_from_slice(buf);
Ok(buf.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
let logs = Logs::default();
let subscriber = tracing_subscriber::fmt()
.with_writer({
let logs = logs.clone();
move || logs.clone()
})
.with_ansi(false)
.finish();
let repos = ReposControl::new(state);
tracing::subscriber::with_default(subscriber, || {
for _ in 0..2 {
let listed = repos.iter(None).collect::<miette::Result<Vec<_>>>()?;
assert_eq!(listed.len(), 3);
assert!(listed.iter().all(|info| info.resync.is_none()));
}
miette::Result::<()>::Ok(())
})?;
// one for each bad entry, not one per listing
let logs = String::from_utf8(logs.0.lock().unwrap().clone()).into_diagnostic()?;
assert_eq!(
logs.matches("skipping an unreadable resync entry").count(),
2,
"{logs}"
);
Ok(())
}
#[cfg(feature = "indexer")]
#[tokio::test]
async fn repo_info_shows_why_a_repo_waits_for_resync() -> miette::Result<()> {
use crate::types::{ResyncErrorKind, ResyncState};
let tmp = tempfile::tempdir().into_diagnostic()?;
let state = Arc::new(AppState::new(&crate::config::Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
})?);
let pds = Url::parse("https://dead.example").into_diagnostic()?;
let stuck = Did::new_static("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa").into_diagnostic()?;
let gone = Did::new_static("did:plc:bbbbbbbbbbbbbbbbbbbbbbbb").into_diagnostic()?;
let queued = Did::new_static("did:plc:cccccccccccccccccccccccc").into_diagnostic()?;
let mut repo = RepoState::backfilling();
repo.pds = Some(pds.as_str().into());
let mut batch = state.db.inner.batch();
for did in [&gone, &queued] {
batch.insert(
&state.db.repos,
keys::repo_key(did),
db::ser_repo_state(&repo)?,
);
}
// never backfilled, so only the identity cache knows its pds
batch.insert(
&state.db.repos,
keys::repo_key(&stuck),
db::ser_repo_state(&RepoState::backfilling())?,
);
state
.resolver
.cache_doc(
stuck.clone(),
crate::resolver::MiniDoc {
pds: pds.clone(),
handle: None,
key: None,
},
)
.await;
batch.insert(
&state.db.indexer.resync,
keys::repo_key(&stuck),
rmp_serde::to_vec(&ResyncState::Error {
kind: ResyncErrorKind::Ratelimited,
retry_count: 2,
next_retry: 1_783_785_600,
})
.into_diagnostic()?,
);
batch.insert(
&state.db.indexer.resync,
keys::repo_key(&gone),
rmp_serde::to_vec(&ResyncState::Gone {
status: RepoStatus::Takendown,
})
.into_diagnostic()?,
);
batch.insert(
&state.db.repo_metadata,
keys::repo_metadata_key(&queued),
db::ser_repo_meta(&RepoMetadata::backfilling(7))?,
);
state
.db
.indexer
.stage_pending_insert(&mut batch, &queued, &keys::pending_key(7));
batch.commit().into_diagnostic()?;
state.throttler.get_handle(&pds).await.record_failure();
state.throttler.get_handle(&pds).await.record_failure();
let repos = ReposControl(state);
let info = repos.info(&stuck).await?.expect("repo exists");
assert_eq!(
serde_json::to_value(&info).into_diagnostic()?["resync"],
serde_json::json!({
"state": "error",
"kind": "ratelimited",
"retry_count": 2,
"next_retry": "2026-07-11T16:00:00Z",
"host_failures": 2,
})
);
let listed = repos
.iter(None)
.map(|info| info.and_then(|i| serde_json::to_value(i).into_diagnostic()))
.collect::<miette::Result<Vec<_>>>()?;
let resyncs: Vec<_> = listed.iter().map(|i| i.get("resync").cloned()).collect();
assert_eq!(
resyncs[0],
Some(serde_json::to_value(&info.resync).into_diagnostic()?)
);
assert_eq!(
resyncs[1],
Some(serde_json::json!({"state": "gone", "status": "takendown"}))
);
assert_eq!(
resyncs[2], None,
"a repo outside the resync queue has no resync"
);
let pending: Vec<_> = listed.iter().map(|i| i["pending"].clone()).collect();
assert_eq!(pending, [false, false, true]);
Ok(())
}
#[cfg(feature = "indexer")]
#[tokio::test]
async fn collections_decodes_big_endian_counts() {
let tmp = tempfile::tempdir().unwrap();
let state = Arc::new(
AppState::new(&crate::config::Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
})
.unwrap(),
);
let did = Did::new_static("did:web:example.com").unwrap();
state
.db
.counts
.insert(
keys::count_collection_key(&did, "app.bsky.feed.post"),
42_u64.to_be_bytes(),
)
.unwrap();
let counts = ReposControl(state).get(&did).collections().await.unwrap();
assert_eq!(counts[&Nsid::new_static("app.bsky.feed.post").unwrap()], 42);
}
}
pub struct ListedRecord {
pub rkey: Rkey,
pub cid: Cid,
pub value: Data,
}
pub struct RecordList {
pub records: Vec<ListedRecord>,
pub cursor: Option<Rkey>,
}
#[derive(Debug, thiserror::Error)]
pub enum MiniDocError {
#[error("repo is not synced yet")]
NotSynced,
#[error("repo not found")]
RepoNotFound,
#[error("could not resolve identity")]
CouldNotResolveIdentity,
#[error("{0}")]
Other(miette::Error),
}
/// a mini doc with a bi-directionally verified handle.
pub struct MiniDoc<'i> {
/// the did.
pub did: Did,
/// the handle. if verification fails or no handle is found,
/// this will be "handle.invalid".
pub handle: Handle,
/// the url of the PDS of this repo.
pub pds: Url,
/// the atproto signing key of this repo.
pub signing_key: DidKey<'i>,
}
struct MiniDocSnapshot {
state: RepoState<'static>,
pending: bool,
gone: bool,
}
struct MiniDocReconcile {
state: RepoState<'static>,
pending: bool,
accepted_doc: bool,
identity_changed: bool,
queued_backfill: bool,
}
#[cfg(feature = "indexer")]
enum GoneRecovery {
Queued,
Pending,
None,
}
/// runs a write to `did`'s repo wherever its writes run: on its indexer shard, so it's in order
/// with everything else written for it, or in place where there's no shard
async fn write_repo<T: Send + 'static>(
state: &Arc<AppState>,
did: &Did,
work: impl FnOnce(&crate::db::Db) -> Result<T> + Send + Sync + 'static,
) -> Result<T> {
#[cfg(feature = "indexer")]
return crate::ingest::indexer::repos::write_repo(state, did, work).await;
#[cfg(not(feature = "indexer"))]
{
let _ = did;
state.db.run(work).await
}
}
/// handle to access data related to this repository.
#[derive(Clone)]
pub struct RepoHandle {
state: Arc<AppState>,
pub did: Did,
}
fn needs_identity_reconcile(
state: &RepoState<'_>,
requested_handle: Option<&Handle>,
gone: bool,
) -> bool {
if state.status == RepoStatus::Deleted {
return false;
}
gone || requested_handle.is_some_and(|requested| state.handle.as_ref() != Some(requested))
|| state.pds.is_none()
|| state.signing_key.is_none()
}
fn read_mini_doc_snapshot(db: &Db, did: &Did) -> Result<Option<MiniDocSnapshot>> {
let did_key = keys::repo_key(did);
let Some(state_bytes) = db.repos.get(&did_key).into_diagnostic()? else {
return Ok(None);
};
let state = db::deser_repo_state(&state_bytes)?.into_static();
#[cfg(feature = "indexer")]
let pending = {
let metadata_key = keys::repo_metadata_key(did);
let metadata = db
.repo_metadata
.get(&metadata_key)
.into_diagnostic()?
.map(|bytes| db::deser_repo_meta(&bytes))
.transpose()?;
metadata
.map(|metadata| {
db.indexer
.is_pending(&keys::pending_key(metadata.index_id))
.into_diagnostic()
})
.transpose()?
.unwrap_or(false)
};
#[cfg(not(feature = "indexer"))]
let pending = false;
#[cfg(feature = "indexer")]
let gone = db
.indexer
.resync
.get(&did_key)
.into_diagnostic()?
.map(|bytes| rmp_serde::from_slice::<crate::types::ResyncState>(&bytes).into_diagnostic())
.transpose()?
.is_some_and(|resync| {
matches!(
resync,
crate::types::ResyncState::Gone { status }
if status != crate::types::RepoStatus::Deleted
)
});
#[cfg(not(feature = "indexer"))]
let gone = false;
Ok(Some(MiniDocSnapshot {
state,
pending,
gone,
}))
}
impl RepoHandle {
pub(crate) async fn state(&self) -> Result<Option<RepoState<'static>>> {
let did_key = keys::repo_key(&self.did);
self.state
.db
.run(move |db| {
let bytes = db.repos.get(&did_key).into_diagnostic()?;
bytes
.as_deref()
.map(db::deser_repo_state)
.transpose()
.map(|opt| opt.map(IntoStatic::into_static))
})
.await
}
/// fetch the current state of this repository.
/// returns `None` if hydrant has never seen this repository.
pub async fn info(&self) -> Result<Option<RepoInfo>> {
let did = self.did.clone().into_static();
let did_key = keys::repo_key(&did);
let state = self.state.clone();
self.state
.db
.run(move |db| {
let Some(state_bytes) = db.repos.get(&did_key).into_diagnostic()? else {
return Ok(None);
};
let repo_state = crate::db::deser_repo_state(&state_bytes)?;
repo_row(&state, did, repo_state).map(Some)
})
.await
}
/// returns the collections of this repository and the number of records it has in each.
pub async fn collections(&self) -> Result<HashMap<Nsid, u64>> {
let did = self.did.clone().into_static();
self.state
.db
.run(move |db| {
let prefix = keys::did_collection_prefix(&did);
let mut res = HashMap::new();
for item in db.counts.prefix(&prefix) {
if res.len() >= 1000 {
break;
}
let (k, v) = item.into_inner().into_diagnostic()?;
let col = k
.strip_prefix(prefix.as_slice())
.ok_or_else(|| miette::miette!("invalid collection count key: {k:?}"))
.and_then(|r| std::str::from_utf8(r).into_diagnostic())
.and_then(|n| Nsid::new(n).into_diagnostic())?
.into_static();
let count = u64::from_be_bytes(
v.as_ref()
.try_into()
.into_diagnostic()
.wrap_err("expected to be count (8 bytes)")?,
);
res.insert(col, count);
}
Ok(res)
})
.await
}
/// returns a bi-directionally validated mini doc.
pub async fn mini_doc(&self) -> Result<MiniDoc<'static>, MiniDocError> {
self.mini_doc_for(None).await
}
pub(crate) async fn mini_doc_for(
&self,
requested_handle: Option<&Handle>,
) -> Result<MiniDoc<'static>, MiniDocError> {
let requested_handle = requested_handle.cloned();
let snapshot = self
.mini_doc_snapshot()
.await
.map_err(MiniDocError::Other)?;
let Some(snapshot) = snapshot else {
return Err(MiniDocError::RepoNotFound);
};
if snapshot.pending {
return Err(MiniDocError::NotSynced);
}
let needs_reconcile =
needs_identity_reconcile(&snapshot.state, requested_handle.as_ref(), snapshot.gone);
if !needs_reconcile {
return self.render_mini_doc(snapshot.state).await;
}
let reconcile_handle = requested_handle
.as_ref()
.filter(|requested| snapshot.state.handle.as_ref() != Some(*requested))
.cloned();
let initial_identity = snapshot.state.identity();
if self
.state
.identity_reconcile_is_cooled_down(
&self.did,
&initial_identity,
reconcile_handle.as_ref(),
snapshot.gone,
)
.await
{
#[cfg(feature = "indexer")]
if snapshot.gone {
match self
.recover_gone_from_stored_state()
.await
.map_err(MiniDocError::Other)?
{
GoneRecovery::Queued => self.state.notify_backfill(),
GoneRecovery::Pending => return Err(MiniDocError::NotSynced),
GoneRecovery::None => {}
}
}
return self.render_mini_doc(snapshot.state).await;
}
// resolution happens off the owner shard. the state is reloaded and the
// trigger is checked again on the shard before applying this result, so a
// slower concurrent resolver cannot overwrite a repair that already won the race.
let fresh = match self.state.resolver.resolve_doc_fresh(&self.did).await {
Ok(fresh) => fresh,
Err(crate::resolver::ResolverError::MissingPds(_)) => {
self.state
.cool_down_identity_reconcile(
self.did.clone().into_static(),
initial_identity,
reconcile_handle,
snapshot.gone,
)
.await;
return Err(MiniDocError::CouldNotResolveIdentity);
}
Err(_) => return Err(MiniDocError::CouldNotResolveIdentity),
};
let fresh_for_db = fresh.clone();
let resolved_at = chrono::Utc::now().timestamp();
let did = self.did.clone().into_static();
let did_for_db = did.clone();
let attempted_handle = reconcile_handle;
let attempted_gone = snapshot.gone;
let initial_pending = snapshot.pending;
#[cfg(feature = "indexer")]
let app = self.state.clone();
let result = write_repo(&self.state, &did, move |db| {
let mut txn = crate::db::Txn::new(db);
let Some(MiniDocSnapshot {
mut state,
pending,
gone,
}) = read_mini_doc_snapshot(db, &did_for_db)?
else {
return Ok(None);
};
let raced = state.identity() != initial_identity;
let current_trigger = needs_identity_reconcile(&state, requested_handle.as_ref(), gone);
// a pending transition made while the network request was in
// flight belongs to another path, and a result applied to a
// backfill would race it.
let pending_race = pending && !initial_pending;
let accepted_doc = !raced && !pending_race && current_trigger;
let identity_changed = if accepted_doc {
state.update_from_resolve(fresh_for_db, resolved_at)
} else {
false
};
// a fetch that found the same identity still vouches for it from now on
if accepted_doc {
txn.put_repo(&did_for_db, &state)?;
}
// a DID document never revives lifecycle state. gone recovery
// only moves the exact durable Gone row into the normal
// pending queue, and the backfill's drain makes it Synced.
#[cfg(feature = "indexer")]
let queued_backfill = if current_trigger && gone && !pending {
queue_backfill(&app, &did_for_db, &mut txn, Some("resolveMiniDoc"))?
== Queued::Queued
} else {
false
};
#[cfg(not(feature = "indexer"))]
let queued_backfill = false;
if accepted_doc || queued_backfill {
txn.commit()?;
}
Ok(Some(MiniDocReconcile {
state,
pending,
accepted_doc,
identity_changed,
queued_backfill,
}))
})
.await
.map_err(MiniDocError::Other)?;
let Some(result) = result else {
return Err(MiniDocError::RepoNotFound);
};
if result.pending && !result.queued_backfill {
return Err(MiniDocError::NotSynced);
}
if result.accepted_doc {
self.state.resolver.cache_doc(did.clone(), fresh).await;
if !result.identity_changed {
self.state
.cool_down_identity_reconcile(
did.clone(),
result.state.identity(),
attempted_handle,
attempted_gone,
)
.await;
}
}
#[cfg(feature = "indexer_stream")]
if result.identity_changed {
crate::ops::announce_identity(&self.state.db, &did, result.state.handle.clone()).await;
}
#[cfg(feature = "indexer")]
if result.queued_backfill {
self.state.notify_backfill();
}
self.render_mini_doc(result.state).await
}
#[cfg(feature = "indexer")]
async fn recover_gone_from_stored_state(&self) -> Result<GoneRecovery> {
let did = self.did.clone().into_static();
let owner = did.clone();
let app = self.state.clone();
write_repo(&self.state, &owner, move |db| {
let mut txn = crate::db::Txn::new(db);
if txn.is_excluded(&did)? {
return Ok(GoneRecovery::None);
}
let Some(snapshot) = read_mini_doc_snapshot(db, &did)? else {
return Ok(GoneRecovery::None);
};
if snapshot.state.status == RepoStatus::Deleted {
return Ok(GoneRecovery::None);
}
if snapshot.pending {
return Ok(GoneRecovery::Pending);
}
if !snapshot.gone {
return Ok(GoneRecovery::None);
}
match queue_backfill(&app, &did, &mut txn, Some("resolveMiniDoc"))? {
Queued::Queued => {
txn.commit()?;
Ok(GoneRecovery::Queued)
}
Queued::Deferred => {
txn.commit()?;
Ok(GoneRecovery::None)
}
Queued::Skipped => Ok(GoneRecovery::None),
}
})
.await
}
async fn mini_doc_snapshot(&self) -> Result<Option<MiniDocSnapshot>> {
let did = self.did.clone().into_static();
self.state
.db
.run(move |db| read_mini_doc_snapshot(db, &did))
.await
}
async fn render_mini_doc(
&self,
info: RepoState<'static>,
) -> Result<MiniDoc<'static>, MiniDocError> {
let pds = info.pds.and_then(|pds| Url::parse(pds.as_ref()).ok());
self.verified_mini_doc(pds, info.signing_key, info.handle)
.await
}
/// a mini doc straight from the identity resolver, for repos that aren't indexed (yet).
/// reads the resolver cache first and writes nothing to the repo db.
#[cfg(feature = "indexer")]
pub(crate) async fn mini_doc_from_resolver(&self) -> Result<MiniDoc<'static>, MiniDocError> {
let doc = self
.state
.resolver
.resolve_doc(&self.did)
.await
.map_err(|_| MiniDocError::CouldNotResolveIdentity)?;
self.verified_mini_doc(Some(doc.pds), doc.key.map(DidKey::from), doc.handle)
.await
}
async fn verified_mini_doc(
&self,
pds: Option<Url>,
signing_key: Option<DidKey<'static>>,
handle: Option<Handle>,
) -> Result<MiniDoc<'static>, MiniDocError> {
let pds = pds.ok_or_else(|| MiniDocError::CouldNotResolveIdentity)?;
let signing_key = signing_key
.ok_or_else(|| MiniDocError::CouldNotResolveIdentity)?
.into_static();
let handle = if let Some(h) = handle {
let is_valid = self
.state
.resolver
.verify_handle(&self.did, &h)
.await
.into_diagnostic()
.map_err(MiniDocError::Other)?;
is_valid.then_some(h).unwrap_or_else(invalid_handle)
} else {
invalid_handle()
};
Ok(MiniDoc {
did: self.did.clone().into_static(),
handle,
pds,
signing_key,
})
}
}