ontoref/crates/ontoref-daemon/src/nats.rs

#[cfg(feature = "nats")]
use std::path::PathBuf;
#[cfg(feature = "nats")]
use std::process::Command;
#[cfg(feature = "nats")]
use std::time::{SystemTime, UNIX_EPOCH};

#[cfg(feature = "nats")]
use anyhow::{anyhow, Result};
#[cfg(feature = "nats")]
use bytes::Bytes;
#[cfg(feature = "nats")]
use platform_nats::{EventStream, NatsConnectionConfig, TopologyConfig};
#[cfg(feature = "nats")]
use serde_json::{json, Value};
#[cfg(feature = "nats")]
use tracing::{info, warn};

/// NATS JetStream publisher for daemon lifecycle events.
///
/// Uses platform-nats `connect_client()` — connection + auth only.
/// Stream/consumer topology comes entirely from `nats/streams.json` (or
/// equivalent), referenced by `nats_events.streams_config` in
/// `.ontoref/config.ncl`.
///
/// Gracefully degrades if NATS is unavailable or disabled in config.
#[cfg(feature = "nats")]
pub struct NatsPublisher {
    stream: EventStream,
    project: String,
    port: u16,
}

#[cfg(feature = "nats")]
impl NatsPublisher {
    /// Connect to NATS JetStream, apply topology from config, bind consumer.
    /// Reads `nats_events` section from `.ontoref/config.ncl`.
    /// Returns `Ok(None)` if disabled or unavailable (graceful degradation).
    pub async fn connect(
        config_path: &PathBuf,
        project: String,
        port: u16,
    ) -> Result<Option<Self>> {
        let config = load_nats_config(config_path)?;

        let nats_section = match config.get("nats_events") {
            Some(section) => section,
            None => return Ok(None),
        };

        let enabled = nats_section
            .get("enabled")
            .and_then(|e| e.as_bool())
            .unwrap_or(false);

        if !enabled {
            return Ok(None);
        }

        let url = nats_section
            .get("url")
            .and_then(|u| u.as_str())
            .unwrap_or("nats://localhost:4222")
            .to_string();

        let nkey_seed = nats_section
            .get("nkey_seed")
            .and_then(|s| s.as_str())
            .map(|s| s.to_string());

        let require_signed = nats_section
            .get("require_signed_messages")
            .and_then(|r| r.as_bool())
            .unwrap_or(false);

        let trusted_nkeys = nats_section
            .get("trusted_nkeys")
            .and_then(|t| t.as_array())
            .map(|arr| {
                arr.iter()
                    .filter_map(|v| v.as_str().map(String::from))
                    .collect()
            })
            .unwrap_or_default();

        let conn_cfg = NatsConnectionConfig {
            url: url.clone(),
            nkey_seed,
            require_signed_messages: require_signed,
            trusted_nkeys,
        };

        let mut stream = match tokio::time::timeout(
            std::time::Duration::from_secs(3),
            EventStream::connect_client(&conn_cfg),
        )
        .await
        {
            Ok(Ok(s)) => s,
            Ok(Err(e)) => {
                warn!(error = %e, url = %url, "NATS connection failed — running without events");
                return Ok(None);
            }
            Err(_) => {
                warn!(url = %url, "NATS connection timed out — running without events");
                return Ok(None);
            }
        };

        info!(url = %url, "NATS connected");

        // Apply topology from streams_config file declared in project config.
        // Fallback: NATS_STREAMS_CONFIG env var.
        let topology_path = nats_section
            .get("streams_config")
            .and_then(|s| s.as_str())
            .map(std::path::PathBuf::from);

        let topology = match TopologyConfig::load(topology_path.as_deref()) {
            Ok(Some(t)) => Some(t),
            Ok(None) => {
                warn!("no topology config found — publish-only mode (no consumer bound)");
                None
            }
            Err(e) => {
                warn!(error = %e, "topology config load failed — publish-only mode");
                None
            }
        };

        if let Some(ref topo) = topology {
            match stream.apply_topology(topo).await {
                Ok(report) => info!(
                    streams = report.streams_applied,
                    consumers = report.consumers_applied,
                    "topology applied"
                ),
                Err(e) => warn!(error = %e, "topology apply failed — publish-only mode"),
            }
        }

        // Bind to daemon consumer on the first stream (convention: "daemon-{project}").
        if let Some(ref topo) = topology {
            if let Some(first_stream) = topo.streams.first() {
                let consumer_name = format!("daemon-{project}");
                if let Err(e) = stream
                    .bind_consumer(&first_stream.name, &consumer_name)
                    .await
                {
                    warn!(error = %e, "failed to bind daemon consumer — pull_events disabled");
                }
            }
        }

        Ok(Some(Self {
            stream,
            project,
            port,
        }))
    }

