/*
 * SPDX-FileCopyrightText: 2026 Coffey Labs
 *
 * SPDX-License-Identifier: AGPL-3.0-only
 */

//! Monitoring acceptance tests, from `docs/spec/features/monitoring.md`, for
//! what the shared suites in `tests/src/telemetry` don't cover. Each check
//! names the test number or requirement.

use crate::utils::{
    account::Account,
    server::{TestServer, TestServerBuilder},
    smtp::SmtpConnection,
};
use common::telemetry::{metrics::store::MetricsStore, tracers::store::TracingStore};
use registry::schema::{
    prelude::{ObjectType, Property},
    structs::{
        Alert, AlertEmail, AlertEvent, AlertEventProperties, CertificateManagement,
        DkimManagement, DnsManagement, Domain, Expression, MetricsStore as MetricsStoreSetting,
        MtaStageAuth, Search, Tenant, TracingStore as TracingStoreSetting, UserRoles,
    },
};
use serde_json::{Value, json};
use std::time::Duration;
use trc::{ClusterEvent, Collector, EventType, MetricType};
use types::id::Id;

const SECRET: &str = "monitoring test user passphrase";

pub async fn test(test: &mut TestServer) {
    println!("Running monitoring tests...");
    let admin = test.account("admin@example.org");
    let base = admin.base_url();

    // Acceptance test 1: history on by default, 30 and 90 days, hourly
    let retention = admin
        .jmap_method_call("x:DataRetention/get", json!({"ids": ["singleton"]}))
        .await;
    let retention = &retention.list()[0];
    assert_eq!(retention["holdTracesFor"], 30 * 86_400_000u64, "test 1: {retention}");
    assert_eq!(retention["holdMetricsFor"], 90 * 86_400_000u64, "test 1");
    assert_eq!(retention["metricsCollectionInterval"]["@type"], "Hourly", "test 1");
    for object in ["x:TracingStore/get", "x:MetricsStore/get"] {
        let store = admin
            .jmap_method_call(object, json!({"ids": ["singleton"]}))
            .await;
        assert_eq!(store.list()[0]["@type"], "Default", "test 1: {object}");
    }

    // Acceptance test 3: two ticks: counters hold the increase, gauges the
    // reading, idle counters write nothing
    test.server
        .metrics_store()
        .purge_metrics(Duration::ZERO)
        .await
        .unwrap();
    test.server.store_metrics().await;
    Collector::update_event_counter(EventType::Cluster(ClusterEvent::PublisherError), 7);
    tokio::time::sleep(Duration::from_millis(1100)).await;
    test.server.store_metrics().await;
    let samples = admin
        .jmap_method_call("x:Metric/get", json!({"ids": null}))
        .await;
    let samples = samples.list();
    let errors = samples
        .iter()
        .filter(|s| s["metric"] == "cluster.publisher-error")
        .collect::<Vec<_>>();
    assert_eq!(errors.last().map(|s| &s["count"]), Some(&json!(7)), "test 3: {errors:?}");
    let memory = samples
        .iter()
        .filter(|s| s["metric"] == "server.memory" && s["@type"] == "Gauge")
        .count();
    assert_eq!(memory, 2, "test 3 / MON-5: a gauge every tick");
    assert!(
        samples.iter().all(|s| s["timestamp"].is_string()),
        "test 3: timestamps"
    );

    // Acceptance test 5: every gauge and histogram in Prometheus output
    // (MON-6: one edition). Histograms appear once they've seen a value
    Collector::update_gauge(MetricType::QueueCount, 3);
    let exported = test.server.export_prometheus_metrics().await.unwrap();
    for name in ["queue_count", "server_memory", "domain_count"] {
        assert!(exported.contains(name), "test 5: {name} in {exported}");
    }
    let histograms = Collector::collect_histograms()
        .map(|h| h.id())
        .collect::<Vec<_>>();
    let gauges = Collector::collect_gauges().map(|g| g.id()).collect::<Vec<_>>();
    assert!(gauges.contains(&MetricType::QueueCount), "test 5: queue.count gauge");
    let _ = histograms;

    // Acceptance tests 7 to 9: what a trace holds
    test.server
        .tracing_store()
        .purge_spans(Duration::ZERO, Some(test.server.search_store()))
        .await
        .unwrap();
    let mut probe = SmtpConnection::connect().await;
    probe.send("QUIT").await;
    let user = admin
        .create_user_account("mon@example.org", SECRET, "Monitoring", &[], vec![])
        .await;
    let mut lmtp = SmtpConnection::connect().await;
    lmtp.ingest(
        "sender@example.org",
        &["mon@example.org"],
        "From: sender@example.org\r\nTo: mon@example.org\r\nSubject: Traced\r\n\r\nHello.\r\n",
    )
    .await;
    lmtp.quit().await;
    tokio::time::sleep(Duration::from_millis(500)).await;
    test.server.notify_task_queue();
    test.wait_for_tasks().await;
    let traces = admin
        .jmap_method_call("x:Trace/get", json!({"ids": null}))
        .await;
    let traces = traces.list().to_vec();
    assert_eq!(traces.len(), 2, "test 7: the probe left no trace: {traces:?}");
    for trace in &traces {
        let events = trace["events"].as_array().cloned().unwrap_or_else(|| {
            trace["events"]
                .as_object()
                .map(|o| o.values().cloned().collect())
                .unwrap_or_default()
        });
        assert!(
            events.iter().all(|e| {
                let name = e["event"].as_str().unwrap_or_default();
                !name.ends_with("raw-input") && !name.ends_with("raw-output")
            }),
            "test 8: no raw I/O"
        );
        assert!(trace["timestamp"].is_string(), "test 9: timestamp");
        assert!(trace["size"].is_number(), "test 9: size");
    }
    assert!(
        traces.iter().any(|t| t["from"] == "sender@example.org"),
        "test 9: from, {traces:?}"
    );
    assert!(
        traces
            .iter()
            .any(|t| t["to"].as_str().is_some_and(|to| to.contains("mon@example.org"))),
        "test 9: to"
    );

    // MON-16: the queueId filter finds the traces that name a queue id (the
    // session that queued the message and its delivery attempt) through the
    // search index, given as a string or a number (the index column is an
    // integer)
    fn queue_ids(value: &Value, out: &mut Vec<u64>) {
        match value {
            Value::Object(map) => {
                if map.get("key").and_then(|k| k.as_str()) == Some("queueId")
                    && let Some(id) = map
                        .get("value")
                        .and_then(|v| v.get("value").unwrap_or(v).as_u64())
                {
                    out.push(id);
                }
                map.values().for_each(|v| queue_ids(v, out));
            }
            Value::Array(list) => list.iter().for_each(|v| queue_ids(v, out)),
            _ => {}
        }
    }
    let with_ids = traces
        .iter()
        .map(|t| {
            let mut ids = Vec::new();
            queue_ids(t, &mut ids);
            (t["id"].as_str().unwrap().to_string(), ids)
        })
        .collect::<Vec<_>>();
    let queue_id = with_ids
        .iter()
        .find_map(|(_, ids)| ids.first().copied())
        .expect("MON-16: a trace with a queue id");
    let mut expected = with_ids
        .iter()
        .filter(|(_, ids)| ids.contains(&queue_id))
        .map(|(id, _)| id.clone())
        .collect::<Vec<_>>();
    expected.sort();
    for filter in [json!(queue_id.to_string()), json!(queue_id)] {
        let response = admin
            .jmap_method_call("x:Trace/query", json!({"filter": {"queueId": filter}}))
            .await;
        let mut found = response
            .0
            .pointer("/methodResponses/0/1/ids")
            .and_then(|ids| ids.as_array())
            .map(|ids| {
                ids.iter()
                    .filter_map(|id| id.as_str().map(str::to_string))
                    .collect::<Vec<_>>()
            })
            .unwrap_or_default();
        found.sort();
        assert_eq!(found, expected, "MON-16: queueId {filter}: {response:?}");
    }
    let response = admin
        .jmap_method_call(
            "x:Trace/query",
            json!({"filter": {"queueId": (queue_id ^ 0x5a5a_5a5a).to_string()}}),
        )
        .await;
    assert_eq!(
        response.0.pointer("/methodResponses/0/1/ids"),
        Some(&json!([])),
        "MON-16: an unknown queue id"
    );

    // Acceptance test 24: destroy removes a trace; create is refused
    let trace_id = traces[0]["id"].as_str().unwrap().to_string();
    let response = admin
        .jmap_method_call("x:Trace/set", json!({"destroy": [trace_id]}))
        .await;
    assert_eq!(
        response.0.pointer("/methodResponses/0/1/destroyed/0"),
        Some(&json!(trace_id)),
        "test 24: {response:?}"
    );
    let response = admin
        .jmap_method_call("x:Trace/set", json!({"create": {"i0": {"events": []}}}))
        .await;
    assert!(
        response.0.pointer("/methodResponses/0/1/notCreated/i0").is_some(),
        "test 24: create refused"
    );

    // Acceptance test 11: trace search off
    admin
        .registry_update_setting(
            Search {
                index_telemetry: false,
                ..Default::default()
            },
            &[Property::IndexTelemetry],
        )
        .await;
    let response = admin
        .jmap_method_call("x:Trace/query", json!({"filter": {"text": "example.org"}}))
        .await;
    assert!(response.to_string().contains("unsupportedFilter"), "test 11: {response:?}");
    let response = admin
        .jmap_method_call(
            "x:Trace/query",
            json!({"filter": {"timestampIsGreaterThan": "2020-01-01T00:00:00Z"}}),
        )
        .await;
    assert!(
        response.0.pointer("/methodResponses/0/1/ids").is_some(),
        "test 11: timestamp still works"
    );
    admin
        .registry_update_setting(
            Search {
                index_telemetry: true,
                ..Default::default()
            },
            &[Property::IndexTelemetry],
        )
        .await;

    // Acceptance tests 14 to 17: live telemetry
    let http = reqwest::Client::builder()
        .danger_accept_invalid_certs(true)
        .timeout(Duration::from_secs(10))
        .build()
        .unwrap();
    let token = |kind: &'static str| {
        let http = http.clone();
        let auth = admin.basic_auth();
        let url = format!("{base}/api/token/{kind}");
        async move {
            let response = http.get(url).header("authorization", auth).send().await.unwrap();
            assert_eq!(response.status(), 200, "test 16: token for {kind}");
            response.text().await.unwrap()
        }
    };
    let metrics_token = token("metrics").await;
    let frame = first_frame(
        &http,
        &format!("{base}/api/live/metrics?token={metrics_token}&metrics=server.memory&interval=1"),
        "metrics",
    )
    .await;
    let parsed: Value = serde_json::from_str(&frame).unwrap();
    assert_eq!(parsed[0]["metric"], "server.memory", "test 15: {frame}");
    assert_eq!(parsed[0]["@type"], "Gauge", "test 15");
    // The same token again, within its minute (MON-23 Decision)
    let _ = first_frame(
        &http,
        &format!("{base}/api/live/metrics?token={metrics_token}&metrics=server.memory&interval=1"),
        "metrics",
    )
    .await;
    // Without a token or credentials: refused
    let refused = http
        .get(format!("{base}/api/live/metrics?token=not-a-token"))
        .send()
        .await
        .unwrap();
    assert_ne!(refused.status(), 200, "test 16: a bad token");

    // Test 14: live tracing sees a matching event, and never raw I/O
    let tracing_token = token("tracing").await;
    let url = format!("{base}/api/live/tracing?token={tracing_token}&filter=mon%40example.org");
    let reader = tokio::spawn({
        let http = http.clone();
        async move { first_frame(&http, &url, "trace").await }
    });
    tokio::time::sleep(Duration::from_millis(500)).await;
    let mut lmtp = SmtpConnection::connect().await;
    lmtp.ingest(
        "sender@example.org",
        &["mon@example.org"],
        "From: sender@example.org\r\nTo: mon@example.org\r\nSubject: Live\r\n\r\nHello.\r\n",
    )
    .await;
    lmtp.quit().await;
    let frame = reader.await.unwrap();
    let events: Value = serde_json::from_str(&frame).unwrap();
    assert!(events.as_array().is_some_and(|e| !e.is_empty()), "test 14: {frame}");
    assert!(!frame.contains("raw-input"), "test 14: no raw I/O");
    assert!(frame.contains("keyValues"), "test 14: x:TraceEvent shape");

    // Test 17: at most eight streams at once
    let mut open = Vec::new();
    for _ in 0..8 {
        let response = http
            .get(format!("{base}/api/live/metrics?token={metrics_token}&interval=60"))
            .send()
            .await
            .unwrap();
        assert_eq!(response.status(), 200, "test 17");
        open.push(response);
    }
    let ninth = http
        .get(format!("{base}/api/live/metrics?token={metrics_token}&interval=60"))
        .send()
        .await
        .unwrap();
    assert_ne!(ninth.status(), 200, "test 17: the ninth stream");
    drop(open);
    tokio::time::sleep(Duration::from_millis(500)).await;

    // Acceptance tests 19 to 21: alerts
    let refused = admin
        .registry_create_object_expect_err(Alert {
            enable: true,
            condition: Expression {
                else_: "metric('no.such-metric') > 1".into(),
                ..Default::default()
            },
            email_alert: AlertEmail::Disabled,
            event_alert: AlertEvent::Disabled,
        })
        .await;
    let _ = refused; // test 21: refused on save
    let alert = admin
        .registry_create_object(Alert {
            enable: true,
            condition: Expression {
                else_: "domain_count > 50".into(),
                ..Default::default()
            },
            email_alert: AlertEmail::Disabled,
            event_alert: AlertEvent::Enabled(AlertEventProperties {
                event_message: "%{domain.count}% domains".to_string().into(),
            }),
        })
        .await;
    let fired = || Collector::read_metric(MetricType::TelemetryAlertEvent);
    let before = fired();
    Collector::update_gauge(MetricType::DomainCount, 100);
    test.server.process_alerts().await.unwrap();
    assert_eq!(fired(), before + 1.0, "test 19: an underscore condition fires");
    test.server.process_alerts().await.unwrap();
    assert_eq!(fired(), before + 1.0, "test 20: once while it holds");
    Collector::update_gauge(MetricType::DomainCount, 1);
    test.server.process_alerts().await.unwrap();
    Collector::update_gauge(MetricType::DomainCount, 100);
    test.server.process_alerts().await.unwrap();
    assert_eq!(fired(), before + 2.0, "test 20: again after it was false");
    Collector::update_gauge(MetricType::DomainCount, 1);
    admin
        .registry_destroy(ObjectType::Alert, [alert])
        .await
        .assert_destroyed(&[alert]);

    // Acceptance test 23: a tenant administrator can't reach telemetry
    let tenant = admin
        .registry_create_object(Tenant {
            name: "mon-t".into(),
            ..Default::default()
        })
        .await;
    let domain = admin
        .registry_create_object(Domain {
            name: "mon-t.example.org".into(),
            is_enabled: true,
            member_tenant_id: Some(tenant),
            certificate_management: CertificateManagement::Manual,
            dns_management: DnsManagement::Manual,
            dkim_management: DkimManagement::Manual,
            ..Default::default()
        })
        .await;
    let t_admin = admin
        .create_user_account("tadmin@mon-t.example.org", SECRET, "T admin", &[], vec![])
        .await;
    admin
        .registry_update_object(
            ObjectType::Account,
            t_admin.id(),
            json!({ Property::Roles: UserRoles::Admin }),
        )
        .await;
    for method in ["x:Trace/query", "x:Metric/query"] {
        let response = t_admin.jmap_method_call(method, json!({})).await;
        assert!(
            response.to_string().contains("forbidden"),
            "test 23: {method} {response:?}"
        );
    }
    let response = http
        .get(format!("{base}/api/token/tracing"))
        .header("authorization", t_admin.basic_auth())
        .send()
        .await
        .unwrap();
    assert_ne!(response.status(), 200, "test 23: no live token");

    // Clean up
    admin.destroy_account(t_admin).await;
    admin
        .registry_destroy(ObjectType::Domain, [domain])
        .await
        .assert_destroyed(&[domain]);
    admin
        .registry_destroy(ObjectType::Tenant, [tenant])
        .await
        .assert_destroyed(&[tenant]);
    admin.destroy_account(user).await;
    test.wait_for_tasks().await;
}

