feat: GraphQL subscription latestRecords

Cargo.toml

@@ -1,18 +1,18 @@
 [workspace]
 resolver = "2"
 members = [
-  "crates/game_api",
-  "crates/records_lib",
-  "crates/socc",
   "crates/admin",
-  "crates/request_filter",
+  "crates/compute-player-map-ranking",
   "crates/dsc_webhook",
-  "crates/migration",
   "crates/entity",
+  "crates/game_api",
   "crates/graphql-api",
   "crates/graphql-schema-generator",
+  "crates/migration",
   "crates/player-map-ranking",
-  "crates/compute-player-map-ranking",
+  "crates/records_lib",
+  "crates/request_filter",
+  "crates/socc",
   "crates/test-env",
 ]
 

crates/game_api/src/auth.rs

@@ -85,8 +85,6 @@ pub mod privilege {
     pub const ADMIN: Flags = 0b1111;
 }
 
-pub const WEB_TOKEN_SESS_KEY: &str = "__obs_web_token";
-
 /// The state string expires in 5 minutes.
 ///
 /// This is typically used to set a timeout for the POST /player/get_token request sent by

crates/game_api/src/configure.rs

@@ -12,7 +12,7 @@ use actix_web::{
 };
 use dsc_webhook::{FormattedRequestHead, WebhookBody, WebhookBodyEmbed, WebhookBodyEmbedField};
 use mkenv::prelude::*;
-use records_lib::{Database, pool::clone_dbconn};
+use records_lib::{Database, pool::clone_dbconn, records_notifier::RecordsNotifier};
 use tracing_actix_web::{DefaultRootSpanBuilder, RequestId};
 
 use crate::{ApiErrorKind, RecordsErrorKindResponse, RecordsResult, Res, TracedError};
@@ -234,18 +234,25 @@ impl tracing_actix_web::RootSpanBuilder for RootSpanBuilder {
     }
 }
 
