use std::fmt::Display;

use jacquard_common::types::string::{Did, Handle};
use jacquard_common::{CowStr, IntoStatic};

pub(crate) use crate::mst::AtpCommit;
#[cfg(feature = "indexer")]
use serde::{Deserialize, Serialize};
use smol_str::ToSmolStr;

use crate::db::types::{DbTid, DidKey};
use crate::resolver::MiniDoc;

#[cfg(any(feature = "indexer_stream", feature = "relay", feature = "jetstream"))]
pub(crate) mod event;
#[cfg(test)]
pub(crate) mod v14;
pub(crate) mod v15;
pub(crate) mod v16;
pub(crate) mod v2;
pub(crate) mod v4;
pub(crate) mod v7;

#[cfg(any(feature = "indexer_stream", feature = "relay", feature = "jetstream"))]
pub(crate) use event::*;
pub(crate) use v15::RepoState;
pub(crate) use v16::*;

/// which record-scoped body copies an operator redaction removes.
#[cfg(feature = "indexer")]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum DeleteBodyTarget {
    Head,
    History,
    All,
}

#[cfg(feature = "indexer")]
impl DeleteBodyTarget {
    pub(crate) fn includes_head(self) -> bool {
        matches!(self, Self::Head | Self::All)
    }

    pub(crate) fn includes_history(self) -> bool {
        matches!(self, Self::History | Self::All)
    }
}

/// a bulk job the owner shard runs in steps, kept on disk while it runs so a restart resumes
/// it. each kind says what the job was started for, so a restart can tell a job that's still
/// due from one that something else made pointless meanwhile
#[cfg(feature = "indexer")]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) enum BulkJob {
    V1(BulkJobV1),
}

#[cfg(feature = "indexer")]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) enum BulkJobV1 {
    /// a backfill writing its records in steps, for the queue entry under this key
    Apply {
        #[serde(with = "serde_bytes")]
        pending_key: Vec<u8>,
    },
    /// an operator erase. an exclude it asked for went into the filter in the txn that
    /// wrote this, so a resumed erase doesn't need to know
    Erase,
    /// a deleted account's records moving to history, as of `death_rev`
    Wipe { death_rev: DbTid },
}

/// idempotent result of deleting indexed body bytes while preserving logical
/// record identity and repository topology.
#[cfg(feature = "indexer")]
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct DeleteBodiesReport {
    pub heads_redacted: u64,
    pub history_redacted: u64,
    pub event_bodies_redacted: u64,
    pub body_bytes_redacted: u64,
    pub buffered_commits_deleted: u64,
    pub buffered_commit_bytes_deleted: u64,
}

impl From<AtpCommit> for Commit {
    fn from(value: AtpCommit) -> 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(self, did: Did) -> Option<AtpCommit> {
        // 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"),
        }
    }
}

#[cfg(feature = "indexer")]
mod indexer {
    use super::*;

    impl RepoMetadata {
        pub fn backfilling(index_id: u64) -> Self {
            Self {
                index_id,
                tracked: true,
                retry_count: 0,
                relay_inactive: None,
                backfilled_rev: None,
                refetch: None,
                refetch_deferred: false,
            }
        }

        /// whether the last car a backfill wrote already has the commit at `rev`
        pub fn car_covers(&self, rev: &jacquard_common::types::tid::Tid) -> bool {
            self.backfilled_rev
                .is_some_and(|car| DbTid::from(rev) <= car)
        }
    }

    impl Allowance {
        /// `Err` is the unix time the window ends, not how long to wait
        pub fn check(
            allowance: &Option<Self>,
            now: i64,
            limit: u32,
            window: i64,
        ) -> Result<(), i64> {
            match allowance.filter(|allowance| now < allowance.since.saturating_add(window)) {
                Some(current) if current.used >= limit => Err(current.since.saturating_add(window)),
                _ => Ok(()),
            }
        }

