eyes-subscriber 0.8.1

//! Boot-time app manifests.
//!
//! Telemetry shows what happened; the manifest tells Eyes what the app's
//! *shape* is — every registered job type, every cron entry with its
//! schedule, and the build version. Apps send one manifest at boot via
//! [`send_manifest`] (or [`send_manifest_from_env`]); each stored manifest
//! doubles as a boot/deploy marker on the server.

use crate::dashboards::NamedDashboard;
use crate::metrics::{MetricThreshold, NamedMetric};
use chrono::{DateTime, Utc};
use serde::Serialize;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
use std::{sync::Arc, time::Duration};
use tokio::{sync::oneshot, task::JoinHandle};
use url::Url;
use uuid::Uuid;

/// The manifest schema version this crate emits.
pub const MANIFEST_VERSION: u32 = 2;

/// The shape of the application, as reported once at boot.
///
/// `manifest_version` and `booted_at` are filled in internally when the
/// manifest is sent ([`MANIFEST_VERSION`] and `Utc::now()` respectively).
///
/// Deliberately `PartialEq` only (no `Eq`, since 0.7.0): the `metrics` field
/// embeds [`NamedMetric`], which carries a `QueryDocument` whose
/// float-bearing IR types intentionally stop at `PartialEq`. `assert_eq!`
/// and `==` keep working; keying a `HashSet`/`HashMap` by `AppManifest` does
/// not compile.
#[derive(Debug, Clone, Default, PartialEq, Serialize)]
#[non_exhaustive]
pub struct AppManifest {
    /// Application build version (e.g. from `CARGO_PKG_VERSION`).
    pub app_version: Option<String>,
    /// Git commit SHA of the running build.
    pub git_sha: Option<String>,
    /// Names of every registered job type.
    pub jobs: Vec<String>,
    /// Client-only selector: which of `jobs` should be marked critical on the
    /// wire. Names not also present in `jobs` are ignored; duplicates are
    /// deduplicated. There is no second top-level wire field — see
    /// [`ManifestPayload::new`].
    pub critical_jobs: Vec<String>,
    /// Every registered cron entry with its schedule.
    pub crons: Vec<CronEntry>,
    /// Public origin used to resolve root-relative monitor targets.
    pub base_url: Option<String>,
    /// `None` does not participate in monitor authority; `Some(vec![])`
    /// explicitly removes every declaration for this app.
    pub monitors: Option<Vec<HttpMonitor>>,
    pub process_instance_id: Option<Uuid>,
    pub process_role: Option<String>,
    pub expected_process_roles: Option<Vec<ExpectedProcessRole>>,
    /// `None` does not participate in named-metric authority; `Some(vec![])`
    /// explicitly removes every declaration for this app.
    pub metrics: Option<Vec<NamedMetric>>,
    /// `None` does not participate in metric-threshold authority;
    /// `Some(vec![])` explicitly removes every declaration for this app
    /// (mirroring `metrics`).
    pub metric_thresholds: Option<Vec<MetricThreshold>>,
    /// `None` does not participate in dashboard authority; `Some(vec![])`
    /// explicitly removes every declaration for this app. A non-empty list
    /// must be paired with a non-null `metrics` array (server-enforced;
    /// publishing dashboards republishes the metric snapshot — see the
    /// server contract).
    pub dashboards: Option<Vec<NamedDashboard>>,
}

impl AppManifest {
    pub fn app_version(mut self, app_version: impl Into<String>) -> Self {
        self.app_version = Some(app_version.into());
        self
    }

    pub fn git_sha(mut self, git_sha: impl Into<String>) -> Self {
        self.git_sha = Some(git_sha.into());
        self
    }

    pub fn jobs(mut self, jobs: Vec<String>) -> Self {
        self.jobs = jobs;
        self
    }

    /// Mark a subset of `jobs` as critical (declares a `critical_job_failed`
    /// run-health monitor for each, once cja adopts process identity). Names
    /// not present in `jobs` are ignored; duplicates are deduplicated.
    pub fn critical_jobs(mut self, jobs: Vec<String>) -> Self {
        self.critical_jobs = jobs;
        self
    }

    pub fn crons(mut self, crons: Vec<CronEntry>) -> Self {
        self.crons = crons;
        self
    }

    pub fn base_url(mut self, base_url: impl Into<String>) -> Self {
        self.base_url = Some(base_url.into());
        self
    }

