use fjall::{Keyspace, OwnedWriteBatch};
use miette::{IntoDiagnostic, Result};

use crate::db::Db;
use crate::pds_discovery::canonical_pds_host;
use crate::pds_meta::HostStatus;

use super::{ChunkBudget, Pass};

const STATUS_PREFIX: &[u8] = b"pds|status|";

/// v11 splits operator bans from observed status. existing durable `Banned`
/// statuses are treated as operator bans, since that is how the pre-v11 value
/// was enforced; the old status row is removed so `unban` can clear the ban.
///
/// bans move first, then remaining status aliases canonicalize.
pub(super) const PASSES: &[Pass] = &[
    Pass {
        name: "pds_bans",
        scan: "filter",
        visit: rewrite_ban,
        budget: ChunkBudget::DEFAULT,
        optional_on_absent: false,
    },
    Pass {
        name: "pds_statuses",
        scan: "filter",
        visit: rewrite_status,
        budget: ChunkBudget::DEFAULT,
        optional_on_absent: false,
    },
];

fn canonical_host(key: &[u8]) -> Option<(String, bool)> {
    let raw = key.strip_prefix(STATUS_PREFIX)?;
    let host = std::str::from_utf8(raw).ok()?;
    let canonical = canonical_pds_host(host).ok()?;
    Some((
        canonical.as_str().to_string(),
        canonical.as_str().as_bytes() == raw,
    ))
}

fn status(value: &[u8]) -> Result<HostStatus> {
    rmp_serde::from_slice(value).into_diagnostic()
}

fn rewrite_ban(
    _db: &Db,
    batch: &mut OwnedWriteBatch,
    scanned: &Keyspace,
    key: &[u8],
    value: &[u8],
) -> Result<usize> {
    let Some((host, _)) = canonical_host(key) else {
        return Ok(0);
    };
    if status(value)? != HostStatus::Banned {
        return Ok(0);
    }

    let ban_key = crate::db::pds_meta::pds_ban_key(&host);
    batch.insert(scanned, ban_key.clone(), []);
    batch.remove(scanned, key);
    Ok(ban_key.len())
}

fn rewrite_status(
    _db: &Db,
    batch: &mut OwnedWriteBatch,
    scanned: &Keyspace,
    key: &[u8],
    value: &[u8],
) -> Result<usize> {
    let Some((host, already_canonical)) = canonical_host(key) else {
        return Ok(0);
    };
    if already_canonical || status(value)? == HostStatus::Banned {
        return Ok(0);
    }

    let new_key = crate::db::pds_meta::pds_status_key(&host);
    batch.insert(scanned, new_key.clone(), value);
    batch.remove(scanned, key);
    Ok(new_key.len() + value.len())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::Config;
    use crate::db::migration::rewind_version_for_test;
    use tempfile::tempdir;

    #[test]
    fn splits_legacy_bans_before_canonicalizing_other_status_aliases() -> Result<()> {
        let tmp = tempdir().into_diagnostic()?;
        let cfg = Config {
            database_path: tmp.path().to_path_buf(),
            ..Default::default()
        };
        {
            let db = Db::open(&cfg)?;
            let mut batch = db.inner.batch();
            let raw = crate::db::pds_meta::pds_status_key("BANNED.EXAMPLE.");
            let canonical = crate::db::pds_meta::pds_status_key("banned.example");
            batch.insert(
                &db.filter,
                raw,
                rmp_serde::to_vec(&HostStatus::Banned).into_diagnostic()?,
            );
            batch.insert(
                &db.filter,
                canonical,
                rmp_serde::to_vec(&HostStatus::Active).into_diagnostic()?,
            );
            batch.commit().into_diagnostic()?;
            rewind_version_for_test(&db, 9)?;
            db.persist()?;
        }

        let db = Db::open(&cfg)?;
        assert!(
            !db.filter
                .contains_key(crate::db::pds_meta::pds_status_key("BANNED.EXAMPLE."))
                .into_diagnostic()?
        );
        let observed = db
            .filter
            .get(crate::db::pds_meta::pds_status_key("banned.example"))
            .into_diagnostic()?
            .expect("non-ban status must remain canonical");
        assert_eq!(status(observed.as_ref())?, HostStatus::Active);
        assert!(
            db.filter
                .contains_key(crate::db::pds_meta::pds_ban_key("banned.example"))
                .into_diagnostic()?
        );
        Ok(())
    }
}