//! What the runner logs of what it does (`knotless::telemetry`): its runs
//! from queued to ended, their steps and containers, its queues'
//! batches, and what went wrong.

use crate::{Subject, Watch};
use knotless::telemetry::{self, Fields, Level, millis};
use std::collections::HashMap;
use std::sync::{LazyLock, Mutex};
use std::time::{Duration, Instant};

/// Logs `message` about the run of `workflow` at `subject`: its ref,
/// commit and pull, and how it ended and how long it took, if it has.
pub(crate) fn run(
    message: &str,
    watch: &Watch,
    subject: &Subject,
    workflow: &str,
    status: Option<&str>,
    took: Option<Duration>,
) {
    let fields = Fields {
        repo: Some(&watch.label),
        workflow: Some(workflow),
        ref_name: Some(&subject.ref_name),
        commit: Some(&subject.commit),
        pull: subject.pull.as_ref().map(|p| p.uri.as_str()),
        count: subject.batch.as_ref().map(|b| b.pulls.len() as u64),
        status,
        duration_ms: took.map(millis),
        ..Fields::default()
    };
    telemetry::note(Level::INFO, message, &fields);
}

/// Logs that `step` of `workflow` started, or, with its `status`, ended.
pub(crate) fn step(
    commit: &str,
    workflow: &str,
    step: &str,
    status: Option<&str>,
    exit_code: Option<i32>,
    took: Option<Duration>,
) {
    let fields = Fields {
        commit: Some(commit),
        workflow: Some(workflow),
        step: Some(step),
        status,
        exit_code: exit_code.map(i64::from),
        duration_ms: took.map(millis),
        ..Fields::default()
    };
    let said = ["step started", "step ended"][usize::from(status.is_some())];
    telemetry::note(Level::INFO, said, &fields);
}

/// Logs `message` about the container `id`, of `image` if known.
pub(crate) fn container(message: &str, id: &str, image: Option<&str>) {
    let fields = Fields {
        container: Some(&id[..id.len().min(12)]),
        source: image,
        ..Fields::default()
    };
    telemetry::note(Level::INFO, message, &fields);
}

/// How often a failure going on is logged again, as `failed` with a count.
const REPEAT_EVERY: Duration = Duration::from_secs(60);

/// How long a failure not seen again is remembered: past it, it is logged
/// at once, as a new one.
const FORGET: Duration = Duration::from_secs(60 * 60);

/// The failures logged, by label and error: when last, and how many
/// times seen since without being logged.
type Failing = HashMap<(String, String), (Instant, u64)>;

static FAILING: LazyLock<Mutex<Failing>> = LazyLock::new(Mutex::default);

/// Logs, at warn, what went wrong with `label`'s work: the first time it
/// goes wrong so, then once each [`REPEAT_EVERY`] while it goes on, with
/// how many times it went wrong so meanwhile. Another error is logged at
/// once, and the same again once a success of `label`'s cleared it.
pub(crate) fn failed(label: &str, err: &anyhow::Error) {
    let error = format!("{err:#}");
    let seen = repeat(&mut FAILING.lock().unwrap(), label, &error, Instant::now());
    let Some(repeats) = seen else {
        return;
    };
    let fields = Fields {
        repo: Some(label),
        error: Some(error),
        count: (repeats > 0).then_some(repeats),
        ..Fields::default()
    };
    telemetry::note(Level::WARN, "failed", &fields);
}

/// Forgets what went wrong with `label`'s work, which went right since.
pub(crate) fn succeeded(label: &str) {
    FAILING.lock().unwrap().retain(|(l, _), _| l != label);
}

/// Whether `error` of `label`, at `now`, is to be logged, with how many
/// times it was seen since last it was; or `None`, counting it.
fn repeat(seen: &mut Failing, label: &str, error: &str, now: Instant) -> Option<u64> {
    seen.retain(|_, (at, _)| now.duration_since(*at) < FORGET);
    let key = (label.to_string(), error.to_string());
    match seen.get_mut(&key) {
        Some((at, skipped)) if now.duration_since(*at) < REPEAT_EVERY => {
            *skipped += 1;
            None
        }
        Some((at, skipped)) => {
            let repeats = *skipped;
            (*at, *skipped) = (now, 0);
            Some(repeats)
        }
        None => {
            seen.insert(key, (now, 0));
            Some(0)
        }
    }
}

/// Logs that a pass over `label`, reconciling it, ended after `took`: at
/// debug, or info if it was slow, past `KNOTLESS_SLOW_MS`.
pub(crate) fn pass(label: &str, took: Duration, ok: bool) {
    let fields = Fields {
        repo: Some(label),
        status: Some(["failed", "ok"][usize::from(ok)]),
        duration_ms: Some(millis(took)),
        ..Fields::default()
    };
    let level = [Level::DEBUG, Level::INFO][usize::from(telemetry::slow(took))];
    telemetry::note(level, "pass ended", &fields);
}

/// Logs, at warn, that a pass over `label` has gone on for `took`, and is
/// at `stage`.
pub(crate) fn long_pass(label: &str, stage: &str, took: Duration) {
    let fields = Fields {
        repo: Some(label),
        stage: Some(stage),
        duration_ms: Some(millis(took)),
        ..Fields::default()
    };
    telemetry::note(Level::WARN, "pass running long", &fields);
}

/// Logs that the disk is `short` of the runner's floor, as a run sheds
/// caches, or as it waits for room once it has.
pub(crate) fn short_of_disk(short: &str, shedding: bool) {
    let fields = Fields {
        reason: Some(short),
        ..Fields::default()
    };
    let message = if shedding {
        "disk short: shedding caches"
    } else {
        "disk short: waiting for room"
    };
    telemetry::note(Level::WARN, message, &fields);
}

