enhance(router): use Hive Console SDK for Supergraph Fetching (#571)

Use `hive-console-sdk` for fetching supergraph from Hive CDN

Documentation changes graphql-hive/console#7283

Author: ardatan

.changeset/use_hive_sdk_for_supergraph.md

@@ -0,0 +1,20 @@
+---
+default: minor
+---
+
+Use `hive-console-sdk` to load supergraph from Hive Console instead of custom implementation.
+
+### Breaking Changes
+
+The configuration for the `hive` supergraph source has been updated. The `timeout` field is now `request_timeout`, and new options `connect_timeout` and `accept_invalid_certs` have been added.
+
+```yaml
+supergraph:
+  source: hive
+  endpoint: "https://cdn.graphql-hive.com/supergraph"
+  key: "YOUR_CDN_KEY"
+  # Old `timeout` is now `request_timeout`
+  request_timeout: 30s
+  # New options
+  connect_timeout: 10s
+  accept_invalid_certs: false

Cargo.lock

@@ -48,6 +48,7 @@ serde_json = { workspace = true }
 
 mimalloc = { version = "0.1.48", features = ["v3"] }
 moka = { version = "0.12.10", features = ["future"] }
+hive-console-sdk = "0.2.0"
 ulid = "1.2.1"
 tokio-util = "0.7.16"
 cookie = "0.18.1"

bin/router/Cargo.toml

@@ -9,6 +9,12 @@ pub enum LoadSupergraphError {
     NetworkError(#[from] reqwest_middleware::Error),
     #[error("Failed to read supergraph from network: {0}")]
     NetworkResponseError(#[from] reqwest::Error),
+    #[error("Failed to lock supergraph: {0}")]
+    LockError(String),
+    #[error("Failed to initialize the loader: {0}")]
+    InitializationError(String),
+    #[error("Invalid configuration: {0}")]
+    InvalidConfiguration(String),
 }
 
 #[derive(Debug)]

bin/router/src/supergraph/base.rs

@@ -1,91 +1,52 @@
 use async_trait::async_trait;
-use http::{
-    header::{ETAG, IF_NONE_MATCH, USER_AGENT},
-    HeaderValue, StatusCode,
+use hive_console_sdk::supergraph_fetcher::{
+    SupergraphFetcher, SupergraphFetcherAsyncState, SupergraphFetcherError,
 };
-use lazy_static::lazy_static;
-use reqwest_middleware::{ClientBuilder, ClientWithMiddleware};
-use reqwest_retry::RetryTransientMiddleware;
-use retry_policies::policies::ExponentialBackoff;
 use std::time::Duration;
-use tokio::sync::RwLock;
 use tracing::{debug, error};
 
 use crate::{
     consts::ROUTER_VERSION,
     supergraph::base::{LoadSupergraphError, ReloadSupergraphResult, SupergraphLoader},
 };
 
-lazy_static! {
-    pub static ref USER_AGENT_VALUE: HeaderValue = {
-        HeaderValue::from_str(&format!("hive-router/{}", ROUTER_VERSION))
-            .expect("failed to construct user-agent")
-    };
-}
-
-static AUTH_HEADER_NAME: &str = "x-hive-cdn-key";
-
 pub struct SupergraphHiveConsoleLoader {
-    endpoint: String,
-    key: String,
-    http_client: ClientWithMiddleware,
+    fetcher: SupergraphFetcher<SupergraphFetcherAsyncState>,
     poll_interval: Duration,
-    timeout: Duration,
-    last_etag: RwLock<Option<HeaderValue>>,
 }
 
-#[async_trait]
-impl SupergraphLoader for SupergraphHiveConsoleLoader {
-    async fn load(&self) -> Result<ReloadSupergraphResult, LoadSupergraphError> {
-        debug!(
-            "Fetching supergraph from Hive Console CDN: '{}'",
-            self.endpoint,
-        );
-
-        let mut req = self
-            .http_client
-            .get(&self.endpoint)
-            .header(AUTH_HEADER_NAME, &self.key)
-            .header(USER_AGENT, USER_AGENT_VALUE.clone())
-            .timeout(self.timeout);
-
-        let mut etag_used = false;
-
-        match self.last_etag.try_read() {
-            Ok(lock_guard) => {
-                if let Some(etag) = lock_guard.as_ref() {
-                    req = req.header(IF_NONE_MATCH, etag);
-                    etag_used = true;
-                }
+impl From<SupergraphFetcherError> for LoadSupergraphError {
+    fn from(err: SupergraphFetcherError) -> Self {
+        match err {
+            SupergraphFetcherError::NetworkError(e) => LoadSupergraphError::NetworkError(e),
+            SupergraphFetcherError::NetworkResponseError(e) => {
+                LoadSupergraphError::NetworkResponseError(e)
             }
-            Err(e) => {
-                error!("Failed to read etag record: {:?}", e);
+            SupergraphFetcherError::Lock(e) => LoadSupergraphError::LockError(e),
+            SupergraphFetcherError::FetcherCreationError(e) => {
+                LoadSupergraphError::InitializationError(e.to_string())
             }
-        };
-
-        let response = req.send().await?.error_for_status()?;
-
-        if etag_used && response.status() == StatusCode::NOT_MODIFIED {
-            Ok(ReloadSupergraphResult::Unchanged)
-        } else {
-            if let Some(new_etag) = response.headers().get(ETAG) {
-                match self.last_etag.try_write() {
-                    Ok(mut v) => {
-                        debug!("saving etag record: {:?}", new_etag);
-                        *v = Some(new_etag.clone());
-                    }
-                    Err(e) => {
-                        error!("Failed to save etag record: {:?}", e);
-                    }
-                }
+            SupergraphFetcherError::InvalidKey(e) => {
+                LoadSupergraphError::InvalidConfiguration(format!("Invalid CDN key: {}", e))
             }
+        }
+    }
+}
 
-            let content = response
-                .text()
-                .await
-                .map_err(LoadSupergraphError::NetworkResponseError)?;
-
-            Ok(ReloadSupergraphResult::Changed { new_sdl: content })
+#[async_trait]
+impl SupergraphLoader for SupergraphHiveConsoleLoader {
+    async fn load(&self) -> Result<ReloadSupergraphResult, LoadSupergraphError> {
+        let fetcher_result = self.fetcher.fetch_supergraph().await;
+        match fetcher_result {
+            // If there was an error fetching the supergraph, propagate it
+            Err(err) => {
+                error!("Error fetching supergraph from Hive Console: {}", err);
+                Err(LoadSupergraphError::from(err))
+            }
+            // If the supergraph has not changed, return Unchanged
+            Ok(None) => Ok(ReloadSupergraphResult::Unchanged),
+            // If there is a new supergraph SDL, return it
+            Ok(Some(sdl)) => Ok(ReloadSupergraphResult::Changed { new_sdl: sdl }),
         }
     }
 
@@ -95,31 +56,34 @@ impl SupergraphLoader for SupergraphHiveConsoleLoader {
 }
 
 impl SupergraphHiveConsoleLoader {
-    pub fn new(
+    pub fn try_new(
         endpoint: String,
         key: &str,
         poll_interval: Duration,
-        timeout: Duration,
-        retry_policy: ExponentialBackoff,
+        connect_timeout: Duration,
+        request_timeout: Duration,
+        accept_invalid_certs: bool,
+        retry_count: u32,
     ) -> Result<Box<Self>, LoadSupergraphError> {
         debug!(
-            "Creating supergraph source from Hive Console CDN: '{}' (poll interval: {}ms, timeout: {}ms)",
+            "Creating supergraph source from Hive Console CDN: '{}' (poll interval: {}ms, request_timeout: {}ms)",
             endpoint,
             poll_interval.as_millis(),
-            timeout.as_millis()
+            request_timeout.as_millis()
         );
-
-        let client = ClientBuilder::new(reqwest::Client::new())
-            .with(RetryTransientMiddleware::new_with_policy(retry_policy))
-            .build();
-
-        Ok(Box::new(Self {
+        let fetcher = SupergraphFetcher::try_new_async(
             endpoint,
-            key: key.to_string(),
-            http_client: client,
+            key,
+            format!("hive-router/{}", ROUTER_VERSION),
+            connect_timeout,
+            request_timeout,
+            accept_invalid_certs,
+            retry_count,
+        )?;
+
+        Ok(Box::new(SupergraphHiveConsoleLoader {
+            fetcher,
             poll_interval,
-            timeout,
-            last_etag: RwLock::new(None),
         }))
     }
 }

bin/router/src/supergraph/hive.rs

@@ -27,22 +27,19 @@ pub fn resolve_from_config(
         SupergraphSource::HiveConsole {
             endpoint,
             key,
-            poll_interval,
+            connect_timeout,
+            request_timeout,
+            accept_invalid_certs,
             retry_policy,
-            timeout,
-        } => {
-            let patched_endpoint = match endpoint.ends_with("/supergraph") {
-                true => endpoint.to_string(),
-                false => format!("{}/supergraph", endpoint),
-            };
-
-            Ok(SupergraphHiveConsoleLoader::new(
-                patched_endpoint,
-                key,
-                *poll_interval,
-                *timeout,
-                retry_policy.into(),
-            )?)
-        }
+            poll_interval,
+        } => Ok(SupergraphHiveConsoleLoader::try_new(
+            endpoint.clone(),
+            key,
+            *poll_interval,
+            *connect_timeout,
+            *request_timeout,
+            *accept_invalid_certs,
+            retry_policy.max_retries,
+        )?),
     }
 }