    pub fn monitors(mut self, monitors: Vec<HttpMonitor>) -> Self {
        self.monitors = Some(monitors);
        self
    }
    pub fn process(mut self, identity: ProcessIdentity) -> Self {
        self.process_instance_id = Some(identity.instance_id());
        self.process_role = Some(identity.role().to_owned());
        self
    }
    pub fn expected_process_roles(mut self, roles: Vec<ExpectedProcessRole>) -> Self {
        self.expected_process_roles = Some(roles);
        self
    }
    pub fn metrics(mut self, metrics: Vec<NamedMetric>) -> Self {
        self.metrics = Some(metrics);
        self
    }
    pub fn metric_thresholds(mut self, v: Vec<MetricThreshold>) -> Self {
        self.metric_thresholds = Some(v);
        self
    }
    pub fn dashboards(mut self, dashboards: Vec<NamedDashboard>) -> Self {
        self.dashboards = Some(dashboards);
        self
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProcessIdentity(Arc<ProcessIdentityInner>);
#[derive(Debug, PartialEq, Eq)]
struct ProcessIdentityInner {
    instance_id: Uuid,
    role: String,
}
impl ProcessIdentity {
    pub fn new(role: impl Into<String>) -> Self {
        Self(Arc::new(ProcessIdentityInner {
            instance_id: Uuid::new_v4(),
            role: role.into(),
        }))
    }
    pub fn instance_id(&self) -> Uuid {
        self.0.instance_id
    }
    pub fn role(&self) -> &str {
        &self.0.role
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ExpectedProcessRole {
    pub role: String,
    pub min_instances: u32,
    pub heartbeat_interval_seconds: u64,
    pub max_staleness_seconds: u64,
    pub evaluation_interval_seconds: u64,
    pub failure_threshold: u32,
    pub shutdown_grace_seconds: u64,
    pub enabled: bool,
}
impl ExpectedProcessRole {
    pub fn new(role: impl Into<String>) -> Self {
        Self {
            role: role.into(),
            min_instances: 1,
            heartbeat_interval_seconds: 30,
            max_staleness_seconds: 120,
            evaluation_interval_seconds: 30,
            failure_threshold: 4,
            shutdown_grace_seconds: 120,
            enabled: true,
        }
    }
    pub fn min_instances(mut self, v: u32) -> Self {
        self.min_instances = v;
        self
    }
    pub fn heartbeat_interval_seconds(mut self, v: u64) -> Self {
        self.heartbeat_interval_seconds = v;
        self
    }
    pub fn max_staleness_seconds(mut self, v: u64) -> Self {
        self.max_staleness_seconds = v;
        self
    }
    pub fn evaluation_interval_seconds(mut self, v: u64) -> Self {
        self.evaluation_interval_seconds = v;
        self
    }
    pub fn failure_threshold(mut self, v: u32) -> Self {
        self.failure_threshold = v;
        self
    }
    pub fn shutdown_grace_seconds(mut self, v: u64) -> Self {
        self.shutdown_grace_seconds = v;
        self
    }
    pub fn enabled(mut self, v: bool) -> Self {
        self.enabled = v;
        self
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "UPPERCASE")]
pub enum HttpMethod {
    Get,
    Head,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct HttpMonitor {
    pub id: String,
    pub target: String,
    pub method: HttpMethod,
    pub interval_seconds: u64,
    pub timeout_seconds: u64,
    pub expected_status_min: u16,
    pub expected_status_max: u16,
    pub failure_threshold: u32,
    pub enabled: bool,
}

impl HttpMonitor {
    pub fn new(id: impl Into<String>, target: impl Into<String>) -> Self {
        Self {
            id: id.into(),
            target: target.into(),
            method: HttpMethod::Get,
            interval_seconds: 60,
            timeout_seconds: 10,
            expected_status_min: 200,
            expected_status_max: 299,
            failure_threshold: 3,
            enabled: true,
        }
    }

    pub fn method(mut self, method: HttpMethod) -> Self {
        self.method = method;
        self
    }
    pub fn interval_seconds(mut self, seconds: u64) -> Self {
        self.interval_seconds = seconds;
        self
    }
    pub fn timeout_seconds(mut self, seconds: u64) -> Self {
        self.timeout_seconds = seconds;
        self
    }
    pub fn expected_status(mut self, min: u16, max: u16) -> Self {
        self.expected_status_min = min;
        self.expected_status_max = max;
        self
    }
    pub fn failure_threshold(mut self, threshold: u32) -> Self {
        self.failure_threshold = threshold;
        self
    }
    pub fn enabled(mut self, enabled: bool) -> Self {
        self.enabled = enabled;
        self
    }
}

#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum MonitorTargetError {
    #[error("relative monitor target requires base_url")]
    MissingBaseUrl,
    #[error("monitor target must be an absolute URL or root-relative path")]
    NonRootRelative,
    #[error("network-path monitor targets are not allowed")]
    NetworkPath,
    #[error("invalid monitor target URL: {0}")]
    InvalidTarget(String),
    #[error("invalid monitor base URL: {0}")]
    InvalidBase(String),
    #[error("only HTTP and HTTPS monitor URLs are supported")]
    UnsupportedScheme,
    #[error("monitor URLs may not contain credentials")]
    Credentials,
    #[error("monitor URLs may not contain fragments")]
    Fragment,
    #[error("monitor URL must contain a host")]
    MissingHost,
    #[error("localhost monitor targets are not allowed")]
    Localhost,
    #[error("monitor target uses a forbidden IP address")]
    ForbiddenIp,
}

/// Resolve and validate an absolute HTTP(S) URL or a root-relative path.
pub fn resolve_monitor_target(
    base_url: Option<&str>,
    target: &str,
) -> Result<Url, MonitorTargetError> {
    if target.starts_with("//") {
        return Err(MonitorTargetError::NetworkPath);
    }
    let url = if target.starts_with('/') {
        // WHATWG parsing treats backslashes as slashes for special schemes and
        // strips ASCII tabs/newlines. Reject those spellings before Url::join
        // can reinterpret a path as an authority.
        if target
            .bytes()
            .any(|byte| matches!(byte, b'\\' | b'\t' | b'\r' | b'\n'))
        {
            return Err(MonitorTargetError::NetworkPath);
        }
        let base = base_url.ok_or(MonitorTargetError::MissingBaseUrl)?;
        let parsed =
            Url::parse(base).map_err(|e| MonitorTargetError::InvalidBase(e.to_string()))?;
        validate_url(&parsed).map_err(|error| match error {
            MonitorTargetError::InvalidTarget(message) => MonitorTargetError::InvalidBase(message),
            other => other,
        })?;
        let joined = parsed
            .join(target)
            .map_err(|e| MonitorTargetError::InvalidTarget(e.to_string()))?;
        if monitor_origin(&joined)? != monitor_origin(&parsed)? {
            return Err(MonitorTargetError::NetworkPath);
        }
        joined
    } else {
        Url::parse(target).map_err(|e| {
            if e == url::ParseError::RelativeUrlWithoutBase {
                MonitorTargetError::NonRootRelative
            } else {
                MonitorTargetError::InvalidTarget(e.to_string())
            }
        })?
    };
    validate_url(&url)?;
    Ok(url)
}

/// Return the normalized HTTP origin used for monitor security comparisons.
///
/// Domain names are lowercase without a trailing root dot and default ports
/// are omitted, so equivalent DNS spellings cannot bypass origin checks.
pub fn monitor_origin(url: &Url) -> Result<String, MonitorTargetError> {
    validate_url(url)?;
    let host = match url.host().ok_or(MonitorTargetError::MissingHost)? {
        url::Host::Domain(name) => name.trim_end_matches('.').to_ascii_lowercase(),
        url::Host::Ipv4(ip) => ip.to_string(),
        url::Host::Ipv6(ip) => format!("[{ip}]"),
    };
    let port = match (url.scheme(), url.port()) {
        ("http", Some(80)) | ("https", Some(443)) | (_, None) => String::new(),
        (_, Some(port)) => format!(":{port}"),
    };
    Ok(format!("{}://{host}{port}", url.scheme()))
}

fn validate_url(url: &Url) -> Result<(), MonitorTargetError> {
    if !matches!(url.scheme(), "http" | "https") {
        return Err(MonitorTargetError::UnsupportedScheme);
    }
    if !url.username().is_empty() || url.password().is_some() {
        return Err(MonitorTargetError::Credentials);
    }
    if url.fragment().is_some() {
        return Err(MonitorTargetError::Fragment);
    }
    let host = url.host().ok_or(MonitorTargetError::MissingHost)?;
    match host {
        url::Host::Domain(name)
            if name.trim_end_matches('.').eq_ignore_ascii_case("localhost")
                || name
                    .trim_end_matches('.')
                    .to_ascii_lowercase()
                    .ends_with(".localhost") =>
        {
            Err(MonitorTargetError::Localhost)
        }
        url::Host::Ipv4(ip) if is_forbidden_monitor_ip(ip.into()) => {
            Err(MonitorTargetError::ForbiddenIp)
        }
        url::Host::Ipv6(ip) if is_forbidden_monitor_ip(ip.into()) => {
            Err(MonitorTargetError::ForbiddenIp)
        }
        _ => Ok(()),
    }
}

/// Returns true when an address is not an acceptable monitor egress target.
/// This is shared by declaration validation and the connection-time resolver.
pub fn is_forbidden_monitor_ip(ip: IpAddr) -> bool {
    match ip {
        IpAddr::V4(ip) => forbidden_v4(ip),
        IpAddr::V6(ip) => forbidden_v6(ip),
    }
}

fn forbidden_v4(ip: Ipv4Addr) -> bool {
    let value = u32::from(ip);
    ip.is_unspecified()
        || ip.is_loopback()
        || ip.is_private()
        || ip.is_link_local()
        || ip.is_multicast()
        || ip.is_broadcast()
        || ip.is_documentation()
        || value >> 24 == 0
        || value & 0xffc0_0000 == 0x6440_0000 // 100.64.0.0/10
        || value & 0xffff_ff00 == 0xc000_0000 // 192.0.0.0/24
        || value & 0xffff_ff00 == 0xc058_6300 // 192.88.99.0/24
        || value & 0xfffe_0000 == 0xc612_0000 // 198.18.0.0/15
        || value & 0xf000_0000 == 0xf000_0000 // 240.0.0.0/4
}

fn forbidden_v6(ip: Ipv6Addr) -> bool {
    let octets = ip.octets();
    let compatible_v4 = octets[..12]
        .iter()
        .all(|byte| *byte == 0)
        .then(|| Ipv4Addr::new(octets[12], octets[13], octets[14], octets[15]));
    ip.is_unspecified()
        || ip.is_loopback()
        || ip.is_multicast()
        || ip.is_unique_local()
        || ip.is_unicast_link_local()
        || ip.to_ipv4_mapped().is_some_and(forbidden_v4)
        || compatible_v4.is_some_and(forbidden_v4)
        // Deny the IANA special-purpose blocks as groups. Some contain narrow
        // protocol exceptions, but none are appropriate arbitrary HTTP egress
        // targets and denying the containing allocation is the safer default.
        || octets[..12] == [0x00, 0x64, 0xff, 0x9b, 0, 0, 0, 0, 0, 0, 0, 0] // 64:ff9b::/96
        || (ip.segments()[0] == 0x0064 && ip.segments()[1] == 0xff9b
            && ip.segments()[2] == 0x0001) // 64:ff9b:1::/48
        || (ip.segments()[0] == 0x0100 && ip.segments()[1..4] == [0, 0, 0]) // 100::/64
        || (ip.segments()[0] == 0x2001 && ip.segments()[1] < 0x0200) // 2001::/23
        || (ip.segments()[0] == 0x2001 && ip.segments()[1] == 0x0db8) // documentation /32
        || ip.segments()[0] == 0x2002 // 6to4, whose embedded IPv4 may be private
        || (ip.segments()[0] == 0x3fff && (ip.segments()[1] & 0xf000) == 0) // documentation /20
        || ip.segments()[0] == 0x5f00 // segment-routing SIDs /16
        || (ip.segments()[0] & 0xffc0) == 0xfec0 // deprecated site-local /10
        || (ip.segments()[0] & 0xe000) != 0x2000 // outside global-unicast 2000::/3
}

/// A single cron registration: its name and schedule string.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct CronEntry {
    pub name: String,
    /// Human/cron schedule string as the app reports it
    /// (e.g. "every 300s" or "0 0 * * *").
    pub schedule: String,
}

/// Error sending a boot manifest.
#[derive(Debug, thiserror::Error)]
pub enum ManifestError {
    /// Bad base URL or missing/invalid environment configuration.
    #[error("Configuration error: {0}")]
    Configuration(String),
    /// The HTTP request itself failed (connect, timeout, ...).
    #[error("HTTP request failed: {0}")]
    Request(#[from] reqwest::Error),
    /// The server answered with a non-2xx status.
    #[error("Server returned error status: {0}")]
    Status(reqwest::StatusCode),
}

/// Wire payload for `POST /api/orgs/:org_id/apps/:app_id/manifest`.
///
/// Jobs are serialized as `{"name": "..."}` objects (not bare strings) so
/// per-job fields can be added later without a server migration.
#[derive(Debug, Serialize)]
struct ManifestPayload<'a> {
    manifest_version: u32,
    app_version: Option<&'a str>,
    git_sha: Option<&'a str>,
    jobs: Vec<JobPayload<'a>>,
    crons: &'a [CronEntry],
    base_url: Option<&'a str>,
    monitors: Option<&'a [HttpMonitor]>,
    process_instance_id: Option<Uuid>,
    process_role: Option<&'a str>,
    expected_process_roles: Option<&'a [ExpectedProcessRole]>,
    metrics: Option<&'a [NamedMetric]>,
    metric_thresholds: Option<&'a [MetricThreshold]>,
    dashboards: Option<&'a [NamedDashboard]>,
    booted_at: DateTime<Utc>,
}

#[derive(Debug, Serialize)]
struct JobPayload<'a> {
    name: &'a str,
    critical: bool,
}

impl<'a> ManifestPayload<'a> {
    fn new(manifest: &'a AppManifest, booted_at: DateTime<Utc>) -> Self {
        let critical: std::collections::HashSet<&str> =
            manifest.critical_jobs.iter().map(String::as_str).collect();
        Self {
            manifest_version: MANIFEST_VERSION,
            app_version: manifest.app_version.as_deref(),
            git_sha: manifest.git_sha.as_deref(),
            jobs: manifest
                .jobs
                .iter()
                .map(|name| JobPayload {
                    name,
                    critical: critical.contains(name.as_str()),
                })
                .collect(),
            crons: &manifest.crons,
            base_url: manifest.base_url.as_deref(),
            monitors: manifest.monitors.as_deref(),
            process_instance_id: manifest.process_instance_id,
            process_role: manifest.process_role.as_deref(),
            expected_process_roles: manifest.expected_process_roles.as_deref(),
            metrics: manifest.metrics.as_deref(),
            metric_thresholds: manifest.metric_thresholds.as_deref(),
            dashboards: manifest.dashboards.as_deref(),
            booted_at,
        }
    }
}

/// Send a boot-time manifest to the Eyes server as a one-shot HTTP POST.
///
/// `booted_at` is stamped with `Utc::now()` at send time and
/// `manifest_version` with [`MANIFEST_VERSION`]. There is no queueing,
/// batching, or retrying involved — this is a single request.
///
/// Callers should fire-and-forget with a warning on failure: a manifest
/// failure must never block app boot. For example:
///
/// ```no_run
/// # use eyes_subscriber::AppManifest;
/// # async fn example(base_url: &str, org_id: uuid::Uuid, app_id: uuid::Uuid) {
/// let manifest = AppManifest::default();
/// if let Err(e) = eyes_subscriber::send_manifest(base_url, org_id, app_id, &manifest, None).await {
///     tracing::warn!("Failed to send app manifest to eyes: {e}");
/// }
/// # }
/// ```
pub async fn send_manifest(
    base_url: &str,
    org_id: Uuid,
    app_id: Uuid,
    manifest: &AppManifest,
    auth_token: Option<&str>,
) -> Result<(), ManifestError> {
    let url = Url::parse(base_url)
        .and_then(|base| base.join(&format!("/api/orgs/{}/apps/{}/manifest", org_id, app_id)))
        .map_err(|e| ManifestError::Configuration(format!("Invalid URL: {}", e)))?;

    let payload = ManifestPayload::new(manifest, Utc::now());

    let mut request = reqwest::Client::new().post(url).json(&payload);
    if let Some(token) = auth_token {
        request = request.bearer_auth(token);
    }
    let response = request.send().await?;

    if !response.status().is_success() {
        return Err(ManifestError::Status(response.status()));
    }

    Ok(())
}

/// [`send_manifest`], reading the destination from the environment:
///
/// - `EYES_URL`: base URL (defaults to `https://eyes.coreyja.com`, matching
///   [`crate::EyesSubscriberBuilder`])
/// - `EYES_ORG_ID`: org UUID (required)
/// - `EYES_APP_ID`: app UUID (required)
/// - `EYES_TOKEN`: bearer token (optional while the server runs in warn mode)
///
/// Returns [`ManifestError::Configuration`] if `EYES_ORG_ID`/`EYES_APP_ID`
/// are unset or not valid UUIDs. As with [`send_manifest`], failures should
/// be logged and ignored by callers — never block app boot on this.
pub async fn send_manifest_from_env(manifest: &AppManifest) -> Result<(), ManifestError> {
    let base_url =
        std::env::var("EYES_URL").unwrap_or_else(|_| "https://eyes.coreyja.com".to_string());

    let org_id = uuid_from_env("EYES_ORG_ID")?;
    let app_id = uuid_from_env("EYES_APP_ID")?;
    let token = token_from_env();

    send_manifest(&base_url, org_id, app_id, manifest, token.as_deref()).await
}

/// `EYES_TOKEN`, trimmed; empty is treated as absent.
fn token_from_env() -> Option<String> {
    std::env::var("EYES_TOKEN")
        .ok()
        .map(|s| s.trim().to_string())
        .filter(|s| !s.is_empty())
}

fn uuid_from_env(var: &str) -> Result<Uuid, ManifestError> {
    let value = std::env::var(var)
        .map_err(|_| ManifestError::Configuration(format!("{var} must be set")))?;
    Uuid::parse_str(value.trim())
        .map_err(|e| ManifestError::Configuration(format!("{var} is not a valid UUID: {e}")))
}

#[derive(Debug, Clone, Serialize)]
pub struct ProcessSignalPayload {
    pub role: String,
    pub app_version: Option<String>,
    pub git_sha: Option<String>,
}
#[derive(Debug, thiserror::Error)]
pub enum ProcessSignalError {
    #[error("Configuration error: {0}")]
    Configuration(String),
    #[error("HTTP request failed: {0}")]
    Request(#[from] reqwest::Error),
    #[error("Server returned error status: {0}")]
    Status(reqwest::StatusCode),
}
/// Everything a process signal (heartbeat or shutdown) needs besides the HTTP
/// client.
///
/// A struct rather than eight positional parameters: adding `auth_token` to
/// the old parameter list pushed both functions past clippy's
/// `too_many_arguments` threshold, and these fields always travel together
/// (they're all fields of [`ProcessHeartbeatConfig`]).
#[derive(Clone, Copy)]
pub struct ProcessSignal<'a> {
    pub base_url: &'a Url,
    pub org_id: Uuid,
    pub app_id: Uuid,
    pub identity: &'a ProcessIdentity,
    pub app_version: Option<&'a str>,
    pub git_sha: Option<&'a str>,
    pub auth_token: Option<&'a str>,
}

// Hand-written so the bearer token is never printed. See `crate::RedactedToken`.
impl std::fmt::Debug for ProcessSignal<'_> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ProcessSignal")
            .field("base_url", &self.base_url)
            .field("org_id", &self.org_id)
            .field("app_id", &self.app_id)
            .field("identity", &self.identity)
            .field("app_version", &self.app_version)
            .field("git_sha", &self.git_sha)
            .field("auth_token", &crate::RedactedToken(self.auth_token))
            .finish()
    }
}

