//! A profile's badges, `dev.knotless.getProfile#badgeView`: counted from
//! the index as the profile is asked for, and from where the account
//! lives, as its DID document and PDS say, looked up now and then. None is
//! stored in anyone's PDS, so none can be claimed by writing a record, and
//! every one comes back whenever the index is rebuilt and refilled.

use super::*;
use crate::index::{Home, Pulls};
use knotless::records::queue::{CONFLICTS, EJECTED, HALVED};

/// `dev.knotless.getProfile#badgeView`.
#[derive(Serialize, Debug, PartialEq)]
pub struct BadgeView {
    kind: &'static str,
    tier: usize,
    count: u64,
    #[serde(skip_serializing_if = "Option::is_none")]
    next: Option<u64>,
}

/// A badge earned once: its one tier.
const ONCE: &[u64] = &[1];

/// Each badge, in the order a profile shows them, and the counts its
/// tiers start at: by group, collaboration, the merge queue, packages and
/// stars, atproto, and those that can no longer be earned.
const BADGES: [(&str, &[u64]); 12] = [
    ("upstream", &[2, 16, 128, 1024]),
    ("second-pair", &[1, 10, 50, 250]),
    ("peacemaker", &[1, 5, 25, 100]),
    ("forked-and-fixed", ONCE),
    ("good-company", &[1, 10, 50, 200]),
    ("smooth-landing", &[1, 10, 50, 250]),
    ("bisected", ONCE),
    ("shipped", &[1, 10, 50, 200]),
    ("stargazer", &[16, 128, 512, 4096]),
    ("carry-your-own", ONCE),
    ("moved-house", ONCE),
    ("first-light", ONCE),
];

/// How many pulls a batch of the merge queue lands together for good
/// company.
const COMPANY: usize = 5;

/// The most accounts a PDS may hold and still be its account's own.
const OWN_PDS_HOLDS: usize = 5;

/// How long what was found of where an account lives is kept before it is
/// looked up again.
const HOME_KEPT_FOR: Duration = Duration::from_secs(24 * 60 * 60);

/// The accounts here before knotless 1.0, a DID a line, as the list stood
/// that day. Until it is written, every account with a repository is.
const FIRST_LIGHT: &str = include_str!("first_light.txt");

/// The badges `counts` earn, one for each of [`BADGES`] in turn: each at
/// the highest tier its count reaches, and none that reaches no tier.
fn badges(counts: [u64; 12]) -> Vec<BadgeView> {
    let earned = BADGES.iter().zip(counts).map(|((kind, at), count)| {
        let tier = at.iter().filter(|&&n| count >= n).count();
        let next = at.get(tier).copied();
        (tier > 0).then_some(BadgeView {
            kind,
            tier,
            count,
            next,
        })
    });
    earned.flatten().collect()
}

/// The badges `did` earned, counted from the index, and from `home`,
/// where its account lives.
pub(super) fn badges_of(index: &Index, did: &str, home: Home) -> Vec<BadgeView> {
    let (shipped, stars, repos) = index.badge_counts(did).unwrap_or_default();
    let mut c = Tally::default();
    for pulls in index.pulls_by(did).unwrap_or_default() {
        let items = forge::assemble(&pulls.repo, &pulls.owner, &pulls.entries);
        let ejected = ejections(&pulls);
        // How many each batch landed, read once the first merged pull
        // names one: a query for each batch read the repository's records
        // as many times over.
        let mut batches = None;
        for item in items.iter().filter(|i| i.author == did) {
            let ejected = ejected
                .get(item.uri.as_str())
                .map_or(&[][..], Vec::as_slice);
            c.authored(&pulls.owner, item, ejected, || {
                batches.get_or_insert_with(|| landed(index, &pulls))
            });
        }
    }
    let first = index.first_comments(did).unwrap_or_default();
    for pulls in index.pulls_commented(did).unwrap_or_default() {
        let items = forge::assemble(&pulls.repo, &pulls.owner, &pulls.entries);
        let merged = items.iter().filter(|i| i.state == "merged");
        let before = |i: &&Item| {
            first
                .get(&i.uri)
                .is_some_and(|f| reviewed(&pulls, i, f, did))
        };
        c.second_pair += merged.filter(before).count() as u64;
    }
    for pulls in index.pulls_from_forks(did).unwrap_or_default() {
        let items = forge::assemble(&pulls.repo, &pulls.owner, &pulls.entries);
        c.forked += items.iter().filter(|i| i.state == "merged").count() as u64;
    }
    badges([
        c.upstream,
        c.second_pair,
        c.peacemaker,
        c.forked,
        c.company,
        c.smooth,
        c.bisected,
        shipped,
        stars,
        home.own as u64,
        home.moved as u64,
        first_light(FIRST_LIGHT, did, repos) as u64,
    ])
}

