road-runner-common 0.31.0

//! Generic client for cex-policy's internal `PolicyInternalService` gRPC —
//! open a quorum proposal for any governed action, or fetch one by id. In-mesh
//! (Istio STRICT mTLS), no app secret: trust is the mesh identity + the
//! NetworkPolicy/AuthorizationPolicy allowlist on the cex-policy side.

pub mod policy_proto {
    tonic::include_proto!("exchange.cex.policy");
}

use policy_proto::policy_internal_service_client::PolicyInternalServiceClient;

use super::error::PolicyFetchError;
use super::port::GovernanceRecordSource;

/// A proposal fetched via `GetProposal` — the payload is included (unlike the
/// public `GET /api/v1/proposals/{id}`), so a governance handler can apply it
/// without a second round-trip.
#[derive(Debug, Clone, PartialEq)]
pub struct GovernanceProposal {
    pub id: String,
    pub policy_id: String,
    /// Resolved governing action (e.g. `"FeeChange"`, `"Listing"`); absent if
    /// the policy lookup failed on cex-policy's side.
    pub action: Option<String>,
    pub status: String,
    /// Parsed proposal payload; `None` if it wasn't valid JSON.
    pub payload: Option<serde_json::Value>,
}

/// A root action fetched via `GetRootAction`. `payload_json` is kept verbatim
/// (not re-serialized) because consumers must verify
/// `sha256(payload_json) == payload_sha256` — and against the hash carried by
/// the `RootActionApplied` event — on the exact bytes cex-policy signed.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct GovernanceRootAction {
    pub action_id: String,
    /// e.g. `AUTHZ_MUTATION`, `SETUP_COMMIT`, `SETUP_AUTHZ`.
    pub kind: String,
    pub payload_json: String,
    /// Lowercase hex sha256 recorded by cex-policy for `payload_json`.
    pub payload_sha256: String,
    /// RFC 3339.
    pub applied_at: String,
}

/// Client for cex-policy's internal proposal API. One instance covers every
/// governed action in the calling service — construct it once and share it
/// (it's cheap to clone; the underlying channel connects lazily and is
/// reference-counted).
#[derive(Clone)]
pub struct PolicyGovernanceClient {
    /// tonic needs a scheme; the configmap convention may already include one.
    target: Option<String>,
}

impl PolicyGovernanceClient {
    /// `url` is the value of e.g. `POLICY_INTERNAL_GRPC_URL`. `None`/empty
    /// leaves the client unconfigured — [`Self::configured`] returns `false`
    /// and every call fails with a clear error rather than panicking, so
    /// callers can fail closed (503) on their governed endpoints.
    pub fn new(url: Option<String>) -> Self {
        let target = url
            .map(|url| url.trim().trim_end_matches('/').to_string())
            .filter(|url| !url.is_empty())
            .map(|url| {
                if url.starts_with("http://") || url.starts_with("https://") {
                    url
                } else {
                    format!("http://{url}")
                }
            });
        Self { target }
    }

    pub fn configured(&self) -> bool {
        self.target.is_some()
    }

    async fn client(
        &self,
    ) -> anyhow::Result<PolicyInternalServiceClient<tonic::transport::Channel>> {
        let target = self
            .target
            .clone()
            .ok_or_else(|| anyhow::anyhow!("cex-policy gRPC target not configured"))?;
        let channel = tonic::transport::Endpoint::from_shared(target)?
            .timeout(std::time::Duration::from_secs(10))
            .connect_timeout(std::time::Duration::from_secs(3))
            .connect_lazy();
        Ok(PolicyInternalServiceClient::new(channel))
    }

