road-runner-common 0.30.0

//! Generic follower for cex-policy's governance stream (`cex_policy.events`,
//! externally-tagged `DomainEvent`s). Every governed action (FeeChange,
//! AuthzChange, AuthRiskRuleset, ...) reacts to the same `QuorumReached`
//! shape; implement [`GovernanceHandler`] once per action and this consumer
//! handles the reconnect loop, event parsing, and commit-after-apply.
//!
//! The permanent root's single-signer actions (`RootActionApplied`, e.g. the
//! one-time `SETUP_COMMIT` manifest) are dispatched the same way to
//! [`GovernanceHandler::apply_root_action`], under the same offset contract.

use std::sync::Arc;

use async_trait::async_trait;
use rdkafka::config::ClientConfig;
use rdkafka::consumer::{CommitMode, Consumer, StreamConsumer};
use rdkafka::message::Message;
use serde::Deserialize;

/// cex-policy's `DomainEvent::RootActionApplied` body: the permanent root
/// applied a single-signer, immediate action. The event carries only the
/// reference and its hash; the payload itself is fetched with
/// [`crate::governance::PolicyGovernanceClient::fetch_root_action`] and must be
/// checked against `payload_sha256` before it is acted on.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
pub struct RootActionAppliedEvent {
    /// `root_actions` row id (UUID).
    pub action_id: String,
    /// e.g. `ADMIN_MUTATION`, `AUTHZ_MUTATION`, `ROOT_CREDENTIAL_REVOKE`,
    /// `SETUP_COMMIT`, `SETUP_AUTHZ`.
    pub kind: String,
    /// Lowercase hex sha256 of the action's canonical payload JSON.
    pub payload_sha256: String,
    /// Keycloak `sub` of the root that signed the action.
    pub root_sub: String,
    /// RFC 3339.
    pub occurred_at: String,
}

/// Applies (or releases) a single governed action once cex-policy resolves its
/// proposal. Implement this for each governed action a service owns; the
/// `action` the handler cares about is whatever it decides to apply in
/// [`Self::apply_approved_proposal`] — a service can either wire up one
/// handler per action (multiple consumers) or dispatch internally on the
/// proposal's resolved `action` field, matching cex-metadata's FeeChange today.
#[async_trait]
pub trait GovernanceHandler: Send + Sync {
    /// A `QuorumReached { status: Approved }` fired for `proposal_id`. Fetch the
    /// proposal (payload included) and apply the mutation it governs. Must be
    /// idempotent — Kafka delivery is at-least-once.
    ///
    /// The return value decides whether the offset advances, so it is a request,
    /// not a report:
    ///
    /// - `Ok(())` — done, **or** permanently undone. Return this for anything a
    ///   second delivery would resolve identically: a proposal for another action,
    ///   a payload missing a field it will never grow, work that needs an operator
    ///   rather than a retry. The event is consumed.
    /// - `Err(_)` — ask me again. The offset is held and the consumer reconnects,
    ///   so this message is redelivered in ~10s. Reserve it for transient causes
    ///   (cex-policy unreachable, the database down), because a handler that keeps
    ///   returning `Err` stalls every later governance decision behind it.
    async fn apply_approved_proposal(&self, proposal_id: &str) -> anyhow::Result<()>;

    /// A `QuorumReached { status: Rejected | Expired | Cancelled }` fired for
    /// `proposal_id`. Default no-op — only override if the proposer side tracks
    /// a `pending_proposal_id` that needs releasing so a new proposal can be
    /// submitted (cex-metadata's FeeChange does this; most handlers don't).
    ///
    /// All three mean the same thing to a handler: the change will not be
    /// applied, so let go of whatever was held for it. Why it will not be
    /// applied — the quorum refused, the clock ran out, or the proposer
    /// withdrew it — belongs in cex-policy's audit trail, not in a marker.
    async fn release_rejected_proposal(&self, _proposal_id: &str) -> anyhow::Result<()> {
        Ok(())
    }

    /// A `RootActionApplied` fired: the permanent root applied a single-signer
    /// action (see [`RootActionAppliedEvent`]). Default no-op — only services
    /// that act on a root action kind (cex-iam for authz, cex-metadata for the
    /// setup manifest's blockchains) override it, and they must ignore kinds
    /// they do not own.
    ///
    /// Same offset contract as [`Self::apply_approved_proposal`]: `Ok(())` for
    /// done or permanently undone (including a hash mismatch, which a retry
    /// cannot fix), `Err(_)` only for transient causes.
    async fn apply_root_action(&self, _event: &RootActionAppliedEvent) -> anyhow::Result<()> {
        Ok(())
    }
}

