hatchery-node 0.2.77

//! Node calls (`mail.*`, `task.*`) over the restreamer uplink.
//!
//! Since A2, C2 no longer forwards signed node calls to core over HTTP: it answers 410.
//! These calls ride a dedicated `m4a` channel on a **separate TCP** uplink from the
//! g4a/PTY bridge (same Frame/Noise/OPEN wire). Core takes the node id from the
//! stream, never from the payload. The legacy HTTP path stays as a fallback for
//! nodes without an uplink.
//!
//! Splitting mail onto its own TCP means a wedged Noise session or handshake retry
//! on ch2 cannot fill the PTY uplink's send buffer.

use crate::restream_session::{NodeSession, NodeUplink, UpstreamKeys};
use crate::uplink_journal::UplinkJournal;
use hatchery_core::ChannelKind;
use hatchery_enroll::NodeIdentity;
use serde_json::Value;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{watch, Mutex};

/// Channel id of the node-call channel; core runs its `m4a` drain consumer on it.
pub const NODE_CALL_CHANNEL: u32 = 2;
const HANDSHAKE: Duration = Duration::from_secs(8);
const CALL_TIMEOUT: Duration = Duration::from_secs(15);
/// After an open/handshake failure, wait before trying again (dedicated mail TCP).
const OPEN_BACKOFF_MIN: Duration = Duration::from_secs(5);
const OPEN_BACKOFF_MAX: Duration = Duration::from_secs(30);
/// How often an idle node-call channel is checked for core's re-handshake request.
const REKEY_CHECK_EVERY: Duration = Duration::from_secs(2);

#[derive(Default)]
struct Link {
    /// Dedicated mail TCP (not the g4a/PTY bridge uplink).
    uplink: Option<Arc<NodeUplink>>,
    session: Option<NodeSession>,
    /// Do not attempt OPEN before this instant.
    open_not_before: Option<Instant>,
    /// Current open-failure backoff (grows 5s → 30s, resets on success).
    open_backoff: Option<Duration>,
}

/// Shared by the door and the node-mail wake; the restream bridge signals presence here.
#[derive(Clone)]
pub struct NodeCalls {
    link: Arc<Mutex<Link>>,
    node_id: String,
    keys: Option<UpstreamKeys>,
    legacy: Option<(SocketAddr, Arc<NodeIdentity>)>,
    /// Enroll HTTP address used only for C2-local config (`node.config`), even when
    /// mail/task calls go over the restream uplink (those get 410 on the HTTP path).
    config_http: Option<(SocketAddr, Arc<NodeIdentity>)>,
}

#[derive(Debug)]
pub enum CallError {
    Refused,
    Unavailable(String),
}

impl NodeCalls {
    pub fn new(
        node_id: String,
        keys: Option<UpstreamKeys>,
        legacy: Option<(SocketAddr, Arc<NodeIdentity>)>,
        config_http: Option<(SocketAddr, Arc<NodeIdentity>)>,
    ) -> Self {
        Self {
            link: Arc::default(),
            node_id,
            keys,
            legacy,
            config_http,
        }
    }

    /// Bridge (re)connect: refresh product URL. Mail TCP is maintained separately
    /// by [`Self::spawn_mail_uplink`] so PTY and mail do not share fate.
    pub async fn set_uplink(&self, uplink: Option<Arc<NodeUplink>>) {
        let connected = uplink.is_some();
        // Drop any stale shared-uplink session handle; mail uses its own TCP.
        {
            let mut l = self.link.lock().await;
            // Do not clear `l.uplink`/`l.session` here — those belong to the mail
            // maintainer. Presence is only for product URL.
            let _ = &mut *l;
        }
        if connected {
            self.refresh_product_url().await;
        } else {
            crate::m4a_accounts::set_c2_product_url(None);
        }
    }

