harmony/harmony-reconciler-contracts/src/fleet.rs

//! Fleet-scale wire-format types.
//!
//! Per-concern payloads on dedicated NATS KV buckets:
//!
//! | Type | Bucket | Cadence |
//! |------|--------|---------|
//! | [`DeviceInfo`] | KV `device-info` | on startup + label/inventory change |
//! | [`DeploymentState`] | KV `device-state` | on reconcile phase transition |
//! | [`HeartbeatPayload`] | KV `device-heartbeat` | every 30 s |
//!
//! The operator watches `device-state` directly — KV watch deliveries
//! are ordered and last-writer-wins, so there's no separate event
//! stream or per-write revision to track.

use std::collections::BTreeMap;
use std::fmt;

use chrono::{DateTime, Utc};
use harmony_types::id::Id;
use serde::{Deserialize, Deserializer, Serialize};

use crate::status::{InventorySnapshot, Phase};

/// Deployment CR `metadata.name`, validated for NATS-subject safety.
///
/// Scope: what identifies a Deployment to the agent. Appears in KV
/// keys (`state.<device>.<deployment>`) and every in-memory map
/// keyed by "which deployment." A raw `String` here would let an
/// invalid name (containing a `.`, splitting into extra subject
/// tokens) break routing at runtime.
///
/// Validation:
/// - Not empty.
/// - No `.` (would alias an extra subject token).
/// - No `*` / `>` (NATS wildcards).
/// - No ASCII whitespace.
/// - ≤ 253 bytes (RFC 1123 max, matches Kubernetes name limit).
///
/// The constructor is fallible; deserialization runs the same
/// validation so malformed payloads are rejected at the wire.
#[derive(Debug, Clone, Hash, PartialEq, Eq, Ord, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct DeploymentName(String);

#[derive(Debug, thiserror::Error, PartialEq, Eq)]
pub enum InvalidDeploymentName {
    #[error("deployment name must not be empty")]
    Empty,
    #[error("deployment name must not exceed 253 bytes")]
    TooLong,
    #[error("deployment name must not contain '.' (would alias an extra NATS subject token)")]
    ContainsDot,
    #[error("deployment name must not contain NATS wildcards '*' or '>'")]
    ContainsWildcard,
    #[error("deployment name must not contain whitespace")]
    ContainsWhitespace,
}

impl DeploymentName {
    pub fn try_new(s: impl Into<String>) -> Result<Self, InvalidDeploymentName> {
        let s = s.into();
        if s.is_empty() {
            return Err(InvalidDeploymentName::Empty);
        }
        if s.len() > 253 {
            return Err(InvalidDeploymentName::TooLong);
        }
        if s.contains('.') {
            return Err(InvalidDeploymentName::ContainsDot);
        }
        if s.contains('*') || s.contains('>') {
            return Err(InvalidDeploymentName::ContainsWildcard);
        }
        if s.chars().any(|c| c.is_ascii_whitespace()) {
            return Err(InvalidDeploymentName::ContainsWhitespace);
        }
        Ok(Self(s))
    }

    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl fmt::Display for DeploymentName {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(&self.0)
    }
}

impl<'de> Deserialize<'de> for DeploymentName {
    fn deserialize<D: Deserializer<'de>>(de: D) -> Result<Self, D::Error> {
        let s = String::deserialize(de)?;
        Self::try_new(s).map_err(serde::de::Error::custom)
    }
}

/// Static-ish per-device facts: routing labels, hardware, agent
/// version. Written to KV key `info.<device_id>` in
/// [`crate::BUCKET_DEVICE_INFO`]. Rewritten by the agent on startup
/// and whenever its labels change — **not** on every heartbeat.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct DeviceInfo {
    pub device_id: Id,
    /// Routing labels. Operator resolves Deployment
    /// `targetSelector.matchLabels` against this map.
    #[serde(default)]
    pub labels: BTreeMap<String, String>,
    /// Hardware / OS snapshot. `None` until the first post-startup
    /// publish.
    #[serde(default)]
    pub inventory: Option<InventorySnapshot>,
    /// RFC 3339 UTC timestamp of this publish.
    pub updated_at: DateTime<Utc>,
}

/// Authoritative current phase for one `(device, deployment)` pair.
/// Written to KV key `state.<device_id>.<deployment>` in
/// [`crate::BUCKET_DEVICE_STATE`]. Deleted when the deployment is
/// removed from the device.
///
/// The operator's KV watch sees every write + delete in order, so
/// this value alone — plus the operator's in-memory belief about
/// the last phase for the pair — is enough to drive the aggregate
/// counters. No separate event stream, no per-write revision.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct DeploymentState {
    pub device_id: Id,
    pub deployment: DeploymentName,
    pub phase: Phase,
    pub last_event_at: DateTime<Utc>,
    #[serde(default)]
    pub last_error: Option<String>,
}