pub struct GovernanceConsumerConfig {
    pub brokers: String,
    /// e.g. `cex_policy.events`.
    pub topic: String,
    /// Consumer group id. Keep this stable across deploys/rewrites — changing
    /// it replays the whole topic from `auto.offset.reset`.
    pub group_id: String,
}

/// Runs forever, reconnecting on any consumer error. Intended to be
/// `tokio::spawn`ed; hold the returned `JoinHandle` if you want to await/abort it.
pub fn spawn_governance_consumer(
    cfg: GovernanceConsumerConfig,
    handler: Arc<dyn GovernanceHandler>,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        loop {
            match build_consumer(&cfg) {
                Ok(consumer) => {
                    #[cfg(feature = "observability")]
                    tracing::info!(topic = %cfg.topic, group = %cfg.group_id, "governance consumer connected");
                    run(&consumer, &handler).await;
                }
                Err(e) => {
                    #[cfg(feature = "observability")]
                    tracing::warn!(error = %e, "governance consumer connect failed; retrying");
                    let _ = e;
                }
            }
            tokio::time::sleep(std::time::Duration::from_secs(10)).await;
        }
    })
}

async fn run(consumer: &StreamConsumer, handler: &Arc<dyn GovernanceHandler>) {
    loop {
        match consumer.recv().await {
            Ok(message) => {
                let applied = match message.payload() {
                    Some(payload) => {
                        handle_with_retry(payload, handler, &HANDLER_RETRY_BACKOFF).await
                    }
                    // Nothing to apply, so nothing to retry.
                    None => true,
                };
                if !applied {
                    // The in-place retries in `handle_with_retry` are spent.
                    // Deliberately leave the offset where it is and drop the
                    // connection, so this message is redelivered.
                    //
                    // The alternative — commit and move on — is what this loop used
                    // to do, and it silently discarded governance decisions the
                    // handler had said it could not apply. Every handler in the
                    // platform was already written against the contract restored
                    // here: `Ok` for "done, or permanently undone", `Err` for
                    // "ask me again". The vault follower's comments state it
                    // outright, and it returns `Ok(())` precisely for the cases
                    // that will never succeed on a retry (a payload with no
                    // subject, a claim that needs an operator) so they cannot wedge
                    // the stream.
                    //
                    // The cost is that a handler which keeps failing stalls the
                    // partition rather than skipping past. That is the right way
                    // round for this stream: a stall stops proposals from applying,
                    // which somebody notices, while a skip loses an approved
                    // decision in a way nothing reports. Reconnect is on a 10s
                    // cycle, so a cex-policy outage resolves itself.
                    break;
                }
                if let Err(e) = consumer.commit_message(&message, CommitMode::Async) {
                    #[cfg(feature = "observability")]
                    tracing::warn!(error = %e, "governance consumer commit failed");
                    let _ = e;
                }
            }
            Err(e) => {
                #[cfg(feature = "observability")]
                tracing::warn!(error = %e, "governance consumer recv error; reconnecting");
                let _ = e;
                break;
            }
        }
    }
}

/// Waits between in-place retries of a handler that asked to be retried
/// (`Err`). Bounded on purpose: about 30 s in total keeps the consumer well
/// inside Kafka's `max.poll.interval.ms`, and a brief cex-policy blip (a pod
/// restart, a rollout) is absorbed without dropping the connection. When the
/// last attempt still fails the offset is held and the consumer reconnects, so
/// the message is redelivered — a longer outage turns into a visible stall,
/// never into a skipped decision.
const HANDLER_RETRY_BACKOFF: [std::time::Duration; 5] = [
    std::time::Duration::from_secs(1),
    std::time::Duration::from_secs(2),
    std::time::Duration::from_secs(4),
    std::time::Duration::from_secs(8),
    std::time::Duration::from_secs(16),
];

/// [`handle`], retried in place with `backoff` between attempts while it asks
/// to be retried. Returns whether the offset may advance; `false` only after
/// every attempt failed. Handlers are idempotent (at-least-once delivery), so
/// repeating one is safe.
async fn handle_with_retry(
    payload: &[u8],
    handler: &Arc<dyn GovernanceHandler>,
    backoff: &[std::time::Duration],
) -> bool {
    if handle(payload, handler).await {
        return true;
    }
    for (attempt, delay) in backoff.iter().enumerate() {
        tokio::time::sleep(*delay).await;
        if handle(payload, handler).await {
            #[cfg(feature = "observability")]
            tracing::info!(
                retries = attempt + 1,
                "governance event applied after retry"
            );
            return true;
        }
        let _ = attempt;
    }
    #[cfg(feature = "observability")]
    tracing::error!(
        attempts = backoff.len() + 1,
        "governance event still failing after in-place retries; holding the offset and reconnecting for redelivery"
    );
    false
}

