//! v13 recounts the per-pds account counts from the repo rows the way v6 does, which repairs
//! a database whose counts drifted from the rows they count

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

use crate::db::Db;
use crate::types::v7;

use super::v6::{counted_host, recount_pds_accounts};

pub(super) fn recount_pds_account_counts(db: &Db, batch: &mut OwnedWriteBatch) -> Result<bool> {
    recount_pds_accounts(db, batch, |value| {
        let state: v7::RepoState = rmp_serde::from_slice(value)
            .into_diagnostic()
            .wrap_err("invalid v12 repo state")?;
        Ok(counted_host(state.active, state.pds.as_deref()))
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::Config;
    use crate::db::keys;
    use crate::db::migration::rewind_version_for_test;
    use jacquard_common::CowStr;
    use jacquard_common::types::did::Did;
    use tempfile::tempdir;

    fn counts_on_disk(db: &Db) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
        db.counts
            .prefix(keys::count_keyspace_key("p|"))
            .map(|guard| {
                let (key, value) = guard.into_inner().into_diagnostic()?;
                Ok((key.to_vec(), value.to_vec()))
            })
            .collect()
    }

    #[test]
    fn repairs_drifted_per_pds_counts_from_the_rows() -> Result<()> {
        let tmp = tempdir().into_diagnostic()?;
        let cfg = Config {
            database_path: tmp.path().to_path_buf(),
            ..Default::default()
        };
        let accounts = |db: &Db, host| db.get_count_sync(&keys::pds_account_count_key(host));
        {
            let db = Db::open(&cfg)?;
            let mut batch = db.inner.batch();
            let rows = [
                ("did:web:one.test", "https://pds.example/", true),
                ("did:web:two.test", "https://pds.example/", true),
                ("did:web:gone.test", "https://gone.example/", false),
                ("did:web:also-gone.test", "https://gone.example/", false),
                ("did:web:fresh.test", "https://fresh.example/", true),
            ];
            for (did, pds, active) in rows {
                let row = v7::RepoState {
                    active,
                    status: v7::RepoStatus::Desynchronized,
                    root: None,
                    last_message_time: None,
                    last_identity_time: None,
                    last_account_time: None,
                    last_updated_at: 0,
                    signing_key: None,
                    pds: Some(CowStr::Borrowed(pds)),
                    handle: None,
                };
                batch.insert(
                    &db.repos,
                    keys::repo_key(&Did::new_static(did).into_diagnostic()?),
                    rmp_serde::to_vec(&row).into_diagnostic()?,
                );
            }
            // what the writers that never counted left behind: one short, one host whose
            // rows all went inactive still counted, one never counted at all
            for (host, count) in [("pds.example", 1), ("gone.example", 2)] {
                crate::db::set_ks_count(&mut batch, &db, &keys::pds_account_count_key(host), count);
            }
            batch.insert(
                &db.counts,
                keys::count_delta_key(1_000_000, &keys::pds_account_count_key("gone.example")),
                3_i64.to_be_bytes(),
            );
            batch.commit().into_diagnostic()?;
            rewind_version_for_test(&db, 9)?;
            db.persist()?;
        }

        let repaired = {
            let db = Db::open(&cfg)?;
            assert_eq!(accounts(&db, "pds.example"), 2);
            assert_eq!(accounts(&db, "gone.example"), 0);
            assert_eq!(accounts(&db, "fresh.example"), 1);
            let on_disk = counts_on_disk(&db)?;
            rewind_version_for_test(&db, 9)?;
            db.persist()?;
            on_disk
        };

        let db = Db::open(&cfg)?;
        assert_eq!(counts_on_disk(&db)?, repaired);
        assert_eq!(accounts(&db, "pds.example"), 2);
        assert_eq!(accounts(&db, "gone.example"), 0);
        assert_eq!(accounts(&db, "fresh.example"), 1);
        Ok(())
    }
}