Gb 7250 add sse client subscription test

.callstack.yml

@@ -0,0 +1,39 @@
+pr_review:
+  # Default: true
+  auto_run: true
+  modules:
+    # Automatically create a description summarizing the changes in pull request. 
+    description:
+      enabled: true
+      diagram: false
+
+    # Find potential bugs in pull request changes or related files.
+    bug_hunter:
+      enabled: true
+      # Include fixes to possible bugs.
+      suggestions: true
+
+    # Suggest improvements to added code.
+    code_suggestions:
+      enabled: true
+
+    # Suggest changes to follow defined code conventions.
+    code_conventions:
+      enabled: false
+      # Describe your code conventions in plain text.
+      conventions: |
+        E.g. Exported variables, functions, classes and methods should be defined before private.
+
+
+    # Point out any typos or grammatical errors in variable names, texts, comments.
+    grammar:
+      enabled: false
+
+    # Suggest performance improvements to added code.
+    performance:
+      enabled: true
+
+    # Find potential security issues in added code.
+    security:
+      enabled: true
+

.github/workflows/callstack-reviewer.yml

@@ -0,0 +1,29 @@
+name: Callstack.ai PR Review
+
+on:
+  workflow_dispatch:
+    inputs:
+      config:
+        type: string
+        description: "config for reviewer"
+        required: true
+      head:
+        type: string
+        description: "head commit sha"
+        required: true
+      base:
+        type: string
+        description: "base commit sha"
+        required: false
+
+jobs:
+  callstack_pr_review_job:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Review PR
+        uses: callstackai/action@main
+        with:
+          config: ${{ inputs.config }}
+          head: ${{ inputs.head }}
+          export: /code/chats.json
+

Cargo.toml

@@ -71,6 +71,7 @@ async-graphql = "7.0.3"
 async-graphql-axum = "7.0.3"
 async-graphql-parser = "7.0.3"
 async-graphql-value = "7.0.3"
+async-sse = "5"
 async-trait = "0.1.80"
 axum = { version = "0.7.5", default-features = false }
 axum-server = { version = "0.6", default-features = false }
@@ -102,6 +103,7 @@ internment = { version = "0.8", features = ["serde", "arc"] }
 itertools = "0.13.0"
 jsonwebtoken = "9.3.0"
 governor = "0.6"
+multipart-stream = "0.1.2"
 num-traits = "0.2.18"
 once_cell = "1.19.0"
 openidconnect = "4.0.0-alpha.1"

engine/crates/gateway-core/Cargo.toml

@@ -19,8 +19,8 @@ workspace = true
 [dependencies]
 async-graphql.workspace = true
 async-runtime.workspace = true
-async-sse = "5.1.0"
-async-trait = "0.1.80"
+async-sse.workspace = true
+async-trait.workspace = true
 blake3.workspace = true
 bytes.workspace = true
 common-types.workspace = true
@@ -36,7 +36,7 @@ headers.workspace = true
 http.workspace = true
 mediatype = "0.19.18"
 mime = "0.3.17"
-multipart-stream = "0.1.2"
+multipart-stream.workspace = true
 operation-normalizer = { path = "../operation-normalizer" }
 partial-caching.workspace = true
 registry-for-cache.workspace = true

engine/crates/integration-tests/Cargo.toml

@@ -12,7 +12,9 @@ async-graphql-parser.workspace = true
 async-graphql.workspace = true
 async-once-cell = "0.5.3"
 async-runtime.workspace = true
+async-sse.workspace = true
 async-trait.workspace = true
+bytes.workspace = true
 crossbeam-queue = "0.3"
 cynic.workspace = true
 cynic-introspection.workspace = true
@@ -32,7 +34,7 @@ headers.workspace = true
 http.workspace = true
 indoc = "2.0.5"
 insta.workspace = true
-multipart-stream = "0.1.2"
+multipart-stream.workspace = true
 names = "0.14.1-dev"
 openidconnect.workspace = true
 reqwest.workspace = true

engine/crates/integration-tests/docker-compose.yml

@@ -1,9 +1,16 @@
 version: '3'
 services:
+  sse-subgraph:
+    restart: unless-stopped
+    build:
+      context: ./data/sse-subgraph
+    ports:
+      - '4092:4092'
+
   # MongoDB
   data-api:
     image: grafbase/mongodb-data-api:latest
-    restart: always
+    restart: unless-stopped
     environment:
       MONGODB_DATABASE_URL: 'mongodb://grafbase:grafbase@mongodb:27017'
     ports:
@@ -15,7 +22,7 @@ services:
 
   mongodb:
     image: mongo:latest
-    restart: always
+    restart: unless-stopped
     environment:
       MONGO_INITDB_ROOT_USERNAME: 'grafbase'
       MONGO_INITDB_ROOT_PASSWORD: 'grafbase'
@@ -28,7 +35,7 @@ services:
   # Postgres
   postgres:
     image: postgres:16
-    restart: always
+    restart: unless-stopped
     command: postgres -c 'max_connections=1000'
     environment:
       POSTGRES_PASSWORD: 'grafbase'
@@ -61,7 +68,7 @@ services:
     environment:
       DSN: 'sqlite:///var/lib/sqlite/db.sqlite?_fk=true'
       URLS_SELF_ISSUER: 'http://127.0.0.1:4444'
-    restart: always
+    restart: unless-stopped
     depends_on:
       - hydra-migrate
     networks:
@@ -104,7 +111,7 @@ services:
       URLS_SELF_ISSUER: 'http://127.0.0.1:4454'
       SERVE_PUBLIC_PORT: '4454'
       SERVE_ADMIN_PORT: '4455'
-    restart: always
+    restart: unless-stopped
     depends_on:
       - hydra-migrate
     networks:

engine/crates/integration-tests/src/federation/mod.rs

@@ -1,24 +1,11 @@
 mod builder;
+mod request;
 
-use std::{
-    any::TypeId,
-    borrow::Cow,
-    collections::HashMap,
-    future::IntoFuture,
-    ops::{Deref, DerefMut},
-    str::FromStr,
-    sync::Arc,
-};
+use std::{any::TypeId, collections::HashMap, sync::Arc};
 
 pub use builder::*;
-use engine::{BatchRequest, Variables};
-use engine_v2::{HttpGraphqlResponse, HttpGraphqlResponseBody};
-use futures::{future::BoxFuture, stream::BoxStream, StreamExt, TryStreamExt};
-use gateway_core::StreamingFormat;
 use graphql_mocks::{MockGraphQlServer, ReceivedRequest};
-use headers::HeaderMapExt;
-use http::{header::Entry, HeaderName, HeaderValue};
-use serde::de::Error;
+pub use request::*;
 
 use crate::engine_v1::GraphQlRequest;
 
@@ -69,176 +56,3 @@ impl TestEngineV2 {
             .collect()
     }
 }