        /// a charge past the window's end starts the next window
        pub fn charge(allowance: &mut Option<Self>, now: i64, amount: u32, window: i64) {
            let current = allowance
                .filter(|allowance| now < allowance.since.saturating_add(window))
                .unwrap_or(Self {
                    since: now,
                    used: 0,
                });
            *allowance = Some(Self {
                used: current.used.saturating_add(amount),
                ..current
            });
        }

        pub fn take(
            allowance: &mut Option<Self>,
            now: i64,
            limit: u32,
            window: i64,
        ) -> Result<(), i64> {
            Self::check(allowance, now, limit, window)?;
            Self::charge(allowance, now, 1, window);
            Ok(())
        }
    }

    impl ResyncState {
        pub fn next_backoff(retry_count: u32) -> i64 {
            // the caller counts the failure before asking, so the first retry waits 2m, then 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 the durable event and record body 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::*;

/// the identity a DID doc owns. it is the only part of a repo state that
/// resolving a doc rewrites, so consumers that cache it have to be told when it
/// moves instead of discovering it later.
#[derive(Debug, Clone, PartialEq)]
pub(crate) struct DocIdentity<'i> {
    handle: Option<Handle>,
    pds: Option<CowStr<'i>>,
    signing_key: Option<DidKey<'i>>,
}

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,
            identity_resolved_at: None,
        }
    }

    #[cfg(any(test, feature = "relay"))]
    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();
    }

    /// the doc-owned identity, the whole of what a resolve can rewrite.
    pub(crate) fn identity(&self) -> DocIdentity<'i> {
        DocIdentity {
            handle: self.handle.clone(),
            pds: self.pds.clone(),
            signing_key: self.signing_key.clone(),
        }
    }

    /// whether `other` has the same doc-owned identity, the same as comparing both
    /// [`Self::identity`]s but without copying them
    #[cfg(any(test, feature = "indexer"))]
    pub(crate) fn same_identity(&self, other: &Self) -> bool {
        (&self.handle, &self.pds, &self.signing_key)
            == (&other.handle, &other.pds, &other.signing_key)
    }

    /// the identity moved, so anyone holding a cached copy must re-resolve it.
    #[cfg(any(test, feature = "indexer_stream"))]
    pub(crate) fn identity_moved_from(&self, previous: &Self) -> bool {
        !self.same_identity(previous)
    }

    /// the account lifecycle moved. a first sighting of a pds counts as a move:
    /// nothing downstream keyed off an identity it never saw.
    #[cfg(any(test, feature = "indexer_stream"))]
    pub(crate) fn account_moved_from(&self, previous: &Self) -> bool {
        self.active != previous.active || self.status != previous.status || previous.pds.is_none()
    }

    pub fn update_from_doc(&mut self, doc: MiniDoc) -> bool {
        let previous = self.identity();
        self.pds = Some(CowStr::Owned(doc.pds.to_smolstr()));
        self.handle = doc.handle;
        self.signing_key = doc.key.map(From::from);
        self.identity() != previous
    }

    /// [`Self::update_from_doc`] with a doc fetched fresh at `at`, unix secs
    pub fn update_from_resolve(&mut self, doc: MiniDoc, at: i64) -> bool {
        self.identity_resolved_at = Some(at);
        self.update_from_doc(doc)
    }

    #[cfg(feature = "indexer")]
    pub(crate) fn set_identity(&mut self, identity: DocIdentity<'i>) {
        let DocIdentity {
            handle,
            pds,
            signing_key,
        } = identity;
        self.handle = handle;
        self.pds = pds;
        self.signing_key = signing_key;
    }

    /// whether going from `before` to `self` saw an account event or changed the account.
    /// the time counts on its own because a newer event can repeat the old status. the
    /// identity has no such check, because the relay moves its clock even for an event it got
    /// no fresh doc for, so there only a changed value counts
    #[cfg(feature = "indexer")]
    pub(crate) fn account_written_since(&self, before: &Self) -> bool {
        self.last_account_time != before.last_account_time
            || (self.active, &self.status) != (before.active, &before.status)
    }

    /// whether going from `before` to `self` resolved the doc or otherwise wrote the identity,
    /// which a refresh that finds the same doc does too, by moving only the resolve time
    #[cfg(feature = "indexer")]
    fn identity_written_since(&self, before: &Self) -> bool {
        !self.same_identity(before) || self.identity_resolved_at != before.identity_resolved_at
    }

    /// puts what one writer changed, going from `before` to `self`, onto `current`, the
    /// row as it is now, so whatever another writer changed in between survives. if both
    /// changed the same part `self` wins, but a root never goes back to an older rev, the
    /// event times only move forward, and the identity (handle, pds, key and when they were
    /// resolved, as one) goes to whichever side resolved it last
    #[cfg(feature = "indexer")]
    pub(crate) fn rebased(self, before: &Self, current: Self) -> Self {
        let account_changed = self.account_written_since(before);
        let identity_changed = self.identity_written_since(before)
            && (!current.identity_written_since(before)
                || self.identity_resolved_at >= current.identity_resolved_at);
        let root_changed = self.root != before.root;

        let (active, status) = if account_changed {
            (self.active, self.status)
        } else {
            (current.active, current.status)
        };
        let (handle, pds, signing_key, identity_resolved_at) = if identity_changed {
            (
                self.handle,
                self.pds,
                self.signing_key,
                self.identity_resolved_at,
            )
        } else {
            (
                current.handle,
                current.pds,
                current.signing_key,
                current.identity_resolved_at,
            )
        };
        let root = match (self.root, current.root) {
            (_, theirs) if !root_changed => theirs,
            (Some(mine), Some(theirs)) if theirs.rev > mine.rev => Some(theirs),
            (mine, _) => mine,
        };
        Self {
            active,
            status,
            root,
            last_message_time: self.last_message_time.max(current.last_message_time),
            last_identity_time: self.last_identity_time.max(current.last_identity_time),
            last_account_time: self.last_account_time.max(current.last_account_time),
            last_updated_at: self.last_updated_at.max(current.last_updated_at),
            signing_key,
            pds,
            handle,
            identity_resolved_at,
        }
    }

    /// when the identity was last resolved. a row from before v15 has no time, and reads as
    /// resolved at a point in the ttl before `epoch` picked by its did, so the rows that were
    /// there at the upgrade come due for a refresh spread over one ttl instead of all at once
    pub(crate) fn identity_resolved_or_spread(&self, did: &Did, epoch: i64, ttl: i64) -> i64 {
        self.identity_resolved_at.unwrap_or_else(|| {
            let spread = (crate::util::hash(&did.as_str()) % ttl.max(1) as u64) as i64;
            epoch - spread
        })
    }
}

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,
            identity_resolved_at: self.identity_resolved_at,
        }
    }
}

