shared_framework/gateway/
mod.rs

1//! Basilisk gateway registration and service-bus connection.
2//!
3//! [`GatewayConnect`] registers this service's path prefixes with the Basilisk
4//! gateway and holds the connected [`BasiliskClient`](BasiliskClient).
5//! [`GatewayConnect::set_up`] is the usual entry point (called by
6//! [`GenericStartup`](crate::app::GenericStartup) during bootstrap);
7//! [`GatewayConnect::instance`] returns the process-wide connection once established.
8//!
9//! ```ignore
10//! let gateway = GatewayConnect::set_up(
11//!     vec!["/api".to_string()],
12//!     "http".to_string(),
13//!     1,
14//!     "token".to_string(),
15//! ).await?;
16//! ```
17
18use std::sync::{Arc, OnceLock};
19use std::time::Duration;
20
21use basilisk_rust_client::{BasiliskClient, BasiliskClientConfig};
22
23use crate::env::AppEnvironment;
24use crate::retry::RetryStrategy;
25
26const MAX_CONNECT_ATTEMPTS: usize = 5;
27
28static INSTANCE: OnceLock<Arc<GatewayConnect>> = OnceLock::new();
29
30/// Gateway connection handle. Wraps the connected Basilisk client, if any.
31#[derive(Clone)]
32pub struct GatewayConnect {
33    /// The connected client. Always `Some` on instances returned by [`GatewayConnect::set_up`].
34    pub client: Option<BasiliskClient>,
35}
36
37impl GatewayConnect {
38    fn new(client: BasiliskClient) -> Self {
39        Self { client: Some(client) }
40    }
41
42    /// Returns the process-wide instance if [`GatewayConnect::set_up`] has succeeded.
43    pub fn instance() -> Option<Arc<Self>> {
44        INSTANCE.get().cloned()
45    }
46
47    /// Returns the connected client, or `None` when there is no connection.
48    pub fn get_client(&self) -> Option<&BasiliskClient> {
49        self.client.as_ref()
50    }
51
52    /// Returns the connected client. Errors when there is no connection (e.g. setup was skipped for empty mount paths).
53    pub fn require_client(&self) -> anyhow::Result<&BasiliskClient> {
54        self.client.as_ref().ok_or_else(|| anyhow::anyhow!("gateway not connected — no BasiliskClient (empty mountPaths)"))
55    }
56
57    /// Registers with the gateway and connects the service bus with exponential-backoff retries.
58    ///
59    /// Reads `BASILISK_HOST`, `BASILISK_BUS_PORT`, `BASILISK_TOKEN`,
60    /// `BASILISK_GATEWAY_URL`, `BASILISK_SERVICE_ID`, and `SERVICE_KEY` from
61    /// [`AppEnvironment`](AppEnvironment) (falling back to the process
62    /// environment), and `SERVER_PORT` for the service port (defaults to `8080`).
63    /// `mount_paths` are registered as the service path prefixes; `scheme`,
64    /// `weight`, and `auth_type` go into the registration. Returns `Ok(None)`
65    /// without connecting when `mount_paths` is empty. Errors when already
66    /// initialized, when required env is missing, or after 5 failed connect attempts.
67    pub async fn set_up(
68        mount_paths: Vec<String>,
69        scheme: impl Into<String>,
70        weight: i32,
71        auth_type: impl Into<String>,
72    ) -> anyhow::Result<Option<Arc<Self>>> {
73        if INSTANCE.get().is_some() {
74            anyhow::bail!("Already initialized");
75        }
76        if mount_paths.is_empty() {
77            // Empty mount paths mean no registration.
78            return Ok(None);
79        }
80
81        // Resolve env — prefer AppEnvironment if initialized, else std::env
82        let (host, port, bus_port, token, base_url, service_id, fingerprint) =
83            resolve_env()?;
84
85        let config = BasiliskClientConfig {
86            gateway_base_url: base_url,
87            bus_host: host.clone(),
88            bus_port,
89            service_id: service_id.clone(),
90            fingerprint,
91            path_prefixes: mount_paths.clone(),
92            scheme: scheme.into(),
93            host,
94            port,
95            weight,
96            registration_auth_type: auth_type.into(),
97            registration_token: token,
98        };
99
100        let client = Self::connect_with_retry(config).await?;
101        let instance = Arc::new(Self::new(client));
102        let _ = INSTANCE.set(instance.clone());
103        Ok(Some(instance))
104    }
105
106    /// Connects with an explicit client config and the same retry policy as [`GatewayConnect::set_up`].
107    ///
108    /// Errors when already initialised or after 5 failed connect attempts.
109    pub async fn set_up_with_config(config: BasiliskClientConfig) -> anyhow::Result<Arc<Self>> {
110        if INSTANCE.get().is_some() {
111            anyhow::bail!("Already initialized");
112        }
113        let client = Self::connect_with_retry(config).await?;
114        let instance = Arc::new(Self::new(client));
115        let _ = INSTANCE.set(instance.clone());
116        Ok(instance)
117    }
118
119    async fn connect_with_retry(config: BasiliskClientConfig) -> anyhow::Result<BasiliskClient> {
120        let strategy = RetryStrategy::new()
121            .with_base_delay(Duration::from_millis(1_000))
122            .with_max_delay(Duration::from_millis(30_000))
123            .with_max_attempts(MAX_CONNECT_ATTEMPTS);
124
125        // RetryStrategy in this crate offers `retry` and `with_exponential_backoff`.
126        // Use `retry` with a closure capturing config.
127        let cfg = config.clone();
128        // We need a clone per attempt, so use retry with closure that clones cfg.
129        let result = strategy
130            .retry(|| {
131                let cfg = cfg.clone();
132                async move { BasiliskClient::connect(cfg).await.map_err(|e| e.to_string()) }
133            })
134            .await;
135
136        match result {
137            Ok(client) => Ok(client),
138            Err(e) => anyhow::bail!("basilisk connect exhausted after {} attempts: {}", e.attempts, e.source),
139        }
140    }
141
142    /// Deregisters this service from the gateway. No-ops when there is no connected client.
143    pub async fn deregister(&self) -> anyhow::Result<()> {
144        if let Some(c) = self.client.as_ref() {
145            c.deregister().await
146        } else {
147            Ok(())
148        }
149    }
150}
151
152fn resolve_env() -> anyhow::Result<(String, u16, u16, String, String, String, String)> {
153    // Try AppEnvironment first, fallback to std::env
154    let env = AppEnvironment::try_get();
155    let get_required = |key: &str| -> anyhow::Result<String> {
156        if let Some(e) = env {
157            // AppEnvironment has typed getters, but we can use get_value or try to read directly
158            // Fall back to std::env if not found in AppEnvironment's map.
159            if let Some(v) = e.get_value(key) {
160                return Ok(v);
161            }
162        }
163        std::env::var(key).map_err(|_| anyhow::anyhow!("missing required env {}", key))
164    };
165
166    let host = get_required("BASILISK_HOST")?;
167    let port: u16 = if let Some(e) = env {
168        e.get_value("SERVER_PORT")
169            .and_then(|v| v.parse().ok())
170            .unwrap_or(8080)
171    } else {
172        std::env::var("SERVER_PORT")
173            .ok()
174            .and_then(|v| v.parse().ok())
175            .unwrap_or(8080)
176    };
177    let bus_port: u16 = get_required("BASILISK_BUS_PORT")?
178        .parse()
179        .map_err(|_| anyhow::anyhow!("invalid BASILISK_BUS_PORT"))?;
180    let token = get_required("BASILISK_TOKEN")?;
181    let base_url = get_required("BASILISK_GATEWAY_URL")?;
182    let service_id = get_required("BASILISK_SERVICE_ID")?;
183    let fingerprint = get_required("SERVICE_KEY")?;
184
185    Ok((host, port, bus_port, token, base_url, service_id, fingerprint))
186}
shared_framework/gateway/mod.rs

shared_framework/gateway/
mod.rs
