road-runner-common 0.22.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 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(),
                min_threshold: None,
                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, None)
            .await
    }

    /// Like [`Self::submit_proposal_for_action`], but requests a quorum floor:
    /// cex-policy raises the effective threshold to `max(policy base,
    /// min_threshold)` (clamped to the eligible voter count) — it can only make
    /// the quorum stricter, never weaker. `None` is identical to the plain
    /// method.
    ///
    /// DEPRECATED (unified-policy migration): value tiering now lives in policy
    /// conditions — prefer [`Self::submit_proposal_for_action_with_context`].
    /// Kept for backward compat during the transition.
    pub async fn submit_proposal_for_action_with_floor(
        &self,
        action: &str,
        proposer: &str,
        payload: &serde_json::Value,
        min_threshold: Option<u8>,
    ) -> anyhow::Result<Option<String>> {
        self.submit_for_action(action, proposer, payload, min_threshold, 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. This is the unified-policy replacement for the
    /// quorum floor: 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, None, context_json)
            .await
    }

    async fn submit_for_action(
        &self,
        action: &str,
        proposer: &str,
        payload: &serde_json::Value,
        min_threshold: Option<u8>,
        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(),
                min_threshold: min_threshold.map(u32::from),
                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()),
        }
    }

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