    /// Open a quorum proposal against a specific, already-known `policy_id`.
    /// Use [`Self::submit_proposal_for_action`] instead when the caller only
    /// knows the action it's proposing — most machine callers should, so that
    /// adopting/reconfiguring governance for that action never requires
    /// redeploying them. Returns the new proposal id.
    pub async fn submit_proposal(
        &self,
        policy_id: &str,
        proposer: &str,
        payload: &serde_json::Value,
    ) -> anyhow::Result<String> {
        let response = self
            .client()
            .await?
            .create_proposal(policy_proto::CreateProposalRequest {
                policy_id: policy_id.to_string(),
                proposer: proposer.to_string(),
                payload_json: serde_json::to_string(payload)?,
                action: String::new(),
                context_json: String::new(),
            })
            .await?
            .into_inner();
        if response.id.is_empty() {
            anyhow::bail!("cex-policy proposal submit returned no id");
        }
        Ok(response.id)
    }

    /// Open a quorum proposal against whichever policy currently governs
    /// `action` (highest-priority enabled policy for it — cex-policy resolves
    /// this server-side on every call). `Ok(None)` means no policy governs
    /// this action right now — the expected, common state for an action
    /// nobody has configured governance for; treat it as "not governed", not
    /// an error. Creating (or re-prioritizing) the governing policy in
    /// cex-policy takes effect on the very next call — no redeploy, no
    /// `*_POLICY_ID` env var to manage.
    pub async fn submit_proposal_for_action(
        &self,
        action: &str,
        proposer: &str,
        payload: &serde_json::Value,
    ) -> anyhow::Result<Option<String>> {
        self.submit_for_action(action, proposer, payload, None)
            .await
    }

    /// Like [`Self::submit_proposal_for_action`], but carries a typed request
    /// `context` (JSON object of field -> value, e.g. `{"value_try":60000}`)
    /// that cex-policy matches against each candidate policy's conditions to
    /// pick the governing tier. Value tiering is expressed as condition-scoped
    /// policies, and the winning tier's own quorum applies. `None`/empty = only
    /// condition-less (base) policies match.
    pub async fn submit_proposal_for_action_with_context(
        &self,
        action: &str,
        proposer: &str,
        payload: &serde_json::Value,
        context: Option<&serde_json::Value>,
    ) -> anyhow::Result<Option<String>> {
        let context_json = context.map(|c| c.to_string());
        self.submit_for_action(action, proposer, payload, context_json)
            .await
    }

    async fn submit_for_action(
        &self,
        action: &str,
        proposer: &str,
        payload: &serde_json::Value,
        context_json: Option<String>,
    ) -> anyhow::Result<Option<String>> {
        let result = self
            .client()
            .await?
            .create_proposal(policy_proto::CreateProposalRequest {
                policy_id: String::new(),
                proposer: proposer.to_string(),
                payload_json: serde_json::to_string(payload)?,
                action: action.to_string(),
                context_json: context_json.unwrap_or_default(),
            })
            .await;
        match result {
            Ok(response) => {
                let id = response.into_inner().id;
                if id.is_empty() {
                    anyhow::bail!("cex-policy proposal submit returned no id");
                }
                Ok(Some(id))
            }
            Err(status) if status.code() == tonic::Code::NotFound => Ok(None),
            Err(status) => Err(status.into()),
        }
    }

    /// Channel for the read RPCs. Unlike [`Self::client`] it fails with a typed
    /// [`PolicyFetchError`], so a consumer can tell "not configured" from
    /// "unreachable" without parsing a message.
    fn read_client(
        &self,
    ) -> Result<PolicyInternalServiceClient<tonic::transport::Channel>, PolicyFetchError> {
        let target = self
            .target
            .clone()
            .ok_or_else(|| PolicyFetchError::NotConfigured("no target URL".to_string()))?;
        let endpoint = tonic::transport::Endpoint::from_shared(target.clone())
            .map_err(|e| PolicyFetchError::NotConfigured(format!("{target}: {e}")))?;
        let channel = endpoint
            .timeout(std::time::Duration::from_secs(10))
            .connect_timeout(std::time::Duration::from_secs(3))
            .connect_lazy();
        Ok(PolicyInternalServiceClient::new(channel))
    }

