Something went wrong. Try again.
very fast at protocol indexer with flexible filtering, xrpc queries, cursor-backed event stream, and more, built on fjall
Something went wrong. Try again.
Rust
at fix-counts
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758use std::fmt::{Debug, Display};
use bytes::Bytes;use jacquard_common::types::cid::IpldCid;use jacquard_common::types::nsid::Nsid;use jacquard_common::types::string::{Did, Rkey};use jacquard_common::types::tid::Tid;use jacquard_common::{CowStr, IntoStatic, types::string::Handle};use jacquard_repo::commit::Commit as AtpCommit;use serde::{Deserialize, Serialize, Serializer};use serde_json::Value;use smol_str::{SmolStr, ToSmolStr};
use crate::db::types::{DbTid, DidKey};
#[cfg(any(feature = "indexer_stream", feature = "jetstream"))]use crate::db::types::TrimmedDid;#[cfg(feature = "indexer_stream")]use crate::db::types::{DbAction, DbRkey};#[cfg(feature = "jetstream")]use crate::ingest::stream::Datetime;use crate::resolver::MiniDoc;
// on-disk schema snapshots for stored types.//// rules:// - versioned modules (`v2`, `v4`, `v7`, ...) are wire-format snapshots, not// convenience namespaces// - once a version is referenced by a shipped migration, do not change its// serialized shape again// - when a stored type changes, add a new versioned module and export the// newest version for live code via `pub(crate) use vN::*`// - migrations should explicitly read the previous version and write the new one// - for example, if `RepoState` changes shape again, add `v8::RepoState`,// keep `v7::RepoState` frozen, export `v8`, and migrate `v7 -> v8`pub(crate) mod v2 { use super::*;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub enum RepoStatus { Backfilling, Synced, Error(SmolStr), Deactivated, Takendown, Suspended, }
#[derive(Debug, Clone, Serialize, Deserialize)] pub(crate) struct Commit { pub version: i64, pub rev: DbTid, pub data: IpldCid, pub prev: Option<IpldCid>, #[serde(with = "jacquard_common::serde_bytes_helper")] pub sig: Bytes, }
#[derive(Debug, Clone, Serialize, Deserialize)] #[serde(bound(deserialize = "'i: 'de"))] pub(crate) struct RepoState<'i> { pub status: RepoStatus, pub root: Option<Commit>, pub last_message_time: Option<i64>, pub last_updated_at: i64, pub tracked: bool, pub index_id: u64, #[serde(borrow)] pub signing_key: Option<DidKey<'i>>, #[serde(borrow)] pub pds: Option<CowStr<'i>>, #[serde(borrow)] pub handle: Option<Handle<'i>>, }}
pub(crate) mod v4 { use super::*; pub(crate) use v2::Commit;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub enum RepoStatus { /// repo is synced to latest commit from what we know of Synced, /// some unclassified fatal error Error(SmolStr), /// user has temporarily paused their overall account. content should /// not be displayed or redistributed, but does not need to be deleted /// from infrastructure. implied time-limited. also the initial state /// for an account after migrating to another pds instance. Deactivated, /// host or service has takendown the account. implied permanent or /// long-term, though may be reverted. Takendown, /// host or service has temporarily paused the account. implied /// time-limited. Suspended, /// user or host has deleted the account, and content should be removed /// from the network. implied permanent or long-term, though may be /// reverted (deleted accounts may reactivate on the same or another /// host). /// /// account is deleted; kept as a tombstone so stale commits arriving from the upstream /// backfill window are not forwarded. active=false per spec. Deleted, /// host detected a repo sync problem. active may be true or false per spec; /// the `active` field on `RepoState` is authoritative. Desynchronized, /// resource rate-limit exceeded. active may be true or false per spec; /// the `active` field on `RepoState` is authoritative. Throttled, }
#[derive(Debug, Clone, Serialize, Deserialize)] #[serde(bound(deserialize = "'i: 'de"))] pub(crate) struct RepoState<'i> { /// whether the upstream considers this account active. /// services should use the `active` flag to control overall account visibility pub active: bool, pub status: RepoStatus, pub root: Option<Commit>, pub last_message_time: Option<i64>, /// this is when we *ingested* any last updates pub last_updated_at: i64, // unix timestamp #[serde(borrow)] pub signing_key: Option<DidKey<'i>>, #[serde(borrow)] pub pds: Option<CowStr<'i>>, #[serde(borrow)] pub handle: Option<Handle<'i>>, }
#[derive(Debug, Clone, Serialize, Deserialize)] pub(crate) struct RepoMetadata { /// whether we are ingesting events for this repo pub tracked: bool, /// index id in pending keyspace (if backfilling) pub index_id: u64, }}
pub(crate) mod v7 { use super::*; pub(crate) use v4::{Commit, RepoMetadata, RepoStatus};
#[derive(Debug, Clone, Serialize, Deserialize)] #[serde(bound(deserialize = "'i: 'de"))] pub(crate) struct RepoState<'i> { /// whether the upstream considers this account active. /// services should use the `active` flag to control overall account visibility pub active: bool, pub status: RepoStatus, pub root: Option<Commit>, /// ms since epoch of the last firehose message we processed for this repo. /// used to deduplicate identity / account events that can arrive from multiple relays at /// different wall-clock times but represent the same underlying PDS event. pub last_message_time: Option<i64>, /// high-water mark for identity events only. /// /// sharing a single clock with commit/sync/account events can suppress valid identity /// updates when those event classes arrive out of order but with overlapping timestamps. pub last_identity_time: Option<i64>, /// high-water mark for account events only. /// /// sharing a single clock with commit/sync/identity events can suppress valid account /// status transitions when those event classes arrive out of order but with overlapping /// timestamps. pub last_account_time: Option<i64>, /// this is when we *ingested* any last updates pub last_updated_at: i64, // unix timestamp #[serde(borrow)] pub signing_key: Option<DidKey<'i>>, #[serde(borrow)] pub pds: Option<CowStr<'i>>, #[serde(borrow)] pub handle: Option<Handle<'i>>, }}
pub(crate) use v7::*;
impl<'c> From<AtpCommit<'c>> for Commit { fn from(value: AtpCommit<'c>) -> Self { Self { data: value.data, prev: value.prev, rev: DbTid::from(&value.rev), sig: value.sig, version: value.version, } }}
impl Commit { pub(crate) fn into_atp_commit<'i>(self, did: Did<'i>) -> Option<AtpCommit<'i>> { // version < 0 is a sentinel used in v2 migration for repos with no commit data if self.version < 0 { return None; } Some(AtpCommit { did, rev: self.rev.to_tid(), data: self.data, prev: self.prev, sig: self.sig, version: self.version, }) }}
impl Display for RepoStatus { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { RepoStatus::Synced => write!(f, "synced"), RepoStatus::Error(e) => write!(f, "error({e})"), RepoStatus::Deactivated => write!(f, "deactivated"), RepoStatus::Takendown => write!(f, "takendown"), RepoStatus::Suspended => write!(f, "suspended"), RepoStatus::Deleted => write!(f, "deleted"), RepoStatus::Desynchronized => write!(f, "desynchronized"), RepoStatus::Throttled => write!(f, "throttled"), } }}
impl RepoMetadata {}
#[cfg(feature = "indexer")]mod indexer { use super::*;
impl RepoMetadata { pub fn backfilling(index_id: u64) -> Self { Self { index_id, tracked: true, } } }
impl ResyncState { pub fn next_backoff(retry_count: u32) -> i64 { // exponential backoff: 1m, 2m, 4m, 8m... up to 1h let base = 60; let cap = 3600; let mult = 2u64.pow(retry_count.min(10)) as i64; let delay = (base * mult).min(cap);
// add +/- 10% jitter let jitter = (rand::random::<f64>() * 0.2 - 0.1) * delay as f64; let delay = (delay as f64 + jitter) as i64;
chrono::Utc::now().timestamp() + delay } }
#[cfg(feature = "indexer_stream")] #[derive(Clone, Debug)] pub(crate) enum BroadcastEvent { Persisted(#[allow(dead_code)] u64), /// a durable record event with optional inline block bytes for live tailing. /// /// used to avoid re-reading `events`/`blocks` from the database when tailing. LiveRecord(std::sync::Arc<super::LiveRecordEvent>), Ephemeral(Box<MarshallableEvt<'static>>), }
#[derive(Debug, PartialEq, Eq, Clone, Copy)] pub(crate) enum GaugeState { Synced, Pending, Resync(Option<ResyncErrorKind>), }
impl GaugeState { pub fn is_resync(&self) -> bool { matches!(self, GaugeState::Resync(_)) } }}
#[cfg(feature = "indexer")]pub(crate) use indexer::*;
impl<'i> RepoState<'i> { pub fn backfilling() -> Self { Self { active: true, status: RepoStatus::Desynchronized, root: None, last_updated_at: chrono::Utc::now().timestamp(), handle: None, pds: None, signing_key: None, last_message_time: None, last_identity_time: None, last_account_time: None, } }
pub fn synced() -> Self { Self { status: RepoStatus::Synced, ..Self::backfilling() } }
// advances the high-water mark to event_ms if it's newer than what we've seen pub fn advance_message_time(&mut self, event_ms: i64) { self.last_message_time = Some(event_ms.max(self.last_message_time.unwrap_or(0))); }
pub fn should_process_identity_time(&self, event_ms: i64) -> bool { self.last_identity_time.is_none_or(|t| event_ms > t) }
pub fn advance_identity_time(&mut self, event_ms: i64) { self.last_identity_time = Some(event_ms.max(self.last_identity_time.unwrap_or(0))); self.advance_message_time(event_ms); }
pub fn should_process_account_time(&self, event_ms: i64) -> bool { self.last_account_time.is_none_or(|t| event_ms > t) }
pub fn advance_account_time(&mut self, event_ms: i64) { self.last_account_time = Some(event_ms.max(self.last_account_time.unwrap_or(0))); self.advance_message_time(event_ms); }
// updates last_updated_at to now pub fn touch(&mut self) { self.last_updated_at = chrono::Utc::now().timestamp(); }
pub fn update_from_doc(&mut self, doc: MiniDoc) -> bool { let new_signing_key = doc.key.map(From::from); let changed = self.pds.as_deref() != Some(doc.pds.as_str()) || self.handle != doc.handle || self.signing_key != new_signing_key; self.pds = Some(CowStr::Owned(doc.pds.to_smolstr())); self.handle = doc.handle; self.signing_key = new_signing_key; changed }}
impl<'i> IntoStatic for RepoState<'i> { type Output = RepoState<'static>;
fn into_static(self) -> Self::Output { RepoState { active: self.active, status: self.status, root: self.root, last_updated_at: self.last_updated_at, handle: self.handle.map(IntoStatic::into_static), pds: self.pds.map(IntoStatic::into_static), signing_key: self.signing_key.map(IntoStatic::into_static), last_message_time: self.last_message_time, last_identity_time: self.last_identity_time, last_account_time: self.last_account_time, } }}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]pub(crate) enum ResyncErrorKind { Ratelimited, Transport, Generic,}
#[derive(Debug, Clone, Serialize, Deserialize)]pub(crate) enum ResyncState { Error { kind: ResyncErrorKind, retry_count: u32, next_retry: i64, // unix timestamp }, Gone { status: RepoStatus, // deactivated, takendown, suspended },}
#[derive(Debug, Serialize, Clone)]pub enum EventType { Record, Identity, Account,}
impl AsRef<str> for EventType { fn as_ref(&self) -> &str { match self { Self::Record => "record", Self::Identity => "identity", Self::Account => "account", } }}
fn event_type_ser_str<S: Serializer>(v: &EventType, s: S) -> Result<S::Ok, S::Error> { s.serialize_str(v.as_ref())}
#[derive(Debug, Serialize, Clone)]pub struct MarshallableEvt<'i> { pub id: u64, #[serde(rename = "type")] #[serde(serialize_with = "event_type_ser_str")] pub kind: EventType, #[serde(borrow)] #[serde(skip_serializing_if = "Option::is_none")] pub record: Option<RecordEvt<'i>>, #[serde(borrow)] #[serde(skip_serializing_if = "Option::is_none")] pub identity: Option<IdentityEvt<'i>>, #[serde(borrow)] #[serde(skip_serializing_if = "Option::is_none")] pub account: Option<AccountEvt<'i>>,}
#[derive(Debug, Serialize, Clone)]pub struct RecordEvt<'i> { pub live: bool, #[serde(borrow)] pub did: Did<'i>, pub rev: Tid, pub collection: Nsid<'i>, pub rkey: Rkey<'i>, pub action: CowStr<'i>, #[serde(skip_serializing_if = "Option::is_none")] pub record: Option<Value>, #[serde(skip_serializing_if = "Option::is_none")] #[serde(serialize_with = "crate::util::opt_cid_serialize_str")] pub cid: Option<IpldCid>,}
#[derive(Debug, Serialize, Clone)]pub struct IdentityEvt<'i> { #[serde(borrow)] pub did: Did<'i>, #[serde(skip_serializing_if = "Option::is_none")] pub handle: Option<Handle<'i>>,}
#[derive(Debug, Serialize, Clone)]pub struct AccountEvt<'i> { #[serde(borrow)] pub did: Did<'i>, pub active: bool, #[serde(skip_serializing_if = "Option::is_none")] pub status: Option<CowStr<'i>>,}
#[cfg(feature = "indexer_stream")]#[derive(Serialize, Deserialize, Clone, Default)]pub(crate) enum StoredData { #[default] Nothing, Ptr(IpldCid), #[serde(with = "jacquard_common::serde_bytes_helper")] Block(Bytes),}
#[cfg(feature = "indexer_stream")]impl StoredData { pub fn is_nothing(&self) -> bool { matches!(self, StoredData::Nothing) }}
#[cfg(feature = "indexer_stream")]impl Debug for StoredData { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { Self::Nothing => f.write_str("nothing"), Self::Block(_) => f.write_str("<block>"), Self::Ptr(cid) => write!(f, "{cid}"), } }}
#[cfg(feature = "indexer_stream")]#[derive(Debug, Serialize, Deserialize, Clone)]#[serde(bound(deserialize = "'i: 'de"))]pub(crate) struct StoredEvent<'i> { #[serde(default)] pub live: bool, #[serde(borrow)] pub did: TrimmedDid<'i>, pub rev: DbTid, #[serde(borrow)] pub collection: CowStr<'i>, pub rkey: DbRkey, pub action: DbAction, #[serde(default)] #[serde(skip_serializing_if = "StoredData::is_nothing")] pub data: StoredData,}
/// a durable record event that is also emitted on the in-memory stream after commit.////// `inline_block` is only used for live tailing to avoid loading the record block from `blocks`./// cursor replay continues to read from the database.#[cfg(feature = "indexer_stream")]#[derive(Debug, Clone)]pub(crate) struct LiveRecordEvent { pub id: u64, pub stored: StoredEvent<'static>, pub inline_block: Option<Bytes>,}
#[inline]#[allow(dead_code)]pub(crate) fn default_true() -> bool { true}
#[cfg(feature = "jetstream")]#[derive(Debug, Serialize, Deserialize, Clone)]#[serde(bound(deserialize = "'i: 'de"))]pub(crate) enum StoredJetstreamEvent<'i> { #[cfg(feature = "indexer_stream")] Commit { #[serde(borrow)] did: TrimmedDid<'i>, #[serde(borrow)] collection: CowStr<'i>, event_id: u64, #[serde(default = "crate::types::default_true")] live: bool, }, #[cfg(feature = "relay")] RelayCommit { #[serde(borrow)] did: TrimmedDid<'i>, #[serde(borrow)] collection: CowStr<'i>, relay_seq: u64, op_index: u32, }, #[cfg(feature = "relay")] RelayAccount { #[serde(borrow)] did: TrimmedDid<'i>, relay_seq: u64, }, #[cfg(feature = "relay")] RelayIdentity { #[serde(borrow)] did: TrimmedDid<'i>, relay_seq: u64, }, Account { #[serde(borrow)] did: TrimmedDid<'i>, active: bool, #[serde(borrow)] status: Option<CowStr<'i>>, seq: i64, time: Datetime, }, Identity { #[serde(borrow)] did: TrimmedDid<'i>, #[serde(borrow)] handle: Option<Handle<'i>>, seq: i64, time: Datetime, },}
#[cfg(feature = "jetstream")]#[derive(Clone, Debug)]pub(crate) struct JetstreamBroadcast { pub id: u64, pub time_us: i64, pub event: StoredJetstreamEvent<'static>, /// pre-serialized json bytes for live tailing. /// replay events read from the database instead. pub ephemeral: Option<bytes::Bytes>,}
#[cfg(feature = "jetstream")]impl<'i> StoredJetstreamEvent<'i> { pub(crate) fn is_live(&self) -> bool { match self { #[cfg(feature = "indexer_stream")] Self::Commit { live, .. } => *live, _ => true, } }
pub(crate) fn did(&self) -> &TrimmedDid<'i> { match self { #[cfg(feature = "indexer_stream")] Self::Commit { did, .. } => did, #[cfg(feature = "relay")] Self::RelayCommit { did, .. } | Self::RelayAccount { did, .. } | Self::RelayIdentity { did, .. } => did, Self::Account { did, .. } | Self::Identity { did, .. } => did, } }
pub(crate) fn collection(&self) -> Option<&str> { match self { #[cfg(feature = "indexer_stream")] Self::Commit { collection, .. } => Some(collection.as_str()), #[cfg(feature = "relay")] Self::RelayCommit { collection, .. } => Some(collection.as_str()), #[cfg(feature = "relay")] Self::RelayAccount { .. } | Self::RelayIdentity { .. } => None, Self::Account { .. } | Self::Identity { .. } => None, } }
pub(crate) fn into_static(self) -> StoredJetstreamEvent<'static> { match self { #[cfg(feature = "indexer_stream")] Self::Commit { did, collection, event_id, live, } => StoredJetstreamEvent::Commit { did: did.into_static(), collection: collection.into_static(), event_id, live, }, #[cfg(feature = "relay")] Self::RelayCommit { did, collection, relay_seq, op_index, } => StoredJetstreamEvent::RelayCommit { did: did.into_static(), collection: collection.into_static(), relay_seq, op_index, }, #[cfg(feature = "relay")] Self::RelayAccount { did, relay_seq } => StoredJetstreamEvent::RelayAccount { did: did.into_static(), relay_seq, }, #[cfg(feature = "relay")] Self::RelayIdentity { did, relay_seq } => StoredJetstreamEvent::RelayIdentity { did: did.into_static(), relay_seq, }, Self::Account { did, active, status, seq, time, } => StoredJetstreamEvent::Account { did: did.into_static(), active, status: status.map(IntoStatic::into_static), seq, time, }, Self::Identity { did, handle, seq, time, } => StoredJetstreamEvent::Identity { did: did.into_static(), handle: handle.map(IntoStatic::into_static), seq, time, }, } }}
#[cfg(feature = "relay")]#[derive(Clone)]pub(crate) enum RelayBroadcast { Persisted(#[allow(dead_code)] u64), #[allow(dead_code)] Ephemeral(u64, bytes::Bytes),}
#[cfg(test)]mod tests { use super::*; use miette::IntoDiagnostic;
#[test] fn identity_dedupe_does_not_depend_on_commit_clock() { let mut state = RepoState::backfilling();
state.advance_message_time(2_000);
assert!(state.should_process_identity_time(1_500)); state.advance_identity_time(1_500); assert_eq!(state.last_message_time, Some(2_000)); assert_eq!(state.last_identity_time, Some(1_500));
assert!(!state.should_process_identity_time(1_500)); assert!(!state.should_process_identity_time(1_499)); assert!(state.should_process_identity_time(1_501)); }
#[test] fn account_dedupe_does_not_depend_on_commit_clock() { let mut state = RepoState::backfilling();
state.advance_message_time(3_000);
assert!(state.should_process_account_time(2_500)); state.advance_account_time(2_500); assert_eq!(state.last_message_time, Some(3_000)); assert_eq!(state.last_account_time, Some(2_500));
assert!(!state.should_process_account_time(2_500)); assert!(!state.should_process_account_time(2_499)); assert!(state.should_process_account_time(2_501)); }
#[test] fn synced_state_is_active_without_backfilling_status() { let state = RepoState::synced(); assert!(state.active); assert_eq!(state.status, RepoStatus::Synced); assert!(state.root.is_none()); }
#[test] fn into_static_preserves_per_event_clocks() -> miette::Result<()> { let mut state = RepoState::backfilling(); state.last_message_time = Some(10); state.last_identity_time = Some(20); state.last_account_time = Some(30); state.handle = Some(Handle::new("alice.test").into_diagnostic()?); state.pds = Some(CowStr::Borrowed("https://pds.example")); state.signing_key = Some(DidKey::from_did_key( "did:key:zQ3shokFTS3brHcDQrn82RUDfCZESWL1ZdCEJwekUDPQiYBme", )?);
let static_state = state.into_static();
assert_eq!(static_state.last_message_time, Some(10)); assert_eq!(static_state.last_identity_time, Some(20)); assert_eq!(static_state.last_account_time, Some(30));
Ok(()) }}