impl ProcessSignal<'_> {
    async fn send(
        &self,
        client: &reqwest::Client,
        endpoint: &str,
    ) -> Result<(), ProcessSignalError> {
        let url = self
            .base_url
            .join(&format!(
                "/api/orgs/{}/apps/{}/process-instances/{}/{endpoint}",
                self.org_id,
                self.app_id,
                self.identity.instance_id()
            ))
            .map_err(|e| ProcessSignalError::Configuration(e.to_string()))?;
        let mut request = client.post(url).json(&ProcessSignalPayload {
            role: self.identity.role().to_owned(),
            app_version: self.app_version.map(str::to_owned),
            git_sha: self.git_sha.map(str::to_owned),
        });
        if let Some(token) = self.auth_token {
            request = request.bearer_auth(token);
        }
        let response = request.send().await?;
        if !response.status().is_success() {
            return Err(ProcessSignalError::Status(response.status()));
        }
        Ok(())
    }
}

pub async fn send_process_heartbeat(
    client: &reqwest::Client,
    signal: ProcessSignal<'_>,
) -> Result<(), ProcessSignalError> {
    signal.send(client, "heartbeat").await
}

pub async fn send_process_shutdown(
    client: &reqwest::Client,
    signal: ProcessSignal<'_>,
) -> Result<(), ProcessSignalError> {
    signal.send(client, "shutdown").await
}