    /// Fetch a proposal (with payload) by id.
    ///
    /// - `Ok(Some(_))` — cex-policy returned the proposal.
    /// - `Ok(None)` — cex-policy answered `NOT_FOUND`. Definitive.
    /// - `Err(_)` — cex-policy did not answer (unreachable, timeout, internal
    ///   error, refused, not configured). **Hold the event**: return `Err` from
    ///   the [`crate::governance::GovernanceHandler`] so the consumer keeps the
    ///   Kafka offset and redelivers. [`PolicyFetchError::is_retryable`] says
    ///   whether a retry is expected to clear it on its own.
    ///
    /// The distinction is the whole point. Versions before 0.26.0 returned
    /// `Ok(None)` for every failure, so the natural handler —
    /// `let Some(p) = fetch(..)? else { return Ok(()) }` — read a one-second
    /// cex-policy outage as "no such proposal", committed the offset, and lost
    /// an approved governance decision with one warn line to show for it.
    pub async fn fetch_proposal(
        &self,
        proposal_id: &str,
    ) -> Result<Option<GovernanceProposal>, PolicyFetchError> {
        let result = self
            .read_client()?
            .get_proposal(policy_proto::GetProposalRequest {
                id: proposal_id.to_string(),
            })
            .await;

        match result {
            Ok(response) => {
                let proposal = response.into_inner();
                let payload = serde_json::from_str(&proposal.payload_json).ok();
                Ok(Some(GovernanceProposal {
                    id: proposal.id,
                    policy_id: proposal.policy_id,
                    action: proposal.action.filter(|action| !action.is_empty()),
                    status: proposal.status,
                    payload,
                }))
            }
            // The one status that is an answer rather than a failure to
            // answer: cex-policy looked and has no such proposal.
            Err(status) if status.code() == tonic::Code::NotFound => {
                #[cfg(feature = "observability")]
                tracing::warn!(proposal_id, "policy internal proposal not found");
                Ok(None)
            }
            Err(status) => {
                let error = PolicyFetchError::from_status("GetProposal", proposal_id, &status);
                #[cfg(feature = "observability")]
                tracing::error!(proposal_id, retryable = error.is_retryable(), error = %error, "policy internal proposal fetch failed");
                Err(error)
            }
        }
    }

    /// Fetch an applied root action (payload included) by id.
    ///
    /// Same contract as [`Self::fetch_proposal`]: `Ok(None)` only when
    /// cex-policy answered NOT_FOUND (unknown id, or never applied); every
    /// transport or server failure is an `Err`, so a handler can hold its
    /// Kafka offset instead of dropping the action. The payload is returned
    /// unverified — checking its hash is the caller's job.
    pub async fn fetch_root_action(
        &self,
        action_id: &str,
    ) -> Result<Option<GovernanceRootAction>, PolicyFetchError> {
        let result = self
            .read_client()?
            .get_root_action(policy_proto::GetRootActionRequest {
                action_id: action_id.to_string(),
            })
            .await;

        match result {
            Ok(response) => {
                let action = response.into_inner();
                Ok(Some(GovernanceRootAction {
                    action_id: action.action_id,
                    kind: action.kind,
                    payload_json: action.payload_json,
                    payload_sha256: action.payload_sha256,
                    applied_at: action.applied_at,
                }))
            }
            Err(status) if status.code() == tonic::Code::NotFound => {
                #[cfg(feature = "observability")]
                tracing::warn!(action_id, "policy internal root action not found");
                Ok(None)
            }
            Err(status) => {
                let error = PolicyFetchError::from_status("GetRootAction", action_id, &status);
                #[cfg(feature = "observability")]
                tracing::error!(action_id, retryable = error.is_retryable(), error = %error, "policy internal root action fetch failed");
                Err(error)
            }
        }
    }

    /// Bootstrap state-machine position (see cex-policy's genesis/handover
    /// ceremony). Every failure is an `Err`: before 0.30.0 this returned
    /// `Ok(None)` on any failure, which a caller could not tell apart from a
    /// real answer.
    pub async fn fetch_bootstrap_state(&self) -> Result<String, PolicyFetchError> {
        self.read_client()?
            .get_bootstrap_state(policy_proto::GetBootstrapStateRequest {})
            .await
            .map(|response| response.into_inner().state)
            .map_err(|status| {
                let error = PolicyFetchError::from_status("GetBootstrapState", "-", &status);
                #[cfg(feature = "observability")]
                tracing::warn!(retryable = error.is_retryable(), error = %error, "policy internal bootstrap-state fetch failed");
                error
            })
    }
}

