//! Triages repositories' issues: has a classifier choose each open one's
//! labels, and writes them as the triager's own `dev.knotless.label`
//! records, with what it judged.
//!
//! The classifier is a decision model (`--decide`, see [`decide`]), as
//! Jev or Clef, or any command, run by `sh -c` (`--classify`): it reads an issue as JSON on stdin, its
//! `subject`, `author`, `title` and `body` and the `labels` to choose
//! among, and writes `{"labels": [...], "model": ..., "reason": ...}` on
//! stdout. Either has `--timeout` to answer. It is asked only of an open
//! issue not judged
//! at its current version, and only for the scopes asked for (`--scopes`)
//! where nobody else has put a label on, so it never second-guesses a
//! person. One that fails, or answers nothing readable, is noted and asked
//! again at the next read.
//!
//! Jetstream is only the doorbell, as it is for knotless-run: an issue
//! written about a watched repository, or the repository's record, sends
//! it to read the repository again once `--settle` has passed, for the
//! appview to index the issue; it reads every repository on connecting,
//! and every `--sweep` besides. `--once` reads each once, and stops.
//!
//! Its labels count only once the owner names its account among the
//! repository's triagers, for those scopes. Until then they count for
//! nothing, and `knotless triager eval <url> <handle>` says how often
//! they agree with people's.
//!
//! With `--discover` it triages, besides the repositories named, every one
//! whose record names it among its triagers, in the scopes it names it a
//! triager of, as knotless-run finds those naming it among their runners:
//! as the appview lists them and private ones' invites say, at the start
//! and every `--sweep`, and as Jetstream brings a repository's record. One
//! that no longer names it is left.

pub mod decide;

use anyhow::{Context, Result, bail};
use decide::Decider;
use knotless::env::Ctx;
use knotless::forge;
use knotless::jetstream;
use knotless::records::nsid;
use knotless::remote::Remote;
use knotless::session;
use knotless::telemetry::{self, Fields, Level};
use knotless::triage::{Choice, Pending, Triage};
use knotless::url::{RepoUrl, SCHEME};
use knotless::xrpc::Xrpc;
use serde::Deserialize;
use std::collections::BTreeMap;
use std::process::Stdio;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::io::AsyncWriteExt;
use tokio::sync::Notify;

const USAGE: &str = "usage: knotless-triage (--decide <url> --model <name> [--threshold <p>] | --classify <command>) --scopes <scope>[,<scope>...] [--jetstream <wss://host>] [--sweep <secs>] [--settle <secs>] [--timeout <secs>] [--once] [--discover] [<handle>/<name>...]";

/// What chooses an issue's labels.
#[derive(Clone, PartialEq)]
pub enum Classifier {
    Decide(Decider),
    /// A command for `sh -c`.
    Command(String),
}

#[derive(Clone)]
pub struct Options {
    pub classify: Classifier,
    /// The scopes of labels it chooses, as `kind` for `kind/bug`.
    pub scopes: Vec<String>,
    pub jetstream: String,
    /// How often every repository is read, whatever Jetstream says.
    pub sweep: Duration,
    /// How long after a ring the repository is read.
    pub settle: Duration,
    /// How long the classifier has for an issue.
    pub timeout: Duration,
    pub once: bool,
    /// Whether to triage, besides `repos`, every repository that names
    /// the triager among its triagers.
    pub discover: bool,
    pub repos: Vec<String>,
}