#[derive(Clone)]
pub struct ProcessHeartbeatConfig {
    pub base_url: Url,
    pub org_id: Uuid,
    pub app_id: Uuid,
    pub identity: ProcessIdentity,
    pub app_version: Option<String>,
    pub git_sha: Option<String>,
    pub heartbeat_interval: Duration,
    pub request_timeout: Duration,
    pub shutdown_timeout: Duration,
    /// Bearer token for the heartbeat/shutdown endpoints. Defaults to
    /// `EYES_TOKEN`; override with [`ProcessHeartbeatConfig::with_token`].
    pub token: Option<String>,
}

// Hand-written so the bearer token is never printed. See `crate::RedactedToken`.
impl std::fmt::Debug for ProcessHeartbeatConfig {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ProcessHeartbeatConfig")
            .field("base_url", &self.base_url)
            .field("org_id", &self.org_id)
            .field("app_id", &self.app_id)
            .field("identity", &self.identity)
            .field("app_version", &self.app_version)
            .field("git_sha", &self.git_sha)
            .field("heartbeat_interval", &self.heartbeat_interval)
            .field("request_timeout", &self.request_timeout)
            .field("shutdown_timeout", &self.shutdown_timeout)
            .field("token", &crate::RedactedToken(self.token.as_deref()))
            .finish()
    }
}