    /// Keep a dedicated TCP + ch2 Noise session open for node-calls.
    /// Reconnects with exponential backoff (5s → 30s) independently of the g4a bridge.
    pub fn spawn_mail_uplink(
        &self,
        addr: SocketAddr,
        keys: UpstreamKeys,
        journal: Option<Arc<UplinkJournal>>,
        mut shutdown: watch::Receiver<bool>,
    ) {
        let this = self.clone();
        tokio::spawn(async move {
            let mut backoff = OPEN_BACKOFF_MIN;
            loop {
                if *shutdown.borrow() {
                    break;
                }
                match this.dial_mail(addr, keys, journal.clone()).await {
                    Ok(uplink) => {
                        tracing::info!(%addr, channel = NODE_CALL_CHANNEL, "mail uplink open (dedicated TCP)");
                        backoff = OPEN_BACKOFF_MIN;
                        // Wait until this TCP dies or shutdown.
                        tokio::select! {
                            _ = uplink.closed() => {
                                tracing::warn!(%addr, "mail uplink closed; reconnecting");
                            }
                            _ = shutdown.changed() => {
                                if *shutdown.borrow() {
                                    break;
                                }
                            }
                        }
                        let mut l = this.link.lock().await;
                        if l.uplink.as_ref().is_some_and(|u| Arc::ptr_eq(u, &uplink)) {
                            l.uplink = None;
                            l.session = None;
                        }
                    }
                    Err(e) => {
                        tracing::error!(
                            error = %e,
                            %addr,
                            backoff_s = backoff.as_secs(),
                            "mail uplink open failed; backing off"
                        );
                        tokio::select! {
                            _ = tokio::time::sleep(backoff) => {}
                            _ = shutdown.changed() => {
                                if *shutdown.borrow() { break; }
                            }
                        }
                        backoff = (backoff.saturating_mul(2)).max(OPEN_BACKOFF_MIN).min(OPEN_BACKOFF_MAX);
                    }
                }
            }
            let mut l = this.link.lock().await;
            l.uplink = None;
            l.session = None;
        });
    }

    async fn dial_mail(
        &self,
        addr: SocketAddr,
        keys: UpstreamKeys,
        journal: Option<Arc<UplinkJournal>>,
    ) -> Result<Arc<NodeUplink>, String> {
        let uplink = NodeUplink::connect(addr, journal).await?;
        let session = uplink
            .open(&self.node_id, NODE_CALL_CHANNEL, ChannelKind::M4a, keys, HANDSHAKE)
            .await?;
        let mut l = self.link.lock().await;
        l.uplink = Some(Arc::clone(&uplink));
        l.session = Some(session);
        l.open_not_before = None;
        l.open_backoff = None;
        Ok(uplink)
    }

    fn next_backoff(current: Option<Duration>) -> Duration {
        match current {
            None => OPEN_BACKOFF_MIN,
            Some(d) => (d.saturating_mul(2)).max(OPEN_BACKOFF_MIN).min(OPEN_BACKOFF_MAX),
        }
    }

    /// Ask C2 for its public config over enroll HTTP and stash `m4a_product_url`.
    pub async fn refresh_product_url(&self) -> bool {
        let Some((c2, identity)) = self.config_http.clone().or_else(|| self.legacy.clone()) else {
            return false;
        };
        match crate::enroll_client::node_call(
            c2,
            &identity,
            &self.node_id,
            hatchery_enroll::call::NODE_CONFIG_OP,
            &serde_json::json!({}),
        )
        .await
        {
            Ok(v) => {
                let url = v
                    .get("m4a_product_url")
                    .and_then(|u| u.as_str())
                    .filter(|u| !u.is_empty())
                    .map(str::to_owned);
                crate::m4a_accounts::set_c2_product_url(url.clone());
                match url {
                    Some(u) => tracing::info!(product = %u, "m4a product url from C2"),
                    None => tracing::info!("m4a product url from C2 is empty; accounts stay local"),
                }
                true
            }
            Err(e) => {
                tracing::warn!(error = ?e, "m4a product url from C2 failed");
                false
            }
        }
    }

    /// [`Self::refresh_product_url`] with retries (5s doubling to 5 min) until C2 answers.
    pub async fn refresh_product_url_retrying(&self) {
        let mut wait = std::time::Duration::from_secs(5);
        while !self.refresh_product_url().await {
            if crate::m4a_accounts::M4aSessionAccounts::resolve_product_url(None).is_some() {
                return;
            }
            tokio::time::sleep(wait).await;
            wait = (wait * 2).min(std::time::Duration::from_secs(300));
        }
    }