/// The options `args` give, with a decision model's token from `env`:
/// `KNOTLESS_DECISION_TOKEN`.
pub fn parse_args(
    args: &[String],
    env: &dyn Fn(&str) -> Option<String>,
) -> Result<Options, String> {
    let (mut command, mut endpoint, mut model, mut threshold) = (None, None, None, 0.5);
    let mut opts = Options {
        classify: Classifier::Command(String::new()),
        scopes: Vec::new(),
        jetstream: "wss://jetstream2.us-east.bsky.network".into(),
        sweep: Duration::from_secs(600),
        settle: Duration::from_secs(5),
        timeout: Duration::from_secs(120),
        once: false,
        discover: false,
        repos: Vec::new(),
    };
    let mut it = args.iter().skip(1);
    while let Some(flag) = it.next() {
        let mut value = || it.next().cloned().ok_or(format!("{flag} takes a value"));
        match flag.as_str() {
            "--classify" => command = Some(value()?),
            "--decide" => endpoint = Some(value()?),
            "--model" => model = Some(value()?),
            "--threshold" => {
                let p = value()?.parse().ok().filter(|p| (0.0..=1.0).contains(p));
                threshold = p.ok_or("--threshold takes a probability, from 0 to 1")?;
            }
            "--scopes" => {
                let scopes = value()?;
                let scopes = scopes.split(',').filter(|s| !s.is_empty());
                opts.scopes = scopes.map(String::from).collect();
            }
            "--jetstream" => opts.jetstream = value()?,
            "--sweep" => opts.sweep = secs(flag, value()?)?,
            "--settle" => opts.settle = secs(flag, value()?)?,
            "--timeout" => opts.timeout = secs(flag, value()?)?,
            "--once" => opts.once = true,
            "--discover" => opts.discover = true,
            other if other.starts_with('-') => return Err(format!("unexpected {other}\n{USAGE}")),
            repo => opts.repos.push(repo.trim_start_matches(SCHEME).to_string()),
        }
    }
    if opts.scopes.is_empty() || (opts.repos.is_empty() && !opts.discover) {
        return Err(format!(
            "name its scopes and a repository, or --discover\n{USAGE}"
        ));
    }
    opts.classify = match (command, endpoint) {
        (Some(command), None) => Classifier::Command(command),
        (None, Some(endpoint)) => Classifier::Decide(Decider {
            endpoint,
            model: model.ok_or("--decide takes --model")?,
            token: env("KNOTLESS_DECISION_TOKEN")
                .ok_or("--decide takes KNOTLESS_DECISION_TOKEN")?,
            threshold,
        }),
        _ => {
            return Err(format!(
                "name one classifier, --decide or --classify\n{USAGE}"
            ));
        }
    };
    Ok(opts)
}

fn secs(flag: &str, value: String) -> Result<Duration, String> {
    let secs = value.parse().map_err(|_| format!("{flag} takes seconds"))?;
    Ok(Duration::from_secs(secs))
}

/// A repository it triages, by its name, and whether it was found naming
/// the triager rather than named on the command line.
#[derive(Clone)]
struct Watch {
    url: RepoUrl,
    found: bool,
}

/// What [`run`] works with.
struct Triager {
    ctx: Ctx,
    opts: Options,
    /// The triager's DID.
    me: String,
    /// What it watches, by owner and key: what records about each name.
    watched: Mutex<BTreeMap<(String, String), Watch>>,
    door: Notify,
}

pub async fn run(opts: Options, ctx: Ctx) -> Result<()> {
    if opts.discover && ctx.env.appview.is_none() {
        bail!(
            "--discover finds repositories through an appview, and KNOTLESS_APPVIEW is set empty"
        );
    }
    let me = session::me(&ctx).await?.did;
    let t = Arc::new(Triager {
        ctx,
        opts,
        me,
        watched: Mutex::default(),
        door: Notify::new(),
    });
    for repo in &t.opts.repos {
        let url = RepoUrl::parse(&format!("{SCHEME}{repo}"))?;
        let id = Triage::open(&t.ctx, &url).await?.repo();
        let watch = Watch { url, found: false };
        t.watched.lock().unwrap().insert(id, watch);
    }
    if t.opts.once {
        if t.opts.discover {
            t.discover().await;
        }
        for watch in t.watching() {
            t.sweep(&watch).await?;
        }
        return Ok(());
    }
    tokio::spawn(follow(t.clone()));
    if t.opts.discover {
        tokio::spawn(discover(t.clone()));
    }
    loop {
        tokio::select! {
            _ = t.door.notified() => tokio::time::sleep(t.opts.settle).await,
            _ = tokio::time::sleep(t.opts.sweep) => {}
        }
        for watch in t.watching() {
            failed(&watch.url, t.sweep(&watch).await);
        }
    }
}

impl Triager {
    /// What it watches now.
    fn watching(&self) -> Vec<Watch> {
        self.watched.lock().unwrap().values().cloned().collect()
    }

