yah-scryer 0.8.42

//! HTTP federation surface — R556-F7-T2.
//!
//! Scryer exposes a small HTTP API that other mesh nodes (and the local hub)
//! call directly. Per W264 §Discovery, yubaba advertises this endpoint via
//! `/services` but is **not** in the data path; consumers connect here.
//!
//! Routes:
//!   - `POST /federate/events`    `{filter, scopes?}` → `{events: [{scope, event}]}`
//!   - `POST /federate/aggregate` `{filter, group_by, since_ms, scopes?}` → `{buckets}`
//!   - `POST /federate/hops`      `{from_unix_nanos, to_unix_nanos}` → `{hops}`
//!   - `GET  /scopes?limit=N`     → `{scopes}`
//!   - `GET  /health`             → `{status:"ok"}`
//!
//! `scopes` omitted = cross-scope rollup (the missing mesh-wide-by-level case
//! at `crates/yah/hub/src/in_process.rs:46`).
//!
//! ## ACL
//!
//! Per W264 §Trust boundary, the operator-tag check sits at scryer's listener,
//! not at yubaba. The calling identity is presented in an `X-Yah-Operator-Tag`
//! header (one tag per occurrence) set by the local Tailscale sidecar that
//! resolved the peer identity via `tailscaled` before forwarding the request.
//! Scryer trusts the header but only because the listener binds the mesh
//! interface — workloads can't reach it without an operator-tagged identity in
//! the first place. [`OperatorTagAcl`] checks `tag:operator` by default.
//!
//! ## Client
//!
//! [`HttpFederationPeer`] is the production `FederationPeer` impl — a small
//! `reqwest`-based client whose `name()` is the node tailnet hostname.
//!
//! @arch:see(.yah/docs/working/W264-kamaji-managed-scryer.md)

use crate::federation::{
    FederationAcl, FederationError, FederationPeer, PeerIdentity, ScopedEvent,
};
use crate::service::{AggregateBucket, EventFilter, Scryer};
use crate::store::ScopeInfo;
use async_trait::async_trait;
use axum::{
    Json, Router,
    extract::{Query, State},
    http::{HeaderMap, StatusCode},
    routing::{get, post},
};
use observation::{EventScope, HopRollup};
use serde::{Deserialize, Serialize};
use std::sync::Arc;

// ─── Wire DTOs ────────────────────────────────────────────────────────────────

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FederateEventsReq {
    pub filter: EventFilter,
    /// When `None`, query rolls up across every scope in the store.
    #[serde(default)]
    pub scopes: Option<Vec<EventScope>>,
}

