use crate::db::types::DbTid;
use crate::db::{keys, ser_repo_state};
use crate::state::AppState;
use crate::types::{GaugeState, RepoMetadata, RepoState, RepoStatus, ResyncState};
use jacquard_api::com_atproto::sync::list_repos::RepoStatus as ListedRepoStatus;
use jacquard_common::IntoStatic;
use jacquard_common::types::did::Did;
use miette::{IntoDiagnostic, Result};
use rand::Rng;
use rand::rngs::SmallRng;
use smol_str::ToSmolStr;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc;
use tracing::{debug, error, info, trace};

use super::{CrawlerStats, InFlightGuard};

const WORKER_CHANNEL_CAPACITY: usize = 64;
const BLOCKING_TASK_TIMEOUT: Duration = Duration::from_secs(30);

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum CapacityDecision {
    Proceed,
    EnterThrottle,
    StayThrottled,
    ReleaseThrottle,
}

fn capacity_decision(
    pending: u64,
    max_pending: usize,
    resume_pending: usize,
    was_throttled: bool,
) -> CapacityDecision {
    if pending > max_pending as u64 {
        if was_throttled {
            CapacityDecision::StayThrottled
        } else {
            CapacityDecision::EnterThrottle
        }
    } else if was_throttled && pending > resume_pending as u64 {
        CapacityDecision::StayThrottled
    } else if was_throttled {
        CapacityDecision::ReleaseThrottle
    } else {
        CapacityDecision::Proceed
    }
}

/// a cursor write to include atomically with the batch.
pub(crate) struct CursorUpdate {
    pub(super) key: Vec<u8>,
    pub(super) value: Option<Vec<u8>>,
}

#[derive(Clone)]
pub(crate) struct RepoListing {
    pub(super) did: Did,
    pub(super) rev: DbTid,
    pub(super) active: bool,
    pub(super) status: Option<ListedRepoStatus>,
    pub(super) create_if_missing: bool,
}

/// a batch of confirmed repos from any crawler source ready to enqueue.
///
/// `guards` hold the DIDs (via `Deref`) and keep their in-flight slots occupied
/// until their lock group is committed to the database. cursor updates commit
/// only after every idempotent group, so a crash can replay but never skip a
/// partially applied page.
pub(crate) struct CrawlerBatch {
    pub(super) guards: Vec<InFlightGuard>,
    pub(super) listings: Vec<RepoListing>,
    pub(super) cursor_updates: Vec<CursorUpdate>,
}

pub(crate) struct CrawlerWorker {
    pub(super) state: Arc<AppState>,
    pub(super) max_pending: usize,
    pub(super) resume_pending: usize,
    pub(super) stats: CrawlerStats,
    pub(super) rx: mpsc::Receiver<CrawlerBatch>,
    pub(super) was_throttled: bool,
}

impl CrawlerWorker {
    pub(crate) fn new(
        state: Arc<AppState>,
        max_pending: usize,
        resume_pending: usize,
        stats: CrawlerStats,
    ) -> (Self, mpsc::Sender<CrawlerBatch>) {
        let (tx, rx) = mpsc::channel(WORKER_CHANNEL_CAPACITY);
        (
            Self {
                state,
                max_pending,
                resume_pending,
                stats,
                rx,
                was_throttled: false,
            },
            tx,
        )
    }

    pub(crate) async fn run(mut self) {
        while let Some(batch) = self.rx.recv().await {
            self.wait_for_capacity().await;
            if let Err(e) = self.enqueue(batch).await {
                error!(err = ?e, "crawler worker: enqueue failed");
            }
        }
    }