/// Acceptance test 26 (compat): INBUXA's settings read back as observed 1,
/// its stored history in the stripped encoding is skipped, not an error, and
/// one purge past its age removes it. Run against a copy of its data with
/// `INBUXA_COMPAT_ADMIN` (`name:password`), `NO_INSERT=1`, and the store's
/// `TMPDIR`/`STORE` pointing at the copy. The purge deletes history, so never
/// point it at the live store.
#[ignore]
#[tokio::test(flavor = "multi_thread")]
pub async fn monitoring_compat() {
    let admin = std::env::var("INBUXA_COMPAT_ADMIN").expect("INBUXA_COMPAT_ADMIN");
    assert!(std::env::var("NO_INSERT").is_ok(), "NO_INSERT must be set");
    let test = TestServerBuilder::new("monitoring_compat")
        .await
        .with_default_listeners()
        .await
        .build_with_opts(false)
        .await;
    let (name, secret) = admin.split_once(':').expect("name:password");
    let admin = Account::new(
        Box::leak(name.to_string().into_boxed_str()),
        Box::leak(secret.to_string().into_boxed_str()),
        &[],
        "Compat admin",
        Id::from(u32::MAX),
    );

    admin.assert_authenticates("INBUXA_COMPAT_ADMIN").await;

    // Observed 1: 30 and 90 days, hourly, both stores Default, no alerts
    let retention = admin
        .jmap_method_call("x:DataRetention/get", json!({"ids": ["singleton"]}))
        .await;
    let retention = &retention.list()[0];
    assert_eq!(retention["holdTracesFor"], 30 * 86_400_000u64, "{retention}");
    assert_eq!(retention["holdMetricsFor"], 90 * 86_400_000u64, "{retention}");
    assert_eq!(retention["metricsCollectionInterval"]["@type"], "Hourly");
    for object in ["x:TracingStore/get", "x:MetricsStore/get"] {
        let store = admin
            .jmap_method_call(object, json!({"ids": ["singleton"]}))
            .await;
        assert_eq!(store.list()[0]["@type"], "Default", "{object}");
    }
    let search = admin
        .jmap_method_call("x:Search/get", json!({"ids": ["singleton"]}))
        .await;
    let search = &search.list()[0];
    assert_eq!(search["indexTelemetry"], true, "{search}");
    let alerts = admin
        .jmap_method_call("x:Alert/get", json!({"ids": null}))
        .await;
    assert!(alerts.list().is_empty(), "{alerts:?}");

    // Old history: read without an error, whatever can't be decoded skipped
    for method in ["x:Trace/get", "x:Metric/get"] {
        let response = admin.jmap_method_call(method, json!({"ids": null})).await;
        assert!(
            response.0.pointer("/methodResponses/0/1/list").is_some(),
            "{method}: {response:?}"
        );
    }

    // One purge past its age, and nothing old is left
    test.server
        .tracing_store()
        .purge_spans(Duration::ZERO, Some(test.server.search_store()))
        .await
        .unwrap();
    test.server
        .metrics_store()
        .purge_metrics(Duration::ZERO)
        .await
        .unwrap();
    for method in ["x:Trace/query", "x:Metric/query"] {
        let response = admin.jmap_method_call(method, json!({})).await;
        assert_eq!(
            response.0.pointer("/methodResponses/0/1/ids"),
            Some(&json!([])),
            "{method}: {response:?}"
        );
    }
}

