Public release release-be8720d92b1c

Source commit: be8720d92b1c2795f3e5903d8c1a65d70adfd095

Public tree identity: sha256:9503bb4414b023fea6500688ef0931a7f6b4c940c509fdf96c8bf3a8950e207c

crates/clusterflux-dap/src/runtime_client/transport.rs

@@ -0,0 +1,53 @@
+use anyhow::{anyhow, Result};
+use clusterflux_control::ControlSession;
+use clusterflux_core::coordinator_wire_request;
+use serde_json::{json, Value};
+
+use crate::virtual_model::AdapterState;
+
+pub(crate) fn client_user_request(state: &AdapterState, mut request: Value) -> Value {
+    let Some(session_secret) = state.client_session_secret.as_deref() else {
+        return request;
+    };
+    if let Value::Object(fields) = &mut request {
+        fields.remove("tenant");
+        fields.remove("project");
+        fields.remove("actor_user");
+    }
+    json!({
+        "type": "authenticated",
+        "session_secret": session_secret,
+        "request": request,
+    })
+}
+
+pub(super) struct CoordinatorSession {
+    session: ControlSession,
+}
+
+impl CoordinatorSession {
+    pub(super) fn connect(addr: &str) -> Result<Self> {
+        Ok(Self {
+            session: ControlSession::connect(addr)?,
+        })
+    }
+
+    pub(super) fn request(&mut self, request: Value) -> Result<Value> {
+        let wire_request = coordinator_wire_request("dap-1", request);
+        let response = self.session.request(&wire_request)?;
+        if response.get("type").and_then(Value::as_str) == Some("error") {
+            return Err(anyhow!(
+                "{}",
+                response
+                    .get("message")
+                    .and_then(Value::as_str)
+                    .unwrap_or("coordinator returned an error")
+            ));
+        }
+        Ok(response)
+    }
+}
+
+pub(super) fn coordinator_request(addr: &str, request: Value) -> Result<Value> {
+    CoordinatorSession::connect(addr)?.request(request)
+}