impl ProcessHeartbeatConfig {
    pub fn new(
        base_url: Url,
        org_id: Uuid,
        app_id: Uuid,
        identity: ProcessIdentity,
        heartbeat_interval: Duration,
    ) -> Self {
        let request_timeout = Duration::from_secs(10).min(heartbeat_interval / 2);
        Self {
            base_url,
            org_id,
            app_id,
            identity,
            app_version: None,
            git_sha: None,
            heartbeat_interval,
            request_timeout,
            shutdown_timeout: Duration::from_secs(5),
            token: token_from_env(),
        }
    }

    /// Override the bearer token, which otherwise defaults to `EYES_TOKEN`.
    pub fn with_token(mut self, token: impl Into<String>) -> Self {
        self.token = Some(token.into());
        self
    }

    /// Borrow this config as a [`ProcessSignal`] for
    /// [`send_process_heartbeat`] / [`send_process_shutdown`].
    pub fn signal(&self) -> ProcessSignal<'_> {
        ProcessSignal {
            base_url: &self.base_url,
            org_id: self.org_id,
            app_id: self.app_id,
            identity: &self.identity,
            app_version: self.app_version.as_deref(),
            git_sha: self.git_sha.as_deref(),
            auth_token: self.token.as_deref(),
        }
    }
    pub fn from_manifest(
        base_url: Url,
        org_id: Uuid,
        app_id: Uuid,
        manifest: &AppManifest,
    ) -> Result<Self, ProcessSignalError> {
        let id = manifest.process_instance_id.ok_or_else(|| {
            ProcessSignalError::Configuration("manifest process identity is required".into())
        })?;
        let role = manifest.process_role.clone().ok_or_else(|| {
            ProcessSignalError::Configuration("manifest process role is required".into())
        })?;
        let matches: Vec<_> = manifest
            .expected_process_roles
            .as_deref()
            .unwrap_or_default()
            .iter()
            .filter(|r| r.enabled && r.role == role)
            .collect();
        if matches.len() != 1 {
            return Err(ProcessSignalError::Configuration(
                "exactly one enabled declaration must match the process role".into(),
            ));
        }
        let mut c = Self::new(
            base_url,
            org_id,
            app_id,
            ProcessIdentity(Arc::new(ProcessIdentityInner {
                instance_id: id,
                role,
            })),
            Duration::from_secs(matches[0].heartbeat_interval_seconds),
        );
        c.app_version = manifest.app_version.clone();
        c.git_sha = manifest.git_sha.clone();
        c.validate()?;
        Ok(c)
    }
    fn validate(&self) -> Result<(), ProcessSignalError> {
        if self.heartbeat_interval.is_zero()
            || self.request_timeout.is_zero()
            || self.request_timeout >= self.heartbeat_interval
        {
            return Err(ProcessSignalError::Configuration(
                "invalid heartbeat interval or request timeout".into(),
            ));
        }
        Ok(())
    }
}
pub struct ProcessHeartbeatHandle {
    cancel: oneshot::Sender<()>,
    task: JoinHandle<()>,
    config: ProcessHeartbeatConfig,
    client: reqwest::Client,
}
pub struct ProcessHeartbeat;
impl ProcessHeartbeat {
    pub fn spawn(
        config: ProcessHeartbeatConfig,
    ) -> Result<ProcessHeartbeatHandle, ProcessSignalError> {
        config.validate()?;
        let client = reqwest::Client::builder()
            .timeout(config.request_timeout)
            .build()?;
        let worker_client = client.clone();
        let worker_config = config.clone();
        let (cancel, mut cancellation) = oneshot::channel();
        let task = tokio::spawn(async move {
            let mut interval = tokio::time::interval(worker_config.heartbeat_interval);
            interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
            loop {
                tokio::select! {_=&mut cancellation=>break,_=interval.tick()=>{let result=send_process_heartbeat(&worker_client,worker_config.signal()).await;if let Err(error)=result{tracing::warn!(%error,"Eyes process heartbeat failed");}}}
            }
        });
        Ok(ProcessHeartbeatHandle {
            cancel,
            task,
            config,
            client,
        })
    }
}
impl ProcessHeartbeatHandle {
    pub async fn shutdown(self) -> Result<(), ProcessSignalError> {
        let _ = self.cancel.send(());
        self.task.abort();
        let send = send_process_shutdown(&self.client, self.config.signal());
        tokio::time::timeout(self.config.shutdown_timeout, send)
            .await
            .map_err(|_| {
                ProcessSignalError::Configuration("process shutdown timed out".into())
            })??;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::dashboards::{DashboardItem, DashboardLink, DashboardSection};

    #[test]
    fn test_manifest_payload_serialization_shape() {
        let manifest = AppManifest::default()
            .app_version("1.2.3")
            .git_sha("abc123")
            .jobs(vec!["SendEmail".to_string(), "RefreshCache".to_string()])
            .critical_jobs(vec!["SendEmail".to_string(), "Unknown".to_string()])
            .crons(vec![CronEntry {
                name: "DailyDigest".to_string(),
                schedule: "0 0 * * *".to_string(),
            }]);

        let booted_at = Utc::now();
        let payload = ManifestPayload::new(&manifest, booted_at);
        let json = serde_json::to_value(&payload).unwrap();

        assert_eq!(
            json,
            serde_json::json!({
                "manifest_version": 2,
                "app_version": "1.2.3",
                "git_sha": "abc123",
                "jobs": [
                    { "name": "SendEmail", "critical": true },
                    { "name": "RefreshCache", "critical": false },
                ],
                "crons": [
                    { "name": "DailyDigest", "schedule": "0 0 * * *" },
                ],
                "base_url": null,
                "monitors": null,
                "process_instance_id": null,
                "process_role": null,
                "expected_process_roles": null,
                "metrics": null,
                "metric_thresholds": null,
                "dashboards": null,
                "booted_at": serde_json::to_value(booted_at).unwrap(),
            })
        );
    }

    #[test]
    fn test_empty_manifest_payload_serialization_shape() {
        let manifest = AppManifest::default();
        let booted_at = Utc::now();
        let json = serde_json::to_value(ManifestPayload::new(&manifest, booted_at)).unwrap();

        assert_eq!(json["manifest_version"], 2);
        assert_eq!(json["app_version"], serde_json::Value::Null);
        assert_eq!(json["git_sha"], serde_json::Value::Null);
        assert_eq!(json["jobs"], serde_json::json!([]));
        assert_eq!(json["crons"], serde_json::json!([]));
        assert_eq!(json["metrics"], serde_json::Value::Null);
        assert_eq!(json["metric_thresholds"], serde_json::Value::Null);
        assert_eq!(json["dashboards"], serde_json::Value::Null);
        assert!(json["booted_at"].is_string());
    }

    /// BLOCKER regression: `ManifestPayload` is the serialization boundary —
    /// it copies fields explicitly, and anything it doesn't copy is silently
    /// dropped. An `AppManifest` with `metric_thresholds` must transmit them.
    #[test]
    fn test_manifest_payload_transmits_metric_thresholds() {
        let threshold =
            crate::metrics::MetricThresholdBuilder::new("err_rate_high", "http.request_count")
                .window_seconds(300)
                .evaluation_interval_seconds(60)
                .critical(crate::metrics::ThresholdComparison::Above, 100.0)
                .build()
                .unwrap();
        let manifest = AppManifest::default().metric_thresholds(vec![threshold]);
        let booted_at = Utc::now();
        let json = serde_json::to_value(ManifestPayload::new(&manifest, booted_at)).unwrap();

        let transmitted = json["metric_thresholds"]
            .as_array()
            .expect("metric_thresholds must serialize as an array");
        assert_eq!(transmitted.len(), 1);
        assert_eq!(transmitted[0]["id"], "err_rate_high");
        assert_eq!(transmitted[0]["metric_id"], "http.request_count");
        assert_eq!(transmitted[0]["critical"]["comparison"], "above");

        // `Some(vec![])` serializes as `[]`, preserving the remove-all
        // authority meaning.
        let manifest = AppManifest::default().metric_thresholds(vec![]);
        let json = serde_json::to_value(ManifestPayload::new(&manifest, booted_at)).unwrap();
        assert_eq!(json["metric_thresholds"], serde_json::json!([]));
    }

    /// BLOCKER regression: a severity threshold built from `1.0f64`
    /// serializes as a floating JSON number (`1.0`, not `1`) — pinning that
    /// the server's value-based (not representation-based) validation is
    /// the load-bearing wire compatibility.
    #[test]
    fn test_metric_threshold_number_representation_is_floating() {
        let threshold = crate::metrics::MetricThresholdBuilder::new("t", "m")
            .window_seconds(60)
            .evaluation_interval_seconds(60)
            .critical(crate::metrics::ThresholdComparison::Above, 1.0)
            .build()
            .unwrap();
        let text = serde_json::to_string(&threshold).unwrap();
        assert!(
            text.contains("\"threshold\":1.0"),
            "serialized {text} must carry 1.0"
        );
        let json = serde_json::to_value(&threshold).unwrap();
        // And the JSON number is a float, not an integer — this is what
        // makes `Number::is_i64()` false on the server side.
        assert!(
            !json["critical"]["threshold"].as_u64().is_some()
                || json["critical"]["threshold"] == serde_json::json!(1.0)
        );
        assert_eq!(json["critical"]["threshold"].as_f64(), Some(1.0));
    }

    /// The serialized `metrics[0]` object, key-for-key against the server's
    /// `ManifestNamedMetric` JSON: if either side drifts, this fails instead
    /// of an ingest-time 400 in production.
    #[test]
    fn test_manifest_payload_metrics_serialization_shape() {
        let request_count = crate::metrics::NamedMetricBuilder::new(
            "http.request_count",
            eyes_query::AggregateFunction::Count,
            None,
        )
        .unwrap()
        .filter_eq("semantic_kind", "http.request")
        .unwrap()
        .time_bucket(300)
        .display_name("Request count")
        .description("HTTP requests served")
        .unit("requests")
        .build()
        .unwrap();
        let latency = crate::metrics::NamedMetricBuilder::new(
            "http.request_latency",
            eyes_query::AggregateFunction::P95,
            Some("duration"),
        )
        .unwrap()
        .filter_eq("semantic_kind", "http.request")
        .unwrap()
        .unit("µs")
        .original_dsl(
            "telemetry | where semantic_kind == \"http.request\" | stats p95(duration) as value",
        )
        .build()
        .unwrap();
        let queue_depth = crate::metrics::NamedMetricBuilder::new(
            "queue.depth",
            eyes_query::AggregateFunction::Max,
            Some("fields.queue_depth"),
        )
        .unwrap()
        .filter_numeric("fields.queue_depth")
        .unwrap()
        .build()
        .unwrap();

        let manifest = AppManifest::default().metrics(vec![request_count, latency, queue_depth]);
        let booted_at = Utc::now();
        let json = serde_json::to_value(ManifestPayload::new(&manifest, booted_at)).unwrap();

        assert_eq!(
            json["metrics"][0],
            serde_json::json!({
                "id": "http.request_count",
                "display_name": "Request count",
                "description": "HTTP requests served",
                "unit": "requests",
                "value_type": "int",
                "result_shape": "time_series",
                "preferred_bucket_seconds": 300,
                "query": {
                    "version": 1,
                    "query": {
                        "source": "telemetry",
                        "stages": [
                            {
                                "stage": "where",
                                "predicate": {
                                    "type": "binary",
                                    "op": "eq",
                                    "left": { "type": "path", "path": { "root": "semantic_kind", "segments": [] } },
                                    "right": { "type": "literal", "value": { "type": "string", "value": "http.request" } }
                                }
                            },
                            {
                                "stage": "aggregate",
                                "aggregates": [{ "function": "count", "argument": null, "alias": "value" }],
                                "groups": [
                                    {
                                        "expr": {
                                            "type": "call",
                                            "function": "time_bucket",
                                            "args": [
                                                { "type": "path", "path": { "root": "timestamp", "segments": [] } },
                                                { "type": "literal", "value": { "type": "duration", "value": 300000000 } }
                                            ]
                                        },
                                        "alias": "bucket"
                                    }
                                ]
                            }
                        ]
                    }
                },
                "original_dsl": null,
            })
        );

        // A duration-valued declaration and one carrying a numeric
        // restriction stage (the `filter_numeric` spelling) round out the
        // key-for-key coverage: value_type duration, scalar shape, and the
        // gte stage Rust builders and direct JSON clients must agree on.
        assert_eq!(json["metrics"][1]["value_type"], "duration");
        assert_eq!(json["metrics"][1]["result_shape"], "scalar");
        assert!(json["metrics"][1]["original_dsl"]
            .as_str()
            .unwrap()
            .contains("p95"));
        assert_eq!(
            json["metrics"][2]["query"]["query"]["stages"][0],
            serde_json::json!({
                "stage": "where",
                "predicate": {
                    "type": "binary",
                    "op": "gte",
                    "left": { "type": "path", "path": { "root": "fields", "segments": [{ "type": "key", "value": "queue_depth" }] } },
                    "right": { "type": "literal", "value": { "type": "float", "value": -1.7976931348623157e308 } }
                }
            })
        );
    }

    /// The serialized `dashboards[0]` object, key-for-key against the
    /// server's `ManifestDashboard` JSON: if either side drifts, this fails
    /// instead of an ingest-time 400 in production. The expected object is
    /// built from a real `NamedDashboard`/`DashboardItem` value — the
    /// payload and the expectation never share a hand-written mental
    /// model.
    #[test]
    fn test_manifest_payload_dashboards_serialization_shape() {
        let dashboard = NamedDashboard::new("api-overview", "API Overview")
            .unwrap()
            .description("Front-door health for the HTTP API")
            .default_range_seconds(3600)
            .section(
                DashboardSection::new()
                    .title("Traffic")
                    .item(
                        DashboardItem::stat("http.request_count.total")
                            .label("Requests")
                            .unit("requests"),
                    )
                    .item(
                        DashboardItem::time_series("http.request_count")
                            .unit("requests")
                            .preferred_bucket_seconds(300),
                    )
                    .item(DashboardItem::links(vec![DashboardLink::new(
                        "Requests view",
                        "/orgs/00000000-0000-0000-0000-000000000000/apps/11111111-1111-1111-1111-111111111111/requests",
                    )
                    .unwrap()])),
            );
        let manifest = AppManifest::default().dashboards(vec![dashboard.clone()]);
        let booted_at = Utc::now();
        let json = serde_json::to_value(ManifestPayload::new(&manifest, booted_at)).unwrap();

        assert_eq!(
            json["dashboards"][0],
            serde_json::to_value(&dashboard).unwrap()
        );
        assert_eq!(
            json["dashboards"][0],
            serde_json::json!({
                "id": "api-overview",
                "title": "API Overview",
                "description": "Front-door health for the HTTP API",
                "default_range_seconds": 3600,
                "sections": [
                    {
                        "title": "Traffic",
                        "description": null,
                        "items": [
                            {
                                "kind": "stat",
                                "query_id": "http.request_count.total",
                                "label": "Requests",
                                "unit": "requests"
                            },
                            {
                                "kind": "time_series",
                                "query_id": "http.request_count",
                                "label": null,
                                "unit": "requests",
                                "preferred_bucket_seconds": 300
                            },
                            {
                                "kind": "links",
                                "links": [
                                    {
                                        "label": "Requests view",
                                        "href": "/orgs/00000000-0000-0000-0000-000000000000/apps/11111111-1111-1111-1111-111111111111/requests"
                                    }
                                ]
                            }
                        ]
                    }
                ]
            })
        );

        // `Some(vec![])` serializes as `[]`, preserving the remove-all
        // authority meaning.
        let manifest = AppManifest::default().dashboards(vec![]);
        let json = serde_json::to_value(ManifestPayload::new(&manifest, booted_at)).unwrap();
        assert_eq!(json["dashboards"], serde_json::json!([]));
    }

    /// Structural equality survived the `Eq` removal: two independently
    /// built manifests with equivalent metrics compare equal, and a query
    /// difference makes them unequal. Also the compile-time guard that the
    /// `PartialEq` derive stays present.
    #[test]
    fn manifest_partial_eq_survives_the_eq_removal() {
        fn build() -> AppManifest {
            let metric = crate::metrics::NamedMetricBuilder::new(
                "http.request_count",
                eyes_query::AggregateFunction::Count,
                None,
            )
            .unwrap()
            .filter_eq("semantic_kind", "http.request")
            .unwrap()
            .time_bucket(300)
            .build()
            .unwrap();
            AppManifest::default()
                .app_version("1.0.0")
                .metrics(vec![metric])
        }
        assert_eq!(build(), build());

        let changed = {
            let metric = crate::metrics::NamedMetricBuilder::new(
                "http.request_count",
                eyes_query::AggregateFunction::Count,
                None,
            )
            .unwrap()
            .filter_eq("semantic_kind", "http.request")
            .unwrap()
            .filter_eq("level", "INFO")
            .unwrap()
            .time_bucket(300)
            .build()
            .unwrap();
            AppManifest::default()
                .app_version("1.0.0")
                .metrics(vec![metric])
        };
        assert_ne!(build(), changed);
    }

    #[test]
    fn resolves_absolute_and_root_relative_targets() {
        assert_eq!(
            resolve_monitor_target(Some("https://example.com/app/"), "/health?full=1")
                .unwrap()
                .as_str(),
            "https://example.com/health?full=1"
        );
        assert_eq!(
            resolve_monitor_target(Some("https://example.com/app"), "/health")
                .unwrap()
                .as_str(),
            "https://example.com/health"
        );
        assert_eq!(
            resolve_monitor_target(None, "https://status.example.net/ping")
                .unwrap()
                .as_str(),
            "https://status.example.net/ping"
        );
    }

    #[test]
    fn rejects_unsafe_or_ambiguous_targets() {
        assert_eq!(
            resolve_monitor_target(None, "/health"),
            Err(MonitorTargetError::MissingBaseUrl)
        );
        assert_eq!(
            resolve_monitor_target(None, "health"),
            Err(MonitorTargetError::NonRootRelative)
        );
        assert_eq!(
            resolve_monitor_target(Some("https://example.com"), "//evil.example"),
            Err(MonitorTargetError::NetworkPath)
        );
        for target in [
            "/\\evil.example/x",
            "/\t/evil.example/x",
            "/\n/evil.example/x",
            "ftp://example.com/a",
            "https://user@example.com/a",
            "https://example.com/a#fragment",
            "http://localhost/a",
            "http://api.localhost/a",
            "http://localhost./a",
            "http://127.0.0.1/a",
            "http://10.0.0.1/a",
            "http://169.254.1.1/a",
            "http://192.0.2.1/a",
            "http://0.1.2.3/a",
            "http://100.64.1.1/a",
            "http://192.0.0.1/a",
            "http://192.88.99.1/a",
            "http://198.18.0.1/a",
            "http://240.0.0.1/a",
            "http://[::1]/a",
            "http://[::7f00:1]/a",
            "http://[fc00::1]/a",
            "http://[2001:db8::1]/a",
            "http://[fec0::1]/a",
            "http://[64:ff9b::c000:201]/a",
            "http://[3fff::1]/a",
            "http://[5f00::1]/a",
        ] {
            assert!(
                resolve_monitor_target(None, target).is_err(),
                "accepted {target}"
            );
        }
        for target in [
            "https://1.1.1.1/a",
            "https://8.8.8.8/a",
            "https://[2606:4700:4700::1111]/a",
        ] {
            assert!(
                resolve_monitor_target(None, target).is_ok(),
                "rejected public target {target}"
            );
        }
    }

    #[test]
    fn monitor_builder_serializes_stable_defaults() {
        let monitor = HttpMonitor::new("public-health", "/health");
        assert_eq!(
            serde_json::to_value(monitor).unwrap(),
            serde_json::json!({
                "id": "public-health", "target": "/health", "method": "GET",
                "interval_seconds": 60, "timeout_seconds": 10,
                "expected_status_min": 200, "expected_status_max": 299,
                "failure_threshold": 3, "enabled": true
            })
        );
    }
}