use crate::db::types::TrimmedDid;
use crate::db::{self, keys};
use crate::state::AppState;
use crate::types::{GaugeState, ResyncState};
use jacquard_common::types::did::Did;
use miette::{IntoDiagnostic, Result};

use std::sync::Arc;
use std::time::Duration;
use tracing::{debug, error, info};

fn queue_resync_if(
    state: &AppState,
    did: &Did,
    eligible: impl FnOnce(&ResyncState) -> bool,
) -> Result<bool> {
    let mut txn = db::Txn::new(&state.db);
    if txn.lock_repo_and_is_excluded(did)? {
        return Ok(false);
    }

    let did_key = keys::repo_key(did);
    let Some(value) = state.db.indexer.resync.get(&did_key).into_diagnostic()? else {
        return Ok(false);
    };
    let resync_state: ResyncState = rmp_serde::from_slice(&value).into_diagnostic()?;
    if !eligible(&resync_state) {
        return Ok(false);
    }

    let metadata_key = keys::repo_metadata_key(did);
    let metadata_bytes = state
        .db
        .repo_metadata
        .get(&metadata_key)
        .into_diagnostic()?
        .ok_or_else(|| miette::miette!("repo metadata not found"))?;
    let mut metadata = crate::db::deser_repo_meta(&metadata_bytes)?;
    let old_pending = keys::pending_key(metadata.index_id);
    metadata.index_id = rand::random();

    txn.batch.remove(&state.db.indexer.resync, &did_key);
    txn.batch.remove(&state.db.indexer.pending, old_pending);
    txn.batch.insert(
        &state.db.indexer.pending,
        keys::pending_key(metadata.index_id),
        &did_key,
    );
    txn.batch.insert(
        &state.db.repo_metadata,
        metadata_key,
        crate::db::ser_repo_meta(&metadata)?,
    );
    txn.transition_lifecycle(did, GaugeState::Pending)?;
    txn.commit()?;
    Ok(true)
}

pub fn queue_gone_backfills(state: &Arc<AppState>) -> Result<()> {
    debug!("scanning for deactivated/takendown repos to retry...");
    let mut candidates = Vec::new();
    for guard in state.db.indexer.resync.iter() {
        let (key, val) = guard.into_inner().into_diagnostic()?;
        let did = match TrimmedDid::try_from(key.as_ref()) {
            Ok(did) => did.to_did(),
            Err(e) => {
                error!(err = %e, "invalid did in db, skipping");
                continue;
            }
        };

        if let Ok(resync_state) = rmp_serde::from_slice::<ResyncState>(&val)
            && matches!(resync_state, ResyncState::Gone { .. })
        {
            candidates.push(did);
        }
    }

    let mut transitions = 0usize;
    for did in candidates {
        debug!(did = %did, "queuing retry for gone repo");
        match queue_resync_if(state, &did, |resync| {
            matches!(resync, ResyncState::Gone { .. })
        }) {
            Ok(true) => transitions += 1,
            Ok(false) => {}
            Err(e) => {
                error!(did = %did, err = %e, "failed to queue gone repo");
                db::check_poisoned_report(&e);
            }
        }
    }
    if transitions > 0 {
        state.notify_backfill();
        info!(count = transitions, "queued gone backfills");
    }
    Ok(())
}

