refactor: move process_request to request handler

Signed-off-by: Gustavo Inacio <[email protected]>

Author: gusinacio

crates/service/src/error.rs

@@ -26,8 +26,6 @@ pub enum IndexerServiceError {
 
     #[error("Issues with provided receipt: {0}")]
     ReceiptError(#[from] tap_core::Error),
-    #[error("Error while processing the request: {0}")]
-    ProcessingError(SubgraphServiceError),
 
     #[error("There was an error while accessing escrow account: {0}")]
     EscrowAccount(#[from] EscrowAccountsError),
@@ -43,7 +41,7 @@ impl IntoResponse for IndexerServiceError {
         }
 
         let status = match self {
-            ReceiptError(_) | EscrowAccount(_) | ProcessingError(_) => StatusCode::BAD_REQUEST,
+            ReceiptError(_) | EscrowAccount(_) => StatusCode::BAD_REQUEST,
             ReceiptNotFound => StatusCode::PAYMENT_REQUIRED,
             DeploymentIdNotFound => StatusCode::INTERNAL_SERVER_ERROR,
             AxumError(_) => StatusCode::BAD_REQUEST,

crates/service/src/routes/request_handler.rs

@@ -1,30 +1,58 @@
 // Copyright 2023-, Edge & Node, GraphOps, and Semiotic Labs.
 // SPDX-License-Identifier: Apache-2.0
 
-use std::sync::Arc;
-
-use crate::error::IndexerServiceError;
+use crate::{
+    error::SubgraphServiceError, middleware::AttestationInput, service::SubgraphServiceState,
+};
 use axum::{
     extract::{Path, State},
+    http::{HeaderValue, Response},
     response::IntoResponse,
 };
+use reqwest::header::CONTENT_TYPE;
 use thegraph_core::DeploymentId;
 use tracing::trace;
 
-use crate::service::IndexerServiceState;
-
 pub async fn request_handler(
-    Path(manifest_id): Path<DeploymentId>,
-    State(state): State<Arc<IndexerServiceState>>,
+    Path(deployment): Path<DeploymentId>,
+    State(state): State<SubgraphServiceState>,
     req: String,
-) -> Result<impl IntoResponse, IndexerServiceError> {
-    trace!("Handling request for deployment `{manifest_id}`");
+) -> Result<impl IntoResponse, SubgraphServiceError> {
+    trace!("Handling request for deployment `{deployment}`");
+
+    let deployment_url = state
+        .graph_node_query_base_url
+        .join(&format!("subgraphs/id/{deployment}"))
+        .map_err(|_| SubgraphServiceError::InvalidDeployment(deployment))?;
 
     let response = state
-        .service_impl
-        .process_request(manifest_id, req)
+        .graph_node_client
+        .post(deployment_url)
+        .body(req.clone())
+        .header(CONTENT_TYPE, HeaderValue::from_static("application/json"))
+        .send()
         .await
-        .map_err(IndexerServiceError::ProcessingError)?;
+        .map_err(SubgraphServiceError::QueryForwardingError)?;
+
+    let attestable = response
+        .headers()
+        .get("graph-attestable")
+        .map_or(false, |value| {
+            value.to_str().map(|value| value == "true").unwrap_or(false)
+        });
+
+    let body = response
+        .text()
+        .await
+        .map_err(SubgraphServiceError::QueryForwardingError)?;
+    let attestation_input = if attestable {
+        AttestationInput::Attestable { req }
+    } else {
+        AttestationInput::NotAttestable
+    };
+
+    let mut response = Response::new(body);
+    response.extensions_mut().insert(attestation_input);
 
     Ok(response)
 }

crates/service/src/service.rs

@@ -4,28 +4,25 @@
 use std::sync::Arc;
 use std::time::Duration;
 
-use super::{error::SubgraphServiceError, routes};
+use super::routes;
 use anyhow::anyhow;
 use async_graphql::{EmptySubscription, Schema};
 use async_graphql_axum::GraphQL;
 use axum::{
-    http::{HeaderValue, Response},
     routing::{post, post_service},
     Router,
 };
 use indexer_config::{Config, DipsConfig};
-use reqwest::{header::CONTENT_TYPE, Url};
+use reqwest::Url;
 use sqlx::PgPool;
 use thegraph_core::attestation::eip712_domain;
-use thegraph_core::DeploymentId;
 
 use crate::{
     cli::Cli,
     database::{
         self,
         dips::{AgreementStore, InMemoryAgreementStore},
     },
-    middleware::AttestationInput,
     routes::dips::Price,
     service::indexer_service::{IndexerServiceOptions, IndexerServiceRelease},
 };
@@ -35,7 +32,6 @@ use tracing::error;
 mod indexer_service;
 mod tap_receipt_header;
 
-pub use indexer_service::{AttestationOutput, IndexerServiceResponse, IndexerServiceState};
 pub use tap_receipt_header::TapReceipt;
 
 #[derive(Clone)]
@@ -48,62 +44,6 @@ pub struct SubgraphServiceState {
     pub graph_node_query_base_url: &'static Url,
 }
 
-pub struct SubgraphService {
-    state: SubgraphServiceState,
-}
-
-impl SubgraphService {
-    fn new(state: SubgraphServiceState) -> Self {
-        Self { state }
-    }
-}
-
-impl SubgraphService {
-    pub async fn process_request(
-        &self,
-        deployment: DeploymentId,
-        req: String,
-    ) -> Result<Response<String>, SubgraphServiceError> {
-        let deployment_url = self
-            .state
-            .graph_node_query_base_url
-            .join(&format!("subgraphs/id/{deployment}"))
-            .map_err(|_| SubgraphServiceError::InvalidDeployment(deployment))?;
-
-        let response = self
-            .state
-            .graph_node_client
-            .post(deployment_url)
-            .body(req.clone())
-            .header(CONTENT_TYPE, HeaderValue::from_static("application/json"))
-            .send()
-            .await
-            .map_err(SubgraphServiceError::QueryForwardingError)?;
-
-        let attestable = response
-            .headers()
-            .get("graph-attestable")
-            .map_or(false, |value| {
-                value.to_str().map(|value| value == "true").unwrap_or(false)
-            });
-
-        let body = response
-            .text()
-            .await
-            .map_err(SubgraphServiceError::QueryForwardingError)?;
-        let attestation_input = if attestable {
-            AttestationInput::Attestable { req }
-        } else {
-            AttestationInput::NotAttestable
-        };
-
-        let mut response = Response::new(body);
-        response.extensions_mut().insert(attestation_input);
-
-        Ok(response)
-    }
-}
-
 /// Run the subgraph indexer service
 pub async fn run() -> anyhow::Result<()> {
     // Parse command line and environment arguments
@@ -184,7 +124,7 @@ pub async fn run() -> anyhow::Result<()> {
         release,
         config,
         url_namespace: "subgraphs",
-        service_impl: SubgraphService::new(state),
+        subgraph_state: state,
         extra_routes: router,
     })
     .await

crates/service/src/service/indexer_service.rs

@@ -6,7 +6,6 @@ use axum::{
     extract::{MatchedPath, Request as ExtractRequest},
     http::{Method, Request},
     middleware::{from_fn, from_fn_with_state},
-    response::IntoResponse,
     routing::{get, post},
     serve, Json, Router, ServiceExt,
 };
@@ -19,11 +18,8 @@ use prometheus::TextEncoder;
 use reqwest::StatusCode;
 use serde::Serialize;
 use sqlx::postgres::PgPoolOptions;
-use std::{
-    collections::HashMap, error::Error, net::SocketAddr, path::PathBuf, sync::Arc, time::Duration,
-};
+use std::{collections::HashMap, net::SocketAddr, path::PathBuf, sync::Arc, time::Duration};
 use tap_core::{manager::Manager, receipt::checks::CheckList, tap_eip712_domain};
-use thegraph_core::Attestation;
 use tokio::{net::TcpListener, signal};
 use tower::ServiceBuilder;
 use tower_governor::{governor::GovernorConfigBuilder, GovernorLayer};
@@ -51,21 +47,7 @@ use crate::{
 };
 use indexer_config::Config;
 
-use super::SubgraphService;
-
-pub trait IndexerServiceResponse {
-    type Data: IntoResponse;
-    type Error: Error;
-
-    fn is_attestable(&self) -> bool;
-    fn as_str(&self) -> Result<&str, Self::Error>;
-    fn finalize(self, attestation: AttestationOutput) -> Self::Data;
-}
-
-pub enum AttestationOutput {
-    Attestation(Option<Attestation>),
-    Attestable,
-}
+use super::SubgraphServiceState;
 
 #[derive(Clone, Serialize)]
 pub struct IndexerServiceRelease {
@@ -89,16 +71,11 @@ impl From<&BuildInfo> for IndexerServiceRelease {
 }
 
 pub struct IndexerServiceOptions {
-    pub service_impl: SubgraphService,
+    pub subgraph_state: SubgraphServiceState,
     pub config: &'static Config,
     pub release: IndexerServiceRelease,
     pub url_namespace: &'static str,
-    pub extra_routes: Router<Arc<IndexerServiceState>>,
-}
-
-pub struct IndexerServiceState {
-    pub config: Config,
-    pub service_impl: SubgraphService,
+    pub extra_routes: Router<SubgraphServiceState>,
 }
 
 const HTTP_CLIENT_TIMEOUT: Duration = Duration::from_secs(30);
@@ -273,11 +250,6 @@ pub async fn run(options: IndexerServiceOptions) -> Result<(), anyhow::Error> {
         CheckList::new(checks),
     )));
 
-    let state = Arc::new(IndexerServiceState {
-        config: options.config.clone(),
-        service_impl: options.service_impl,
-    });
-
     // Rate limits by allowing bursts of 10 requests and requiring 100ms of
     // time between consecutive requests after that, effectively rate
     // limiting to 10 req/s.
@@ -358,8 +330,6 @@ pub async fn run(options: IndexerServiceOptions) -> Result<(), anyhow::Error> {
         }
     }
 
-    misc_routes = misc_routes.with_state(state.clone());
-
     let mut request_handler_route = post(request_handler);
 
     // inject auth
@@ -421,7 +391,7 @@ pub async fn run(options: IndexerServiceOptions) -> Result<(), anyhow::Error> {
                 .expect("Failed to set up `/{url_namespace}/id/:id` route"),
             request_handler_route,
         )
-        .with_state(state.clone());
+        .with_state(options.subgraph_state.clone());
 
     let router = NormalizePath::trim_trailing_slash(
         misc_routes
@@ -457,7 +427,7 @@ pub async fn run(options: IndexerServiceOptions) -> Result<(), anyhow::Error> {
                          _span: &tracing::Span| {},
                     ),
             )
-            .with_state(state),
+            .with_state(options.subgraph_state),
     );
 
     serve_metrics(options.config.metrics.get_socket_addr());