-pub fn configure(cfg: &mut web::ServiceConfig, db: Database) {
+pub fn configure(cfg: &mut web::ServiceConfig, db: Database, records_notifier: RecordsNotifier) {
     let client = reqwest::Client::builder()
         .timeout(Duration::from_secs(5))
         .build()
         .unwrap();
 
+    let records_subscription = records_notifier.get_subscription();
+
     cfg.app_data(web::Data::new(crate::AuthState::default()))
         .app_data(client.clone())
         .app_data(clone_dbconn(&db.sql_conn))
         .app_data(db.redis_pool.clone())
         .app_data(db.clone())
-        .service(crate::graphql_route(db.clone(), client))
+        .app_data(records_notifier)
+        .service(crate::graphql_route(
+            db.clone(),
+            client,
+            records_subscription,
+        ))
         .service(crate::api_route())
         .default_service(web::to(not_found));
 }

crates/game_api/src/env.rs

@@ -126,15 +126,6 @@ mkenv::make_config! {
             default_val_fmt: "empty",
         },
 
-        pub gql_endpoint: {
-            var_name: "GQL_ENDPOINT",
-            layers: [
-                or_default_val(|| "/graphql".to_owned()),
-            ],
-            description: "The route to the GraphQL endpoint (e.g. /graphql)",
-            default_val_fmt: "/graphql",
-        },
-
         pub wh_rank_compute_err: {
             var_name: "WEBHOOK_RANK_COMPUTE_ERROR",
             layers: [or_default()],

crates/game_api/src/graphql.rs

@@ -1,92 +1,160 @@
-use actix_session::Session;
-use actix_web::{HttpRequest, web};
-use actix_web::{HttpResponse, Resource, Responder};
-use async_graphql::ErrorExtensionValues;
+use std::sync::Arc;
+
+use actix_http::RequestHead;
+use actix_web::{HttpRequest, Scope, guard, web};
+use actix_web::{HttpResponse, Responder};
 use async_graphql::http::{GraphQLPlaygroundConfig, playground_source};
-use async_graphql_actix_web::GraphQLRequest;
+use async_graphql::{ErrorExtensionValues, Executor};
+use async_graphql_actix_web::{GraphQLRequest, GraphQLSubscription};
+use futures::StreamExt;
+use futures::stream::BoxStream;
 use graphql_api::error::{ApiGqlError, ApiGqlErrorKind};
 use graphql_api::schema::{Schema, create_schema};
-use mkenv::prelude::*;
 use records_lib::Database;
+use records_lib::records_notifier::LatestRecordsSubscription;
 use reqwest::Client;
 use tracing_actix_web::RequestId;
 
-use crate::auth::{WEB_TOKEN_SESS_KEY, WebToken};
-use crate::{ApiErrorKind, RecordsResult, Res, configure, internal};
+use crate::{ApiErrorKind, RecordsResult, Res, configure};
 
-async fn index_graphql(
+#[derive(Clone)]
+struct ExecutorInventory {
+    req_head: RequestHead,
     request_id: RequestId,
-    client: Res<reqwest::Client>,
-    req: HttpRequest,
-    session: Session,
-    schema: Res<Schema>,
-    GraphQLRequest(request): GraphQLRequest,
-) -> RecordsResult<impl Responder> {
-    let web_token = session
-        .get::<WebToken>(WEB_TOKEN_SESS_KEY)
-        .map_err(|e| internal!("unable to retrieve web token: {e}"))?;
-
-    let request = {
-        if let Some(web_token) = web_token {
-            request.data(web_token)
-        } else {
-            request
-        }
-    };
+    client: reqwest::Client,
+}
 
-    let mut result = schema.execute(request).await;
+impl ExecutorInventory {
+    fn mask_internal_errors(
+        &self,
+        mut response: async_graphql::Response,
+    ) -> async_graphql::Response {
+        for error in &mut response.errors {
+            tracing::error!("Error encountered when processing GraphQL request: {error:?}");
 
-    for error in &mut result.errors {
-        tracing::error!("Error encountered when processing GraphQL request: {error:?}");
+            let api_error = error.source::<ApiGqlError>().cloned();
 
-        let api_error = error.source::<ApiGqlError>().cloned();
+            let extensions = error
+                .extensions
+                .get_or_insert_with(ErrorExtensionValues::default);
 
-        let extensions = error
-            .extensions
-            .get_or_insert_with(ErrorExtensionValues::default);
+            // Don't expose internal server errors
+            if let Some(err) = api_error
+                && let ApiGqlErrorKind::Lib(records_err) = err.kind()
+            {
+                let err = ApiErrorKind::Lib(records_err);
+                let (err_type, status_code) = err.get_err_type_and_status_code();
 
-        // Don't expose internal server errors
-        if let Some(err) = api_error
-            && let ApiGqlErrorKind::Lib(records_err) = err.kind()
-        {
-            let err = ApiErrorKind::Lib(records_err);
-            let (err_type, status_code) = err.get_err_type_and_status_code();
+                let mapped_err_type =
+                    if (100..200).contains(&err_type) || status_code.is_server_error() {
+                        error.message = "Internal server error".to_owned();
+                        configure::send_internal_err_msg_detached(
+                            self.client.clone(),
+                            self.req_head.clone(),
+                            self.request_id,
+                            err,
+                        );
 
-            let mapped_err_type = if (100..200).contains(&err_type) || status_code.is_server_error()
-            {
-                error.message = "Internal server error".to_owned();
-                configure::send_internal_err_msg_detached(
-                    client.0.clone(),
-                    req.head().clone(),
-                    request_id,
-                    err,
-                );
-
-                105 // Unknown type
-            } else {
-                err_type
-            };
-
-            extensions.set("error_code", mapped_err_type);
+                        105 // Unknown type
+                    } else {
+                        err_type
+                    };
+
+                extensions.set("error_code", mapped_err_type);
+            }
+
+            extensions.set("request_id", self.request_id.to_string());
         }
 
-        extensions.set("request_id", request_id.to_string());
+        response
+    }
+}
+
+#[derive(Clone)]
+struct GraphqlApiExecutor {
+    schema: Schema,
+    inventory: ExecutorInventory,
+}
+
+impl Executor for GraphqlApiExecutor {
+    async fn execute(&self, request: async_graphql::Request) -> async_graphql::Response {
+        let result = Executor::execute(&self.schema, request).await;
+        self.inventory.mask_internal_errors(result)
     }
 
+    fn execute_stream(
+        &self,
+        request: async_graphql::Request,
+        session_data: Option<Arc<async_graphql::Data>>,
+    ) -> BoxStream<'static, async_graphql::Response> {
+        let inventory = self.inventory.clone();
+        Executor::execute_stream(&self.schema, request, session_data)
+            .map(move |response| inventory.mask_internal_errors(response))
+            .boxed()
+    }
+}
+
+async fn index_graphql(
+    request_id: RequestId,
+    client: Res<reqwest::Client>,
+    req: HttpRequest,
+    schema: Res<Schema>,
+    GraphQLRequest(request): GraphQLRequest,
+) -> RecordsResult<impl Responder> {
+    let executor = GraphqlApiExecutor {
+        schema: schema.0,
+        inventory: ExecutorInventory {
+            req_head: req.head().clone(),
+            request_id,
+            client: client.0,
+        },
+    };
+
+    let result = executor.execute(request).await;
+
     Ok(web::Json(result))
 }
 
 async fn index_playground() -> impl Responder {
     HttpResponse::Ok()
         .content_type("text/html; charset=utf-8")
-        .body(playground_source(GraphQLPlaygroundConfig::new(
-            &crate::env().gql_endpoint.get(),
-        )))
+        .body(playground_source(
+            GraphQLPlaygroundConfig::new("/graphql")
+                .subscription_endpoint("/graphql/subscriptions"),
+        ))
+}
+
+async fn index_subscriptions(
+    request_id: RequestId,
+    client: Res<reqwest::Client>,
+    schema: Res<Schema>,
+    req: HttpRequest,
+    payload: web::Payload,
+) -> Result<impl Responder, actix_web::Error> {
+    GraphQLSubscription::new(GraphqlApiExecutor {
+        schema: schema.0,
+        inventory: ExecutorInventory {
+            req_head: req.head().clone(),
+            request_id,
+            client: client.0,
+        },
+    })
+    .start(&req, payload)
 }
 
-pub fn graphql_route(db: Database, client: Client) -> Resource {
-    web::resource("/graphql")
-        .app_data(create_schema(db, client))
-        .route(web::get().to(index_playground))
-        .route(web::post().to(index_graphql))
+pub fn graphql_route(
+    db: Database,
+    client: Client,
+    records_sub: LatestRecordsSubscription,
+) -> Scope {
+    web::scope("/graphql")
+        .app_data(create_schema(db, client, records_sub))
+        .route("", web::get().to(index_playground))
+        .route("", web::post().to(index_graphql))
+        .route(
+            "/subscriptions",
+            web::get()
+                .guard(guard::Header("upgrade", "websocket"))
+                .to(index_subscriptions),
+        )
 }