did-hosting-server 0.9.0

//! Stats collection and sync to control plane.
//!
//! Hot-path operations (`record_resolve`, `record_update`) accumulate in the
//! in-memory `StatsCollector`. A periodic sync task drains the per-DID deltas
//! and pushes them to the control plane, which holds the authoritative totals.
//!
//! Stats are **not** persisted to disk on the server. On restart the counters
//! start at zero and deltas are additive on the control plane, so there is no
//! double-counting.

pub use did_hosting_common::server::stats_collector::{StatsAggregate, StatsCollector};

use affinidi_messaging_didcomm_service::DIDCommService;
use std::sync::atomic::{AtomicU64, Ordering};
use tracing::debug;

/// Monotonic sequence counter for stats sync idempotency.
static SYNC_SEQ: AtomicU64 = AtomicU64::new(0);

/// Push per-DID stat deltas to the control plane over the messaging transport.
///
/// The binding (TSP or DIDComm) follows the control plane's DID document;
/// there is no HTTPS side channel to the control plane.
///
/// Sent as a signed `.../server/stats-sync/0.1` trust task: the control plane
/// credits the deltas to the server whose DID signed the document, so an
/// unsigned push — the bare-message form this replaced — is refused there.
pub async fn sync_to_control_messaging(
    svc: &DIDCommService,
    state: &crate::server::AppState,
    server_did: &str,
    control_did: &str,
    collector: &StatsCollector,
) {
    use did_hosting_common::didcomm_types::MSG_STATS_SYNC;
    use did_hosting_common::server::didcomm_profile::TransportFallback;
    use did_hosting_common::server::trust_tasks::send::{build_signed_request, send_trust_task};
    use serde_json::json;

    // Resolve the signer before draining, so a node that cannot sign keeps
    // its deltas for a later tick instead of discarding them.
    let Some(identity) = state.identity.as_deref() else {
        debug!("stats sync skipped: service identity not loaded, cannot sign");
        return;
    };
    let signer = match did_hosting_common::server::trust_tasks::identity_signing_secret(
        identity, server_did,
    ) {
        Ok(s) => s,
        Err(e) => {
            debug!(error = %e, "stats sync skipped: no signing key");
            return;
        }
    };

    let deltas = collector.drain_for_sync();
    if deltas.is_empty() {
        return;
    }

    let seq = SYNC_SEQ.fetch_add(1, Ordering::Relaxed);

    let did_deltas: Vec<_> = deltas
        .iter()
        .map(|d| {
            json!({
                "mnemonic": d.mnemonic,
                "resolve_delta": d.resolve_delta,
                "update_delta": d.update_delta,
                "last_resolved_at": d.last_resolved_at,
                "last_updated_at": d.last_updated_at,
            })
        })
        .collect();

    let body = json!({
        "server_did": server_did,
        "seq": seq,
        "did_deltas": did_deltas,
    });

    let fallback = TransportFallback::from_config(
        state.config.mediator_did.as_deref(),
        state.config.features.tsp,
    );
    let sent =
        match build_signed_request(MSG_STATS_SYNC, server_did, control_did, body, &signer).await {
            Ok(doc) => send_trust_task(
                svc,
                "server",
                server_did,
                control_did,
                &doc,
                &fallback,
                state.did_resolver.as_ref(),
            )
            .await
            .map(|_| ()),
            Err(e) => Err(e),
        };
    if let Err(e) = sent {
        debug!(error = %e, "failed to sync stats to control plane (trust task)");
    }
}