/// cex-policy's `DomainEvent` is externally tagged:
/// `{ "QuorumReached": { proposal_id, status, occurred_at } }`.
///
/// Returns whether the offset may advance. A malformed event, an event of another
/// kind, and a handler that returned `Ok` are all "advance": none of them will read
/// differently on a second delivery. Only a handler asking to be retried holds the
/// offset back.
async fn handle(payload: &[u8], handler: &Arc<dyn GovernanceHandler>) -> bool {
    let event: serde_json::Value = match serde_json::from_slice(payload) {
        Ok(event) => event,
        Err(e) => {
            #[cfg(feature = "observability")]
            tracing::warn!(error = %e, "unparseable policy event");
            let _ = e;
            return true;
        }
    };
    if let Some(root_action) = event.get("RootActionApplied") {
        return handle_root_action(root_action, handler).await;
    }
    let Some(quorum) = event.get("QuorumReached").and_then(|v| v.as_object()) else {
        return true;
    };
    let Some(proposal_id) = quorum.get("proposal_id").and_then(|v| v.as_str()) else {
        return true;
    };
    let status = quorum.get("status").and_then(|v| v.as_str());

    let result = match status {
        Some("Approved") => handler.apply_approved_proposal(proposal_id).await,
        // `Cancelled` is the proposer withdrawing their own proposal. It has to
        // release like a rejection: a marker that outlives the proposal locks
        // its subject out of ever being proposed again, with nothing anywhere
        // saying why — the same permanent lock the expiry case was fixed for.
        Some("Rejected") | Some("Expired") | Some("Cancelled") => {
            handler.release_rejected_proposal(proposal_id).await
        }
        _ => Ok(()),
    };
    match result {
        Ok(()) => true,
        Err(e) => {
            // Error, not warn: an approved decision that has not been applied is a
            // governance outage, and it stays one until this stops appearing.
            #[cfg(feature = "observability")]
            tracing::error!(
                proposal_id, ?status, error = %e,
                "applying QuorumReached failed; holding the offset and retrying"
            );
            let _ = (e, status);
            false
        }
    }
}

/// `{ "RootActionApplied": { action_id, kind, payload_sha256, root_sub, occurred_at } }`.
/// Same return contract as [`handle`]: a malformed body advances (it reads the
/// same on every delivery), only a handler `Err` holds the offset.
async fn handle_root_action(
    body: &serde_json::Value,
    handler: &Arc<dyn GovernanceHandler>,
) -> bool {
    let event: RootActionAppliedEvent = match serde_json::from_value(body.clone()) {
        Ok(event) => event,
        Err(e) => {
            #[cfg(feature = "observability")]
            tracing::warn!(error = %e, "malformed RootActionApplied event");
            let _ = e;
            return true;
        }
    };
    match handler.apply_root_action(&event).await {
        Ok(()) => true,
        Err(e) => {
            #[cfg(feature = "observability")]
            tracing::error!(
                action_id = %event.action_id, kind = %event.kind, error = %e,
                "applying RootActionApplied failed; holding the offset and retrying"
            );
            let _ = e;
            false
        }
    }
}