/// What the pulls an account took part in count toward its badges.
#[derive(Default)]
struct Tally {
    upstream: u64,
    second_pair: u64,
    peacemaker: u64,
    forked: u64,
    company: u64,
    smooth: u64,
    bisected: u64,
}

impl Tally {
    /// Counts `item`, a pull the account opened on the repository `owner`
    /// keeps: merged into another's repository; landed by the merge
    /// queue, never turned away, or in a batch of many; turned away for a
    /// conflict and landed after all; or found by halving a batch.
    /// `ejected` says why the runner turned it away each time, and
    /// `batches` how many pulls each batch of its repository landed.
    fn authored<'a>(
        &mut self,
        owner: &str,
        item: &Item,
        ejected: &[&str],
        batches: impl FnOnce() -> &'a HashMap<String, usize>,
    ) {
        let merged = item.state == "merged";
        self.upstream += (merged && owner != item.author) as u64;
        let conflicted = ejected.iter().any(|r| r.contains(CONFLICTS));
        self.peacemaker += (merged && conflicted) as u64;
        self.bisected += ejected.iter().any(|r| r.contains(HALVED)) as u64;
        let Some(batch) = item.merged_batch.as_ref().filter(|_| merged) else {
            return;
        };
        self.smooth += ejected.is_empty() as u64;
        let landed = batches().get(batch).copied().unwrap_or_default();
        self.company += (landed >= COMPANY) as u64;
    }
}

/// Why the runner of the repository of `pulls` turned each pull away from
/// its merge queue, each time it did, by the pull's URI.
fn ejections(pulls: &Pulls) -> HashMap<&str, Vec<&str>> {
    let mut ejected: HashMap<&str, Vec<&str>> = HashMap::new();
    let queue = pulls
        .entries
        .iter()
        .filter(|e| e.uri.contains("/dev.knotless.queue/"));
    for e in queue {
        let turned = e.value["state"] == EJECTED && pulls.repo.is_runner(&e.author);
        if let Some(subject) = e.value["subject"].as_str().filter(|_| turned) {
            let reason = e.value["reason"].as_str().unwrap_or_default();
            ejected.entry(subject).or_default().push(reason);
        }
    }
    ejected
}

/// How many pulls the merge queue landed in each batch on the repository
/// of `pulls`, as those who may merge said, by the batch's tip.
fn landed(index: &Index, pulls: &Pulls) -> HashMap<String, usize> {
    let states = index.batch_states(&pulls.owner, &pulls.repo.key);
    let states = states.unwrap_or_default();
    let merged = states.iter().filter(|e| {
        let at = e.uri.rsplit('/').next().unwrap_or_default();
        let by = forge::allowed(&pulls.repo, &pulls.owner, "", &e.author, "merged", at);
        e.value["state"] == "merged" && by
    });
    let mut batches: HashMap<&str, HashSet<&str>> = HashMap::new();
    for e in merged {
        if let (Some(batch), Some(pull)) = (e.value["batch"].as_str(), e.value["subject"].as_str())
        {
            batches.entry(batch).or_default().insert(pull);
        }
    }
    let counted = batches
        .into_iter()
        .map(|(b, pulls)| (b.to_string(), pulls.len()));
    counted.collect()
}