    /// blocks until the pending queue has capacity.  mirrors the hysteresis logic
    /// that was previously duplicated in each crawler loop:
    ///   - above `max_pending`: hard stop, poll every 5s
    ///   - after crossing `max_pending`: cooldown until below `resume_pending`
    ///   - below `resume_pending`: proceed (or release throttle)
    async fn wait_for_capacity(&mut self) {
        loop {
            let pending = self.state.db.get_count("pending").await;
            match capacity_decision(
                pending,
                self.max_pending,
                self.resume_pending,
                self.was_throttled,
            ) {
                CapacityDecision::Proceed => break,
                CapacityDecision::EnterThrottle => {
                    debug!(
                        pending,
                        max = self.max_pending,
                        "throttling: above max pending"
                    );
                    self.was_throttled = true;
                    self.stats.set_throttled(true);
                    tokio::time::sleep(Duration::from_secs(5)).await;
                }
                CapacityDecision::StayThrottled => {
                    tokio::time::sleep(Duration::from_secs(5)).await;
                }
                CapacityDecision::ReleaseThrottle => {
                    self.was_throttled = false;
                    self.stats.set_throttled(false);
                    info!("throttling released");
                    break;
                }
            }
        }
    }

    /// filters already-known repos, commits them to `repos` + `pending` (plus
    /// an optional cursor update), then releases the in-flight guards.
    async fn enqueue(&mut self, batch: CrawlerBatch) -> Result<()> {
        let CrawlerBatch {
            guards,
            listings,
            cursor_updates,
        } = batch;

        // nothing to insert but still need to commit the cursor if present.
        if guards.is_empty() && listings.is_empty() {
            if !cursor_updates.is_empty() {
                self.commit_cursors(cursor_updates).await?;
            }
            return Ok(());
        }

        // Group by the one record lock each DID needs. Acquiring a whole crawler
        // page at once usually captures nearly all 256 locks and turns one long
        // operator redaction into a global ingestion convoy.
        let app_state = self.state.clone();
        let surviving = app_state
            .db
            .run(move |db| {
                let mut rng: SmallRng = rand::make_rng();
                let mut listing_groups = std::collections::BTreeMap::<u8, Vec<_>>::new();
                for listing in listings {
                    listing_groups
                        .entry(crate::db::record_lock_index_for_did(&listing.did))
                        .or_default()
                        .push(listing);
                }
                let mut guard_groups = std::collections::BTreeMap::<u8, Vec<_>>::new();
                for guard in guards {
                    guard_groups
                        .entry(crate::db::record_lock_index_for_did(&guard))
                        .or_default()
                        .push(guard);
                }
                let lock_indices = listing_groups
                    .keys()
                    .chain(guard_groups.keys())
                    .copied()
                    .collect::<std::collections::BTreeSet<_>>();
                let mut surviving = Vec::new();
                let mut changed_listings = Vec::new();

                for lock_index in lock_indices {
                    let mut txn = crate::db::Txn::new(db);
                    txn.hold_record_lock_index(lock_index);
                    for listing in listing_groups.remove(&lock_index).unwrap_or_default() {
                        if reconcile_listing(&mut txn, &listing)?.emits_account() {
                            changed_listings.push(listing);
                        }
                    }
                    for guard in guard_groups.remove(&lock_index).unwrap_or_default() {
                        if db
                            .filter
                            .contains_key(crate::db::filter::exclude_key(guard.as_str())?)
                            .into_diagnostic()?
                        {
                            continue;
                        }
                        let did_key = keys::repo_key(&guard);
                        let metadata_key = keys::repo_metadata_key(&guard);
                        if db.repos.contains_key(&did_key).into_diagnostic()? {
                            continue;
                        }
                        let state = RepoState::backfilling();
                        let metadata = RepoMetadata::backfilling(rng.next_u64());
                        txn.batch
                            .insert(&db.repos, &did_key, ser_repo_state(&state)?);
                        txn.batch.insert(
                            &db.repo_metadata,
                            &metadata_key,
                            crate::db::ser_repo_meta(&metadata)?,
                        );
                        #[cfg(feature = "indexer")]
                        txn.batch.insert(
                            &db.indexer.pending,
                            keys::pending_key(metadata.index_id),
                            &did_key,
                        );
                        txn.batch
                            .remove(&db.crawler, keys::crawler_retry_key(&guard));
                        trace!(did = %*guard, "enqueuing repo");
                        txn.counts.add_repos(1);
                        #[cfg(feature = "indexer")]
                        txn.transition_lifecycle(&guard, crate::types::GaugeState::Pending)?;
                        surviving.push(guard);
                    }
                    txn.commit()?;
                }

                let mut txn = crate::db::Txn::new(db);
                for cursor in cursor_updates {
                    stage_cursor(&mut txn.batch, db, cursor);
                }
                txn.commit()?;
                Ok((surviving, changed_listings))
            })
            .await
            .inspect_err(|e| error!(err = ?e, "enqueue batch commit failed"))
            .unwrap_or_default();

        let (surviving, changed_listings) = surviving;
        let count = surviving.len();
        // release in-flight slots now that the batch is committed
        drop(surviving);

        if count > 0 {
            self.stats.record_processed(count);
            #[cfg(feature = "indexer")]
            self.state.notify_backfill();
        }

        #[cfg(feature = "indexer_stream")]
        for listing in changed_listings {
            let evt = crate::types::AccountEvt {
                did: listing.did,
                active: listing.active,
                status: listing
                    .status
                    .map(|status| jacquard_common::CowStr::Owned(status.as_str().to_smolstr())),
            };
            let _ = self
                .state
                .db
                .stream
                .event_tx
                .send(crate::ops::make_account_event(&self.state.db, evt));
        }
        #[cfg(not(feature = "indexer_stream"))]
        drop(changed_listings);

        Ok(())
    }

