did-hosting-control 0.9.0

//! Control-plane infrastructure trust tasks: server registration, health, and
//! stats sync.
//!
//! These are the control↔server ops that used to exist only as legacy `MSG_*`
//! DIDComm messages, which made them DIDComm-only and therefore invisible to a
//! TSP-only server. As trust-task documents they route through the same
//! [`crate::messaging::dispatch_trust_task_doc`] entry that TSP, the DIDComm
//! envelope, and HTTPS all share — so the binding is chosen from the peer's DID
//! document rather than hard-coded.
//!
//! ## One Type URI per op, on every wire
//!
//! The `MSG_*` constants in `didcomm_types` are *already* canonical Trust-Task
//! Type URIs, with the `_ACK` / `_PONG` forms being the `#response` variant of
//! their request:
//!
//! ```text
//! MSG_SERVER_REGISTER      .../spec/did-management/server/register/0.1
//! MSG_SERVER_REGISTER_ACK  .../spec/did-management/server/register/0.1#response
//! MSG_HEALTH_PING          .../spec/did-management/server/health/0.1
//! MSG_HEALTH_PONG          .../spec/did-management/server/health/0.1#response
//! MSG_STATS_SYNC           .../spec/did-management/server/stats-sync/0.1
//! MSG_STATS_ACK            .../spec/did-management/server/stats-sync/0.1#response
//! ```
//!
//! So we reuse them verbatim as document Type URIs. The op has one identity
//! whether it travels as a DIDComm `typ`, inside a trust-task envelope, or as a
//! raw TSP frame — and `TrustTask::respond_with` *derives* the reply URI rather
//! than us restating it. `send::msg_constants_are_request_response_type_uri_pairs`
//! pins that.
//!
//! Note the unused `TASK_SERVER_HEALTH_PING_1_0` / `_PONG_1_0` constants in
//! `did_hosting_tasks` are **not** used here: they sit on the `/did-hosting/`
//! authority with no `/spec/` segment, so they don't parse as Type URIs at all,
//! and they wrongly model the response as a separate URI instead of a fragment.
//! They are route-header decorators for the HTTPS surface, nothing more.
//!
//! ## Authorisation
//!
//! Every document reaching [`dispatch`] has already passed
//! [`crate::messaging::dispatch_trust_task_doc`]'s proof gate, so `signer` is
//! the DID whose key signed it — not a transport's report. Registration, stats
//! sync, health pongs and domain acks then require that DID to hold the
//! `Service` role in the ACL; sync acks are only logged. These ops don't run
//! the typed §7.2 pipeline because upstream has no typed payload specs for
//! them; the proof gate supplies what that pipeline would (issuer binding,
//! audience, freshness, replay).

use serde_json::Value;
use tracing::warn;

use did_hosting_common::didcomm_types::{
    MSG_HEALTH_PONG, MSG_REPLICA_DOMAIN_ASSIGN_ACK, MSG_REPLICA_DOMAIN_PURGE_ACK,
    MSG_REPLICA_DOMAIN_UNASSIGN_ACK, MSG_REPLICA_DOMAIN_UPSERT_ACK, MSG_SERVER_REGISTER,
    MSG_SERVER_REGISTER_ACK, MSG_STATS_ACK, MSG_STATS_SYNC, MSG_SYNC_BATCH_ACK,
    MSG_SYNC_DELETE_ACK, MSG_SYNC_UPDATE_ACK,
};
use did_hosting_common::server::didcomm_profile::ObservedTransport;

use crate::registry;
use crate::server::AppState;

/// Does this Type URI belong to the infrastructure ops handled here?
///
/// Compared on the full string, fragment included: `register/0.1` is a request
/// we act on, while `register/0.1#response` is an ack *we* emit and must never
/// route back into ourselves. The sync and domain `#response` acks are the
/// other way round — edges emit them, we receive them.
pub fn owns(type_uri: &str) -> bool {
    matches!(
        type_uri,
        MSG_SERVER_REGISTER
            | MSG_HEALTH_PONG
            | MSG_STATS_SYNC
            | MSG_SYNC_UPDATE_ACK
            | MSG_SYNC_BATCH_ACK
            | MSG_SYNC_DELETE_ACK
            | MSG_REPLICA_DOMAIN_ASSIGN_ACK
            | MSG_REPLICA_DOMAIN_UNASSIGN_ACK
            | MSG_REPLICA_DOMAIN_PURGE_ACK
            | MSG_REPLICA_DOMAIN_UPSERT_ACK
    )
}

/// Handle an infrastructure trust task signed by `sender` (the proven issuer).
///
/// Returns the serialised response document, or `None` when the op is terminal
/// (a health pong is an answer, not a question).
///
/// `via` is the transport the document actually arrived on, derived from the
/// caller's `TransportHandler::binding_uri()`. `None` when the binding is one
/// we don't recognise — recorded as "unknown" rather than guessed. The dispatch
/// itself stays transport-agnostic; this is observability only.
pub async fn dispatch(
    state: &AppState,
    sender: &str,
    via: Option<ObservedTransport>,
    doc: trust_tasks_rs::TrustTask<Value>,
) -> Option<Value> {
    let response = dispatch_inner(state, sender, doc).await;

    // *After* the dispatch, deliberately. A registration creates the instance,
    // so recording beforehand would find nothing to write to and the very
    // message that enrolled the server would go unobserved.
    if let Some(via) = via {
        registry::record_inbound_transport(
            &state.registry_ks,
            sender,
            via,
            crate::auth::session::now_epoch(),
        )
        .await;
    }

    response
}

