//! `summarize-stats` subcommand: scan every per-repo `RepoStats` row and
//! sum it into one set of grand totals.
//!
//! One `s<pds>\0<did>` row exists per successfully-fetched repo (it is
//! overwritten on each successful re-fetch), so these totals describe
//! the most recent capture of every `(pds, did)` pair we hold — the
//! same population `count-repos` reports as `total_pairs`.
//!
//! `total_wire_bytes` is the figure to compare against on-disk DB size:
//! it is the sum of raw HTTP body bytes received (post-`Content-Encoding`
//! is `car_bytes`; the rocksdb `records` CF then stores the *decoded*
//! blocks re-compressed with lz4/zstd, so on-disk < wire is expected).

use std::collections::BTreeMap;

use crate::storage::{Db, DbError, repo_stats};

#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error("storage: {0}")]
    Storage(#[from] DbError),
}

#[derive(Debug, Default)]
pub struct Summary {
    /// Number of `s/` rows = repos fetched at least once.
    pub repos: u64,
    /// Sum of raw HTTP body bytes (on-the-wire CAR size).
    pub total_wire_bytes: u64,
    /// Sum of decoded CAR bytes (post-`Content-Encoding`).
    pub total_car_bytes: u64,
    pub total_record_count: u64,
    pub total_backlink_count: u64,
    pub total_blob_ref_count: u64,
    /// Sum of declared blob `size` fields.
    pub total_blob_size: u64,
    pub total_unsized_blob_count: u64,
    pub total_blob_no_mime_count: u64,
    /// Repo count grouped by the `Content-Encoding` the server used
    /// (`"identity"` for none). Shows how much of the corpus arrived
    /// already-compressed, which explains a small on-disk footprint.
    pub by_encoding: BTreeMap<String, u64>,
}

pub fn run(db: &Db) -> Result<Summary, Error> {
    let rows = repo_stats::scan_all(db)?;
    let mut s = Summary {
        repos: rows.len() as u64,
        ..Summary::default()
    };
    for row in rows {
        let st = row.stats;
        s.total_wire_bytes += st.wire_bytes;
        s.total_car_bytes += st.car_bytes;
        s.total_record_count += st.record_count;
        s.total_backlink_count += st.backlink_count;
        s.total_blob_ref_count += st.blob_ref_count;
        s.total_blob_size += st.blob_total_size;
        s.total_unsized_blob_count += st.unsized_blob_count;
        s.total_blob_no_mime_count += st.blob_no_mime_count;
        let enc = st.content_encoding.unwrap_or_else(|| "identity".into());
        *s.by_encoding.entry(enc).or_insert(0) += 1;
    }
    Ok(s)
}

/// Render a byte count as a binary-prefixed string, e.g. `5.01 GiB`.
/// For human-readable logging alongside the exact byte total.
pub fn human_bytes(n: u64) -> String {
    const UNITS: [&str; 6] = ["B", "KiB", "MiB", "GiB", "TiB", "PiB"];
    let mut v = n as f64;
    let mut u = 0;
    while v >= 1024.0 && u < UNITS.len() - 1 {
        v /= 1024.0;
        u += 1;
    }
    if u == 0 {
        format!("{n} B")
    } else {
        format!("{v:.2} {}", UNITS[u])
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::storage::open_temporary;
    use crate::storage::repo_stats::RepoStats;
    use hubble_pds::Hostname;
    use rocksdb::WriteBatch;

    fn h(s: &str) -> Hostname {
        Hostname::new(s).unwrap()
    }

    #[test]
    fn human_bytes_scales() {
        assert_eq!(human_bytes(0), "0 B");
        assert_eq!(human_bytes(512), "512 B");
        assert_eq!(human_bytes(1024), "1.00 KiB");
        assert_eq!(human_bytes(5 * 1024 * 1024 * 1024), "5.00 GiB");
    }

    #[test]
    fn empty_db_is_all_zero() {
        let (db, _g) = open_temporary();
        let s = run(&db).unwrap();
        assert_eq!(s.repos, 0);
        assert_eq!(s.total_wire_bytes, 0);
        assert!(s.by_encoding.is_empty());
    }

    #[test]
    fn sums_across_rows_and_groups_by_encoding() {
        let (db, _g) = open_temporary();
        let a = h("a.example");
        let b = h("b.example");

        let r1 = RepoStats {
            content_encoding: Some("zstd".into()),
            wire_bytes: 1_000,
            car_bytes: 4_000,
            record_count: 10,
            backlink_count: 3,
            blob_ref_count: 2,
            blob_total_size: 500,
            unsized_blob_count: 1,
            blob_no_mime_count: 1,
            ..RepoStats::default()
        };
        let r2 = RepoStats {
            content_encoding: None,
            wire_bytes: 250,
            car_bytes: 250,
            record_count: 5,
            ..RepoStats::default()
        };
        let r3 = RepoStats {
            content_encoding: Some("zstd".into()),
            wire_bytes: 750,
            car_bytes: 2_000,
            record_count: 7,
            ..RepoStats::default()
        };

        let mut batch = WriteBatch::default();
        repo_stats::put_into(&mut batch, &db, &a, "did:plc:1", &r1).unwrap();
        repo_stats::put_into(&mut batch, &db, &a, "did:plc:2", &r2).unwrap();
        repo_stats::put_into(&mut batch, &db, &b, "did:plc:1", &r3).unwrap();
        db.inner.write(batch).unwrap();

        let s = run(&db).unwrap();
        assert_eq!(s.repos, 3);
        assert_eq!(s.total_wire_bytes, 2_000);
        assert_eq!(s.total_car_bytes, 6_250);
        assert_eq!(s.total_record_count, 22);
        assert_eq!(s.total_backlink_count, 3);
        assert_eq!(s.total_blob_ref_count, 2);
        assert_eq!(s.total_blob_size, 500);
        assert_eq!(s.total_unsized_blob_count, 1);
        assert_eq!(s.total_blob_no_mime_count, 1);
        assert_eq!(s.by_encoding.get("zstd"), Some(&2));
        assert_eq!(s.by_encoding.get("identity"), Some(&1));
    }
}