    pub async fn publish_started(&self) -> Result<()> {
        let payload = json!({
            "project": self.project,
            "port": self.port,
            "timestamp": iso8601_now(),
        });
        self.stream
            .publish("ecosystem.daemon.started", Bytes::from(payload.to_string()))
            .await?;
        info!(port = self.port, "published daemon.started");
        Ok(())
    }

    pub async fn publish_stopped(&self, uptime_secs: u64) -> Result<()> {
        let payload = json!({
            "project": self.project,
            "uptime_seconds": uptime_secs,
            "timestamp": iso8601_now(),
        });
        self.stream
            .publish("ecosystem.daemon.stopped", Bytes::from(payload.to_string()))
            .await?;
        info!(uptime = uptime_secs, "published daemon.stopped");
        Ok(())
    }

    /// Publish a file change notification for a specific project and event
    /// type.
    pub async fn publish_notification(
        &self,
        project: &str,
        event: &crate::notifications::NotificationEvent,
        files: &[String],
    ) -> Result<()> {
        let subject = format!("ecosystem.{project}.{}", event.nats_suffix());
        let payload = json!({
            "project": project,
            "event": format!("{event:?}"),
            "files": files,
            "timestamp": iso8601_now(),
        });
        self.stream
            .publish(&subject, Bytes::from(payload.to_string()))
            .await?;
        info!(subject = %subject, files = files.len(), "published notification");
        Ok(())
    }

    /// Publish an actor registration event.
    pub async fn publish_actor_registered(
        &self,
        token: &str,
        actor_type: &str,
        project: &str,
    ) -> Result<()> {
        let payload = json!({
            "token": token,
            "actor_type": actor_type,
            "project": project,
            "timestamp": iso8601_now(),
        });
        self.stream
            .publish(
                "ecosystem.actor.registered",
                Bytes::from(payload.to_string()),
            )
            .await?;
        info!(token = %token, "published actor.registered");
        Ok(())
    }

    /// Publish an actor deregistration event.
    pub async fn publish_actor_deregistered(&self, token: &str, reason: &str) -> Result<()> {
        let payload = json!({
            "token": token,
            "reason": reason,
            "timestamp": iso8601_now(),
        });
        self.stream
            .publish(
                "ecosystem.actor.deregistered",
                Bytes::from(payload.to_string()),
            )
            .await?;
        info!(token = %token, reason = %reason, "published actor.deregistered");
        Ok(())
    }

    /// Publish a file change event for a project (general, not requiring ack).
    pub async fn publish_file_changed(&self, project: &str, files: &[String]) -> Result<()> {
        let subject = format!("ecosystem.{project}.file.changed");
        let payload = json!({
            "project": project,
            "files": files,
            "timestamp": iso8601_now(),
        });
        self.stream
            .publish(&subject, Bytes::from(payload.to_string()))
            .await?;
        Ok(())
    }

    pub async fn publish_cache_invalidated(&self, reason: &str) -> Result<()> {
        let payload = json!({
            "project": self.project,
            "reason": reason,
            "affected_keys": [],
            "timestamp": iso8601_now(),
        });
        self.stream
            .publish(
                "ecosystem.daemon.cache.invalidated",
                Bytes::from(payload.to_string()),
            )
            .await?;
        info!(reason = %reason, "published daemon.cache.invalidated");
        Ok(())
    }

