use fjall::Slice;
use jacquard_common::IntoStatic;
use jacquard_common::types::did::Did;
use miette::{IntoDiagnostic, Result};
use std::sync::Arc;
use tracing::{debug, error, warn};

use crate::backfill::admission::BackfillAdmission;
use crate::backfill::client::ThrottledHttpClient;
use crate::backfill::error::BackfillError;
use crate::config::BackfillStrategy;
use crate::db::{Txn as DbTxn, keys};
use crate::ingest::indexer::{IndexerMessage, IndexerTx};
use crate::state::AppState;
use crate::types::{GaugeState, RepoState, RepoStatus, ResyncErrorKind, ResyncState};

use super::process::process_did;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum TaskDisposition {
    Finished,
    Deferred,
}

fn stage_excluded_pending_cleanup(
    txn: &mut DbTxn<'_>,
    db: &crate::db::Db,
    did: &Did,
    pending_key: &[u8],
) -> Result<bool> {
    if !txn.lock_repo_and_is_excluded(did)? {
        return Ok(false);
    }
    txn.transition_pending_key(did, pending_key, GaugeState::Synced)?;
    txn.batch.remove(&db.indexer.pending, pending_key);
    Ok(true)
}

async fn remove_pending_if_excluded(
    state: &AppState,
    did: &Did,
    pending_key: Slice,
) -> Result<bool> {
    let did = did.clone();
    state
        .db
        .run(move |db| {
            let mut txn = DbTxn::new(db);
            if !stage_excluded_pending_cleanup(&mut txn, db, &did, pending_key.as_ref())? {
                return Ok(false);
            }
            txn.commit()?;
            Ok(true)
        })
        .await
}

async fn remove_pending(state: &AppState, pending_key: Slice) -> Result<()> {
    state
        .db
        .run(move |db| {
            let mut batch = db.inner.batch();
            batch.remove(&db.indexer.pending, pending_key);
            batch.commit().into_diagnostic()
        })
        .await
}