/// Logs `message` about a batch of `pulls` pulls into `branch` of
/// `label`, at `tip`, and why, if said.
pub(crate) fn batch(
    message: &str,
    label: &str,
    branch: &str,
    tip: &str,
    pulls: usize,
    why: Option<&str>,
) {
    let fields = Fields {
        repo: Some(label),
        ref_name: Some(branch),
        commit: Some(tip),
        count: Some(pulls as u64),
        reason: why,
        ..Fields::default()
    };
    telemetry::note(Level::INFO, message, &fields);
}

/// Logs that the pull `uri` was ejected from `label`'s queue, and why.
pub(crate) fn ejected(label: &str, uri: &str, why: &str) {
    let fields = Fields {
        repo: Some(label),
        pull: Some(uri),
        reason: Some(why),
        ..Fields::default()
    };
    telemetry::note(Level::INFO, "pull ejected", &fields);
}

/// Logs a look at `label`'s queue records, which took `took`: at debug,
/// or info if it found them changed.
pub(crate) fn noticed(label: &str, took: Duration, changed: bool) {
    let fields = Fields {
        repo: Some(label),
        duration_ms: Some(millis(took)),
        ..Fields::default()
    };
    let level = [Level::DEBUG, Level::INFO][usize::from(changed)];
    telemetry::note(level, "queue noticed", &fields);
}

/// Logs, at warn, a write to the PDS tried again after `wait`.
pub(crate) fn retried(err: &anyhow::Error, wait: Duration) {
    let fields = Fields {
        error: Some(format!("{err:#}")),
        duration_ms: Some(millis(wait)),
        ..Fields::default()
    };
    telemetry::note(Level::WARN, "pds write retried", &fields);
}

/// Logs that Jetstream, at `url`, connected, or could not be or dropped.
pub(crate) fn feed(url: &str, connected: bool) {
    let fields = Fields {
        feed: Some("jetstream"),
        source: Some(url),
        connected: Some(connected),
        ..Fields::default()
    };
    let (level, said) =
        [(Level::WARN, "feed down"), (Level::INFO, "feed connected")][usize::from(connected)];
    telemetry::note(level, said, &fields);
}

/// Logs what the runner watches, as whom.
pub(crate) fn watching(repos: &[String], did: &str) {
    let repos = repos.join(" ");
    let fields = Fields {
        repo: Some(&repos),
        did: Some(did),
        ..Fields::default()
    };
    telemetry::note(Level::INFO, "watching", &fields);
}

/// Logs that `label` was found naming this runner, and is watched.
pub(crate) fn found(label: &str) {
    let fields = Fields {
        repo: Some(label),
        ..Fields::default()
    };
    telemetry::note(Level::INFO, "repository found", &fields);
}

/// Logs that `label`, found before, no longer names this runner.
pub(crate) fn left(label: &str) {
    let fields = Fields {
        repo: Some(label),
        ..Fields::default()
    };
    telemetry::note(Level::INFO, "repository left", &fields);
}

/// Logs that `gone` checks of `label` were deleted.
pub(crate) fn pruned(label: &str, gone: usize) {
    let fields = Fields {
        repo: Some(label),
        count: Some(gone as u64),
        ..Fields::default()
    };
    telemetry::note(Level::INFO, "checks pruned", &fields);
}

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

    #[test]
    fn notes() {
        // Nothing is set up to log here: each only has to run.
        step("a/b", "test", "build", None, None, None);
        step(
            "a/b",
            "test",
            "build",
            Some("success"),
            Some(0),
            Some(Duration::ZERO),
        );
        container("container started", "0123456789abcdef", Some("rust"));
        container("container stopped", "short", None);
        failed("a/b", &anyhow::anyhow!("no PDS"));
        failed("a/b", &anyhow::anyhow!("no PDS"));
        succeeded("a/b");
        pass("a/b", Duration::ZERO, true);
        pass("a/b", Duration::from_secs(5), false);
        long_pass("a/b", "landing", Duration::from_secs(300));
        batch("batch started", "a/b", "main", "abc", 2, None);
        ejected("a/b", "at://did:plc:x/dev.knotless.pull/1", "conflicts");
        noticed("a/b", Duration::ZERO, true);
        retried(&anyhow::anyhow!("503"), Duration::from_secs(1));
        feed("ws://127.0.0.1:9", false);
        feed("ws://127.0.0.1:9", true);
        watching(&["a/b".into()], "did:plc:x");
        found("did:plc:x/b");
        left("did:plc:x/b");
        short_of_disk("1.0 GiB free", true);
        short_of_disk("1.0 GiB free", false);
    }

    #[test]
    fn failures_going_on_are_logged_now_and_then() {
        let mut seen = Failing::new();
        let start = Instant::now();
        let at = |secs| start + Duration::from_secs(secs);
        // The first time, at once; the same again only each minute, with
        // how often it was seen meanwhile.
        assert_eq!(repeat(&mut seen, "a/b", "gone", at(0)), Some(0));
        assert_eq!(repeat(&mut seen, "a/b", "gone", at(5)), None);
        assert_eq!(repeat(&mut seen, "a/b", "gone", at(10)), None);
        assert_eq!(repeat(&mut seen, "a/b", "gone", at(60)), Some(2));
        assert_eq!(repeat(&mut seen, "a/b", "gone", at(61)), None);
        // Another error, or another repository's, is logged at once.
        assert_eq!(repeat(&mut seen, "a/b", "refused", at(62)), Some(0));
        assert_eq!(repeat(&mut seen, "c/d", "gone", at(62)), Some(0));
        // One not seen for an hour is new again.
        assert_eq!(repeat(&mut seen, "a/b", "gone", at(2 * 60 * 60)), Some(0));
        assert_eq!(seen.len(), 1);
    }
}