#[async_trait::async_trait]
impl GovernanceRecordSource for PolicyGovernanceClient {
    async fn fetch_proposal(
        &self,
        proposal_id: &str,
    ) -> Result<Option<GovernanceProposal>, PolicyFetchError> {
        PolicyGovernanceClient::fetch_proposal(self, proposal_id).await
    }

    async fn fetch_root_action(
        &self,
        action_id: &str,
    ) -> Result<Option<GovernanceRootAction>, PolicyFetchError> {
        PolicyGovernanceClient::fetch_root_action(self, action_id).await
    }
}

#[cfg(test)]
mod tests {
    //! Runs the real client against an in-process fake cex-policy that answers
    //! each RPC with a chosen gRPC status, so the three outcomes are pinned on
    //! the wire, not on a hand-written mock of the client.

    use super::policy_proto::policy_internal_service_server::{
        PolicyInternalService, PolicyInternalServiceServer,
    };
    use super::policy_proto::*;
    use super::*;
    use tonic::{Code, Request, Response, Status};

    /// `None` = answer successfully; `Some(code)` = fail with that code.
    struct FakePolicy {
        fail_with: Option<Code>,
    }

    impl FakePolicy {
        fn check(&self) -> Result<(), Status> {
            match self.fail_with {
                None => Ok(()),
                Some(code) => Err(Status::new(code, "fake cex-policy")),
            }
        }
    }

    #[tonic::async_trait]
    impl PolicyInternalService for FakePolicy {
        async fn get_proposal(
            &self,
            request: Request<GetProposalRequest>,
        ) -> Result<Response<GetProposalResponse>, Status> {
            self.check()?;
            Ok(Response::new(GetProposalResponse {
                id: request.into_inner().id,
                policy_id: "policy-1".into(),
                status: "Approved".into(),
                payload_json: r#"{"k":"v"}"#.into(),
                action: Some("FeeChange".into()),
            }))
        }

        async fn get_bootstrap_state(
            &self,
            _request: Request<GetBootstrapStateRequest>,
        ) -> Result<Response<GetBootstrapStateResponse>, Status> {
            self.check()?;
            Ok(Response::new(GetBootstrapStateResponse {
                state: "SEALED".into(),
                ..Default::default()
            }))
        }

        async fn create_proposal(
            &self,
            _request: Request<CreateProposalRequest>,
        ) -> Result<Response<CreateProposalResponse>, Status> {
            Err(Status::unimplemented("not used by these tests"))
        }

        async fn get_admin_credential_key(
            &self,
            _request: Request<GetAdminCredentialKeyRequest>,
        ) -> Result<Response<AdminCredentialKey>, Status> {
            Err(Status::unimplemented("not used by these tests"))
        }

        async fn resolve_admin_by_sub(
            &self,
            _request: Request<ResolveAdminBySubRequest>,
        ) -> Result<Response<ResolvedAdmin>, Status> {
            Err(Status::unimplemented("not used by these tests"))
        }

        async fn get_root_action(
            &self,
            request: Request<GetRootActionRequest>,
        ) -> Result<Response<GetRootActionResponse>, Status> {
            self.check()?;
            Ok(Response::new(GetRootActionResponse {
                action_id: request.into_inner().action_id,
                kind: "SETUP_COMMIT".into(),
                payload_json: "{}".into(),
                payload_sha256: "ab".repeat(32),
                applied_at: "2026-10-27T00:00:00Z".into(),
            }))
        }
    }