pub fn retry_worker(state: Arc<AppState>) {
    let db = &state.db;
    info!("retry worker started");
    loop {
        std::thread::sleep(Duration::from_secs(60));

        let now = chrono::Utc::now().timestamp();
        let mut candidates = Vec::new();
        for guard in db.indexer.resync.iter() {
            let (key, value) = match guard.into_inner() {
                Ok(t) => t,
                Err(e) => {
                    error!(err = %e, "failed to get resync state");
                    db::check_poisoned(&e);
                    continue;
                }
            };
            let did = match TrimmedDid::try_from(key.as_ref()) {
                Ok(did) => did.to_did(),
                Err(e) => {
                    error!(err = %e, "invalid did in db, skipping");
                    continue;
                }
            };

            match rmp_serde::from_slice::<ResyncState>(&value) {
                Ok(ResyncState::Error { next_retry, .. }) if next_retry <= now => {
                    candidates.push(did);
                }
                Ok(_) => {}
                Err(e) => {
                    error!(did = %did, err = %e, "failed to deserialize resync state");
                }
            }
        }

        let mut transitions = 0usize;
        for did in candidates {
            let did_key = keys::repo_key(&did);
            let is_pds_throttled = db
                .repos
                .get(&did_key)
                .ok()
                .flatten()
                .and_then(|bytes| {
                    let repo = rmp_serde::from_slice::<crate::types::RepoState>(&bytes).ok()?;
                    repo.pds
                        .as_ref()
                        .and_then(|pds| url::Url::parse(pds.as_ref()).ok())
                })
                .is_some_and(|pds| {
                    state
                        .throttler
                        .snapshot(&pds)
                        .is_throttled(chrono::Utc::now().timestamp())
                });
            if is_pds_throttled {
                continue;
            }

            debug!(did = %did, "retrying backfill");
            match queue_resync_if(
                &state,
                &did,
                |resync| matches!(resync, ResyncState::Error { next_retry, .. } if *next_retry <= now),
            ) {
                Ok(true) => transitions += 1,
                Ok(false) => {}
                Err(e) => {
                    error!(did = %did, err = %e, "failed to queue retry");
                    db::check_poisoned_report(&e);
                }
            }
        }
        if transitions > 0 {
            state.notify_backfill();
            info!(count = transitions, "queued retries");
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::Config;
    use crate::types::{RepoMetadata, RepoState};

    fn state() -> (tempfile::TempDir, AppState) {
        let tmp = tempfile::tempdir().unwrap();
        let config = Config {
            database_path: tmp.path().to_path_buf(),
            ..Default::default()
        };
        let state = AppState::new(&config).unwrap();
        (tmp, state)
    }

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

    fn stage_retry(state: &AppState, excluded: bool) {
        let did = did();
        let mut batch = state.db.inner.batch();
        batch.insert(
            &state.db.repos,
            keys::repo_key(&did),
            crate::db::ser_repo_state(&RepoState::synced()).unwrap(),
        );
        batch.insert(
            &state.db.repo_metadata,
            keys::repo_metadata_key(&did),
            crate::db::ser_repo_meta(&RepoMetadata::backfilling(7)).unwrap(),
        );
        batch.insert(
            &state.db.indexer.resync,
            keys::repo_key(&did),
            rmp_serde::to_vec(&ResyncState::Gone {
                status: crate::types::RepoStatus::Deactivated,
            })
            .unwrap(),
        );
        if excluded {
            batch.insert(
                &state.db.filter,
                crate::db::filter::exclude_key(did.as_str()).unwrap(),
                [],
            );
        }
        batch.commit().unwrap();
    }

    #[test]
    fn excluded_repo_is_not_requeued() {
        let (_tmp, state) = state();
        stage_retry(&state, true);

        assert!(
            !queue_resync_if(&state, &did(), |resync| {
                matches!(resync, ResyncState::Gone { .. })
            })
            .unwrap()
        );
        assert!(
            state
                .db
                .indexer
                .resync
                .contains_key(keys::repo_key(&did()))
                .unwrap()
        );
        assert!(state.db.indexer.pending.is_empty().unwrap());
    }

    #[test]
    fn eligible_repo_moves_from_resync_to_pending() {
        let (_tmp, state) = state();
        stage_retry(&state, false);

        assert!(
            queue_resync_if(&state, &did(), |resync| {
                matches!(resync, ResyncState::Gone { .. })
            })
            .unwrap()
        );
        assert!(
            !state
                .db
                .indexer
                .resync
                .contains_key(keys::repo_key(&did()))
                .unwrap()
        );
        assert_eq!(state.db.indexer.pending.len().unwrap(), 1);
    }
}