pub(super) async fn did_task(
    state: &Arc<AppState>,
    http: ThrottledHttpClient,
    buffer_tx: IndexerTx,
    did: &Did,
    pending_key: Slice,
    admission: BackfillAdmission,
    _permit: tokio::sync::OwnedSemaphorePermit,
    verify_signatures: bool,
    strategy: BackfillStrategy,
) -> Result<TaskDisposition, BackfillError> {
    let db = &state.db;

    // Drop stale/excluded work before resolver, admission, or network I/O. A
    // repository drop may race task spawning, and orphan pending keys are not
    // necessarily the one named by current metadata.
    if remove_pending_if_excluded(state, did, pending_key.clone()).await? {
        return Ok(TaskDisposition::Finished);
    }

    match process_did(
        state,
        &http,
        did,
        pending_key.clone(),
        &admission,
        verify_signatures,
        strategy,
    )
    .await
    {
        Ok(Some(imported)) => {
            let applied = state
                .db
                .run({
                    let did = did.clone();
                    let pending_key = pending_key.clone();
                    move |db| {
                        let did_key = keys::repo_key(&did);
                        let mut txn = DbTxn::new(db);
                        if stage_excluded_pending_cleanup(&mut txn, db, &did, pending_key.as_ref())?
                        {
                            txn.commit()?;
                            return Ok::<_, miette::Report>(false);
                        }
                        let applied = txn.transition_pending_key(
                            &did,
                            pending_key.as_ref(),
                            GaugeState::Synced,
                        )?;
                        txn.batch.remove(&db.indexer.pending, pending_key.clone());
                        if applied {
                            txn.batch.remove(&db.indexer.resync, &did_key);
                        }
                        txn.commit()?;
                        Ok::<_, miette::Report>(applied)
                    }
                })
                .await?;

            if !applied {
                return Ok(TaskDisposition::Finished);
            }

            state
                .backfill_drained
                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);

            let state = state.clone();
            state
                .db
                .run(move |db| {
                    db.inner
                        .persist(fjall::PersistMode::Buffer)
                        .into_diagnostic()
                })
                .await?;

            #[cfg(feature = "indexer_stream")]
            imported.emit(&state, did);
            #[cfg(not(feature = "indexer_stream"))]
            drop(imported);

            if let Err(e) = buffer_tx
                .send(IndexerMessage::BackfillFinished(did.clone()))
                .await
            {
                error!(err = %e, "failed to send BackfillFinished");
            }
            Ok(TaskDisposition::Finished)
        }
        Ok(None) => {
            remove_pending(state, pending_key).await?;
            Ok(TaskDisposition::Finished)
        }
        Err(BackfillError::PdsBusy) => Ok(TaskDisposition::Deferred),
        Err(BackfillError::UnsafeEndpoint(reason)) => {
            // The identity document is the only authority that can repair a
            // permanently unsafe PDS. Park this queue item rather than hot
            // retrying the same invalid endpoint. A later identity event can
            // enqueue fresh work with new evidence.
            warn!(%reason, "unsafe PDS endpoint, parking backfill until identity evidence changes");
            remove_pending(state, pending_key).await?;
            Ok(TaskDisposition::Finished)
        }
        Err(BackfillError::Deleted) => {
            warn!("orphaned pending entry, cleaning up");
            state
                .db
                .run({
                    let did = did.clone();
                    let pending_key = pending_key.clone();
                    move |db| {
                        let mut txn = DbTxn::new(db);
                        if stage_excluded_pending_cleanup(&mut txn, db, &did, pending_key.as_ref())?
                        {
                            txn.commit()?;
                            return Ok::<_, miette::Report>(());
                        }
                        txn.transition_pending_key(&did, pending_key.as_ref(), GaugeState::Synced)?;
                        txn.batch.remove(&db.indexer.pending, pending_key);
                        txn.commit()?;
                        Ok::<_, miette::Report>(())
                    }
                })
                .await?;
            state
                .backfill_drained
                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
            Ok(TaskDisposition::Finished)
        }
        Err(e) => {
            match &e {
                BackfillError::Ratelimited | BackfillError::PreemptivelyThrottled => {
                    debug!("too many requests");
                }
                BackfillError::Transport(reason) => {
                    error!(%reason, "transport error");
                }
                BackfillError::Generic(e) => {
                    error!(err = %e, "failed");
                }
                BackfillError::RepoNotFound => {
                    warn!("repo not found, retrying through existing backoff");
                }
                BackfillError::Deleted => unreachable!("already handled"),
                BackfillError::PdsBusy => unreachable!("already handled"),
                BackfillError::UnsafeEndpoint(_) => unreachable!("handled above"),
            }

            let error_kind = match &e {
                BackfillError::Ratelimited | BackfillError::PreemptivelyThrottled => {
                    ResyncErrorKind::Ratelimited
                }
                BackfillError::Transport(_) => ResyncErrorKind::Transport,
                BackfillError::Generic(_) => ResyncErrorKind::Generic,
                BackfillError::RepoNotFound => ResyncErrorKind::Generic,
                BackfillError::Deleted => unreachable!("already handled"),
                BackfillError::PdsBusy => unreachable!("already handled"),
                BackfillError::UnsafeEndpoint(_) => unreachable!("handled above"),
            };

            let did_key = keys::repo_key(did);

            // 1. get current retry count
            let did_key_clone = did_key.clone();
            let existing_state = db
                .run(move |db| {
                    db.indexer
                        .resync
                        .get(&did_key_clone)
                        .into_diagnostic()
                        .and_then(|b| {
                            b.map(|b| rmp_serde::from_slice::<ResyncState>(&b).into_diagnostic())
                                .transpose()
                        })
                })
                .await?;

            let mut retry_count = match existing_state {
                Some(ResyncState::Error { retry_count, .. }) => retry_count,
                Some(ResyncState::Gone { .. }) => {
                    remove_pending(state, pending_key).await?;
                    return Ok(TaskDisposition::Finished);
                }
                None => 0,
            };

            // Calculate new stats
            let next_retry = if matches!(e, BackfillError::PreemptivelyThrottled) {
                chrono::Utc::now().timestamp() + 60
            } else {
                retry_count += 1;
                ResyncState::next_backoff(retry_count)
            };

            let resync_state = ResyncState::Error {
                kind: error_kind,
                retry_count,
                next_retry,
            };
            let error_string = e.to_string();

            let applied = state
                .db
                .run({
                    let did_key = did_key.into_static();
                    let did = did.clone();
                    let pending_key = pending_key.clone();
                    move |db| {
                        // 3. save to resync
                        let serialized_resync_state =
                            rmp_serde::to_vec(&resync_state).into_diagnostic()?;

                        // 4. and update the main repo state
                        let serialized_repo_state =
                            if let Some(state_bytes) = db.repos.get(&did_key).into_diagnostic()? {
                                let mut state: RepoState =
                                    rmp_serde::from_slice(&state_bytes).into_diagnostic()?;
                                state.active = true;
                                state.status = RepoStatus::Error(error_string.into());
                                Some(rmp_serde::to_vec(&state).into_diagnostic()?)
                            } else {
                                None
                            };
                        let mut txn = DbTxn::new(db);
                        if stage_excluded_pending_cleanup(&mut txn, db, &did, pending_key.as_ref())?
                        {
                            txn.commit()?;
                            return Ok::<_, miette::Report>(false);
                        }
                        let applied = txn.transition_pending_key(
                            &did,
                            pending_key.as_ref(),
                            GaugeState::Resync(Some(error_kind)),
                        )?;
                        txn.batch.remove(&db.indexer.pending, pending_key.clone());
                        if applied {
                            txn.batch
                                .insert(&db.indexer.resync, &did_key, serialized_resync_state);
                            if let Some(state_bytes) = serialized_repo_state {
                                txn.batch.insert(&db.repos, &did_key, state_bytes);
                            }
                        }
                        txn.commit()?;
                        Ok::<_, miette::Report>(applied)
                    }
                })
                .await?;

            if applied {
                state
                    .backfill_drained
                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
            }

            Err(e)
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::backfill::admission::BackfillAdmission;
    use crate::config::Config;
    use crate::db;
    use crate::db::types::DbRkey;
    use crate::ingest::indexer::FirehoseWorker;
    use crate::types::RepoMetadata;
    use axum::Router;
    use axum::routing::get;
    use bytes::Bytes;
    use jacquard_common::CowStr;
    use jacquard_common::types::cid::IpldCid;
    use jacquard_common::types::string::Handle;
    use jacquard_common::types::tid::Tid;
    use jacquard_repo::commit::Commit as AtpCommit;
    use jacquard_repo::mst::Mst;
    use jacquard_repo::{BlockStore, MemoryBlockStore};
    use reqwest::StatusCode;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use tokio::sync::{RwLock, Semaphore};
    use url::Url;

    #[tokio::test]
    async fn excluded_orphan_pending_entry_is_removed_before_work() -> miette::Result<()> {
        let tmp = tempfile::tempdir().into_diagnostic()?;
        let config = Config {
            database_path: tmp.path().to_path_buf(),
            ..Default::default()
        };
        let state = AppState::new(&config)?;
        let did = Did::new("did:plc:ewvi7nxzyoun6zhxrhs64oiz")?.into_static();
        let pending_key = keys::pending_key(7);
        let mut batch = state.db.inner.batch();
        batch.insert(
            &state.db.filter,
            crate::db::filter::exclude_key(did.as_str())?,
            [],
        );
        batch.insert(&state.db.indexer.pending, pending_key, keys::repo_key(&did));
        batch.commit().into_diagnostic()?;

        assert!(
            remove_pending_if_excluded(&state, &did, fjall::Slice::from(pending_key.as_slice()),)
                .await?
        );
        assert!(
            state
                .db
                .indexer
                .pending
                .get(pending_key)
                .into_diagnostic()?
                .is_none()
        );
        Ok(())
    }

    // --- full-backfill `RepoNotFound` boundary ---------------------------------------------
    //
    // `getRepo` `RepoNotFound` is a fetch result, not lifecycle evidence. these tests pin the
    // non-destructive policy at the task boundary: the snapshot survives, the queue parks a
    // retryable error, and only a later valid CAR or an authoritative deletion moves state.

    const DID: &str = "did:plc:ewvi7nxzyoun6zhxrhs64oiz";
    const NOT_FOUND_BODY: &[u8] = br#"{"error":"RepoNotFound","message":"not here"}"#;
    const COLLECTION: &str = "app.bsky.feed.post";

    fn post(text: &str) -> Vec<u8> {
        serde_ipld_dagcbor::to_vec(&serde_json::json!({
            "$type": COLLECTION,
            "text": text,
        }))
        .unwrap()
    }

    struct PdsStub {
        base: Url,
        address: std::net::SocketAddr,
        response: Arc<RwLock<(StatusCode, Bytes)>>,
        requests: Arc<AtomicUsize>,
    }

    impl PdsStub {
        async fn spawn() -> Self {
            let response = Arc::new(RwLock::new((
                StatusCode::BAD_REQUEST,
                Bytes::from_static(NOT_FOUND_BODY),
            )));
            let requests = Arc::new(AtomicUsize::new(0));
            let app = Router::new().route(
                "/xrpc/com.atproto.sync.getRepo",
                get({
                    let response = response.clone();
                    let requests = requests.clone();
                    move || {
                        let response = response.clone();
                        let requests = requests.clone();
                        async move {
                            requests.fetch_add(1, Ordering::SeqCst);
                            let (status, body) = response.read().await.clone();
                            (status, body)
                        }
                    }
                }),
            );
            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let address = listener.local_addr().unwrap();
            // Keep the endpoint public by syntax while pinning its socket in
            // the test client; no DNS or external network is involved.
            let base = Url::parse("http://pds.example/").unwrap();
            tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
            Self {
                base,
                address,
                response,
                requests,
            }
        }

        async fn serve_car(&self, car: Bytes) {
            *self.response.write().await = (StatusCode::OK, car);
        }

        fn requests(&self) -> usize {
            self.requests.load(Ordering::SeqCst)
        }
    }

    async fn spawn_doc_stub(doc: String) -> Url {
        let app = Router::new().fallback({
            move |_uri: axum::http::Uri| {
                let doc = doc.clone();
                async move {
                    (
                        [(axum::http::header::CONTENT_TYPE, "application/json")],
                        doc,
                    )
                }
            }
        });
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let base = Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap();
        tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
        base
    }

    fn did_doc(base: &Url) -> String {
        serde_json::json!({
            "@context": ["https://www.w3.org/ns/did/v1"],
            "id": DID,
            "service": [{
                "id": "#atproto_pds",
                "type": "AtprotoPersonalDataServer",
                "serviceEndpoint": base.as_str(),
            }],
        })
        .to_string()
    }

    fn rooted(rev: &str) -> miette::Result<RepoState<'static>> {
        let mut state = RepoState::synced();
        state.root = Some(crate::types::Commit {
            version: 3,
            rev: crate::db::types::DbTid::from(&Tid::new(rev).into_diagnostic()?),
            data: jacquard_repo::mst::util::compute_cid(b"seeded-root").into_diagnostic()?,
            prev: None,
            sig: Bytes::from_static(b"signature"),
        });
        state.pds = Some(CowStr::Borrowed("https://old-pds.example"));
        state.handle = Some(Handle::new_static("alice.test").into_diagnostic()?);
        state.last_message_time = Some(101);
        state.last_updated_at = 5;
        Ok(state.into_static())
    }

    struct BuiltCar {
        bytes: Bytes,
        mst_root: IpldCid,
        rev: Tid,
    }

    async fn build_car(rev: &str, records: &[(&str, &[u8])]) -> miette::Result<BuiltCar> {
        let store = Arc::new(MemoryBlockStore::new());
        let mut mst = Mst::new(store.clone());
        for &(rkey, body) in records {
            let cid = jacquard_repo::mst::util::compute_cid(body).into_diagnostic()?;
            mst.add_mut(&format!("{COLLECTION}/{rkey}"), cid)
                .await
                .into_diagnostic()?;
            store
                .put_many([(cid, Bytes::copy_from_slice(body))])
                .await
                .into_diagnostic()?;
        }
        let mst_root = mst.persist().await.into_diagnostic()?;
        let rev = Tid::new(rev).into_diagnostic()?;
        let did: Did = Did::new_static(DID).into_diagnostic()?;
        let commit = AtpCommit {
            did,
            version: 3,
            data: mst_root,
            rev: rev.clone(),
            prev: None,
            sig: Bytes::from_static(b"signature"),
        };
        let commit_cid = commit.to_cid().into_diagnostic()?;
        let commit_cbor = commit.to_cbor().into_diagnostic()?;
        store
            .put_many([(commit_cid, Bytes::from(commit_cbor.clone()))])
            .await
            .into_diagnostic()?;

        let mut bytes = Vec::new();
        let mut writer =
            iroh_car::CarWriter::new(iroh_car::CarHeader::new_v1(vec![commit_cid]), &mut bytes);
        writer
            .write(commit_cid, &commit_cbor)
            .await
            .into_diagnostic()?;
        mst.write_blocks_to_car(&mut writer)
            .await
            .into_diagnostic()?;
        writer.finish().await.into_diagnostic()?;
        Ok(BuiltCar {
            bytes: Bytes::from(bytes),
            mst_root,
            rev,
        })
    }

    struct Fixture {
        state: Arc<AppState>,
        did: Did,
        pds: PdsStub,
        _tmp: tempfile::TempDir,
    }

    impl Fixture {
        async fn new() -> miette::Result<Self> {
            let pds = PdsStub::spawn().await;
            let endpoint = pds.base.clone();
            Self::new_with_pds(pds, endpoint).await
        }

        async fn new_with_pds(pds: PdsStub, pds_endpoint: Url) -> miette::Result<Self> {
            let tmp = tempfile::tempdir().into_diagnostic()?;
            let doc_base = spawn_doc_stub(did_doc(&pds_endpoint)).await;
            let public_plc = Url::parse("http://public.example/").unwrap();
            let config = Config {
                database_path: tmp.path().to_path_buf(),
                plc_urls: vec![public_plc.clone()],
                identity_cache_size: 64,
                ..Default::default()
            };
            let mut app_state = AppState::new(&config)?;
            let public_client = reqwest::Client::builder()
                .no_proxy()
                .resolve(
                    "public.example",
                    std::net::SocketAddr::from((
                        std::net::Ipv4Addr::LOCALHOST,
                        doc_base.port().unwrap(),
                    )),
                )
                .build()
                .into_diagnostic()?;
            let resolver = crate::resolver::Resolver::with_plc_urls_and_client(
                vec![public_plc.clone()],
                64,
                crate::net::PublicHttpClient::test(public_client),
            );
            app_state.resolver = resolver;
            Ok(Self {
                state: Arc::new(app_state),
                did: Did::new_static(DID).into_diagnostic()?,
                pds,
                _tmp: tmp,
            })
        }

        fn seed(
            &self,
            repo: &RepoState<'static>,
            index_id: u64,
            records: &[(&str, &[u8])],
        ) -> miette::Result<()> {
            let mut batch = self.state.db.inner.batch();
            batch.insert(
                &self.state.db.repos,
                keys::repo_key(&self.did),
                db::ser_repo_state(repo)?,
            );
            batch.insert(
                &self.state.db.repo_metadata,
                keys::repo_metadata_key(&self.did),
                db::ser_repo_meta(&RepoMetadata::backfilling(index_id))?,
            );
            batch.insert(
                &self.state.db.indexer.pending,
                keys::pending_key(index_id),
                keys::repo_key(&self.did),
            );
            for &(rkey, body) in records {
                self.state.db.indexer.stage_record(
                    &mut batch,
                    keys::record_key(&self.did, COLLECTION, &DbRkey::new(rkey)),
                    body,
                );
            }
            batch.commit().into_diagnostic()
        }

        /// mirrors the retry worker: a completed queue entry gives way to a fresh pending key.
        fn reindex_pending(&self, from: u64, to: u64) -> miette::Result<()> {
            let mut batch = self.state.db.inner.batch();
            batch.remove(&self.state.db.indexer.pending, keys::pending_key(from));
            batch.insert(
                &self.state.db.indexer.pending,
                keys::pending_key(to),
                keys::repo_key(&self.did),
            );
            batch.insert(
                &self.state.db.repo_metadata,
                keys::repo_metadata_key(&self.did),
                db::ser_repo_meta(&RepoMetadata::backfilling(to))?,
            );
            batch.commit().into_diagnostic()
        }

        fn clear_resync(&self) -> miette::Result<()> {
            self.state
                .db
                .indexer
                .resync
                .remove(keys::repo_key(&self.did))
                .into_diagnostic()?;
            Ok(())
        }

        /// the authoritative deletion primitive plus its `Deleted` tombstone: the same pair the
        /// firehose and list-repos paths apply under the repository lock.
        fn delete_authoritatively(&self, repo: &RepoState<'static>) -> miette::Result<()> {
            let mut txn = DbTxn::new(&self.state.db);
            crate::ops::delete_repo(&mut txn, &self.state.db, &self.did, repo)?;
            crate::ops::transition_repo(
                &mut txn.batch,
                &self.state.db,
                &mut Vec::new(),
                &self.did,
                repo.clone(),
                RepoStatus::Deleted,
            )?;
            txn.commit()
        }

        async fn run(&self, pending_key: [u8; 8]) -> Result<TaskDisposition, BackfillError> {
            let (tx, _worker) = FirehoseWorker::new(self.state.clone(), 1);
            let client = reqwest::Client::builder()
                .resolve("pds.example", self.pds.address)
                .build()
                .into_diagnostic()?;
            let http = ThrottledHttpClient::new(
                crate::net::PublicHttpClient::test(client),
                self.state.throttler.clone(),
            );
            let admission = BackfillAdmission::new(4, 4);
            let permit = Arc::new(Semaphore::new(1)).acquire_owned().await.unwrap();
            did_task(
                &self.state,
                http,
                tx,
                &self.did,
                Slice::from(pending_key),
                admission,
                permit,
                false,
                BackfillStrategy::Full,
            )
            .await
        }

        fn repo(&self) -> miette::Result<RepoState<'static>> {
            let bytes = self
                .state
                .db
                .repos
                .get(keys::repo_key(&self.did))
                .into_diagnostic()?
                .expect("repository state row")
                .to_vec();
            Ok(db::deser_repo_state(&bytes)?.into_static())
        }

        fn resync(&self) -> miette::Result<Option<ResyncState>> {
            self.state
                .db
                .indexer
                .resync
                .get(keys::repo_key(&self.did))
                .into_diagnostic()?
                .map(|bytes| rmp_serde::from_slice::<ResyncState>(&bytes).into_diagnostic())
                .transpose()
        }

        fn pending_keys(&self) -> miette::Result<Vec<Vec<u8>>> {
            self.state
                .db
                .indexer
                .pending
                .iter()
                .map(|guard| Ok(guard.into_inner().into_diagnostic()?.0.to_vec()))
                .collect()
        }

        fn record_rkeys(&self) -> miette::Result<Vec<String>> {
            let prefix = keys::record_prefix_did(&self.did);
            self.state
                .db
                .indexer
                .record_prefix(&prefix)
                .map(|guard| {
                    let (key, _) = guard.into_inner().into_diagnostic()?;
                    let (_, rkey) = keys::split_record_suffix(&key[prefix.len()..])?;
                    Ok(rkey.to_string())
                })
                .collect()
        }
    }

    #[tokio::test]
    async fn unsafe_pds_endpoint_is_parked_until_fresh_identity_evidence() -> miette::Result<()> {
        let pds = PdsStub::spawn().await;
        let fixture = Fixture::new_with_pds(pds, Url::parse("http://127.0.0.1/").unwrap()).await?;
        fixture.seed(&RepoState::backfilling().into_static(), 13, &[])?;

        assert_eq!(
            fixture.run(keys::pending_key(13)).await?,
            TaskDisposition::Finished,
            "unsafe endpoint work must park instead of entering retry backoff"
        );
        assert!(fixture.pending_keys()?.is_empty());
        assert!(fixture.resync()?.is_none());
        assert_eq!(
            fixture.pds.requests(),
            0,
            "unsafe endpoint must not connect"
        );
        Ok(())
    }

    #[tokio::test]
    async fn repo_not_found_keeps_a_rooted_snapshot_and_parks_a_retry() -> miette::Result<()> {
        let fixture = Fixture::new().await?;
        let seeded = rooted("3jzfcijpj2z2a")?;
        let body = post("a");
        fixture.seed(&seeded, 7, &[("a", body.as_slice())])?;

        let err = fixture.run(keys::pending_key(7)).await.unwrap_err();

        assert!(
            matches!(err, BackfillError::RepoNotFound),
            "unexpected error: {err}"
        );
        let after = fixture.repo()?;
        assert_eq!(
            after.root.as_ref().map(|root| root.data),
            seeded.root.as_ref().map(|root| root.data),
            "a non-authoritative fetch result must not clear the root"
        );
        assert!(
            matches!(&after.status, RepoStatus::Error(_)),
            "expected a retryable error status, got {:?}",
            after.status
        );
        assert!(after.active);
        assert_eq!(after.last_message_time, Some(101));
        assert_eq!(fixture.record_rkeys()?, ["a"]);
        assert!(
            fixture
                .state
                .db
                .repo_metadata
                .get(keys::repo_metadata_key(&fixture.did))
                .into_diagnostic()?
                .is_some(),
            "metadata must survive"
        );
        assert!(
            fixture.pending_keys()?.is_empty(),
            "parked work must leave pending, or the worker spins"
        );

        let Some(ResyncState::Error {
            kind,
            retry_count,
            next_retry,
        }) = fixture.resync()?
        else {
            panic!("expected a durable retryable error");
        };
        assert_eq!(kind, ResyncErrorKind::Generic);
        assert_eq!(retry_count, 1);
        assert!(
            next_retry > chrono::Utc::now().timestamp(),
            "the retry worker may only requeue a due error"
        );
        Ok(())
    }

    #[tokio::test]
    async fn repo_not_found_on_a_never_rooted_repo_uses_the_same_retry_policy() -> miette::Result<()>
    {
        let fixture = Fixture::new().await?;
        fixture.seed(&RepoState::backfilling().into_static(), 9, &[])?;

        let err = fixture.run(keys::pending_key(9)).await.unwrap_err();

        assert!(
            matches!(err, BackfillError::RepoNotFound),
            "unexpected error: {err}"
        );
        let after = fixture.repo()?;
        assert!(after.root.is_none(), "no root may be invented");
        assert!(
            matches!(&after.status, RepoStatus::Error(_)),
            "expected a retryable error status, got {:?}",
            after.status
        );
        assert!(fixture.record_rkeys()?.is_empty());
        assert!(fixture.pending_keys()?.is_empty());
        assert!(matches!(
            fixture.resync()?,
            Some(ResyncState::Error {
                kind: ResyncErrorKind::Generic,
                ..
            })
        ));
        Ok(())
    }

    #[tokio::test]
    async fn repo_not_found_cannot_overwrite_a_replaced_pending_entry() -> miette::Result<()> {
        let fixture = Fixture::new().await?;
        let seeded = rooted("3jzfcijpj2z2a")?;
        let body = post("a");
        fixture.seed(&seeded, 7, &[("a", body.as_slice())])?;
        fixture.reindex_pending(7, 11)?;

        let err = fixture.run(keys::pending_key(7)).await.unwrap_err();

        assert!(
            matches!(err, BackfillError::RepoNotFound),
            "unexpected error: {err}"
        );
        assert_eq!(fixture.pending_keys()?, [keys::pending_key(11).to_vec()]);
        assert!(
            fixture.resync()?.is_none(),
            "stale work must not park a retry against a newer queue entry"
        );
        let after = fixture.repo()?;
        assert_eq!(after.status, RepoStatus::Synced);
        assert_eq!(
            after.root.as_ref().map(|root| root.data),
            seeded.root.as_ref().map(|root| root.data)
        );
        assert_eq!(fixture.record_rkeys()?, ["a"]);
        Ok(())
    }

    #[tokio::test]
    async fn authoritative_deletion_is_not_resurrected_by_a_late_repo_not_found()
    -> miette::Result<()> {
        let fixture = Fixture::new().await?;
        let seeded = rooted("3jzfcijpj2z2a")?;
        let body = post("a");
        fixture.seed(&seeded, 7, &[("a", body.as_slice())])?;
        fixture.delete_authoritatively(&seeded)?;

        let err = fixture.run(keys::pending_key(7)).await.unwrap_err();

        assert!(
            matches!(err, BackfillError::RepoNotFound),
            "unexpected error: {err}"
        );
        let after = fixture.repo()?;
        assert_eq!(after.status, RepoStatus::Deleted);
        assert!(!after.active, "the tombstone must stay inactive");
        assert!(
            fixture.record_rkeys()?.is_empty(),
            "erased records must stay erased"
        );
        assert!(fixture.pending_keys()?.is_empty());
        assert!(
            fixture.resync()?.is_none(),
            "a deleted repo must not be re-queued"
        );
        Ok(())
    }

    #[tokio::test]
    async fn a_later_valid_car_replaces_the_error_after_requeue() -> miette::Result<()> {
        let fixture = Fixture::new().await?;
        let seeded = rooted("3jzfcijpj2z2a")?;
        let first = post("a");
        fixture.seed(&seeded, 7, &[("a", first.as_slice())])?;
        assert!(matches!(
            fixture.run(keys::pending_key(7)).await.unwrap_err(),
            BackfillError::RepoNotFound
        ));
        assert_eq!(fixture.pds.requests(), 1);

        fixture.clear_resync()?;
        fixture.reindex_pending(7, 21)?;
        let second = post("b");
        let car = build_car(
            "3jzfcijpj2z2b",
            &[("a", first.as_slice()), ("b", second.as_slice())],
        )
        .await?;
        fixture.pds.serve_car(car.bytes.clone()).await;

        assert_eq!(
            fixture.run(keys::pending_key(21)).await.unwrap(),
            TaskDisposition::Finished
        );

        let after = fixture.repo()?;
        assert_eq!(
            after.root.as_ref().map(|root| root.data),
            Some(car.mst_root)
        );
        assert_eq!(
            after.root.as_ref().map(|root| root.rev.to_tid()),
            Some(car.rev.clone())
        );
        assert_eq!(fixture.record_rkeys()?, ["a", "b"]);
        assert!(
            fixture.resync()?.is_none(),
            "a successful import clears the error row"
        );
        assert!(fixture.pending_keys()?.is_empty());
        assert_eq!(fixture.pds.requests(), 2);
        Ok(())
    }
}