    /// Has the classifier judge each of `watch`'s issues waiting for it,
    /// in the scopes asked for: of one found, only those the repository
    /// names it a triager of. One found that no longer names it, or that
    /// cannot be read, is left: discovery finds it again if it still does.
    async fn sweep(&self, watch: &Watch) -> Result<()> {
        let scopes = &self.opts.scopes;
        let opened = Triage::open(&self.ctx, &watch.url).await;
        if !watch.found {
            return judge(&self.ctx, &self.opts, &opened?, scopes).await;
        }
        let granted = opened.as_ref().ok().and_then(|t| t.granted(scopes));
        let (Ok(triage), Some(granted)) = (opened, granted) else {
            self.watched
                .lock()
                .unwrap()
                .retain(|_, w| w.url != watch.url);
            said(&watch.url, "left");
            return Ok(());
        };
        judge(&self.ctx, &self.opts, &triage, &granted).await
    }

    /// Watches `did`'s repository `name` if its record names this
    /// triager and it is not watched already by that name: by another, it
    /// was renamed, and is watched by this one instead.
    async fn consider(&self, did: &str, name: &str) -> Result<()> {
        let url = RepoUrl {
            authority: did.to_string(),
            name: name.to_string(),
        };
        let remote = Remote::open(&self.ctx, &url).await?;
        let Some(repo) = remote.repo().await? else {
            return Ok(());
        };
        if repo.value.triager(&self.me).is_none() {
            return Ok(());
        }
        let url = RepoUrl {
            authority: did.to_string(),
            name: repo.rkey,
        };
        {
            let mut watched = self.watched.lock().unwrap();
            let id = (did.to_string(), repo.value.key);
            // One on the command line is watched as it is.
            if watched.get(&id).is_some_and(|w| !w.found || w.url == url) {
                return Ok(());
            }
            let watch = Watch {
                url: url.clone(),
                found: true,
            };
            watched.insert(id, watch);
        }
        said(&url, "found");
        self.door.notify_one();
        Ok(())
    }

    /// Considers each repository the appview lists as naming this
    /// triager, and each private one that invited it.
    async fn discover(&self) {
        let listed = self.listed().await;
        let invited = self.invited().await;
        for found in [listed, invited] {
            match found {
                Ok(found) => {
                    for (did, name) in found {
                        let url = RepoUrl {
                            authority: did.clone(),
                            name: name.clone(),
                        };
                        failed(&url, self.consider(&did, &name).await);
                    }
                }
                Err(err) => {
                    let f = Fields {
                        error: Some(format!("{err:#}")),
                        ..Default::default()
                    };
                    telemetry::note(Level::WARN, "finding repositories failed", &f);
                }
            }
        }
    }

    /// The repositories the appview lists as naming this triager, by
    /// owner and name.
    async fn listed(&self) -> Result<Vec<(String, String)>> {
        let appview = self.ctx.env.appview.clone().unwrap_or_default();
        let xrpc = Xrpc::new(self.ctx.http.clone(), &appview);
        let mut found = Vec::new();
        // The page to ask for next, if any: the first, from no cursor.
        let mut next: Option<Option<String>> = Some(None);
        while let Some(cursor) = next {
            let mut params = vec![("triager", self.me.as_str()), ("limit", "500")];
            params.extend(std::iter::repeat("cursor").zip(cursor.as_deref()));
            let page: Listing = xrpc.query("dev.knotless.listRepos", &params, None).await?;
            found.extend(page.repos.into_iter().map(|r| (r.did, r.name)));
            next = page.cursor.map(Some);
        }
        Ok(found)
    }

    /// The private repositories that invited this triager, all of them:
    /// it keeps nothing, so reads them from the first each time, and one
    /// left a while ago is found again.
    async fn invited(&self) -> Result<Vec<(String, String)>> {
        let me = session::me(&self.ctx).await?;
        let mut found = Vec::new();
        let mut cursor = None;
        loop {
            let page = knotless::invites::list(&self.ctx, &me, cursor.as_deref()).await?;
            if page.invites.is_empty() {
                return Ok(found);
            }
            found.extend(page.invites.into_iter().map(|i| (i.did, i.name)));
            cursor = page.cursor;
        }
    }