async fn dispatch_inner(
    state: &AppState,
    sender: &str,
    doc: trust_tasks_rs::TrustTask<Value>,
) -> Option<Value> {
    match doc.type_uri.to_string().as_str() {
        MSG_SERVER_REGISTER => {
            let reply_id = uuid::Uuid::new_v4().to_string();
            match crate::messaging::do_server_register(state, sender, &doc.payload).await {
                Ok(ack) => {
                    let resp = doc.respond_with(reply_id, ack);
                    debug_assert_eq!(resp.type_uri.to_string(), MSG_SERVER_REGISTER_ACK);
                    Some(serde_json::to_value(&resp).expect("ack document serialises"))
                }
                Err(rej) => {
                    let err = doc.reject_with(
                        reply_id,
                        trust_tasks_rs::RejectReason::PermissionDenied {
                            reason: rej.comment.clone(),
                        },
                    );
                    warn!(
                        sender,
                        code = rej.code,
                        comment = %rej.comment,
                        "server registration rejected (trust task)"
                    );
                    Some(serde_json::to_value(&err).expect("error document serialises"))
                }
            }
        }
        MSG_STATS_SYNC => {
            let reply_id = uuid::Uuid::new_v4().to_string();
            match crate::messaging::do_stats_sync(state, sender, &doc.payload).await {
                Ok(ack) => {
                    let resp = doc.respond_with(reply_id, ack);
                    debug_assert_eq!(resp.type_uri.to_string(), MSG_STATS_ACK);
                    Some(serde_json::to_value(&resp).expect("ack document serialises"))
                }
                Err(rej) => {
                    let err = doc.reject_with(
                        reply_id,
                        trust_tasks_rs::RejectReason::PermissionDenied {
                            reason: rej.comment.clone(),
                        },
                    );
                    Some(serde_json::to_value(&err).expect("error document serialises"))
                }
            }
        }
        MSG_HEALTH_PONG => {
            crate::messaging::do_health_pong(state, sender, &doc.payload).await;
            None
        }
        uri @ (MSG_SYNC_UPDATE_ACK | MSG_SYNC_BATCH_ACK | MSG_SYNC_DELETE_ACK) => {
            crate::messaging::do_sync_ack(sender, uri, &doc.payload);
            acknowledge_outbox(state, sender, &doc).await;
            None
        }
        uri @ (MSG_REPLICA_DOMAIN_ASSIGN_ACK
        | MSG_REPLICA_DOMAIN_UNASSIGN_ACK
        | MSG_REPLICA_DOMAIN_PURGE_ACK) => {
            crate::messaging::do_domain_ack(state, sender, uri, &doc.payload).await;
            acknowledge_outbox(state, sender, &doc).await;
            None
        }
        // The upsert ack carries nothing the registry tracks; it only settles
        // the outbox entry.
        MSG_REPLICA_DOMAIN_UPSERT_ACK => {
            acknowledge_outbox(state, sender, &doc).await;
            None
        }
        // `owns` gates this; a mismatch means the two drifted.
        other => {
            warn!(type_uri = other, "trust_tasks_infra: unowned type URI");
            None
        }
    }
}

/// Settle the outbox entry an acknowledgement answers: the document it threads
/// to, from the server that signed it. Only a Service-role signer settles
/// anything — `sender` is the proven issuer.
pub(crate) async fn acknowledge_outbox(
    state: &AppState,
    sender: &str,
    doc: &trust_tasks_rs::TrustTask<Value>,
) {
    let thread = doc.thread_id.as_deref().or_else(|| {
        doc.payload
            .get("inResponseTo")
            .and_then(|r| r.get("id"))
            .and_then(Value::as_str)
    });
    let Some(thread) = thread else {
        return;
    };
    // A registered Service settles what it was sent; so does a configured
    // watcher (see `server_push::is_configured_watcher`). The outbox only ever
    // settles an entry on its own target's signature either way.
    if !crate::server_push::is_configured_watcher(state, sender)
        && !matches!(
            crate::acl::check_acl(&state.acl_ks, sender).await,
            Ok(crate::acl::Role::Service)
        )
    {
        warn!(sender, "acknowledgement ignored: Service role required");
        return;
    }
    match crate::outbox::acknowledge(&state.store, sender, thread).await {
        Ok(true) => {
            crate::registry::record_ack(
                &state.registry_ks,
                sender,
                crate::auth::session::now_epoch(),
            )
            .await;
            state.outbox_notify.notify_one();
        }
        Ok(false) => tracing::debug!(sender, thread, "acknowledgement for no awaited entry"),
        Err(e) => warn!(sender, error = %e, "failed to settle acknowledged outbox entry"),
    }
}