/// `{events: [{scope, event}]}` — each row carries the scope it was stored
/// under (R585-F2). The envelope is what lets a cross-scope rollup
/// (`scopes: None`) say *which* task run or service each event came from;
/// before it, every consumer downstream rendered the scope blank.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FederateEventsResp {
    pub events: Vec<ScopedEvent>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FederateAggregateReq {
    pub filter: EventFilter,
    pub group_by: String,
    #[serde(default)]
    pub since_ms: u64,
    #[serde(default)]
    pub scopes: Option<Vec<EventScope>>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BucketDto {
    pub key: String,
    pub count: u64,
}

impl From<AggregateBucket> for BucketDto {
    fn from(b: AggregateBucket) -> Self {
        Self { key: b.key, count: b.count }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FederateAggregateResp {
    pub buckets: Vec<BucketDto>,
}

/// `POST /federate/hops` request — R893-F19.
///
/// A half-open nanosecond range, not a `since_ms` like the aggregate route.
/// Spans are stamped in `start_unix_nanos` and rolled up into 60s windows
/// aligned on that same clock (`observation::rollup_window_start`), so a
/// relative window would have to be re-based against the peer's clock on the
/// peer's side — which silently changes the answer when two nodes disagree
/// about now. The caller owns the range it asked for; the peer only aligns it.
/// `bucket_nanos` splits the range into consecutive sub-windows instead of
/// merging it into one cell per key, so a caller can see a hop's latency
/// *shape* move over the window rather than only its endpoint quantiles. Each
/// returned [`HopRollup`] carries the sub-window it covers in its own
/// `window_start_unix_nanos` / `window_end_unix_nanos`, so the grid is
/// self-describing and the caller re-merges to totals with an exact histogram
/// add. `None` = one cell per key over the whole range.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FederateHopsReq {
    pub from_unix_nanos: u64,
    pub to_unix_nanos: u64,
    #[serde(default)]
    pub bucket_nanos: Option<u64>,
}

/// `{hops: [HopRollup]}` — one cell per `(scope, peer, kind)` on this node,
/// already merged across the window's 60s rollup buckets.
///
/// The full [`HopRollup`] crosses the wire, histogram included, rather than
/// pre-computed quantiles. A caller federating several nodes must merge
/// histograms element-wise; averaging two nodes' p95 is not the p95 of their
/// union (R893-F15).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FederateHopsResp {
    pub hops: Vec<HopRollup>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ScopeInfoDto {
    pub scope: EventScope,
    pub event_count: i64,
    pub last_offset_ms: i64,
}

impl From<ScopeInfo> for ScopeInfoDto {
    fn from(s: ScopeInfo) -> Self {
        Self { scope: s.scope, event_count: s.event_count, last_offset_ms: s.last_offset_ms }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ScopesResp {
    pub scopes: Vec<ScopeInfoDto>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HealthResp {
    pub status: String,
}

#[derive(Debug, Clone, Copy, Deserialize)]
pub struct ScopesQuery {
    #[serde(default)]
    pub limit: Option<usize>,
}

// ─── Header constants ─────────────────────────────────────────────────────────

/// Header name carrying the calling identity's Tailscale operator tags.
/// Set by the local Tailscale sidecar after `WhoIs` resolution.
pub const OPERATOR_TAG_HEADER: &str = "x-yah-operator-tag";

// ─── Server ───────────────────────────────────────────────────────────────────

/// Shared state for the federation router.
pub struct FederationState {
    pub scryer: Arc<Scryer>,
    pub acl: Arc<dyn FederationAcl>,
}

impl FederationState {
    pub fn new(scryer: Arc<Scryer>, acl: Arc<dyn FederationAcl>) -> Arc<Self> {
        Arc::new(Self { scryer, acl })
    }
}

/// Build the axum router for the federation surface.
pub fn router(state: Arc<FederationState>) -> Router {
    Router::new()
        .route("/federate/events", post(handle_events))
        .route("/federate/aggregate", post(handle_aggregate))
        .route("/federate/hops", post(handle_hops))
        .route("/scopes", get(handle_scopes))
        .route("/health", get(handle_health))
        .with_state(state)
}

fn identity_from(headers: &HeaderMap) -> PeerIdentity {
    let mut id = PeerIdentity::default();
    for value in headers.get_all(OPERATOR_TAG_HEADER) {
        if let Ok(s) = value.to_str() {
            id = id.with_tag(s.to_string());
        }
    }
    id
}

fn ensure_authorized(
    state: &FederationState,
    headers: &HeaderMap,
) -> Result<(), (StatusCode, Json<ErrorBody>)> {
    let identity = identity_from(headers);
    if state.acl.is_authorized(&identity) {
        Ok(())
    } else {
        Err((
            StatusCode::FORBIDDEN,
            Json(ErrorBody {
                error: "unauthorized: operator tag required".to_string(),
            }),
        ))
    }
}

#[derive(Debug, Serialize)]
struct ErrorBody {
    error: String,
}

async fn handle_health() -> Json<HealthResp> {
    Json(HealthResp { status: "ok".to_string() })
}

async fn handle_scopes(
    State(state): State<Arc<FederationState>>,
    headers: HeaderMap,
    Query(q): Query<ScopesQuery>,
) -> Result<Json<ScopesResp>, (StatusCode, Json<ErrorBody>)> {
    ensure_authorized(&state, &headers)?;
    let limit = q.limit.unwrap_or(1000).min(10_000);
    let scopes = state.scryer.list_scopes(limit).map_err(scryer_err)?;
    Ok(Json(ScopesResp {
        scopes: scopes.into_iter().map(ScopeInfoDto::from).collect(),
    }))
}

async fn handle_events(
    State(state): State<Arc<FederationState>>,
    headers: HeaderMap,
    Json(req): Json<FederateEventsReq>,
) -> Result<Json<FederateEventsResp>, (StatusCode, Json<ErrorBody>)> {
    ensure_authorized(&state, &headers)?;
    let events = match req.scopes {
        None => state
            .scryer
            .events_all_scoped(&req.filter)
            .await
            .map_err(scryer_err)?,
        Some(scopes) => {
            let mut acc: Vec<ScopedEvent> = Vec::new();
            for scope in &scopes {
                let part = ScopedEvent::tag_all(
                    scope,
                    state
                        .scryer
                        .events(scope, &req.filter)
                        .await
                        .map_err(scryer_err)?,
                );
                acc = crate::federation::merge_ordered(acc, part);
            }
            acc
        }
    };
    Ok(Json(FederateEventsResp { events }))
}

async fn handle_aggregate(
    State(state): State<Arc<FederationState>>,
    headers: HeaderMap,
    Json(req): Json<FederateAggregateReq>,
) -> Result<Json<FederateAggregateResp>, (StatusCode, Json<ErrorBody>)> {
    ensure_authorized(&state, &headers)?;
    let buckets = match req.scopes {
        None => state
            .scryer
            .aggregate_all(req.since_ms, &req.group_by)
            .map_err(scryer_err)?,
        Some(scopes) => {
            let mut counts: std::collections::HashMap<String, u64> =
                std::collections::HashMap::new();
            for scope in &scopes {
                for b in state
                    .scryer
                    .aggregate(scope, req.since_ms, &req.group_by)
                    .map_err(scryer_err)?
                {
                    *counts.entry(b.key).or_insert(0) += b.count;
                }
            }
            let mut buckets: Vec<AggregateBucket> = counts
                .into_iter()
                .map(|(key, count)| AggregateBucket { key, count })
                .collect();
            buckets.sort_by(|a, b| b.count.cmp(&a.count).then(a.key.cmp(&b.key)));
            buckets
        }
    };
    Ok(Json(FederateAggregateResp {
        buckets: buckets.into_iter().map(BucketDto::from).collect(),
    }))
}

async fn handle_hops(
    State(state): State<Arc<FederationState>>,
    headers: HeaderMap,
    Json(req): Json<FederateHopsReq>,
) -> Result<Json<FederateHopsResp>, (StatusCode, Json<ErrorBody>)> {
    ensure_authorized(&state, &headers)?;
    let hops = state
        .scryer
        .hop_rollups(req.from_unix_nanos, req.to_unix_nanos, req.bucket_nanos)
        .map_err(scryer_err)?;
    Ok(Json(FederateHopsResp { hops }))
}

fn scryer_err(err: crate::service::ScryerError) -> (StatusCode, Json<ErrorBody>) {
    (
        StatusCode::INTERNAL_SERVER_ERROR,
        Json(ErrorBody { error: err.to_string() }),
    )
}

// ─── Client ───────────────────────────────────────────────────────────────────

/// Production `FederationPeer` impl — `reqwest` POST to a peer scryer's
/// `/federate/events`. The local sidecar (or hub) is expected to add the
/// `X-Yah-Operator-Tag` header before this client's request leaves the host.
pub struct HttpFederationPeer {
    name: String,
    base_url: String,
    operator_tag: String,
    client: reqwest::Client,
}

impl HttpFederationPeer {
    /// `name` is the peer's tailnet hostname (used for logging + rule matching);
    /// `base_url` is the scryer endpoint root (e.g. `http://100.64.0.7:6543`).
    pub fn new(
        name: impl Into<String>,
        base_url: impl Into<String>,
        operator_tag: impl Into<String>,
    ) -> Result<Self, reqwest::Error> {
        Ok(Self {
            name: name.into(),
            base_url: base_url.into(),
            operator_tag: operator_tag.into(),
            client: reqwest::Client::builder().build()?,
        })
    }

    pub fn with_client(
        name: impl Into<String>,
        base_url: impl Into<String>,
        operator_tag: impl Into<String>,
        client: reqwest::Client,
    ) -> Self {
        Self {
            name: name.into(),
            base_url: base_url.into(),
            operator_tag: operator_tag.into(),
            client,
        }
    }

    /// Cross-scope aggregate query — convenience for hub callers that need the
    /// rollup but don't go through the `FederationPeer` trait (which is
    /// events-only by design).
    pub async fn aggregate(
        &self,
        since_ms: u64,
        group_by: &str,
    ) -> Result<Vec<BucketDto>, FederationError> {
        let req = FederateAggregateReq {
            filter: EventFilter::default(),
            group_by: group_by.to_string(),
            since_ms,
            scopes: None,
        };
        let url = format!("{}/federate/aggregate", self.base_url.trim_end_matches('/'));
        let resp = self
            .client
            .post(&url)
            .header(OPERATOR_TAG_HEADER, &self.operator_tag)
            .json(&req)
            .send()
            .await
            .map_err(|e| FederationError::Rpc(e.to_string()))?;
        if !resp.status().is_success() {
            return Err(FederationError::Rpc(format!(
                "aggregate http {}: {}",
                resp.status(),
                resp.text().await.unwrap_or_default()
            )));
        }
        let body: FederateAggregateResp = resp
            .json()
            .await
            .map_err(|e| FederationError::Rpc(e.to_string()))?;
        Ok(body.buckets)
    }

    /// Hop-matrix query — the same "convenience for hub callers" shape as
    /// [`Self::aggregate`], and off the `FederationPeer` trait for the same
    /// reason: the trait is events-only by design, and a span rollup is a
    /// different signal with a different time base.
    ///
    /// Returns whole [`HopRollup`]s so the caller can merge histograms across
    /// nodes exactly (R893-F15). The window is nanoseconds since the Unix
    /// epoch, half-open.
    pub async fn hop_rollups(
        &self,
        from_unix_nanos: u64,
        to_unix_nanos: u64,
        bucket_nanos: Option<u64>,
    ) -> Result<Vec<HopRollup>, FederationError> {
        let req = FederateHopsReq { from_unix_nanos, to_unix_nanos, bucket_nanos };
        let url = format!("{}/federate/hops", self.base_url.trim_end_matches('/'));
        let resp = self
            .client
            .post(&url)
            .header(OPERATOR_TAG_HEADER, &self.operator_tag)
            .json(&req)
            .send()
            .await
            .map_err(|e| FederationError::Rpc(e.to_string()))?;
        if !resp.status().is_success() {
            return Err(FederationError::Rpc(format!(
                "hops http {}: {}",
                resp.status(),
                resp.text().await.unwrap_or_default()
            )));
        }
        let body: FederateHopsResp = resp
            .json()
            .await
            .map_err(|e| FederationError::Rpc(e.to_string()))?;
        Ok(body.hops)
    }
}

#[async_trait]
impl FederationPeer for HttpFederationPeer {
    fn name(&self) -> &str {
        &self.name
    }

    async fn events(&self, filter: &EventFilter) -> Result<Vec<ScopedEvent>, FederationError> {
        let req = FederateEventsReq { filter: filter.clone(), scopes: None };
        let url = format!("{}/federate/events", self.base_url.trim_end_matches('/'));
        let resp = self
            .client
            .post(&url)
            .header(OPERATOR_TAG_HEADER, &self.operator_tag)
            .json(&req)
            .send()
            .await
            .map_err(|e| FederationError::Rpc(e.to_string()))?;
        if !resp.status().is_success() {
            return Err(FederationError::Rpc(format!(
                "events http {}: {}",
                resp.status(),
                resp.text().await.unwrap_or_default()
            )));
        }
        let body: FederateEventsResp = resp
            .json()
            .await
            .map_err(|e| FederationError::Rpc(e.to_string()))?;
        Ok(body.events)
    }
}

/// Convenience: bind a tokio listener on `addr` and serve the federation
/// router. Returns the local socket address and a join handle on success so
/// callers (kamaji service manifest, integration tests) can shut it down.
pub async fn serve(
    state: Arc<FederationState>,
    addr: std::net::SocketAddr,
) -> Result<(std::net::SocketAddr, tokio::task::JoinHandle<()>), std::io::Error> {
    let listener = tokio::net::TcpListener::bind(addr).await?;
    let local = listener.local_addr()?;
    let app = router(state);
    let handle = tokio::spawn(async move {
        let _ = axum::serve(listener, app).await;
    });
    Ok((local, handle))
}