#[cfg(feature = "indexer")]
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
pub(crate) enum ResyncErrorKind {
    Ratelimited,
    Transport,
    Generic,
}

#[cfg(feature = "indexer")]
#[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
    },
}

#[cfg(test)]
mod tests {
    use super::*;
    use miette::IntoDiagnostic;

    #[test]
    fn identity_moves_only_for_doc_owned_fields() -> miette::Result<()> {
        let mut previous = RepoState::synced();
        previous.pds = Some(CowStr::Borrowed("https://pds.example"));
        previous.handle = Some(Handle::new_static("alice.test").into_diagnostic()?);

        let mut state = previous.clone();
        state.advance_message_time(1_000);
        state.touch();
        state.root = previous.root.clone();
        assert!(!state.identity_moved_from(&previous));
        assert!(!state.account_moved_from(&previous));

        let mut handle_moved = previous.clone();
        handle_moved.handle = Some(Handle::new_static("carol.test").into_diagnostic()?);
        assert!(handle_moved.identity_moved_from(&previous));
        assert!(!handle_moved.account_moved_from(&previous));

        let mut pds_moved = previous.clone();
        pds_moved.pds = Some(CowStr::Borrowed("https://other.example"));
        assert!(pds_moved.identity_moved_from(&previous));

        let mut key_moved = previous.clone();
        key_moved.signing_key = Some(DidKey::from_did_key(
            "did:key:zQ3shokFTS3brHcDQrn82RUDfCZESWL1ZdCEJwekUDPQiYBme",
        )?);
        assert!(key_moved.identity_moved_from(&previous));

        let mut account_moved = previous.clone();
        account_moved.status = RepoStatus::Throttled;
        assert!(account_moved.account_moved_from(&previous));
        assert!(!account_moved.identity_moved_from(&previous));

        Ok(())
    }