/// Tiny liveness ping. Written to KV key `heartbeat.<device_id>` in
/// [`crate::BUCKET_DEVICE_HEARTBEAT`].
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct HeartbeatPayload {
    pub device_id: Id,
    pub at: DateTime<Utc>,
}

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

    fn ts(s: &str) -> DateTime<Utc> {
        DateTime::parse_from_rfc3339(s).unwrap().with_timezone(&Utc)
    }

    fn dn(s: &str) -> DeploymentName {
        DeploymentName::try_new(s).expect("valid")
    }

    #[test]
    fn deployment_name_accepts_rfc1123() {
        assert!(DeploymentName::try_new("hello-world").is_ok());
        assert!(DeploymentName::try_new("a").is_ok());
        assert!(DeploymentName::try_new("a-b-c-1-2-3").is_ok());
    }

    #[test]
    fn deployment_name_rejects_dot() {
        assert_eq!(
            DeploymentName::try_new("hello.world"),
            Err(InvalidDeploymentName::ContainsDot)
        );
    }

    #[test]
    fn deployment_name_rejects_nats_wildcards() {
        assert_eq!(
            DeploymentName::try_new("hello*"),
            Err(InvalidDeploymentName::ContainsWildcard)
        );
        assert_eq!(
            DeploymentName::try_new("hello>"),
            Err(InvalidDeploymentName::ContainsWildcard)
        );
    }

    #[test]
    fn deployment_name_rejects_empty_and_too_long() {
        assert_eq!(
            DeploymentName::try_new(""),
            Err(InvalidDeploymentName::Empty)
        );
        assert_eq!(
            DeploymentName::try_new("x".repeat(254)),
            Err(InvalidDeploymentName::TooLong)
        );
    }

    #[test]
    fn deployment_name_rejects_whitespace() {
        assert_eq!(
            DeploymentName::try_new("hello world"),
            Err(InvalidDeploymentName::ContainsWhitespace)
        );
        assert_eq!(
            DeploymentName::try_new("hello\tworld"),
            Err(InvalidDeploymentName::ContainsWhitespace)
        );
    }

    #[test]
    fn deployment_name_deserialization_validates() {
        let json = r#""bad.name""#;
        let result: Result<DeploymentName, _> = serde_json::from_str(json);
        assert!(result.is_err());
    }

    #[test]
    fn deployment_name_roundtrip() {
        let name = dn("hello-world");
        let json = serde_json::to_string(&name).unwrap();
        assert_eq!(json, r#""hello-world""#);
        let back: DeploymentName = serde_json::from_str(&json).unwrap();
        assert_eq!(name, back);
    }

    #[test]
    fn deployment_state_roundtrip() {
        let original = DeploymentState {
            device_id: Id::from("pi-01".to_string()),
            deployment: dn("hello-web"),
            phase: Phase::Failed,
            last_event_at: ts("2026-04-22T10:05:00Z"),
            last_error: Some("image pull 429".to_string()),
        };
        let json = serde_json::to_string(&original).unwrap();
        let back: DeploymentState = serde_json::from_str(&json).unwrap();
        assert_eq!(original, back);
    }

    #[test]
    fn heartbeat_is_tiny() {
        let hb = HeartbeatPayload {
            device_id: Id::from("pi-01".to_string()),
            at: ts("2026-04-22T10:00:30Z"),
        };
        let bytes = serde_json::to_vec(&hb).unwrap();
        assert!(
            bytes.len() < 96,
            "heartbeat payload grew to {} bytes: {}",
            bytes.len(),
            String::from_utf8_lossy(&bytes),
        );
    }

    #[test]
    fn device_info_roundtrip() {
        let original = DeviceInfo {
            device_id: Id::from("pi-01".to_string()),
            labels: BTreeMap::from([("group".to_string(), "site-a".to_string())]),
            inventory: Some(InventorySnapshot {
                hostname: "pi-01".to_string(),
                arch: "aarch64".to_string(),
                os: "Ubuntu 24.04".to_string(),
                kernel: "6.8.0".to_string(),
                cpu_cores: 4,
                memory_mb: 8192,
                agent_version: "0.1.0".to_string(),
            }),
            updated_at: ts("2026-04-22T10:00:00Z"),
        };
        let json = serde_json::to_string(&original).unwrap();
        let back: DeviceInfo = serde_json::from_str(&json).unwrap();
        assert_eq!(original, back);
    }
}