    /// Pull pending messages from the bound JetStream consumer.
    /// Returns (subject, parsed_json) for each valid message.
    /// Returns empty vec if no consumer is bound (publish-only mode).
    pub async fn pull_events(&self, max_msgs: usize) -> Result<Vec<(String, Value)>> {
        let batch = match self.stream.pull_batch(max_msgs).await {
            Ok(b) => b,
            Err(e) => {
                let msg = e.to_string();
                if msg.contains("no consumer bound") {
                    return Ok(Vec::new());
                }
                return Err(e);
            }
        };

        let mut events = Vec::with_capacity(batch.len());

        for (subject, payload_bytes, msg) in batch {
            match serde_json::from_slice::<Value>(&payload_bytes) {
                Ok(json) => events.push((subject, json)),
                Err(e) => {
                    warn!(error = %e, subject = %subject, "invalid JSON in NATS message — skipping");
                }
            }
            let _ = msg.ack().await;
        }

        Ok(events)
    }

    /// Extract (mode_id, params) from a reflection.request payload.
    pub fn parse_reflection_request(payload: &Value) -> Option<(String, Value)> {
        let mode_id = payload.get("mode_id")?.as_str()?.to_string();
        let params = payload
            .get("params")
            .cloned()
            .unwrap_or(Value::Object(Default::default()));
        Some((mode_id, params))
    }
}

/// Load project NATS config from .ontoref/config.ncl via nickel export.
#[cfg(feature = "nats")]
fn load_nats_config(config_path: &PathBuf) -> Result<Value> {
    if !config_path.exists() {
        return Ok(json!({}));
    }

    let output = Command::new("nickel")
        .arg("export")
        .arg(config_path)
        .output()
        .map_err(|e| anyhow!("running nickel export: {e}"))?;

    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr);
        return Err(anyhow!("nickel export failed: {stderr}"));
    }

    serde_json::from_slice(&output.stdout).map_err(|e| anyhow!("parsing nickel export output: {e}"))
}

/// ISO 8601 timestamp (UTC) without external dependency.
#[cfg(feature = "nats")]
fn iso8601_now() -> String {
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default();

    let secs = now.as_secs();
    let micros = now.subsec_micros();

    let days = secs / 86400;
    let day_secs = secs % 86400;

    // Gregorian approximation (valid for 1970–2099)
    let year = 1970 + (days / 365);
    let month = ((days % 365) / 30) + 1;
    let day = ((days % 365) % 30) + 1;

    format!(
        "{:04}-{:02}-{:02}T{:02}:{:02}:{:02}.{:06}Z",
        year,
        month,
        day,
        day_secs / 3600,
        (day_secs % 3600) / 60,
        day_secs % 60,
        micros,
    )
}

// ── No-op implementation when nats feature is disabled ────────────────

#[cfg(not(feature = "nats"))]
pub struct NatsPublisher;

#[cfg(not(feature = "nats"))]
impl NatsPublisher {
    pub async fn connect(
        _config_path: &std::path::PathBuf,
        _project: String,
        _port: u16,
    ) -> anyhow::Result<Option<Self>> {
        Ok(None)
    }

    pub async fn publish_started(&self) -> anyhow::Result<()> {
        Ok(())
    }

    pub async fn publish_stopped(&self, _uptime_secs: u64) -> anyhow::Result<()> {
        Ok(())
    }

    pub async fn publish_notification(
        &self,
        _project: &str,
        _event: &crate::notifications::NotificationEvent,
        _files: &[String],
    ) -> anyhow::Result<()> {
        Ok(())
    }

    pub async fn publish_actor_registered(
        &self,
        _token: &str,
        _actor_type: &str,
        _project: &str,
    ) -> anyhow::Result<()> {
        Ok(())
    }

    pub async fn publish_actor_deregistered(
        &self,
        _token: &str,
        _reason: &str,
    ) -> anyhow::Result<()> {
        Ok(())
    }

    pub async fn publish_file_changed(
        &self,
        _project: &str,
        _files: &[String],
    ) -> anyhow::Result<()> {
        Ok(())
    }

    pub async fn publish_cache_invalidated(&self, _reason: &str) -> anyhow::Result<()> {
        Ok(())
    }
}