    #[test]
    fn first_sighting_of_a_pds_moves_the_account() {
        let mut previous = RepoState::backfilling();
        previous.pds = None;

        let mut state = previous.clone();
        state.pds = Some(CowStr::Borrowed("https://pds.example"));

        assert!(state.account_moved_from(&previous));
        assert!(state.identity_moved_from(&previous));
    }

    #[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_static("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(())
    }

    #[cfg(feature = "indexer")]
    fn commit_at(rev: &str) -> Commit {
        Commit {
            version: 3,
            rev: crate::db::types::DbTid::from(
                &jacquard_common::types::tid::Tid::new(rev).unwrap(),
            ),
            data: crate::car::compute_cid(rev.as_bytes()),
            prev: None,
            sig: bytes::Bytes::from_static(b"signature"),
        }
    }

    #[cfg(feature = "indexer")]
    #[test]
    fn a_rebase_keeps_what_only_the_other_writer_changed() -> miette::Result<()> {
        let mut before = RepoState::synced();
        before.root = Some(commit_at("3jzfcijpj2z2a"));
        before.handle = Some(Handle::new_static("alice.test").into_diagnostic()?);
        before.last_message_time = Some(10);

        let mut mine = before.clone();
        mine.root = Some(commit_at("3jzfcijpj2z2b"));
        mine.last_message_time = Some(30);

        let mut theirs = before.clone();
        theirs.active = false;
        theirs.status = RepoStatus::Deactivated;
        theirs.handle = Some(Handle::new_static("bob.test").into_diagnostic()?);
        theirs.last_message_time = Some(20);
        theirs.last_account_time = Some(20);

        let merged = mine.rebased(&before, theirs);
        assert_eq!(merged.root, Some(commit_at("3jzfcijpj2z2b")));
        assert!(!merged.active);
        assert_eq!(merged.status, RepoStatus::Deactivated);
        assert_eq!(merged.handle.as_ref().map(Handle::as_str), Some("bob.test"));
        assert_eq!(merged.last_message_time, Some(30));
        assert_eq!(merged.last_account_time, Some(20));
        Ok(())
    }

    #[cfg(feature = "indexer")]
    #[test]
    fn a_rebase_lets_this_writer_win_a_conflict_but_never_moves_the_root_back() {
        let before = RepoState::synced();

        let mut mine = before.clone();
        mine.status = RepoStatus::Takendown;
        mine.active = false;
        mine.root = Some(commit_at("3jzfcijpj2z2a"));

        let mut theirs = before.clone();
        theirs.status = RepoStatus::Throttled;
        theirs.root = Some(commit_at("3jzfcijpj2z2b"));

        let merged = mine.rebased(&before, theirs);
        assert_eq!(merged.status, RepoStatus::Takendown);
        assert!(!merged.active);
        assert_eq!(merged.root, Some(commit_at("3jzfcijpj2z2b")));
    }

    #[cfg(feature = "indexer")]
    #[test]
    fn a_rebase_keeps_a_newer_event_that_repeats_the_old_value() {
        let mut before = RepoState::synced();
        before.active = false;
        before.status = RepoStatus::Deactivated;
        before.last_account_time = Some(10);

        let mut mine = before.clone();
        mine.last_account_time = Some(20);

        let mut theirs = before.clone();
        theirs.active = true;
        theirs.status = RepoStatus::Desynchronized;

        let merged = mine.rebased(&before, theirs);
        assert!(!merged.active);
        assert_eq!(merged.status, RepoStatus::Deactivated);
        assert_eq!(merged.last_account_time, Some(20));
    }

    #[cfg(feature = "indexer")]
    #[test]
    fn an_identity_clock_alone_doesnt_beat_a_newer_identity() {
        let mut before = RepoState::synced();
        before.pds = Some(CowStr::Borrowed("https://a.example/"));
        before.last_identity_time = Some(10);

        // the relay handling an identity event it got no fresh doc for
        let mut mine = before.clone();
        mine.last_identity_time = Some(20);

        // a backfill persisting the doc it resolved
        let mut theirs = before.clone();
        theirs.pds = Some(CowStr::Borrowed("https://b.example/"));

        let merged = mine.rebased(&before, theirs);
        assert_eq!(merged.pds.as_deref(), Some("https://b.example/"));
        assert_eq!(merged.last_identity_time, Some(20));
    }

    #[test]
    fn rows_without_a_resolve_time_spread_over_one_ttl() -> miette::Result<()> {
        let ttl = 86_400;
        let epoch = 1_700_000_000;
        let ages = (0..1000)
            .map(|i| {
                let did = Did::new_owned(format!("did:plc:{i:0>24}")).into_diagnostic()?;
                let state = RepoState::backfilling();
                let at = state.identity_resolved_or_spread(&did, epoch, ttl);
                assert_eq!(at, state.identity_resolved_or_spread(&did, epoch, ttl));
                Ok(epoch - at)
            })
            .collect::<miette::Result<Vec<_>>>()?;
        assert!(ages.iter().all(|age| (0..ttl).contains(age)));
        // a tenth of the ttl each, give or take
        for slot in 0..10 {
            let in_slot = ages.iter().filter(|age| **age * 10 / ttl == slot).count();
            assert!((50..150).contains(&in_slot), "slot {slot} has {in_slot}");
        }

        let mut stamped = RepoState::backfilling();
        stamped.identity_resolved_at = Some(42);
        let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").into_diagnostic()?;
        assert_eq!(stamped.identity_resolved_or_spread(&did, epoch, ttl), 42);
        Ok(())
    }

    #[cfg(feature = "indexer")]
    fn resolved(pds: &'static str, at: Option<i64>) -> RepoState<'static> {
        RepoState {
            pds: Some(CowStr::Borrowed(pds)),
            identity_resolved_at: at,
            ..RepoState::synced()
        }
    }

    #[cfg(feature = "indexer")]
    #[test]
    fn a_refresh_that_finds_the_same_doc_keeps_its_stamp() {
        let before = resolved("https://a.example/", Some(10));
        let mine = resolved("https://a.example/", Some(20));
        let mut theirs = before.clone();
        theirs.last_message_time = Some(5);

        let merged = mine.rebased(&before, theirs);
        assert_eq!(merged.identity_resolved_at, Some(20));
        assert_eq!(merged.last_message_time, Some(5));
    }

    #[cfg(feature = "indexer")]
    #[test]
    fn the_newest_resolve_wins_the_whole_identity() {
        let before = resolved("https://a.example/", Some(10));

        // the relay resolved b, then a backfill that resolved after it wrote c
        let relay = resolved("https://b.example/", Some(20));
        let backfill = resolved("https://c.example/", Some(30));
        let merged = relay.clone().rebased(&before, backfill.clone());
        assert_eq!(merged.pds.as_deref(), Some("https://c.example/"));
        assert_eq!(merged.identity_resolved_at, Some(30));

        // and the other way around
        let merged = backfill.rebased(&before, relay);
        assert_eq!(merged.pds.as_deref(), Some("https://c.example/"));
        assert_eq!(merged.identity_resolved_at, Some(30));

        // a writer that never stamps still beats a row nobody else touched
        let unstamped = resolved("https://d.example/", Some(10));
        let mut moved = unstamped.clone();
        moved.pds = Some(CowStr::Borrowed("https://e.example/"));
        let merged = moved.rebased(&unstamped, unstamped.clone());
        assert_eq!(merged.pds.as_deref(), Some("https://e.example/"));
    }

    #[cfg(feature = "indexer")]
    #[test]
    fn marking_an_identity_stale_sticks_unless_someone_resolved_since() {
        let before = resolved("https://a.example/", Some(10));
        let stale = resolved("https://a.example/", Some(0));

        let merged = stale.clone().rebased(&before, before.clone());
        assert_eq!(merged.identity_resolved_at, Some(0));

        let fresh = resolved("https://a.example/", Some(30));
        let merged = stale.rebased(&before, fresh);
        assert_eq!(merged.identity_resolved_at, Some(30));
    }

    #[test]
    fn commit_and_repo_status_conversion_contracts() {
        let did: Did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap();
        let rev = jacquard_common::types::tid::Tid::new("3jzfcijpj2z2a").unwrap();
        let data = crate::car::compute_cid(b"root");
        let prev = crate::car::compute_cid(b"prev");
        let atp = AtpCommit {
            did: did.clone(),
            version: 3,
            data,
            rev: rev.clone(),
            prev: Some(prev),
            sig: bytes::Bytes::from_static(b"\x01\x02\x03"),
        };

        let stored = Commit::from(atp);
        assert_eq!(stored.version, 3);
        assert_eq!(stored.data, data);
        assert_eq!(stored.prev, Some(prev));
        assert_eq!(stored.rev.to_tid(), rev);
        assert_eq!(stored.sig.as_ref(), b"\x01\x02\x03");

        let restored = stored.into_atp_commit(did.clone()).unwrap();
        assert_eq!(restored.did, did);
        assert_eq!(restored.version, 3);
        assert_eq!(restored.data, data);
        assert_eq!(restored.prev, Some(prev));
        assert_eq!(restored.rev, rev);
        assert_eq!(restored.sig.as_ref(), b"\x01\x02\x03");

        let mut sentinel = Commit::from(restored);
        sentinel.version = -1;
        assert!(sentinel.into_atp_commit(did).is_none());

        for (status, display) in [
            (RepoStatus::Synced, "synced"),
            (RepoStatus::Error("stinkpot".into()), "error(stinkpot)"),
            (RepoStatus::Deactivated, "deactivated"),
            (RepoStatus::Takendown, "takendown"),
            (RepoStatus::Suspended, "suspended"),
            (RepoStatus::Deleted, "deleted"),
            (RepoStatus::Desynchronized, "desynchronized"),
            (RepoStatus::Throttled, "throttled"),
        ] {
            assert_eq!(status.to_string(), display);
        }
    }
    #[cfg(feature = "indexer")]
    #[test]
    fn bulk_job_golden_fixtures() {
        let jobs = [
            BulkJob::V1(BulkJobV1::Apply {
                pending_key: vec![0, 0, 0, 0, 0, 0, 0, 7],
            }),
            BulkJob::V1(BulkJobV1::Erase),
            BulkJob::V1(BulkJobV1::Wipe {
                death_rev: DbTid::new_from_bytes(42_u64.to_be_bytes()),
            }),
        ];
        let encoded = jobs
            .iter()
            .map(|job| crate::util::hex::encode(rmp_serde::to_vec(job).unwrap()))
            .collect::<Vec<_>>();
        assert_eq!(
            encoded,
            [
                "81a2563181a54170706c7991c4080000000000000007",
                "81a25631a54572617365",
                "81a2563181a45769706591c408000000000000002a",
            ]
        );
        for job in jobs {
            let bytes = rmp_serde::to_vec(&job).unwrap();
            assert_eq!(rmp_serde::from_slice::<BulkJob>(&bytes).unwrap(), job);
        }
    }

    #[cfg(feature = "indexer")]
    #[test]
    fn current_persisted_state_golden_fixtures() {
        let states = [
            ResyncState::Error {
                kind: ResyncErrorKind::Ratelimited,
                retry_count: 7,
                next_retry: -42,
            },
            ResyncState::Error {
                kind: ResyncErrorKind::Transport,
                retry_count: 8,
                next_retry: 0,
            },
            ResyncState::Error {
                kind: ResyncErrorKind::Generic,
                retry_count: 9,
                next_retry: 42,
            },
            ResyncState::Gone {
                status: RepoStatus::Deleted,
            },
        ];
        let encoded = states
            .iter()
            .map(|state| rmp_serde::to_vec(state).unwrap())
            .collect::<Vec<_>>();
        assert_eq!(
            encoded.iter().map(crate::util::hex::encode).collect::<Vec<_>>(),
            [
                "81a54572726f7293ab526174656c696d6974656407d0d6",
                "81a54572726f7293a95472616e73706f72740800",
                "81a54572726f7293a747656e65726963092a",
                "81a4476f6e6591a744656c65746564",
            ]
        );
        assert!(matches!(
            rmp_serde::from_slice::<ResyncState>(&encoded[0]).unwrap(),
            ResyncState::Error {
                kind: ResyncErrorKind::Ratelimited,
                retry_count: 7,
                next_retry: -42,
            }
        ));
        assert!(matches!(
            rmp_serde::from_slice::<ResyncState>(&encoded[1]).unwrap(),
            ResyncState::Error {
                kind: ResyncErrorKind::Transport,
                retry_count: 8,
                next_retry: 0,
            }
        ));
        assert!(matches!(
            rmp_serde::from_slice::<ResyncState>(&encoded[2]).unwrap(),
            ResyncState::Error {
                kind: ResyncErrorKind::Generic,
                retry_count: 9,
                next_retry: 42,
            }
        ));
        assert!(matches!(
            rmp_serde::from_slice::<ResyncState>(&encoded[3]).unwrap(),
            ResyncState::Gone {
                status: RepoStatus::Deleted,
            }
        ));

        let metadata = RepoMetadata {
            tracked: true,
            index_id: 7,
            retry_count: 2,
            relay_inactive: None,
            backfilled_rev: None,
            refetch: None,
            refetch_deferred: false,
        };
        let encoded = rmp_serde::to_vec(&metadata).unwrap();
        assert_eq!(crate::util::hex::encode(&encoded), "97c30702c0c0c0c2");
        let decoded: RepoMetadata = rmp_serde::from_slice(&encoded).unwrap();
        assert!(decoded.tracked);
        assert_eq!(decoded.index_id, 7);
        assert_eq!(decoded.retry_count, 2);
    }

    #[cfg(feature = "indexer")]
    #[test]
    fn operator_redaction_json_contracts() {
        for (target, expected) in [
            (DeleteBodyTarget::Head, "\"head\""),
            (DeleteBodyTarget::History, "\"history\""),
            (DeleteBodyTarget::All, "\"all\""),
        ] {
            let encoded = serde_json::to_string(&target).unwrap();
            assert_eq!(encoded, expected);
            assert_eq!(
                serde_json::from_str::<DeleteBodyTarget>(&encoded).unwrap(),
                target
            );
        }

        let report = DeleteBodiesReport {
            heads_redacted: 1,
            history_redacted: 7,
            event_bodies_redacted: 3,
            body_bytes_redacted: 11,
            buffered_commits_deleted: 2,
            buffered_commit_bytes_deleted: 13,
        };
        let encoded = serde_json::to_value(&report).unwrap();
        assert_eq!(encoded["heads_redacted"], 1);
        assert_eq!(encoded["history_redacted"], 7);
        assert_eq!(encoded["event_bodies_redacted"], 3);
        assert_eq!(encoded["body_bytes_redacted"], 11);
        assert_eq!(encoded["buffered_commits_deleted"], 2);
        assert_eq!(encoded["buffered_commit_bytes_deleted"], 13);
    }
}