//! `export-pairs` subcommand: write every captured `(did, host)` pair —
//! i.e. every repo whose CAR was successfully fetched and stored (a
//! `C<pds>\0<did>` commit row exists).
//!
//! Same population as `count-repos`' `total_pairs`. Output is TSV,
//! one `did\thost` line per captured repo, written to `--out` (a file
//! rather than stdout so the process's stdout logs don't interleave).
//! DIDs and hostnames never contain tabs or newlines, so TSV is safe
//! and trivially consumable with `cut` / `awk` / `sort`.
//!
//! Rows come out sorted by `host` then `did` (the `C/` key order from
//! [`commit::all_pairs`]).

use std::io::{BufWriter, Write};
use std::path::Path;

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

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

    #[error("io: {0}")]
    Io(#[from] std::io::Error),
}

/// Write all captured `(did, host)` pairs as TSV to `out_path`.
/// Returns the number of pairs written.
pub fn run(db: &Db, out_path: &Path) -> Result<u64, Error> {
    let pairs = commit::all_pairs(db)?; // Vec<(pds, did)>, C/-key sorted
    let file = std::fs::File::create(out_path)?;
    let mut w = BufWriter::new(file);
    let mut n = 0u64;
    for (pds, did) in pairs {
        writeln!(w, "{did}\t{pds}")?;
        n += 1;
    }
    w.flush()?;
    Ok(n)
}

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

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

    fn sample() -> commit::StoredCommit {
        commit::StoredCommit {
            did: "did:plc:x".into(),
            version: 3,
            data: "bafyreidkjxqphmt3yhvdgmwx5p64q7s7tcmw3rj7ynvtu7tnp7nb6yurxe".into(),
            rev: "3kabc".into(),
            prev: None,
            sig: vec![0xde, 0xad],
        }
    }

    #[test]
    fn empty_db_writes_empty_file() {
        let (db, _g) = open_temporary();
        let dir = tempfile::tempdir().unwrap();
        let out = dir.path().join("pairs.tsv");
        let n = run(&db, &out).unwrap();
        assert_eq!(n, 0);
        assert_eq!(std::fs::read_to_string(&out).unwrap(), "");
    }

    #[test]
    fn writes_did_tab_host_per_captured_repo() {
        let (db, _g) = open_temporary();
        let a = h("a.example");
        let b = h("b.example");
        let mut batch = WriteBatch::default();
        commit::put_into(&mut batch, &db, &a, "did:plc:1", &sample()).unwrap();
        commit::put_into(&mut batch, &db, &a, "did:plc:2", &sample()).unwrap();
        commit::put_into(&mut batch, &db, &b, "did:plc:1", &sample()).unwrap();
        db.inner.write(batch).unwrap();

        let dir = tempfile::tempdir().unwrap();
        let out = dir.path().join("pairs.tsv");
        let n = run(&db, &out).unwrap();
        assert_eq!(n, 3);

        let mut lines: Vec<String> = std::fs::read_to_string(&out)
            .unwrap()
            .lines()
            .map(|s| s.to_string())
            .collect();
        lines.sort();
        assert_eq!(
            lines,
            vec![
                "did:plc:1\ta.example",
                "did:plc:1\tb.example",
                "did:plc:2\ta.example",
            ]
        );
    }
}