/// Whether `did` first commented on `item`, at `first`, before it merged:
/// before the first time anyone else said it merged.
fn reviewed(pulls: &Pulls, item: &Item, first: &str, did: &str) -> bool {
    let merges = pulls.entries.iter().filter(|e| {
        e.uri.contains("/dev.knotless.state/")
            && e.value["subject"] == item.uri.as_str()
            && e.value["state"] == "merged"
            && e.author != did
    });
    let merged = merges
        .filter_map(|e| when(e.value["createdAt"].as_str()?))
        .min();
    let first = when(first);
    first
        .zip(merged)
        .is_some_and(|(first, merged)| first < merged)
}

/// A record's time, as it says it.
fn when(at: &str) -> Option<jiff::Timestamp> {
    at.parse().ok()
}

/// Whether `did` was here before knotless 1.0: as `list` says, once it is
/// written, and until then, whether it has any of `repos`.
fn first_light(list: &str, did: &str, repos: u64) -> bool {
    let listed: Vec<&str> = list
        .lines()
        .map(str::trim)
        .filter(|l| !l.is_empty())
        .collect();
    let unwritten = listed.is_empty() && repos > 0;
    unwritten || listed.contains(&did)
}

/// Where `remote`'s account lives, as its badges count it: as `index` last
/// found it, and looked up again once that is a day old. One never looked
/// up is looked up now.
pub(super) async fn home(app: &App, index: Arc<Index>, remote: &Remote) -> Home {
    let (did, pds) = (remote.did().to_string(), remote.identity.pds.clone());
    let Some((home, checked)) = index.home(&did).ok().flatten() else {
        return look(app.ctx.clone(), index, did, pds).await;
    };
    if now_secs() - checked >= HOME_KEPT_FOR.as_secs() as i64 {
        // Marked as looked up now, so the pages showing it meanwhile look
        // it up once between them.
        let _ = index.set_home(&did, home, now_secs());
        tokio::spawn(look(app.ctx.clone(), index, did, pds));
    }
    home
}

/// Looks up now where `did`'s account lives, at `pds`, and keeps what was
/// found in `index`. Neither its DID's history nor its PDS answering, it
/// keeps nothing, to be looked up again.
async fn look(
    ctx: knotless::env::Ctx,
    index: Arc<crate::index::Index>,
    did: String,
    pds: String,
) -> Home {
    let (history, holds) = tokio::join!(history(&ctx, &did), holds(&ctx, &pds));
    let answered = history.is_ok() || holds.is_ok();
    let home = Home {
        moved: history.is_ok_and(|log| moved(&log)),
        own: holds.is_ok_and(|repos| own(&repos, &did)),
    };
    let _ = answered.then(|| index.set_home(&did, home, now_secs()));
    home
}

/// The history of `did`'s DID document, as its directory keeps it: only a
/// did:plc has one.
async fn history(ctx: &knotless::env::Ctx, did: &str) -> anyhow::Result<serde_json::Value> {
    anyhow::ensure!(did.starts_with("did:plc:"), "{did} keeps no history");
    let url = format!("{}/{did}/log/audit", ctx.env.plc_url);
    let res = ctx.http.get(url).send().await?.error_for_status()?;
    Ok(res.json().await?)
}

/// Whether the history `log` of a DID document named more than one PDS:
/// the account moved house.
fn moved(log: &serde_json::Value) -> bool {
    let ops = log.as_array().into_iter().flatten();
    let ops = ops
        .filter(|e| e["nullified"] != true)
        .map(|e| &e["operation"]);
    let named = ops.filter_map(|op| {
        let pds = op["services"]["atproto_pds"]["endpoint"].as_str();
        pds.or(op["service"].as_str())
    });
    let homes: HashSet<&str> = named.map(|p| p.trim_end_matches('/')).collect();
    homes.len() > 1
}