    /// Starts a fake cex-policy on an ephemeral port and returns a client
    /// pointed at it.
    async fn client_against(fail_with: Option<Code>) -> PolicyGovernanceClient {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let incoming =
            tonic::transport::server::TcpIncoming::from_listener(listener, true, None).unwrap();
        tokio::spawn(
            tonic::transport::Server::builder()
                .add_service(PolicyInternalServiceServer::new(FakePolicy { fail_with }))
                .serve_with_incoming(incoming),
        );
        PolicyGovernanceClient::new(Some(format!("http://{addr}")))
    }

    const TRANSIENT: [Code; 3] = [Code::Unavailable, Code::DeadlineExceeded, Code::Internal];

    #[tokio::test]
    async fn an_answered_proposal_is_returned() {
        let client = client_against(None).await;
        let proposal = client.fetch_proposal("p-1").await.unwrap().unwrap();
        assert_eq!(proposal.id, "p-1");
        assert_eq!(proposal.status, "Approved");
        assert_eq!(proposal.action.as_deref(), Some("FeeChange"));
        assert_eq!(proposal.payload, Some(serde_json::json!({"k": "v"})));
    }

    #[tokio::test]
    async fn not_found_is_the_only_definitive_absence() {
        let client = client_against(Some(Code::NotFound)).await;
        assert_eq!(client.fetch_proposal("p-1").await.unwrap(), None);
        assert_eq!(client.fetch_root_action("a-1").await.unwrap(), None);
    }

    #[tokio::test]
    async fn transient_statuses_are_retryable_errors_not_absence() {
        for code in TRANSIENT {
            let client = client_against(Some(code)).await;

            let error = client
                .fetch_proposal("p-1")
                .await
                .expect_err("proposal must not be Ok");
            assert!(error.is_retryable(), "{code:?}: {error}");
            assert_eq!(error.code(), Some(code));

            let error = client
                .fetch_root_action("a-1")
                .await
                .expect_err("root action must not be Ok");
            assert!(error.is_retryable(), "{code:?}: {error}");

            let error = client
                .fetch_bootstrap_state()
                .await
                .expect_err("bootstrap must not be Ok");
            assert!(error.is_retryable(), "{code:?}: {error}");
        }
    }

    #[tokio::test]
    async fn a_refusal_is_an_error_but_not_retryable() {
        let client = client_against(Some(Code::PermissionDenied)).await;
        let error = client
            .fetch_proposal("p-1")
            .await
            .expect_err("must not be Ok");
        assert!(!error.is_retryable());
        assert!(matches!(
            error,
            PolicyFetchError::Refused {
                code: Code::PermissionDenied,
                ..
            }
        ));
    }

    #[tokio::test]
    async fn an_unreachable_cex_policy_is_a_retryable_error() {
        // Bind then drop, so the port is known to have no listener.
        let addr = {
            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            listener.local_addr().unwrap()
        };
        let client = PolicyGovernanceClient::new(Some(format!("http://{addr}")));
        let error = client
            .fetch_proposal("p-1")
            .await
            .expect_err("must not be Ok");
        assert!(error.is_retryable(), "{error}");
    }

    #[tokio::test]
    async fn an_unconfigured_client_fails_closed() {
        let client = PolicyGovernanceClient::new(None);
        let error = client
            .fetch_proposal("p-1")
            .await
            .expect_err("must not be Ok");
        assert!(matches!(error, PolicyFetchError::NotConfigured(_)));
        assert!(!error.is_retryable());
    }

    #[tokio::test]
    async fn the_port_delegates_to_the_client() {
        let source: std::sync::Arc<dyn GovernanceRecordSource> =
            std::sync::Arc::new(client_against(Some(Code::Unavailable)).await);
        assert!(source
            .fetch_proposal("p-1")
            .await
            .unwrap_err()
            .is_retryable());
        assert!(source
            .fetch_root_action("a-1")
            .await
            .unwrap_err()
            .is_retryable());
    }

    #[tokio::test]
    async fn a_root_action_is_returned_verbatim() {
        let client = client_against(None).await;
        let action = client.fetch_root_action("a-1").await.unwrap().unwrap();
        assert_eq!(action.action_id, "a-1");
        assert_eq!(action.kind, "SETUP_COMMIT");
        assert_eq!(action.payload_json, "{}");
    }
}