    async fn commit_cursors(&self, cursors: Vec<CursorUpdate>) -> Result<()> {
        let state = self.state.clone();
        tokio::time::timeout(
            BLOCKING_TASK_TIMEOUT,
            state.db.run(move |db| {
                let mut txn = crate::db::Txn::new(db);
                for cursor in cursors {
                    stage_cursor(&mut txn.batch, db, cursor);
                }
                txn.commit()
            }),
        )
        .await
        .map_err(|_| miette::miette!("cursor-only commit timed out"))??;
        Ok(())
    }
}

fn stage_cursor(batch: &mut fjall::OwnedWriteBatch, db: &crate::db::Db, cursor: CursorUpdate) {
    match cursor.value {
        Some(value) => batch.insert(&db.cursors, cursor.key, value),
        None => batch.remove(&db.cursors, cursor.key),
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ReconcileOutcome {
    Unchanged,
    StateOnly,
    AccountChanged,
}

impl ReconcileOutcome {
    fn emits_account(self) -> bool {
        self == Self::AccountChanged
    }
}

fn reconcile_listing(
    txn: &mut crate::db::Txn<'_>,
    listing: &RepoListing,
) -> Result<ReconcileOutcome> {
    // production callers pre-acquire this listing's one lock group; this local
    // check keeps direct callers on the same record-lock -> lifecycle hierarchy.
    let db = txn.db;
    if txn.lock_repo_and_is_excluded(&listing.did)? {
        return Ok(ReconcileOutcome::Unchanged);
    }
    let did_key = keys::repo_key(&listing.did);
    let Some(state_bytes) = db.repos.get(&did_key).into_diagnostic()? else {
        if listing.active || !listing.create_if_missing {
            return Ok(ReconcileOutcome::Unchanged);
        }
        let status = listed_repo_status(listing.status.as_ref());
        let mut state = RepoState::backfilling();
        state.active = false;
        state.status = status.clone();
        if status != RepoStatus::Deleted {
            txn.batch.insert(
                &db.indexer.resync,
                &did_key,
                rmp_serde::to_vec(&ResyncState::Gone {
                    status: status.clone(),
                })
                .into_diagnostic()?,
            );
            txn.transition_lifecycle(&listing.did, GaugeState::Resync(None))?;
        }
        txn.batch
            .insert(&db.repos, &did_key, ser_repo_state(&state)?);
        txn.counts.add_repos(1);
        return Ok(ReconcileOutcome::StateOnly);
    };
    let mut state = crate::db::deser_repo_state(&state_bytes)?.into_static();

    if listing.active {
        let listing_is_newer = state
            .root
            .as_ref()
            .is_none_or(|root| listing.rev.as_bytes() > root.rev.as_bytes());
        if state.active && !listing_is_newer {
            return Ok(ReconcileOutcome::Unchanged);
        }
        if listing_is_newer && listing_is_pending(db, &listing.did)? {
            return Ok(ReconcileOutcome::Unchanged);
        }

        let was_active = state.active;
        txn.batch.remove(&db.indexer.resync, &did_key);
        if !listing_is_newer {
            txn.transition_lifecycle(&listing.did, GaugeState::Synced)?;
            state.status = RepoStatus::Synced;
        } else {
            enqueue_listing_backfill(txn, &listing.did, &did_key)?;
            txn.transition_lifecycle(&listing.did, GaugeState::Pending)?;
            state.status = RepoStatus::Desynchronized;
        }
        state.active = true;
        state.touch();
        txn.batch
            .insert(&db.repos, did_key, ser_repo_state(&state)?);
        return Ok(if was_active {
            ReconcileOutcome::StateOnly
        } else {
            ReconcileOutcome::AccountChanged
        });
    }

    let status = listed_repo_status(listing.status.as_ref());
    let account_changed = state.active || state.status != status;
    if !account_changed && !listing_is_pending(db, &listing.did)? {
        return Ok(ReconcileOutcome::Unchanged);
    }

    txn.transition_lifecycle(
        &listing.did,
        if status == RepoStatus::Deleted {
            GaugeState::Synced
        } else {
            GaugeState::Resync(None)
        },
    )?;
    if status == RepoStatus::Deleted {
        crate::ops::delete_repo(&mut *txn, db, &listing.did, &state)?;
        state.root = None;
    } else {
        if let Some(metadata_bytes) = db
            .repo_metadata
            .get(keys::repo_metadata_key(&listing.did))
            .into_diagnostic()?
        {
            let metadata = crate::db::deser_repo_meta(&metadata_bytes)?;
            txn.batch
                .remove(&db.indexer.pending, keys::pending_key(metadata.index_id));
        }
        txn.batch.insert(
            &db.indexer.resync,
            &did_key,
            rmp_serde::to_vec(&ResyncState::Gone {
                status: status.clone(),
            })
            .into_diagnostic()?,
        );
    }

    state.active = false;
    state.status = status;
    state.touch();
    txn.batch
        .insert(&db.repos, did_key, ser_repo_state(&state)?);
    Ok(if account_changed {
        ReconcileOutcome::AccountChanged
    } else {
        ReconcileOutcome::StateOnly
    })
}

fn listing_is_pending(db: &crate::db::Db, did: &Did) -> Result<bool> {
    let Some(metadata_bytes) = db
        .repo_metadata
        .get(keys::repo_metadata_key(did))
        .into_diagnostic()?
    else {
        return Ok(false);
    };
    let metadata = crate::db::deser_repo_meta(&metadata_bytes)?;
    Ok(db
        .indexer
        .pending
        .get(keys::pending_key(metadata.index_id))
        .into_diagnostic()?
        .is_some())
}

fn enqueue_listing_backfill(txn: &mut crate::db::Txn<'_>, did: &Did, did_key: &[u8]) -> Result<()> {
    let db = txn.db;
    let metadata_key = keys::repo_metadata_key(did);
    let existing = db
        .repo_metadata
        .get(&metadata_key)
        .into_diagnostic()?
        .map(|bytes| crate::db::deser_repo_meta(&bytes))
        .transpose()?;
    if let Some(metadata) = &existing {
        txn.batch
            .remove(&db.indexer.pending, keys::pending_key(metadata.index_id));
    }
    let mut metadata = existing.unwrap_or_else(|| RepoMetadata::backfilling(rand::random()));
    metadata.index_id = rand::random();
    txn.batch.insert(
        &db.repo_metadata,
        metadata_key,
        crate::db::ser_repo_meta(&metadata)?,
    );
    txn.batch.insert(
        &db.indexer.pending,
        keys::pending_key(metadata.index_id),
        did_key,
    );
    Ok(())
}

fn listed_repo_status(status: Option<&ListedRepoStatus>) -> RepoStatus {
    match status {
        Some(ListedRepoStatus::Takendown) => RepoStatus::Takendown,
        Some(ListedRepoStatus::Suspended) => RepoStatus::Suspended,
        Some(ListedRepoStatus::Deleted) => RepoStatus::Deleted,
        Some(ListedRepoStatus::Deactivated) => RepoStatus::Deactivated,
        Some(ListedRepoStatus::Desynchronized) => RepoStatus::Desynchronized,
        Some(ListedRepoStatus::Throttled) => RepoStatus::Throttled,
        Some(ListedRepoStatus::Other(status)) => RepoStatus::Error(status.to_smolstr()),
        None => RepoStatus::Error("inactive".into()),
    }
}

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

    #[test]
    fn pending_between_resume_and_max_does_not_start_throttling() {
        assert_eq!(
            capacity_decision(75, 100, 50, false),
            CapacityDecision::Proceed
        );
        assert_eq!(
            capacity_decision(100, 100, 50, false),
            CapacityDecision::Proceed
        );
    }

    #[test]
    fn pending_above_max_starts_throttling() {
        assert_eq!(
            capacity_decision(101, 100, 50, false),
            CapacityDecision::EnterThrottle
        );
    }

    #[test]
    fn throttling_stays_active_until_resume_boundary() {
        assert_eq!(
            capacity_decision(100, 100, 50, true),
            CapacityDecision::StayThrottled
        );
        assert_eq!(
            capacity_decision(51, 100, 50, true),
            CapacityDecision::StayThrottled
        );
        assert_eq!(
            capacity_decision(50, 100, 50, true),
            CapacityDecision::ReleaseThrottle
        );
        assert_eq!(
            capacity_decision(0, 100, 50, true),
            CapacityDecision::ReleaseThrottle
        );
    }

    fn did() -> Did {
        Did::new_static("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa").unwrap()
    }

    fn rev(value: u8) -> DbTid {
        let mut bytes = [0; 8];
        bytes[7] = value;
        DbTid::new_from_bytes(bytes)
    }

    fn state_with_root(rev: DbTid) -> RepoState<'static> {
        let mut state = RepoState::synced();
        state.root = Some(crate::types::Commit {
            version: 3,
            rev,
            data: jacquard_repo::mst::util::compute_cid(b"root").unwrap(),
            prev: None,
            sig: Bytes::new(),
        });
        state
    }

    fn open_db() -> Result<(tempfile::TempDir, crate::db::Db)> {
        let tmp = tempfile::tempdir().into_diagnostic()?;
        let config = crate::config::Config {
            database_path: tmp.path().to_path_buf(),
            ..Default::default()
        };
        let db = crate::db::Db::open(&config)?;
        Ok((tmp, db))
    }

    fn insert_repo(
        db: &crate::db::Db,
        state: &RepoState<'_>,
        index_id: u64,
        pending: bool,
    ) -> Result<()> {
        let did = did();
        let did_key = keys::repo_key(&did);
        db.repos
            .insert(&did_key, ser_repo_state(state)?)
            .into_diagnostic()?;
        db.repo_metadata
            .insert(
                keys::repo_metadata_key(&did),
                crate::db::ser_repo_meta(&RepoMetadata::backfilling(index_id))?,
            )
            .into_diagnostic()?;
        if pending {
            db.indexer
                .pending
                .insert(keys::pending_key(index_id), did_key)
                .into_diagnostic()?;
        }
        Ok(())
    }

    fn apply(db: &crate::db::Db, listing: &RepoListing) -> Result<ReconcileOutcome> {
        let mut txn = crate::db::Txn::new(db);
        let changed = reconcile_listing(&mut txn, listing)?;
        txn.commit()?;
        Ok(changed)
    }

    fn read_state(db: &crate::db::Db) -> Result<RepoState<'static>> {
        let bytes = db
            .repos
            .get(keys::repo_key(&did()))
            .into_diagnostic()?
            .expect("repo state should exist");
        Ok(crate::db::deser_repo_state(&bytes)?.into_static())
    }

    #[test]
    fn inactive_listing_removes_pending_backfill() -> Result<()> {
        let (_tmp, db) = open_db()?;
        insert_repo(&db, &RepoState::backfilling(), 7, true)?;
        let listing = RepoListing {
            did: did(),
            rev: rev(1),
            active: false,
            status: Some(ListedRepoStatus::Deactivated),
            create_if_missing: false,
        };

        assert_eq!(apply(&db, &listing)?, ReconcileOutcome::AccountChanged);
        assert!(
            db.indexer
                .pending
                .get(keys::pending_key(7))
                .into_diagnostic()?
                .is_none()
        );
        let state = read_state(&db)?;
        assert!(!state.active);
        assert_eq!(state.status, RepoStatus::Deactivated);
        assert_eq!(apply(&db, &listing)?, ReconcileOutcome::Unchanged);
        Ok(())
    }

    #[test]
    fn relay_first_persists_unknown_inactive_listing_without_backfill() -> Result<()> {
        let (_tmp, db) = open_db()?;
        let listing = RepoListing {
            did: did(),
            rev: rev(1),
            active: false,
            status: Some(ListedRepoStatus::Takendown),
            create_if_missing: true,
        };

        assert_eq!(apply(&db, &listing)?, ReconcileOutcome::StateOnly);
        let state = read_state(&db)?;
        assert!(!state.active);
        assert_eq!(state.status, RepoStatus::Takendown);
        assert_eq!(db.indexer.pending.iter().count(), 0);
        assert!(
            db.indexer
                .resync
                .contains_key(keys::repo_key(&listing.did))
                .into_diagnostic()?
        );
        assert_eq!(apply(&db, &listing)?, ReconcileOutcome::Unchanged);
        Ok(())
    }

    #[test]
    fn current_active_listing_does_not_reenqueue() -> Result<()> {
        let (_tmp, db) = open_db()?;
        insert_repo(&db, &state_with_root(rev(2)), 7, false)?;
        let listing = RepoListing {
            did: did(),
            rev: rev(2),
            active: true,
            status: None,
            create_if_missing: false,
        };

        assert_eq!(apply(&db, &listing)?, ReconcileOutcome::Unchanged);
        assert_eq!(db.indexer.pending.iter().count(), 0);
        Ok(())
    }

    #[test]
    fn unchanged_inactive_listing_repairs_stale_pending_entry() -> Result<()> {
        let (_tmp, db) = open_db()?;
        let mut state = RepoState::backfilling();
        state.active = false;
        state.status = RepoStatus::Deactivated;
        insert_repo(&db, &state, 7, true)?;
        let listing = RepoListing {
            did: did(),
            rev: rev(1),
            active: false,
            status: Some(ListedRepoStatus::Deactivated),
            create_if_missing: false,
        };

        assert_eq!(apply(&db, &listing)?, ReconcileOutcome::StateOnly);
        assert_eq!(db.indexer.pending.iter().count(), 0);
        Ok(())
    }

    #[test]
    fn newer_active_listing_reenqueues_missed_revision() -> Result<()> {
        let (_tmp, db) = open_db()?;
        insert_repo(&db, &state_with_root(rev(1)), 7, false)?;
        let listing = RepoListing {
            did: did(),
            rev: rev(2),
            active: true,
            status: None,
            create_if_missing: false,
        };

        assert_eq!(apply(&db, &listing)?, ReconcileOutcome::StateOnly);
        assert_eq!(db.indexer.pending.iter().count(), 1);
        let state = read_state(&db)?;
        assert!(state.active);
        assert_eq!(state.status, RepoStatus::Desynchronized);
        Ok(())
    }

    #[test]
    fn terminal_cursor_update_clears_resume_point() -> Result<()> {
        let (_tmp, db) = open_db()?;
        let key = b"crawler_cursor|relay".to_vec();
        db.cursors.insert(&key, b"old").into_diagnostic()?;
        let mut txn = crate::db::Txn::new(&db);
        stage_cursor(
            &mut txn.batch,
            &db,
            CursorUpdate {
                key: key.clone(),
                value: None,
            },
        );
        txn.commit()?;
        assert!(db.cursors.get(key).into_diagnostic()?.is_none());
        Ok(())
    }

    #[test]
    fn cursor_batch_can_advance_one_marker_and_clear_another() -> Result<()> {
        let (_tmp, db) = open_db()?;
        let old_key = b"relay_first_cursor|relay|hosts".to_vec();
        let marker_key = b"relay_first_cursor|relay|handled".to_vec();
        db.cursors.insert(&old_key, b"old").into_diagnostic()?;
        let mut txn = crate::db::Txn::new(&db);
        for cursor in [
            CursorUpdate {
                key: marker_key.clone(),
                value: Some(7u64.to_be_bytes().to_vec()),
            },
            CursorUpdate {
                key: old_key.clone(),
                value: None,
            },
        ] {
            stage_cursor(&mut txn.batch, &db, cursor);
        }
        txn.commit()?;

        assert_eq!(
            db.cursors.get(marker_key).into_diagnostic()?.as_deref(),
            Some(7u64.to_be_bytes().as_slice())
        );
        assert!(db.cursors.get(old_key).into_diagnostic()?.is_none());
        Ok(())
    }

    #[tokio::test]
    #[cfg(feature = "indexer_stream")]
    async fn changed_listing_emits_one_account_event() -> Result<()> {
        let tmp = tempfile::tempdir().into_diagnostic()?;
        let config = crate::config::Config {
            database_path: tmp.path().to_path_buf(),
            ..Default::default()
        };
        let state = Arc::new(AppState::new(&config)?);
        insert_repo(&state.db, &RepoState::backfilling(), 7, true)?;
        let stats = CrawlerStats::new(
            state.clone(),
            Vec::new(),
            crate::util::throttle::Throttler::new(1, 1),
        );
        let (mut worker, _tx) = CrawlerWorker::new(state.clone(), 100, 50, stats);
        let mut events = state.db.stream.event_tx.subscribe();
        let listing = RepoListing {
            did: did(),
            rev: rev(1),
            active: false,
            status: Some(ListedRepoStatus::Deactivated),
            create_if_missing: false,
        };

        worker
            .enqueue(CrawlerBatch {
                guards: Vec::new(),
                listings: vec![listing.clone()],
                cursor_updates: Vec::new(),
            })
            .await?;
        let crate::types::BroadcastEvent::Ephemeral(event) = events.recv().await.unwrap() else {
            panic!("expected account event");
        };
        let account = event.account.expect("account payload");
        assert!(!account.active);
        assert_eq!(account.status.as_deref(), Some("deactivated"));

        worker
            .enqueue(CrawlerBatch {
                guards: Vec::new(),
                listings: vec![listing],
                cursor_updates: Vec::new(),
            })
            .await?;
        assert!(events.try_recv().is_err());
        Ok(())
    }

    #[tokio::test]
    #[cfg(feature = "indexer_stream")]
    async fn missed_revision_does_not_emit_account_event() -> Result<()> {
        let tmp = tempfile::tempdir().into_diagnostic()?;
        let config = crate::config::Config {
            database_path: tmp.path().to_path_buf(),
            ..Default::default()
        };
        let state = Arc::new(AppState::new(&config)?);
        insert_repo(&state.db, &state_with_root(rev(1)), 7, false)?;
        let stats = CrawlerStats::new(
            state.clone(),
            Vec::new(),
            crate::util::throttle::Throttler::new(1, 1),
        );
        let (mut worker, _tx) = CrawlerWorker::new(state.clone(), 100, 50, stats);
        let mut events = state.db.stream.event_tx.subscribe();

        worker
            .enqueue(CrawlerBatch {
                guards: Vec::new(),
                listings: vec![RepoListing {
                    did: did(),
                    rev: rev(2),
                    active: true,
                    status: None,
                    create_if_missing: false,
                }],
                cursor_updates: Vec::new(),
            })
            .await?;

        assert!(events.try_recv().is_err());
        assert_eq!(state.db.indexer.pending.iter().count(), 1);
        Ok(())
    }
}