    /// Whether `event` writes an issue about a repository it watches, or a
    /// watched owner's repository record, as when labels are defined.
    fn rings(&self, event: &jetstream::Event) -> bool {
        let watched = self.watched.lock().unwrap();
        let did = event.did.as_deref().unwrap_or_default();
        event.ops.iter().any(|op| match op.collection.as_str() {
            nsid::ISSUE => (op.record.as_ref()).is_some_and(|r| {
                watched
                    .keys()
                    .any(|(owner, key)| forge::is_about(r, owner, key))
            }),
            nsid::REPO => watched.keys().any(|(owner, _)| owner == did),
            _ => false,
        })
    }

    /// With `--discover`, considers each repository whose record `event`
    /// writes, which may name this triager now.
    fn found(self: &Arc<Self>, event: &jetstream::Event) {
        let Some(did) = event.did.clone().filter(|_| self.opts.discover) else {
            return;
        };
        let written = event.ops.iter().filter(|op| op.record.is_some());
        for op in written.filter(|op| op.collection == nsid::REPO) {
            let (t, did, name) = (self.clone(), did.clone(), op.rkey.clone());
            tokio::spawn(async move {
                let url = RepoUrl {
                    authority: did.clone(),
                    name: name.clone(),
                };
                failed(&url, t.consider(&did, &name).await);
            });
        }
    }
}

/// Considers, at the start and every `--sweep`, each repository that
/// names the triager: Jetstream brings public ones named meanwhile
/// ([`Triager::found`]), but not always, and never private ones.
async fn discover(t: Arc<Triager>) {
    loop {
        t.discover().await;
        tokio::time::sleep(t.opts.sweep).await;
    }
}

/// Logs `message` of `url`.
fn said(url: &RepoUrl, message: &str) {
    let repo = url.to_string();
    let f = Fields {
        repo: Some(&repo),
        ..Default::default()
    };
    telemetry::note(Level::INFO, message, &f);
}

/// Logs that reading `url` failed, if `done` says it did.
fn failed(url: &RepoUrl, done: Result<()>) {
    if let Err(err) = done {
        let repo = url.to_string();
        let f = Fields {
            repo: Some(&repo),
            error: Some(format!("{err:#}")),
            ..Default::default()
        };
        telemetry::note(Level::WARN, "reading the repository failed", &f);
    }
}

/// A page of the appview's `dev.knotless.listRepos`, as discovery reads it.
#[derive(Deserialize)]
struct Listing {
    repos: Vec<Listed>,
    cursor: Option<String>,
}

#[derive(Deserialize)]
struct Listed {
    did: String,
    name: String,
}

/// Has the classifier judge each of `triage`'s issues waiting for it in
/// `scopes`, and writes what it chose.
async fn judge(ctx: &Ctx, opts: &Options, triage: &Triage, scopes: &[String]) -> Result<()> {
    if scopes.is_empty() {
        return Ok(());
    }
    let repo = triage.url().to_string();
    for issue in triage.pending(ctx, scopes).await? {
        let judged = match classify(ctx, &opts.classify, opts.timeout, &issue).await {
            Ok(choice) => triage.judge(&issue, &choice).await,
            Err(err) => Err(err),
        };
        let mut f = Fields {
            repo: Some(&repo),
            key: Some(&issue.subject),
            ..Default::default()
        };
        match judged {
            Ok(labels) => {
                let labels = labels.join(",");
                f.reason = Some(&labels);
                telemetry::note(Level::INFO, "judged", &f);
            }
            Err(err) => {
                f.error = Some(format!("{err:#}"));
                telemetry::note(Level::WARN, "judging failed", &f);
            }
        }
    }
    Ok(())
}

/// What `classifier` chooses for `issue`, in `timeout` or not at all.
async fn classify(
    ctx: &Ctx,
    classifier: &Classifier,
    timeout: Duration,
    issue: &Pending,
) -> Result<Choice> {
    let chosen = async {
        match classifier {
            Classifier::Decide(d) => decide::classify(&ctx.http, d, issue).await,
            Classifier::Command(command) => run_command(command, issue).await,
        }
    };
    let chosen = tokio::time::timeout(timeout, chosen).await;
    chosen.context("the classifier took too long")?
}

