road-runner-common 0.15.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;

/// 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)]
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>,
}

/// 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 `policy_id` (the specific policy governing
    /// the action being proposed — a service typically has one policy id per
    /// governed action, configured per-deployment). 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)?,
            })
            .await?
            .into_inner();
        if response.id.is_empty() {
            anyhow::bail!("cex-policy proposal submit returned no id");
        }
        Ok(response.id)
    }

    /// Fetch a proposal (with payload) by id. `Ok(None)` on any transport/lookup
    /// failure — callers treat a fetch failure as "try again later", not as a
    /// hard error, since it's driven by an at-least-once Kafka event.
    pub async fn fetch_proposal(
        &self,
        proposal_id: &str,
    ) -> anyhow::Result<Option<GovernanceProposal>> {
        let result = self
            .client()
            .await?
            .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,
                }))
            }
            Err(status) => {
                #[cfg(feature = "observability")]
                tracing::warn!(proposal_id, error = %status, "policy internal proposal fetch failed");
                let _ = status;
                Ok(None)
            }
        }
    }

    /// Bootstrap state-machine position (see cex-policy's genesis/handover
    /// ceremony). `Ok(None)` on transport failure, same fail-soft contract as
    /// [`Self::fetch_proposal`].
    pub async fn fetch_bootstrap_state(&self) -> anyhow::Result<Option<String>> {
        let result = self
            .client()
            .await?
            .get_bootstrap_state(policy_proto::GetBootstrapStateRequest {})
            .await;
        match result {
            Ok(response) => Ok(Some(response.into_inner().state)),
            Err(status) => {
                #[cfg(feature = "observability")]
                tracing::warn!(error = %status, "policy internal bootstrap-state fetch failed");
                let _ = status;
                Ok(None)
            }
        }
    }
}