-
-#[must_use]
-pub struct ExecutionRequest {
-    request: GraphQlRequest,
-    #[allow(dead_code)]
-    headers: Vec<(String, String)>,
-    engine: Arc<engine_v2::Engine<TestRuntime>>,
-}
-
-impl ExecutionRequest {
-    pub fn by_client(self, name: &'static str, version: &'static str) -> Self {
-        self.header("x-grafbase-client-name", name)
-            .header("x-grafbase-client-version", version)
-    }
-
-    /// Adds a header into the request
-    pub fn header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
-        self.headers.push((name.into(), value.into()));
-        self
-    }
-
-    pub fn variables(mut self, variables: impl serde::Serialize) -> Self {
-        self.request.variables = Some(Variables::from_json(
-            serde_json::to_value(variables).expect("variables to be serializable"),
-        ));
-        self
-    }
-
-    pub fn extensions(mut self, extensions: impl serde::Serialize) -> Self {
-        self.request.extensions =
-            serde_json::from_value(serde_json::to_value(extensions).expect("extensions to be serializable"))
-                .expect("extensions to be deserializable");
-        self
-    }
-
-    fn http_headers(&self) -> http::HeaderMap {
-        let mut headers = http::HeaderMap::new();
-
-        for (key, value) in &self.headers {
-            let key = HeaderName::from_str(key).unwrap();
-            let value = HeaderValue::from_str(value).unwrap();
-
-            if let Entry::Occupied(mut e) = headers.entry(key.clone()) {
-                e.append(value);
-            } else {
-                headers.insert(key, value);
-            }
-        }
-
-        headers
-    }
-
-    pub fn into_multipart_stream(self) -> MultipartStreamRequest {
-        MultipartStreamRequest(self)
-    }
-}
-
-impl IntoFuture for ExecutionRequest {
-    type Output = GraphqlResponse;
-
-    type IntoFuture = BoxFuture<'static, Self::Output>;
-
-    fn into_future(self) -> Self::IntoFuture {
-        let headers = self.http_headers();
-        let request = BatchRequest::Single(self.request.into_engine_request());
-        Box::pin(async move { self.engine.execute(headers, request).await.try_into().unwrap() })
-    }
-}
-
-pub struct MultipartStreamRequest(ExecutionRequest);
-
-impl MultipartStreamRequest {
-    pub async fn collect<B>(self) -> B
-    where
-        B: Default + Extend<serde_json::Value>,
-    {
-        self.await.stream.collect().await
-    }
-}
-
-impl IntoFuture for MultipartStreamRequest {
-    type Output = GraphqlStreamingResponse;
-
-    type IntoFuture = BoxFuture<'static, Self::Output>;
-
-    fn into_future(self) -> Self::IntoFuture {
-        let mut headers = self.0.http_headers();
-        headers.typed_insert(StreamingFormat::IncrementalDelivery);
-        let request = BatchRequest::Single(self.0.request.into_engine_request());
-        Box::pin(async move {
-            let response = self.0.engine.execute(headers, request).await;
-            let stream = multipart_stream::parse(response.body.into_stream().map_ok(Into::into), "-")
-                .map(|result| serde_json::from_slice(&result.unwrap().body).unwrap());
-            GraphqlStreamingResponse {
-                stream: Box::pin(stream),
-                headers: response.headers,
-            }
-        })
-    }
-}
-
-pub struct GraphqlStreamingResponse {
-    pub stream: BoxStream<'static, serde_json::Value>,
-    pub headers: http::HeaderMap,
-}
-
-#[derive(serde::Serialize, Debug)]
-pub struct GraphqlResponse {
-    #[serde(flatten)]
-    pub body: serde_json::Value,
-    #[serde(skip)]
-    pub headers: http::HeaderMap,
-}
-
-impl TryFrom<HttpGraphqlResponse> for GraphqlResponse {
-    type Error = serde_json::Error;
-
-    fn try_from(response: HttpGraphqlResponse) -> Result<Self, Self::Error> {
-        Ok(GraphqlResponse {
-            body: match response.body {
-                HttpGraphqlResponseBody::Bytes(bytes) => serde_json::from_slice(bytes.as_ref())?,
-                HttpGraphqlResponseBody::Stream(_) => {
-                    return Err(serde_json::Error::custom("Unexpected stream response body"))?
-                }
-            },
-            headers: response.headers,
-        })
-    }
-}
-
-impl std::fmt::Display for GraphqlResponse {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}", serde_json::to_string_pretty(&self.body).unwrap())
-    }
-}
-
-impl Deref for GraphqlResponse {
-    type Target = serde_json::Value;
-
-    fn deref(&self) -> &Self::Target {
-        &self.body
-    }
-}
-
-impl DerefMut for GraphqlResponse {
-    fn deref_mut(&mut self) -> &mut Self::Target {
-        &mut self.body
-    }
-}
-
-impl GraphqlResponse {
-    pub fn into_value(self) -> serde_json::Value {
-        self.body
-    }
-
-    #[track_caller]
-    pub fn into_data(self) -> serde_json::Value {
-        assert!(self.errors().is_empty(), "{self:#?}");
-
-        match self.body {
-            serde_json::Value::Object(mut value) => value.remove("data"),
-            _ => None,
-        }
-        .unwrap_or_default()
-    }
-
-    pub fn errors(&self) -> Cow<'_, Vec<serde_json::Value>> {
-        self.body["errors"]
-            .as_array()
-            .map(Cow::Borrowed)
-            .unwrap_or_else(|| Cow::Owned(Vec::new()))
-    }
-}