/// What `command` chooses for `issue`, given it on stdin.
async fn run_command(command: &str, issue: &Pending) -> Result<Choice> {
    let mut child = tokio::process::Command::new("sh")
        .args(["-c", command])
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .kill_on_drop(true)
        .spawn()?;
    let mut stdin = child.stdin.take().expect("stdin is piped");
    // One that reads none of it answers all the same.
    let _ = stdin.write_all(&serde_json::to_vec(issue)?).await;
    drop(stdin);
    let out = child.wait_with_output().await?;
    if !out.status.success() {
        bail!("the classifier failed: {}", out.status);
    }
    serde_json::from_slice(&out.stdout)
        .context("the classifier's answer is not {labels, model, reason}")
}

/// Rings `t`'s door on connecting to Jetstream, and on each event that
/// [`Triager::rings`] for, reconnecting whenever the connection drops;
/// with `--discover`, has it consider each repository record written.
fn follow(t: Arc<Triager>) -> impl Future<Output = ()> {
    let base = t.opts.jetstream.clone();
    jetstream::follow(base, jetstream::IDLE, Doorbell { t })
}

/// Jetstream as the triager hears it: a doorbell for the repositories it
/// reads again, so each connection starts at the stream's present.
struct Doorbell {
    t: Arc<Triager>,
}

impl jetstream::Follower for Doorbell {
    fn connected(&mut self, up: bool) {
        if up {
            self.t.door.notify_one();
        }
    }

    fn event(&mut self, event: jetstream::Event, _frame: &str) {
        if self.t.rings(&event) {
            self.t.door.notify_one();
        }
        self.t.found(&event);
    }
}

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

    fn args(line: &str) -> Vec<String> {
        line.split(' ').map(String::from).collect()
    }

    #[test]
    fn options() {
        let none = |_: &str| None;
        let line = "knotless-triage --classify ./c --scopes kind,,area --jetstream ws://j --sweep 9 --settle 1 --timeout 3 --once knotless://a.example/p b.example/q";
        let opts = parse_args(&args(line), &none).unwrap();
        assert!(opts.classify == Classifier::Command("./c".into()));
        assert_eq!(opts.scopes, ["kind", "area"]);
        assert_eq!(opts.jetstream, "ws://j");
        let durations = [opts.sweep, opts.settle, opts.timeout];
        assert_eq!(durations.map(|d| d.as_secs()), [9, 1, 3]);
        assert!(opts.once && !opts.discover);
        assert_eq!(opts.repos, ["a.example/p", "b.example/q"]);
        let line = "knotless-triage --classify ./c --scopes kind --discover";
        let found = parse_args(&args(line), &none).unwrap();
        assert!(found.discover && found.repos.is_empty());
        let env = |k: &str| (k == "KNOTLESS_DECISION_TOKEN").then(|| "t0ken".to_string());
        let line = "knotless-triage --decide https://d --model jev-latest --threshold 0.8 --scopes kind a.example/p";
        let decider = Decider {
            endpoint: "https://d".into(),
            token: "t0ken".into(),
            model: "jev-latest".into(),
            threshold: 0.8,
        };
        let opts = parse_args(&args(line), &env).unwrap();
        assert!(opts.classify == Classifier::Decide(decider));
        for (bad, says) in [
            (
                "knotless-triage --classify ./c --scopes kind",
                "name its scopes",
            ),
            (
                "knotless-triage --classify ./c a.example/p",
                "name its scopes",
            ),
            (
                "knotless-triage --scopes kind a.example/p",
                "name one classifier",
            ),
            (
                "knotless-triage --decide u --classify ./c --scopes k a.example/p",
                "name one classifier",
            ),
            (
                "knotless-triage --decide u --scopes k a.example/p",
                "--decide takes --model",
            ),
            (
                "knotless-triage --decide u --model m --scopes k a.example/p",
                "--decide takes KNOTLESS_DECISION_TOKEN",
            ),
            (
                "knotless-triage --threshold 2",
                "--threshold takes a probability",
            ),
            (
                "knotless-triage --threshold x",
                "--threshold takes a probability",
            ),
            ("knotless-triage --classify", "--classify takes a value"),
            ("knotless-triage --sweep soon", "--sweep takes seconds"),
            ("knotless-triage --bogus", "unexpected --bogus"),
        ] {
            let err = parse_args(&args(bad), &none).err().unwrap();
            assert!(err.starts_with(says), "{bad}: {err}");
        }
    }
}