Add support for bearer auth header for private subgraph endpoints

crates/driver/example.toml

@@ -82,6 +82,7 @@ max-order-age = "1m"
 # [[liquidity.balancer-v2]] # Balancer V2 configuration
 # preset = "balancer-v2"
 # graph-url = "http://localhost:1234" # which subgraph url to fetch the data from
+# api-key = "<goldsky_api_key>" # optional bearer token for private subgraph endpoints (e.g. Goldsky /api/private/...)
 # pool-deny-list = [] # optional
 
 # [[liquidity.balancer-v2]] # Custom Balancer V2 configuration
@@ -94,6 +95,7 @@ max-order-age = "1m"
 # [[liquidity.uniswap-v3]] # Uniswap V3 configuration
 # preset = "uniswap-v3"
 # graph-url = "http://localhost:1234" # which subgraph url to fetch the data from
+# api-key = "<goldsky_api_key>" # optional bearer token for private subgraph endpoints (e.g. Goldsky /api/private/...)
 # max_pools_to_initialize = 100 # how many of the deepest pools to initialise on startup
 
 # [[liquidity.uniswap-v3]] # Custom Uniswap V3 configuration

crates/driver/src/boundary/liquidity/balancer/v2/mod.rs

@@ -177,6 +177,7 @@ async fn init_liquidity(
             boundary::liquidity::cache_config(),
             block_stream.clone(),
             boundary::liquidity::http_client(),
+            config.api_key.clone(),
             web3.clone(),
             &contracts,
             config.pool_deny_list.to_vec(),

crates/driver/src/boundary/liquidity/uniswap/v3.rs

@@ -120,6 +120,7 @@ async fn init_liquidity(
             &config.graph_url,
             web3.clone(),
             boundary::liquidity::http_client(),
+            config.api_key.clone(),
             block_retriever,
             config.max_pools_to_initialize,
             config.max_pools_per_tick_query,

crates/driver/src/infra/config/file/load.rs

@@ -224,10 +224,12 @@ pub async fn load(chain: Chain, path: &Path) -> infra::Config {
                         preset,
                         max_pools_to_initialize,
                         graph_url,
+                        api_key,
                         reinit_interval,
                         max_pools_per_tick_query,
                     } => liquidity::config::UniswapV3 {
                         max_pools_to_initialize,
+                        api_key,
                         reinit_interval,
                         ..match preset {
                             file::UniswapV3Preset::UniswapV3 => {
@@ -244,12 +246,14 @@ pub async fn load(chain: Chain, path: &Path) -> infra::Config {
                         router,
                         max_pools_to_initialize,
                         graph_url,
+                        api_key,
                         reinit_interval,
                         max_pools_per_tick_query,
                     } => liquidity::config::UniswapV3 {
                         router: router.into(),
                         max_pools_to_initialize,
                         graph_url,
+                        api_key,
                         reinit_interval,
                         max_pools_per_tick_query,
                     },
@@ -265,9 +269,11 @@ pub async fn load(chain: Chain, path: &Path) -> infra::Config {
                         preset,
                         pool_deny_list,
                         graph_url,
+                        api_key,
                         reinit_interval,
                     } => liquidity::config::BalancerV2 {
                         pool_deny_list: pool_deny_list.clone(),
+                        api_key,
                         reinit_interval,
                         ..match preset {
                             file::BalancerV2Preset::BalancerV2 => {
@@ -285,6 +291,7 @@ pub async fn load(chain: Chain, path: &Path) -> infra::Config {
                         composable_stable,
                         pool_deny_list,
                         graph_url,
+                        api_key,
                         reinit_interval,
                     } => liquidity::config::BalancerV2 {
                         vault: vault.into(),
@@ -295,6 +302,7 @@ pub async fn load(chain: Chain, path: &Path) -> infra::Config {
                         composable_stable,
                         pool_deny_list: pool_deny_list.clone(),
                         graph_url,
+                        api_key,
                         reinit_interval,
                     },
                 })

crates/driver/src/infra/config/file/mod.rs

@@ -531,6 +531,12 @@ enum UniswapV3Config {
 
         graph_url: Url,
 
+        /// Optional bearer token used to access private subgraph endpoints
+        /// (e.g. Goldsky `/api/private/...`). When set, requests carry the
+        /// `Authorization: Bearer <api-key>` header.
+        #[serde(default)]
+        api_key: Option<String>,
+
         /// How many pool IDs can be present in a where clause of a Tick query
         /// at once. Some subgraphs are overloaded and throw errors when
         /// there are too many.
@@ -561,6 +567,12 @@ enum UniswapV3Config {
         /// The URL used to connect to uniswap v3 subgraph client.
         graph_url: Url,
 
+        /// Optional bearer token used to access private subgraph endpoints
+        /// (e.g. Goldsky `/api/private/...`). When set, requests carry the
+        /// `Authorization: Bearer <api-key>` header.
+        #[serde(default)]
+        api_key: Option<String>,
+
         /// How often the liquidity source should be reinitialized to get
         /// access to new pools.
         #[serde(with = "humantime_serde", default = "default_reinit_interval")]
@@ -598,6 +610,12 @@ enum BalancerV2Config {
         /// The URL used to connect to balancer v2 subgraph client.
         graph_url: Url,
 
+        /// Optional bearer token used to access private subgraph endpoints
+        /// (e.g. Goldsky `/api/private/...`). When set, requests carry the
+        /// `Authorization: Bearer <api-key>` header.
+        #[serde(default)]
+        api_key: Option<String>,
+
         /// How often the liquidity source should be reinitialized to get
         /// access to new pools.
         #[serde(with = "humantime_serde", default = "default_reinit_interval")]
@@ -639,6 +657,12 @@ enum BalancerV2Config {
         /// The URL used to connect to balancer v2 subgraph client.
         graph_url: Url,
 
+        /// Optional bearer token used to access private subgraph endpoints
+        /// (e.g. Goldsky `/api/private/...`). When set, requests carry the
+        /// `Authorization: Bearer <api-key>` header.
+        #[serde(default)]
+        api_key: Option<String>,
+
         /// How often the liquidity source should be reinitialized to get
         /// access to new pools.
         #[serde(with = "humantime_serde", default = "default_reinit_interval")]

crates/driver/src/infra/liquidity/config.rs

@@ -170,6 +170,12 @@ pub struct UniswapV3 {
     /// The URL used to connect to uniswap v3 subgraph client.
     pub graph_url: Url,
 
+    /// Optional bearer token sent as `Authorization: Bearer <api_key>` with
+    /// every subgraph request. Used to access private subgraph endpoints
+    /// (e.g. Goldsky `/api/private/...`).
+    #[debug(ignore)]
+    pub api_key: Option<String>,
+
     /// How often the liquidity source should be reinitialized to
     /// become aware of new pools.
     pub reinit_interval: Option<Duration>,
@@ -192,6 +198,7 @@ impl UniswapV3 {
             router: contracts::UniswapV3SwapRouterV2::deployment_address(&chain.id())?.into(),
             max_pools_to_initialize: 100,
             graph_url: graph_url.clone(),
+            api_key: None,
             reinit_interval: None,
             max_pools_per_tick_query,
         })
@@ -229,6 +236,12 @@ pub struct BalancerV2 {
     /// The base URL used to connect to balancer v2 subgraph client.
     pub graph_url: Url,
 
+    /// Optional bearer token sent as `Authorization: Bearer <api_key>` with
+    /// every subgraph request. Used to access private subgraph endpoints
+    /// (e.g. Goldsky `/api/private/...`).
+    #[debug(ignore)]
+    pub api_key: Option<String>,
+
     /// How often the liquidty source should be re-initialized to become
     /// aware of new pools.
     pub reinit_interval: Option<Duration>,
@@ -285,6 +298,7 @@ impl BalancerV2 {
             ),
             pool_deny_list: Vec::new(),
             graph_url: graph_url.clone(),
+            api_key: None,
             reinit_interval: None,
         })
     }

crates/liquidity-sources/src/balancer_v2/graph_api.rs

@@ -35,10 +35,15 @@ pub struct BalancerSubgraphClient(SubgraphClient);
 
 impl BalancerSubgraphClient {
     /// Creates a new Balancer subgraph client with full subgraph URL.
-    pub fn from_subgraph_url(subgraph_url: &Url, client: Client) -> Result<Self> {
+    pub fn from_subgraph_url(
+        subgraph_url: &Url,
+        client: Client,
+        api_key: Option<String>,
+    ) -> Result<Self> {
         Ok(Self(SubgraphClient::try_new(
             subgraph_url.clone(),
             client,
+            api_key,
             usize::MAX,
         )?))
     }

crates/liquidity-sources/src/balancer_v2/pool_fetching/mod.rs

@@ -240,11 +240,13 @@ impl BalancerPoolFetcher {
         config: CacheConfig,
         block_stream: CurrentBlockWatcher,
         client: Client,
+        api_key: Option<String>,
         web3: Web3,
         contracts: &BalancerContracts,
         deny_listed_pool_ids: Vec<B256>,
     ) -> Result<Self> {
-        let pool_initializer = BalancerSubgraphClient::from_subgraph_url(subgraph_url, client)?;
+        let pool_initializer =
+            BalancerSubgraphClient::from_subgraph_url(subgraph_url, client, api_key)?;
         let web3 = web3.labeled("balancerV2");
         let fetcher = Arc::new(Cache::new(
             create_aggregate_pool_fetcher(

crates/liquidity-sources/src/subgraph.rs

@@ -16,6 +16,11 @@ const MAX_NUMBER_OF_ATTEMPTS_DEFAULT: usize = 10;
 pub struct SubgraphClient {
     client: Client,
     subgraph_url: Url,
+    /// Optional bearer token used to authenticate requests against private
+    /// subgraph endpoints (e.g. Goldsky private endpoints under
+    /// `/api/private/...`). When set, it is sent as the `Authorization:
+    /// Bearer <token>` header on every request.
+    api_key: Option<String>,
     max_pools_per_tick_query: usize,
     max_number_of_attempts: usize,
 }
@@ -32,14 +37,20 @@ pub struct Data<T> {
 
 impl SubgraphClient {
     /// Creates a new subgraph client from the specified organization and name.
+    ///
+    /// If `api_key` is provided, every request sent by this client will carry
+    /// an `Authorization: Bearer <api_key>` header so that private subgraph
+    /// endpoints (e.g. Goldsky private endpoints) can be used.
     pub fn try_new(
         subgraph_url: Url,
         client: Client,
+        api_key: Option<String>,
         max_pools_per_tick_query: usize,
     ) -> Result<Self> {
         Ok(Self {
             client,
             subgraph_url,
+            api_key,
             max_pools_per_tick_query,
             max_number_of_attempts: MAX_NUMBER_OF_ATTEMPTS_DEFAULT,
         })
@@ -74,9 +85,11 @@ impl SubgraphClient {
     where
         T: DeserializeOwned,
     {
-        match self
-            .client
-            .post(self.subgraph_url.clone())
+        let mut request = self.client.post(self.subgraph_url.clone());
+        if let Some(api_key) = &self.api_key {
+            request = request.bearer_auth(api_key);
+        }
+        match request
             .json(&Query {
                 query,
                 variables: variables.clone(),

crates/liquidity-sources/src/uniswap_v3/graph_api.rs

@@ -66,10 +66,15 @@ impl UniV3SubgraphClient {
     pub async fn from_subgraph_url(
         subgraph_url: &Url,
         client: Client,
+        api_key: Option<String>,
         max_pools_per_tick_query: usize,
     ) -> Result<Self> {
-        let subgraph_client =
-            SubgraphClient::try_new(subgraph_url.clone(), client, max_pools_per_tick_query)?;
+        let subgraph_client = SubgraphClient::try_new(
+            subgraph_url.clone(),
+            client,
+            api_key,
+            max_pools_per_tick_query,
+        )?;
 
         Ok(Self {
             client: subgraph_client,

crates/liquidity-sources/src/uniswap_v3/pool_fetching.rs

@@ -138,12 +138,17 @@ impl PoolsCheckpointHandler {
     pub async fn new(
         subgraph_url: &Url,
         client: Client,
+        api_key: Option<String>,
         max_pools_to_initialize_cache: usize,
         max_pools_per_tick_query: usize,
     ) -> Result<Self> {
-        let graph_api =
-            UniV3SubgraphClient::from_subgraph_url(subgraph_url, client, max_pools_per_tick_query)
-                .await?;
+        let graph_api = UniV3SubgraphClient::from_subgraph_url(
+            subgraph_url,
+            client,
+            api_key,
+            max_pools_per_tick_query,
+        )
+        .await?;
         let mut registered_pools = graph_api.get_registered_pools().await?;
         tracing::debug!(
             block = %registered_pools.fetched_block_number, pools = %registered_pools.pools.len(),
@@ -285,6 +290,7 @@ impl UniswapV3PoolFetcher {
         subgraph_url: &Url,
         web3: Web3,
         client: Client,
+        api_key: Option<String>,
         block_retriever: Arc<dyn BlockRetrieving>,
         max_pools_to_initialize: usize,
         max_pools_per_tick_query: usize,
@@ -293,6 +299,7 @@ impl UniswapV3PoolFetcher {
         let checkpoint = PoolsCheckpointHandler::new(
             subgraph_url,
             client,
+            api_key,
             max_pools_to_initialize,
             max_pools_per_tick_query,
         )