fn build_consumer(cfg: &GovernanceConsumerConfig) -> anyhow::Result<StreamConsumer> {
    let settings = {
        let mut s = crate::kafka::KafkaSettings::from_env();
        s.brokers = cfg.brokers.clone();
        s
    };
    let mut client = ClientConfig::new();
    for (k, v) in settings.security_settings() {
        client.set(k, v);
    }
    let consumer: StreamConsumer = client
        .set("group.id", &cfg.group_id)
        .set("enable.auto.commit", "false")
        .set("auto.offset.reset", "earliest")
        .set("isolation.level", "read_committed")
        .create()?;
    consumer.subscribe(&[cfg.topic.as_str()])?;
    Ok(consumer)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};

    /// The offset contract is the whole safety property of this loop, and it was
    /// wrong once in the direction that loses data: the commit used to happen
    /// regardless of what the handler returned, so a handler saying "I could not
    /// apply this" had its approved governance decision discarded. Every handler in
    /// the platform is written against the contract these tests pin.
    #[derive(Default)]
    struct FakeHandler {
        applied: AtomicUsize,
        released: AtomicUsize,
        root_actions: std::sync::Mutex<Vec<RootActionAppliedEvent>>,
        fail_apply: bool,
        fail_root_action: bool,
    }

    #[async_trait]
    impl GovernanceHandler for FakeHandler {
        async fn apply_approved_proposal(&self, _proposal_id: &str) -> anyhow::Result<()> {
            self.applied.fetch_add(1, Ordering::SeqCst);
            if self.fail_apply {
                anyhow::bail!("cex-policy unreachable");
            }
            Ok(())
        }

        async fn release_rejected_proposal(&self, _proposal_id: &str) -> anyhow::Result<()> {
            self.released.fetch_add(1, Ordering::SeqCst);
            Ok(())
        }

        async fn apply_root_action(&self, event: &RootActionAppliedEvent) -> anyhow::Result<()> {
            self.root_actions.lock().unwrap().push(event.clone());
            if self.fail_root_action {
                anyhow::bail!("cex-policy unreachable");
            }
            Ok(())
        }
    }

    /// Implements only the required method, so every other hook is the default.
    struct MinimalHandler;

    #[async_trait]
    impl GovernanceHandler for MinimalHandler {
        async fn apply_approved_proposal(&self, _proposal_id: &str) -> anyhow::Result<()> {
            Ok(())
        }
    }

    fn root_action_event(kind: &str) -> Vec<u8> {
        serde_json::to_vec(&serde_json::json!({
            "RootActionApplied": {
                "action_id": "22222222-2222-4222-8222-222222222222",
                "kind": kind,
                "payload_sha256": "ab".repeat(32),
                "root_sub": "root-sub",
                "occurred_at": "2026-10-23T00:00:00Z"
            }
        }))
        .unwrap()
    }

    fn quorum_event(status: &str) -> Vec<u8> {
        serde_json::to_vec(&serde_json::json!({
            "QuorumReached": {
                "proposal_id": "11111111-1111-4111-8111-111111111111",
                "status": status,
                "occurred_at": "2026-08-08T00:00:00Z"
            }
        }))
        .unwrap()
    }

    #[tokio::test]
    async fn an_applied_proposal_advances_the_offset() {
        let handler: Arc<dyn GovernanceHandler> = Arc::new(FakeHandler::default());
        assert!(handle(&quorum_event("Approved"), &handler).await);
    }

    #[tokio::test]
    async fn a_handler_asking_to_retry_holds_the_offset() {
        let handler: Arc<dyn GovernanceHandler> =
            Arc::new(FakeHandler { fail_apply: true, ..Default::default() });
        assert!(
            !handle(&quorum_event("Approved"), &handler).await,
            "an approved decision the handler could not apply must be redelivered, not discarded"
        );
    }

    /// Fails the first `failures` apply attempts, then succeeds — cex-policy
    /// unreachable for a moment, then back.
    struct FlakyHandler {
        failures: usize,
        attempts: AtomicUsize,
    }

    #[async_trait]
    impl GovernanceHandler for FlakyHandler {
        async fn apply_approved_proposal(&self, _proposal_id: &str) -> anyhow::Result<()> {
            let attempt = self.attempts.fetch_add(1, Ordering::SeqCst);
            if attempt < self.failures {
                anyhow::bail!("cex-policy GetProposal failed transiently (Unavailable)");
            }
            Ok(())
        }
    }

    const NO_WAIT: [std::time::Duration; 3] = [std::time::Duration::ZERO; 3];

    #[tokio::test]
    async fn a_transient_failure_is_retried_in_place_and_applied_after_recovery() {
        let flaky = Arc::new(FlakyHandler {
            failures: 2,
            attempts: AtomicUsize::new(0),
        });
        let handler: Arc<dyn GovernanceHandler> = flaky.clone();
        assert!(handle_with_retry(&quorum_event("Approved"), &handler, &NO_WAIT).await);
        assert_eq!(
            flaky.attempts.load(Ordering::SeqCst),
            3,
            "two failures, then the apply"
        );
    }

    #[tokio::test]
    async fn a_failure_outlasting_the_retries_holds_the_offset() {
        let flaky = Arc::new(FlakyHandler {
            failures: usize::MAX,
            attempts: AtomicUsize::new(0),
        });
        let handler: Arc<dyn GovernanceHandler> = flaky.clone();
        assert!(
            !handle_with_retry(&quorum_event("Approved"), &handler, &NO_WAIT).await,
            "an approval that never applied must stay uncommitted for redelivery"
        );
        assert_eq!(
            flaky.attempts.load(Ordering::SeqCst),
            NO_WAIT.len() + 1,
            "bounded"
        );
    }

    #[tokio::test]
    async fn a_successful_apply_is_not_repeated() {
        let flaky = Arc::new(FlakyHandler {
            failures: 0,
            attempts: AtomicUsize::new(0),
        });
        let handler: Arc<dyn GovernanceHandler> = flaky.clone();
        assert!(handle_with_retry(&quorum_event("Approved"), &handler, &NO_WAIT).await);
        assert_eq!(flaky.attempts.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn rejected_expired_and_cancelled_go_to_the_release_path() {
        for status in ["Rejected", "Expired", "Cancelled"] {
            let fake = Arc::new(FakeHandler::default());
            let handler: Arc<dyn GovernanceHandler> = fake.clone();
            assert!(handle(&quorum_event(status), &handler).await);
            assert_eq!(fake.released.load(Ordering::SeqCst), 1, "{status}");
            assert_eq!(fake.applied.load(Ordering::SeqCst), 0, "{status}");
        }
    }

    /// None of these will read differently on a second delivery, so holding the
    /// offset for them would wedge the stream on a message nothing can consume.
    #[tokio::test]
    async fn unconsumable_events_advance_rather_than_wedge_the_stream() {
        let fake = Arc::new(FakeHandler::default());
        let handler: Arc<dyn GovernanceHandler> = fake.clone();

        assert!(handle(b"not json at all", &handler).await);
        assert!(handle(br#"{"BootstrapSealed":{"occurred_at":"2026-08-08T00:00:00Z"}}"#, &handler).await);
        assert!(handle(br#"{"QuorumReached":{"status":"Approved"}}"#, &handler).await);
        assert!(handle(&quorum_event("Pending"), &handler).await);

        assert_eq!(fake.applied.load(Ordering::SeqCst), 0);
        assert_eq!(fake.released.load(Ordering::SeqCst), 0);
    }

    #[tokio::test]
    async fn root_action_applied_is_dispatched_with_its_fields() {
        let fake = Arc::new(FakeHandler::default());
        let handler: Arc<dyn GovernanceHandler> = fake.clone();
        assert!(handle(&root_action_event("SETUP_COMMIT"), &handler).await);

        let seen = fake.root_actions.lock().unwrap().clone();
        assert_eq!(
            seen,
            vec![RootActionAppliedEvent {
                action_id: "22222222-2222-4222-8222-222222222222".into(),
                kind: "SETUP_COMMIT".into(),
                payload_sha256: "ab".repeat(32),
                root_sub: "root-sub".into(),
                occurred_at: "2026-10-23T00:00:00Z".into(),
            }]
        );
        assert_eq!(fake.applied.load(Ordering::SeqCst), 0);
        assert_eq!(fake.released.load(Ordering::SeqCst), 0);
    }

    #[tokio::test]
    async fn a_root_action_handler_asking_to_retry_holds_the_offset() {
        let handler: Arc<dyn GovernanceHandler> = Arc::new(FakeHandler {
            fail_root_action: true,
            ..Default::default()
        });
        assert!(!handle(&root_action_event("SETUP_COMMIT"), &handler).await);
    }

    #[tokio::test]
    async fn the_default_root_action_hook_is_a_no_op_that_advances() {
        let handler: Arc<dyn GovernanceHandler> = Arc::new(MinimalHandler);
        assert!(handle(&root_action_event("AUTHZ_MUTATION"), &handler).await);
        assert!(MinimalHandler
            .apply_root_action(&RootActionAppliedEvent {
                action_id: "x".into(),
                kind: "SETUP_COMMIT".into(),
                payload_sha256: String::new(),
                root_sub: String::new(),
                occurred_at: String::new(),
            })
            .await
            .is_ok());
    }

    #[tokio::test]
    async fn a_malformed_root_action_advances_without_dispatch() {
        let fake = Arc::new(FakeHandler::default());
        let handler: Arc<dyn GovernanceHandler> = fake.clone();
        assert!(
            handle(
                br#"{"RootActionApplied":{"kind":"SETUP_COMMIT"}}"#,
                &handler
            )
            .await
        );
        assert!(handle(br#"{"RootActionApplied":"nope"}"#, &handler).await);
        assert!(fake.root_actions.lock().unwrap().is_empty());
    }
}