    fn sign(&self, op: &str, payload: &Value) -> Option<Value> {
        let (_, identity) = self.config_http.as_ref().or(self.legacy.as_ref())?;
        let ts = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).ok()?.as_millis() as u64;
        let call = hatchery_enroll::call::SignedCall::sign(identity, &self.node_id, op, payload, ts).ok()?;
        serde_json::to_value(call).ok()
    }

    /// The node-call channel only reads downlink while a call waits for its reply, so a
    /// core restart (core then holds no keys and asks for a re-handshake) went unnoticed
    /// until the next call failed. Check the idle channel every few seconds and
    /// re-handshake as soon as core asks.
    pub fn spawn_rekey_watch(&self, mut shutdown: watch::Receiver<bool>) {
        let this = self.clone();
        tokio::spawn(async move {
            loop {
                tokio::select! {
                    _ = tokio::time::sleep(REKEY_CHECK_EVERY) => {}
                    _ = shutdown.changed() => if *shutdown.borrow() { return },
                }
                let Ok(mut l) = this.link.try_lock() else { continue };
                if l.session.as_mut().is_some_and(NodeSession::stale_downlink) {
                    tracing::info!(channel = NODE_CALL_CHANNEL, "core asked for a new Noise session; re-handshaking");
                    l.session = None;
                    l.open_not_before = None;
                    l.open_backoff = None;
                    this.open_locked(&mut l).await;
                }
            }
        });
    }

    async fn open_locked(&self, l: &mut Link) {
        if let (Some(uplink), Some(keys)) = (l.uplink.clone(), self.keys) {
            match uplink.open(&self.node_id, NODE_CALL_CHANNEL, ChannelKind::M4a, keys, HANDSHAKE).await {
                Ok(s) => {
                    tracing::info!("node calls over the uplink: channel open");
                    l.session = Some(s);
                    l.open_not_before = None;
                    l.open_backoff = None;
                }
                Err(e) => {
                    let wait = Self::next_backoff(l.open_backoff);
                    tracing::warn!(error = %e, "node-call channel open failed; backing off {}s", wait.as_secs());
                    l.open_backoff = Some(wait);
                    l.open_not_before = Some(Instant::now() + wait);
                }
            }
        }
    }

    pub async fn call(&self, op: &str, payload: &Value) -> Result<Value, CallError> {
        // A stale session (core restarted) costs one immediate re-handshake, not the call.
        for attempt in 0..2 {
            match self.call_once(op, payload).await {
                Err(CallError::Unavailable(e)) if attempt == 0 && e.contains("decrypt") => {
                    tracing::info!(op, "node call hit a stale Noise session; re-handshaking and retrying");
                    let mut l = self.link.lock().await;
                    l.open_not_before = None;
                    l.open_backoff = None;
                }
                other => return other,
            }
        }
        unreachable!("the second attempt returns")
    }

    async fn call_once(&self, op: &str, payload: &Value) -> Result<Value, CallError> {
        if self.keys.is_some() {
            let mut l = self.link.lock().await;
            if l.session.as_mut().is_some_and(NodeSession::stale_downlink) {
                tracing::info!(channel = NODE_CALL_CHANNEL, "core asked for a new Noise session; re-handshaking");
                l.session = None;
                l.open_not_before = None;
                l.open_backoff = None;
            }
            if l.uplink.is_some() {
                if l.session.is_none() {
                    let gated = l.open_not_before.is_some_and(|t| Instant::now() < t);
                    if !gated {
                        if let (Some(uplink), Some(keys)) = (l.uplink.clone(), self.keys) {
                            match uplink
                                .open(&self.node_id, NODE_CALL_CHANNEL, ChannelKind::M4a, keys, HANDSHAKE)
                                .await
                            {
                                Ok(s) => {
                                    tracing::info!("node calls over the uplink: channel open");
                                    l.session = Some(s);
                                    l.open_not_before = None;
                                    l.open_backoff = None;
                                }
                                Err(e) => {
                                    let wait = Self::next_backoff(l.open_backoff);
                                    tracing::warn!(
                                        error = %e,
                                        "node-call channel open failed; backing off {}s",
                                        wait.as_secs()
                                    );
                                    l.open_backoff = Some(wait);
                                    l.open_not_before = Some(Instant::now() + wait);
                                }
                            }
                        }
                    }
                }
                if let Some(session) = l.session.as_mut() {
                    // The node signs every call with its enroll identity; core verifies it.
                    let signed = self.sign(op, payload);
                    return match session.call_signed(op, payload, signed.as_ref(), CALL_TIMEOUT).await {
                        Ok(v) => Ok(v),
                        Err(e) if e.starts_with("403") => Err(CallError::Refused),
                        Err(e)
                            if e.chars().take(3).all(|c| c.is_ascii_digit()) && e.contains(':') =>
                        {
                            Err(CallError::Unavailable(e))
                        }
                        Err(e) => {
                            l.session = None;
                            let wait = Self::next_backoff(l.open_backoff);
                            l.open_backoff = Some(wait);
                            l.open_not_before = Some(Instant::now() + wait);
                            Err(CallError::Unavailable(e))
                        }
                    };
                }
            }
        }
        match &self.legacy {
            Some((c2, identity)) => {
                match crate::enroll_client::node_call(*c2, identity, &self.node_id, op, payload).await
                {
                    Ok(v) => Ok(v),
                    Err(crate::enroll_client::EnrollClientError::Refused) => Err(CallError::Refused),
                    Err(e) => Err(CallError::Unavailable(format!("{e:?}"))),
                }
            }
            None => Err(CallError::Unavailable("no uplink".into())),
        }
    }
}