/// The first `event:` frame of that name on an event stream, its data.
async fn first_frame(http: &reqwest::Client, url: &str, event: &str) -> String {
    let mut response = http.get(url).send().await.unwrap();
    assert_eq!(response.status(), 200, "{url}");
    let mut buffer = String::new();
    let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
    loop {
        let chunk = tokio::time::timeout_at(deadline, response.chunk())
            .await
            .unwrap_or_else(|_| panic!("no {event} frame from {url}: {buffer}"))
            .unwrap()
            .expect("stream ended");
        buffer.push_str(&String::from_utf8_lossy(&chunk));
        while let Some(end) = buffer.find("\n\n") {
            let frame = buffer[..end].to_string();
            buffer.drain(..end + 2);
            if frame.starts_with(&format!("event: {event}\n"))
                && let Some(data) = frame.split_once("\ndata: ").map(|(_, d)| d.to_string())
            {
                return data;
            }
        }
    }
}

/// Runs the monitoring tests alone:
/// `cargo test -p tests monitoring_tests -- --ignored`.
#[ignore]
#[tokio::test(flavor = "multi_thread")]
pub async fn monitoring_tests() {
    let mut test = TestServerBuilder::new("monitoring_tests")
        .await
        .with_default_listeners()
        .await
        .with_object(MetricsStoreSetting::Default)
        .await
        .with_object(TracingStoreSetting::Default)
        .await
        .with_object(MtaStageAuth {
            require: Expression {
                else_: "false".to_string(),
                ..Default::default()
            },
            ..Default::default()
        })
        .await
        .build()
        .await;
    let admin = test.create_admin_account("admin@example.org").await;
    test.insert_account(admin);
    self::test(&mut test).await;
    if test.is_reset() {
        test.temp_dir.delete();
    }
}