/// The first few accounts `pds` holds, as it lists them: one more than an
/// account's own may hold.
async fn holds(ctx: &knotless::env::Ctx, pds: &str) -> anyhow::Result<Vec<String>> {
    #[derive(Deserialize)]
    struct Repos {
        repos: Vec<Listed>,
    }
    #[derive(Deserialize)]
    struct Listed {
        did: String,
    }
    let xrpc = knotless::xrpc::Xrpc::new(ctx.http.clone(), pds);
    let limit = (OWN_PDS_HOLDS + 1).to_string();
    let params = [("limit", limit.as_str())];
    let listed: Repos = xrpc
        .query("com.atproto.sync.listRepos", &params, None)
        .await?;
    Ok(listed.repos.into_iter().map(|r| r.did).collect())
}

/// Whether a PDS holding `repos` is `did`'s own: it holds `did`, and only a
/// few accounts besides.
fn own(repos: &[String], did: &str) -> bool {
    repos.len() <= OWN_PDS_HOLDS && repos.iter().any(|r| r == did)
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[test]
    fn badges_take_the_highest_tier_reached() {
        let shown = |counts| {
            badges(counts)
                .into_iter()
                .map(|b| (b.kind, b.tier, b.next))
                .collect::<Vec<_>>()
        };
        assert_eq!(shown([1, 0, 0, 0, 0, 0, 0, 0, 15, 0, 0, 0]), []);
        assert_eq!(
            shown([2, 0, 0, 3, 0, 0, 0, 10, 4096, 0, 0, 1]),
            [
                ("upstream", 1, Some(16)),
                ("forked-and-fixed", 1, None),
                ("shipped", 2, Some(50)),
                ("stargazer", 4, None),
                ("first-light", 1, None)
            ]
        );
    }

    #[test]
    fn first_light_is_every_account_with_a_repository_until_the_list() {
        assert!(first_light("", "did:a", 1));
        assert!(!first_light("\n", "did:a", 0));
        let list = "did:a\n did:b \n";
        assert!(first_light(list, "did:b", 0));
        assert!(!first_light(list, "did:c", 3));
    }

    #[test]
    fn moved_house_is_a_history_naming_two_pdses() {
        let op = |pds: &str| json!({"operation": {"services": {"atproto_pds": {"endpoint": pds}}}});
        let legacy = json!({"operation": {"type": "create", "service": "https://a.example/"}});
        assert!(!moved(&json!([legacy, op("https://a.example")])));
        assert!(moved(&json!([legacy, op("https://b.example")])));
        // An operation undone does not count.
        let undone = json!({"nullified": true, "operation": {"services": {"atproto_pds": {"endpoint": "https://b.example"}}}});
        assert!(!moved(&json!([op("https://a.example"), undone])));
        assert!(!moved(&json!({"not": "a log"})));
    }

    #[test]
    fn carry_your_own_is_a_pds_of_a_few_accounts() {
        let repos = |n: usize| (0..n).map(|i| format!("did:{i}")).collect::<Vec<_>>();
        assert!(own(&repos(1), "did:0"));
        assert!(own(&repos(5), "did:4"));
        assert!(!own(&repos(6), "did:0"));
        assert!(!own(&repos(2), "did:x"));
    }

    #[tokio::test]
    async fn a_home_not_found_is_kept_for_no_one() {
        let dir = tempfile::tempdir().unwrap();
        let index = Arc::new(Index::open(&dir.path().join("i.db")).unwrap());
        let nowhere = "http://127.0.0.1:1";
        let env =
            knotless::env::Env::from_vars(|k| (k == "KNOTLESS_PLC_URL").then(|| nowhere.into()));
        let ctx = knotless::env::Ctx::new(env);
        for did in ["did:plc:a", "did:web:a.example"] {
            let home = look(ctx.clone(), index.clone(), did.into(), nowhere.into()).await;
            assert_eq!(home, Home::default());
            assert_eq!(index.home(did).unwrap(), None);
        }
    }

    /// An index with did:a's repository `r`, whose runner is did:ci.
    fn index() -> (tempfile::TempDir, Index, serde_json::Value) {
        let dir = tempfile::tempdir().unwrap();
        let index = Index::open(&dir.path().join("i.db")).unwrap();
        let owned = json!({"name": "r", "key": "k", "createdAt": "2026-01-01T00:00:00Z",
            "runners": [{"did": "did:ci"}]});
        index.put("did:a", nsid::REPO, "r", &owned).unwrap();
        let repo = json!({"uri": "at://did:a/dev.knotless.repo/r", "key": "k"});
        (dir, index, repo)
    }

    fn pull(index: &Index, who: &str, n: &str, head: &serde_json::Value) {
        let repo = json!({"uri": "at://did:a/dev.knotless.repo/r", "key": "k"});
        let pull = json!({"repo": repo, "title": n, "base": "main",
            "head": {"repo": head, "branch": "topic"}, "createdAt": "2026-01-02T00:00:00Z"});
        index.put(who, nsid::PULL, n, &pull).unwrap();
    }

    fn state(index: &Index, who: &str, rkey: &str, subject: &str, extra: serde_json::Value) {
        let repo = json!({"uri": "at://did:a/dev.knotless.repo/r", "key": "k"});
        let mut s = json!({"repo": repo, "state": "merged", "commit": "c",
            "subject": subject, "createdAt": "2026-01-03T00:00:00Z"});
        s.as_object_mut()
            .unwrap()
            .extend(extra.as_object().unwrap().clone());
        index.put(who, nsid::STATE, rkey, &s).unwrap();
    }

    fn ids(index: &Index, did: &str) -> Vec<(&'static str, u64)> {
        badges_of(index, did, Home::default())
            .iter()
            .map(|b| (b.kind, b.count))
            .collect()
    }

    #[test]
    fn badges_count_pulls_merged_elsewhere() {
        let (_dir, index, repo) = index();
        for n in ["1", "2", "3"] {
            pull(&index, "did:b", n, &repo);
        }
        let uri = |n: &str| format!("at://did:b/dev.knotless.pull/{n}");
        state(&index, "did:a", "3kaaaaaaaaaa2", &uri("1"), json!({}));
        state(&index, "did:a", "3kaaaaaaaaaa3", &uri("2"), json!({}));
        // Only the owner's word merges: the author's own does not.
        state(&index, "did:b", "3kaaaaaaaaaa4", &uri("3"), json!({}));
        assert_eq!(ids(&index, "did:b"), [("upstream", 2)]);
        // The owner's own pulls are not upstream of anyone; its repository
        // is here before 1.0.
        assert_eq!(ids(&index, "did:a"), [("first-light", 1)]);
        // Releases, and stars on its most starred repository, each
        // account's once.
        let release = json!({"repo": "k", "package": "p", "version": "1.0.0", "files": [],
            "createdAt": "2026-01-01T00:00:00Z"});
        index.put("did:a", nsid::RELEASE, "v", &release).unwrap();
        let star = json!({"subject": repo, "createdAt": "2026-01-04T00:00:00Z"});
        for (who, rkey) in [("did:b", "1"), ("did:b", "2"), ("did:c", "1")] {
            index.put(who, nsid::STAR, rkey, &star).unwrap();
        }
        assert_eq!(index.badge_counts("did:a").unwrap(), (1, 2, 1));
        assert_eq!(ids(&index, "did:a"), [("shipped", 1), ("first-light", 1)]);
        let home = Home {
            moved: true,
            own: true,
        };
        let kinds: Vec<_> = badges_of(&index, "did:c", home)
            .iter()
            .map(|b| b.kind)
            .collect();
        assert_eq!(kinds, ["carry-your-own", "moved-house"]);
    }

    #[test]
    fn badges_count_what_the_merge_queue_did() {
        let (_dir, index, repo) = index();
        let uri = |n: &str| format!("at://did:a/dev.knotless.pull/{n}");
        let queue = |who: &str, rkey: &str, n: &str, reason: &str| {
            let q = json!({"repo": repo, "subject": uri(n), "state": "ejected",
                "reason": reason, "createdAt": "2026-01-02T12:00:00Z"});
            index.put(who, nsid::QUEUE, rkey, &q).unwrap();
        };
        // Pulls 1 to 5 land in one batch, 6 alone; 7 is turned away.
        for n in ["1", "2", "3", "4", "5", "6", "7"] {
            pull(&index, "did:a", n, &repo);
        }
        for (rkey, n) in ["1", "2", "3", "4", "5"].iter().enumerate() {
            let rkey = format!("3kaaaaaaaaab{rkey}");
            state(
                &index,
                "did:ci",
                &rkey,
                &uri(n),
                json!({"batch": "b".repeat(40)}),
            );
        }
        state(
            &index,
            "did:ci",
            "3kaaaaaaaaac2",
            &uri("6"),
            json!({"batch": "d".repeat(40)}),
        );
        // One not allowed to merge saying 6 landed with the five does not
        // make its batch five.
        let b = json!({"batch": "d".repeat(40)});
        state(&index, "did:x", "3kaaaaaaaaac3", &uri("1"), b);
        // Pull 1 was turned away for a conflict before it landed.
        let conflict = format!("it does not apply on main at 0123: {CONFLICTS} a.rs");
        queue("did:ci", "3kaaaaaaaaad1", "1", &conflict);
        // Pull 7 was found by halving a batch, and not merged; a record by
        // anyone but the runner says nothing.
        queue(
            "did:ci",
            "3kaaaaaaaaad2",
            "7",
            &format!("its batch failed. {HALVED}"),
        );
        queue(
            "did:x",
            "3kaaaaaaaaad3",
            "6",
            &format!("its batch failed. {HALVED}"),
        );
        assert_eq!(
            ids(&index, "did:a"),
            [
                ("peacemaker", 1),
                ("good-company", 5),
                ("smooth-landing", 5),
                ("bisected", 1),
                ("first-light", 1)
            ]
        );
    }

    #[test]
    fn badges_count_reviews_and_pulls_from_forks() {
        let (_dir, index, repo) = index();
        let fork = json!({"name": "r", "key": "f", "createdAt": "2026-01-01T00:00:00Z",
            "forkOf": repo});
        index.put("did:b", nsid::REPO, "f", &fork).unwrap();
        let forked = json!({"uri": "at://did:b/dev.knotless.repo/f", "key": "f"});
        pull(&index, "did:b", "1", &forked);
        pull(&index, "did:b", "2", &forked);
        pull(&index, "did:b", "3", &repo);
        let uri = |n: &str| format!("at://did:b/dev.knotless.pull/{n}");
        state(&index, "did:a", "3kaaaaaaaaaa2", &uri("1"), json!({}));
        state(&index, "did:a", "3kaaaaaaaaaa3", &uri("3"), json!({}));
        let comment = |who: &str, rkey: &str, n: &str, at: &str| {
            let c = json!({"repo": repo, "subject": uri(n), "body": "lgtm", "createdAt": at});
            index.put(who, nsid::COMMENT, rkey, &c).unwrap();
        };
        // did:c read 1 before it merged and 3 only after; 2 never merged.
        comment("did:c", "1", "1", "2026-01-02T10:00:00Z");
        comment("did:c", "2", "1", "2026-01-04T10:00:00Z");
        comment("did:c", "3", "3", "2026-01-05T00:00:00Z");
        comment("did:c", "4", "2", "2026-01-02T10:00:00Z");
        assert_eq!(ids(&index, "did:c"), [("second-pair", 1)]);
        // did:a's repository had a pull from its fork merged; 3 came from
        // its own repository, not a fork.
        assert_eq!(
            ids(&index, "did:a"),
            [("forked-and-fixed", 1), ("first-light", 1)]
        );
    }
}