feat(apollo-router-plugin): persisted operations (app deployments) plugin with Hive, update usage to API-Version v2 (#5732)

.changeset/proud-news-invite.md

@@ -0,0 +1,5 @@
+---
+'hive-apollo-router-plugin': minor
+---
+
+Updated Apollo-Router custom plugin for Hive to use Usage reporting spec v2. [Learn more](https://the-guild.dev/graphql/hive/docs/specs/usage-reports)

.changeset/rotten-islands-joke.md

@@ -0,0 +1,5 @@
+---
+'hive-apollo-router-plugin': minor
+---
+
+Add support for persisted documents using Hive App Deployments. [Learn more](https://the-guild.dev/graphql/hive/product-updates/2024-07-30-persisted-documents-app-deployments-preview)

packages/libraries/router/Cargo.toml

@@ -19,10 +19,17 @@ path = "src/lib.rs"
 [dependencies]
 apollo-router = { version = "^1.13.0" }
 thiserror = "1.0.57"
-reqwest = { version = "0.12.0", default-features = false, features = ["rustls-tls", "blocking", "json"] }
+reqwest = { version = "0.12.0", default-features = false, features = [
+  "rustls-tls",
+  "blocking",
+  "json",
+] }
+reqwest-retry = "0.7.0"
+reqwest-middleware = "0.4.0"
 sha2 = { version = "0.10.8", features = ["std"] }
 anyhow = "1"
 tracing = "0.1"
+hyper = { version = "0.14.28", features = ["server", "client", "stream"] }
 async-trait = "0.1.77"
 futures = { version = "0.3.30", features = ["thread-pool"] }
 schemars = { version = "0.8", features = ["url"] }
@@ -32,7 +39,16 @@ tokio = { version = "1.36.0", features = ["full"] }
 tower = { version = "0.4.13", features = ["full"] }
 http = "0.2"
 graphql-parser = { version = "0.5.0", package = "graphql-parser-hive-fork" }
-graphql-tools = { version = "0.4.0", features = ["graphql_parser_fork"], default-features = false }
+graphql-tools = { version = "0.4.0", features = [
+  "graphql_parser_fork",
+], default-features = false }
 lru = "^0.12.1"
 md5 = "0.7.0"
 rand = "0.8.5"
+
+[dev-dependencies]
+httpmock = "0.7.0"
+jsonschema = { version = "0.26.1", default-features = false, features = [
+  "resolve-file",
+] }
+lazy_static = "1.5.0"

packages/libraries/router/README.md

@@ -13,6 +13,8 @@ At the moment, the following are implemented:
 - [Fetching Supergraph from Hive CDN](https://the-guild.dev/graphql/hive/docs/high-availability-cdn)
 - [Sending usage information](https://the-guild.dev/graphql/hive/docs/schema-registry/usage-reporting)
   from a running Apollo Router instance to Hive
+- Persisted Operations using Hive's
+  [App Deployments](https://the-guild.dev/graphql/hive/docs/schema-registry/app-deployments)
 
 This project is constructed as a Rust project that implements Apollo-Router plugin interface.
 

packages/libraries/router/router.yaml

@@ -7,3 +7,4 @@ supergraph:
   introspection: true
 plugins:
   hive.usage: {}
+  hive.persisted_documents: {}

packages/libraries/router/src/agent.rs

@@ -23,6 +23,7 @@ pub struct Report {
 #[derive(Serialize, Debug)]
 struct OperationMapRecord {
     operation: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
     operationName: Option<String>,
     fields: Vec<String>,
 }
@@ -33,7 +34,10 @@ struct Operation {
     operationMapKey: String,
     timestamp: u64,
     execution: Execution,
+    #[serde(skip_serializing_if = "Option::is_none")]
     metadata: Option<Metadata>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    persistedDocumentHash: Option<String>,
 }
 
 #[allow(non_snake_case)]
@@ -46,12 +50,15 @@ struct Execution {
 
 #[derive(Serialize, Debug)]
 struct Metadata {
+    #[serde(skip_serializing_if = "Option::is_none")]
     client: Option<ClientInfo>,
 }
 
 #[derive(Serialize, Debug)]
 struct ClientInfo {
+    #[serde(skip_serializing_if = "Option::is_none")]
     name: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
     version: Option<String>,
 }
 
@@ -65,6 +72,7 @@ pub struct ExecutionReport {
     pub errors: usize,
     pub operation_body: String,
     pub operation_name: Option<String>,
+    pub persisted_document_hash: Option<String>,
 }
 
 #[derive(Debug, Clone)]
@@ -145,6 +153,7 @@ impl UsageAgent {
         connect_timeout: u64,
         request_timeout: u64,
         accept_invalid_certs: bool,
+        flush_interval: Duration,
     ) -> Self {
         let schema = parse_schema::<String>(&schema)
             .expect("Failed to parse schema")
@@ -173,7 +182,7 @@ impl UsageAgent {
 
         tokio::task::spawn(async move {
             loop {
-                tokio::time::sleep(Duration::from_secs(5)).await;
+                tokio::time::sleep(flush_interval).await;
 
                 let agent_ref = agent_for_interval.lock().await.clone();
                 tokio::task::spawn(async move {
@@ -197,21 +206,13 @@ impl UsageAgent {
             let operation = self
                 .processor
                 .lock()
-                .map_err(|e| {
-                    AgentError::Lock(
-                        e.to_string(),
-                    )
-                })?
+                .map_err(|e| AgentError::Lock(e.to_string()))?
                 .process(
                     &op.operation_body,
                     &self
                         .state
                         .lock()
-                        .map_err(|e| {
-                            AgentError::Lock(
-                                e.to_string(),
-                            )
-                        })?
+                        .map_err(|e| AgentError::Lock(e.to_string()))?
                         .schema,
                 );
             match operation {
@@ -230,6 +231,21 @@ impl UsageAgent {
                     match operation {
                         Some(operation) => {
                             let hash = operation.hash;
+
+                            let client_name = non_empty_string(op.client_name);
+                            let client_version = non_empty_string(op.client_version);
+
+                            let metadata: Option<Metadata> =
+                                if client_name.is_some() || client_version.is_some() {
+                                    Some(Metadata {
+                                        client: Some(ClientInfo {
+                                            name: client_name,
+                                            version: client_version,
+                                        }),
+                                    })
+                                } else {
+                                    None
+                                };
                             report.operations.push(Operation {
                                 operationMapKey: hash.clone(),
                                 timestamp: op.timestamp,
@@ -238,12 +254,8 @@ impl UsageAgent {
                                     duration: op.duration.as_nanos(),
                                     errorsTotal: op.errors,
                                 },
-                                metadata: Some(Metadata {
-                                    client: Some(ClientInfo {
-                                        name: non_empty_string(op.client_name),
-                                        version: non_empty_string(op.client_version),
-                                    }),
-                                }),
+                                persistedDocumentHash: op.persisted_document_hash,
+                                metadata,
                             });
                             if !report.map.contains_key(&hash) {
                                 report.map.insert(
@@ -272,9 +284,7 @@ impl UsageAgent {
         let size = self
             .state
             .lock()
-            .map_err(|e| {
-                AgentError::Lock(e.to_string())
-            })?
+            .map_err(|e| AgentError::Lock(e.to_string()))?
             .push(execution_report);
 
         self.flush_if_full(size)?;
@@ -285,13 +295,14 @@ impl UsageAgent {
     pub async fn send_report(&self, report: Report) -> Result<(), AgentError> {
         const DELAY_BETWEEN_TRIES: Duration = Duration::from_millis(500);
         const MAX_TRIES: u8 = 3;
-
         let mut error_message = "unexpected error".to_string();
 
         for _ in 0..MAX_TRIES {
+            // Based on https://the-guild.dev/graphql/hive/docs/specs/usage-reports#data-structure
             let resp = self
                 .client
                 .post(self.endpoint.clone())
+                .header("X-Usage-API-Version", "2")
                 .header(
                     reqwest::header::AUTHORIZATION,
                     format!("Bearer {}", self.token.clone()),

packages/libraries/router/src/graphql.rs

@@ -236,7 +236,6 @@ impl<'a> OperationVisitor<'a, SchemaCoordinatesContext> for SchemaCoordinatesVis
                     match arg_value {
                         Value::Enum(value) => {
                             let value_str = value.to_string();
-                            println!("Coordinate: {input_type_name}.{value_str}");
                             ctx.schema_coordinates
                                 .insert(format!("{input_type_name}.{value_str}").to_string());
                         }

packages/libraries/router/src/lib.rs

@@ -1,5 +1,6 @@
 mod agent;
 mod graphql;
+pub mod persisted_documents;
 pub mod registry;
 pub mod registry_logger;
 pub mod usage;

packages/libraries/router/src/main.rs

@@ -1,16 +1,25 @@
 // Specify the modules our binary should include -- https://twitter.com/YassinEldeeb7/status/1468680104243077128
 mod agent;
 mod graphql;
+mod persisted_documents;
 mod registry;
 mod registry_logger;
 mod usage;
 
+use apollo_router::register_plugin;
+use persisted_documents::PersistedDocumentsPlugin;
 use registry::HiveRegistry;
-use usage::register;
+use usage::UsagePlugin;
+
+// Register the Hive plugin
+pub fn register_plugins() {
+    register_plugin!("hive", "usage", UsagePlugin);
+    register_plugin!("hive", "persisted_documents", PersistedDocumentsPlugin);
+}
 
 fn main() {
-    // Register the usage reporting plugin
-    register();
+    // Register the Hive plugins in Apollo Router
+    register_plugins();
 
     // Initialize the Hive Registry and start the Apollo Router
     match HiveRegistry::new(None).and(apollo_router::main()) {

packages/libraries/router/src/persisted_documents.rs

@@ -0,0 +1,782 @@
+use apollo_router::graphql;
+use apollo_router::graphql::Error;
+use apollo_router::layers::ServiceBuilderExt;
+use apollo_router::plugin::Plugin;
+use apollo_router::plugin::PluginInit;
+use apollo_router::services::router;
+use apollo_router::services::router::Body;
+use apollo_router::Context;
+use core::ops::Drop;
+use std::env;
+use futures::FutureExt;
+use http::StatusCode;
+use lru::LruCache;
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+use std::num::NonZeroUsize;
+use std::ops::ControlFlow;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tower::{BoxError, ServiceBuilder, ServiceExt};
+use tracing::{debug, info, warn};
+use reqwest_middleware::{ClientBuilder, ClientWithMiddleware};
+use reqwest_retry::{RetryTransientMiddleware, policies::ExponentialBackoff};
+
+pub struct PersistedDocumentsPlugin {
+    persisted_documents_manager: Arc<PersistedDocumentsManager>,
+    configuration: Config,
+}
+
+pub(crate) static PERSISTED_DOCUMENT_HASH_KEY: &str = "hive::persisted_document_hash";
+
+#[derive(Clone, Debug, Deserialize, JsonSchema, Default)]
+pub struct Config {
+    enabled: Option<bool>,
+    /// GraphQL Hive persisted documents CDN endpoint URL.
+    endpoint: Option<String>,
+    /// GraphQL Hive persisted documents CDN access token.
+    key: Option<String>,
+    /// Whether arbitrary documents should be allowed along-side persisted documents.
+    /// default: false
+    allow_arbitrary_documents: Option<bool>,
+    /// A timeout for only the connect phase of a request to GraphQL Hive
+    /// Unit: seconds
+    /// Default: 5
+    connect_timeout: Option<u64>,
+    /// Retry count for the request to CDN request
+    /// Default: 3
+    retry_count: Option<u32>,
+    /// A timeout for the entire request to GraphQL Hive
+    /// Unit: seconds
+    /// Default: 15
+    request_timeout: Option<u64>,
+    /// Accept invalid SSL certificates
+    /// default: false
+    accept_invalid_certs: Option<bool>,
+    /// Configuration for the size of the in-memory caching of persisted documents.
+    /// Default: 1000
+    cache_size: Option<usize>,
+}
+
+impl PersistedDocumentsPlugin {
+    #[cfg(test)]
+    fn new(config: Config) -> Self {
+        PersistedDocumentsPlugin {
+            configuration: config.clone(),
+            persisted_documents_manager: Arc::new(PersistedDocumentsManager::new(&config)),
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl Plugin for PersistedDocumentsPlugin {
+    type Config = Config;
+
+    async fn new(init: PluginInit<Config>) -> Result<Self, BoxError> {
+        let mut config = init.config.clone();
+        if init.config.endpoint.is_none() {
+            if let Ok(endpoint) = env::var("HIVE_CDN_ENDPOINT") {
+                config.endpoint = Some(str::replace(&endpoint, "/supergraph", ""));
+            }
+        }
+
+        if init.config.key.is_none() {
+            if let Ok(key) = env::var("HIVE_CDN_KEY") {
+                config.key = Some(key);
+            }
+        }
+
+        Ok(PersistedDocumentsPlugin {
+            configuration: config.clone(),
+            persisted_documents_manager: Arc::new(PersistedDocumentsManager::new(&config)),
+        })
+    }
+
+    fn router_service(&self, service: router::BoxService) -> router::BoxService {
+        let enabled = self.configuration.enabled.unwrap_or(true);
+        let allow_arbitrary_documents = self
+            .configuration
+            .allow_arbitrary_documents
+            .unwrap_or(false);
+        let mgr_ref = self.persisted_documents_manager.clone();
+
+        if enabled {
+            ServiceBuilder::new()
+                .oneshot_checkpoint_async(move |req: router::Request| {
+                    let mgr = mgr_ref.clone();
+
+                    async move {
+                        let (parts, body) = req.router_request.into_parts();
+                        let bytes: hyper::body::Bytes = hyper::body::to_bytes(body)
+                            .await
+                            .map_err(PersistedDocumentsError::FailedToReadBody)?;
+
+                        let payload = PersistedDocumentsManager::extract_document_id(&bytes);
+
+                        let mut payload = match payload {
+                            Ok(payload) => payload,
+                            Err(e) => {
+                                return Ok(ControlFlow::Break(e.to_router_response(req.context)));
+                            }
+                        };
+
+                        if payload.original_req.query.is_some() {
+                            if allow_arbitrary_documents {
+                                let roll_req: router::Request = (
+                                    http::Request::<Body>::from_parts(parts, bytes.into()),
+                                    req.context,
+                                )
+                                    .into();
+
+                                return Ok(ControlFlow::Continue(roll_req));
+                            } else {
+                                return Ok(ControlFlow::Break(
+                                    PersistedDocumentsError::PersistedDocumentRequired
+                                        .to_router_response(req.context),
+                                ));
+                            }
+                        }
+
+                        if payload.document_id.is_none() {
+                            return Ok(ControlFlow::Break(
+                                PersistedDocumentsError::KeyNotFound
+                                    .to_router_response(req.context),
+                            ));
+                        }
+
+                        match payload.document_id.as_ref() {
+                            None => {
+                                return Ok(ControlFlow::Break(
+                                    PersistedDocumentsError::PersistedDocumentRequired
+                                        .to_router_response(req.context),
+                                ));
+                            }
+                            Some(document_id) => match mgr.resolve_document(document_id).await {
+                                Ok(document) => {
+                                    info!("Document found in persisted documents: {}", document);
+
+                                    if req
+                                        .context
+                                        .insert(PERSISTED_DOCUMENT_HASH_KEY, document_id.clone())
+                                        .is_err()
+                                    {
+                                        warn!("failed to extend router context with persisted document hash key");
+                                    }
+                                    
+                                    payload.original_req.query = Some(document);
+
+                                    let mut bytes: Vec<u8> = Vec::new();
+                                    serde_json::to_writer(&mut bytes, &payload).unwrap();
+
+                                    let roll_req: router::Request = (
+                                        http::Request::<Body>::from_parts(parts, bytes.into()),
+                                        req.context,
+                                    )
+                                        .into();
+
+                                    Ok(ControlFlow::Continue(roll_req))
+                                }
+                                Err(e) => {
+                                    return Ok(ControlFlow::Break(
+                                        e.to_router_response(req.context),
+                                    ));
+                                }
+                            },
+                        }
+                    }
+                    .boxed()
+                })
+                .service(service)
+                .boxed()
+        } else {
+            service
+        }
+    }
+}
+
+impl Drop for PersistedDocumentsPlugin {
+    fn drop(&mut self) {
+        debug!("PersistedDocumentsPlugin has been dropped!");
+    }
+}
+
+#[derive(Debug)]
+struct PersistedDocumentsManager {
+    agent: ClientWithMiddleware,
+    cache: Arc<Mutex<LruCache<String, String>>>,
+    config: Config,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum PersistedDocumentsError {
+    #[error("Failed to read body: {0}")]
+    FailedToReadBody(hyper::Error),
+    #[error("Failed to parse body: {0}")]
+    FailedToParseBody(serde_json::Error),
+    #[error("Persisted document not found.")]
+    DocumentNotFound,
+    #[error("Failed to locate the persisted document key in request.")]
+    KeyNotFound,
+    #[error("Failed to validate persisted document")]
+    FailedToFetchFromCDN(reqwest_middleware::Error),
+    #[error("Failed to read CDN response body")]
+    FailedToReadCDNResponse(reqwest::Error),
+    #[error("No persisted document provided, or document id cannot be resolved.")]
+    PersistedDocumentRequired,
+}
+
+impl PersistedDocumentsError {
+    fn message(&self) -> String {
+        self.to_string()
+    }
+
+    fn code(&self) -> String {
+        match self {
+            PersistedDocumentsError::FailedToReadBody(_) => "FAILED_TO_READ_BODY".into(),
+            PersistedDocumentsError::FailedToParseBody(_) => "FAILED_TO_PARSE_BODY".into(),
+            PersistedDocumentsError::DocumentNotFound => "PERSISTED_DOCUMENT_NOT_FOUND".into(),
+            PersistedDocumentsError::KeyNotFound => "PERSISTED_DOCUMENT_KEY_NOT_FOUND".into(),
+            PersistedDocumentsError::FailedToFetchFromCDN(_) => "FAILED_TO_FETCH_FROM_CDN".into(),
+            PersistedDocumentsError::FailedToReadCDNResponse(_) => "FAILED_TO_READ_CDN_RESPONSE".into(),
+            PersistedDocumentsError::PersistedDocumentRequired => {
+                "PERSISTED_DOCUMENT_REQUIRED".into()
+            }
+        }
+    }
+
+    fn to_router_response(&self, ctx: Context) -> router::Response {
+        let errors = vec![Error::builder()
+            .message(self.message())
+            .extension_code(self.code())
+            .build()];
+
+        router::Response::error_builder()
+            .errors(errors)
+            .status_code(StatusCode::OK)
+            .context(ctx)
+            .build()
+            .unwrap()
+    }
+}
+
+impl PersistedDocumentsManager {
+    fn new(config: &Config) -> Self {
+        let retry_policy = ExponentialBackoff::builder().build_with_max_retries(config.retry_count.unwrap_or(3));
+
+        let reqwest_agent = reqwest::Client::builder()
+        .danger_accept_invalid_certs(config.accept_invalid_certs.unwrap_or(false))
+        .connect_timeout(Duration::from_secs(config.connect_timeout.unwrap_or(5)))
+        .timeout(Duration::from_secs(config.request_timeout.unwrap_or(15))).build().expect("Failed to create reqwest client");
+        let agent = ClientBuilder::new(reqwest_agent).with(RetryTransientMiddleware::new_with_policy(retry_policy))
+            .build();
+
+        let cache_size = config.cache_size.unwrap_or(1000);
+        let cache = Arc::new(Mutex::new(LruCache::<String, String>::new(
+            NonZeroUsize::new(cache_size).unwrap(),
+        )));
+
+        Self {
+            agent,
+            cache,
+            config: config.clone(),
+        }
+    }
+
+    /// Extracts the document id from the request body.
+    /// In case of a parsing error, it returns the error.
+    /// This will also try to parse other GraphQL-related (see `original_req`) fields in order to 
+    /// pass it to the next layer.
+    fn extract_document_id(
+        body: &hyper::body::Bytes,
+    ) -> Result<ExpectedBodyStructure, PersistedDocumentsError> {
+        serde_json::from_slice::<ExpectedBodyStructure>(&body)
+            .map_err(PersistedDocumentsError::FailedToParseBody)
+    }
+
+    /// Resolves the document from the cache, or from the CDN
+    async fn resolve_document(&self, document_id: &str) -> Result<String, PersistedDocumentsError> {
+        let cached_record = self.cache.lock().await.get(document_id).cloned();
+
+        match cached_record {
+            Some(document) => {
+                debug!("Document {} found in cache: {}", document_id, document);
+
+                return Ok(document);
+            }
+            None => {
+                debug!(
+                    "Document {} not found in cache. Fetching from CDN",
+                    document_id
+                );
+                let cdn_document_id = str::replace(document_id, "~", "/");
+                let cdn_artifact_url = format!("{}/apps/{}", self.config.endpoint.as_ref().unwrap(), cdn_document_id);
+                info!(
+                    "Fetching document {} from CDN: {}",
+                    document_id, cdn_artifact_url
+                );
+                let cdn_response = self
+                    .agent
+                    .get(cdn_artifact_url)
+                    .header("X-Hive-CDN-Key", self.config.key.as_ref().unwrap())
+                    .send()
+                    .await;
+
+                match cdn_response {
+                    Ok(response) => {
+                        if response.status().is_success() {
+                            let document = response.text().await.map_err(PersistedDocumentsError::FailedToReadCDNResponse)?;
+                            debug!(
+                                "Document fetched from CDN: {}, storing in local cache",
+                                document
+                            );
+                            self.cache
+                                .lock()
+                                .await
+                                .put(document_id.into(), document.clone());
+
+                            return Ok(document);
+                        }
+
+                        warn!(
+                            "Document fetch from CDN failed: HTTP {}, Body: {:?}",
+                            response.status(),
+                            response.text().await.unwrap_or_else(|_| "Unavailable".to_string())
+                        );
+
+                        return Err(PersistedDocumentsError::DocumentNotFound);
+                    }
+                    Err(e) => {
+                        warn!("Failed to fetch document from CDN: {:?}", e);
+
+                        return Err(PersistedDocumentsError::FailedToFetchFromCDN(e));
+                    }
+                }
+            }
+        }
+    }
+}
+
+/// Expected body structure for the router incoming requests
+/// This is used to extract the document id and the original request as-is (see `flatten` attribute)
+#[derive(Debug, Serialize, Deserialize, Clone)]
+struct ExpectedBodyStructure {
+    /// This field is set to optional in order to prevent parsing errors
+    /// At runtime later, the plugin will double check the value.
+    #[serde(rename = "documentId")]
+    #[serde(skip_serializing)]
+    document_id: Option<String>,
+    /// The rest of the GraphQL request, flattened to keep the original structure.
+    #[serde(flatten)]
+    original_req: graphql::Request,
+}
+
+/// To test this plugin, we do the following:
+/// 1. Create the plugin instance
+/// 2. Link it to a mocked router service that reflects
+///    back the body (to validate that the plugin is working and passes the body correctly)
+/// 3. Run HTTP mock to create a mock Hive CDN server
+#[cfg(test)]
+mod hive_persisted_documents_tests {
+    use apollo_router::plugin::test::MockRouterService;
+    use futures::executor::block_on;
+    use http::Method;
+    use httpmock::{Method::GET, Mock, MockServer};
+    use serde_json::json;
+
+    use super::*;
+
+    /// Creates a regular GraphQL request with a very simple GraphQL query:
+    /// { "query": "query { __typename }" }
+    fn create_regular_request() -> router::Request {
+        let mut r = graphql::Request::default();
+
+        r.query = Some("query { __typename }".into());
+
+        router::Request::fake_builder()
+            .method(Method::POST)
+            .body(Body::from(serde_json::to_string(&r).unwrap()))
+            .header("content-type", "application/json")
+            .build()
+            .unwrap()
+    }
+
+    /// Creates a persisted document request with a document id and optional variables.
+    /// The document id is used to fetch the persisted document from the CDN.
+    /// { "documentId": "123", "variables": { ... } }
+    fn create_persisted_request(
+        document_id: &str,
+        variables: Option<serde_json::Value>,
+    ) -> router::Request {
+        let body = json!({
+            "documentId": document_id,
+            "variables": variables,
+        });
+
+        let body_str = serde_json::to_string(&body).unwrap();
+
+        router::Request::fake_builder()
+            .body(body_str)
+            .header("content-type", "application/json")
+            .build()
+            .unwrap()
+    }
+
+    /// Creates an "invalid" persisted request with an empty JSON object body.
+    fn create_invalid_req() -> router::Request {
+        router::Request::fake_builder()
+            .method(Method::POST)
+            .body(serde_json::to_string(&json!({})).unwrap())
+            .header("content-type", "application/json")
+            .build()
+            .unwrap()
+    }
+
+
+    struct PersistedDocumentsCDNMock {
+        server: MockServer,
+    }
+
+    impl PersistedDocumentsCDNMock {
+        fn new() -> Self {
+            let server = MockServer::start();
+
+            Self { server }
+        }
+
+        fn endpoint(&self) -> String {
+            self.server.url("")
+        }
+
+        /// Registers a valid artifact URL with an actual GraphQL document
+        fn add_valid(&self, document_id: &str) -> Mock {
+            let valid_artifact_url = format!("/apps/{}", str::replace(document_id, "~", "/"));
+            let document = "query { __typename }";
+            let mock = self.server.mock(|when, then| {
+                when.method(GET).path(valid_artifact_url);
+                then.status(200)
+                    .header("content-type", "text/plain")
+                    .body(document);
+            });
+
+            mock
+        }
+    }
+
+    async fn get_body(router_req: router::Request) -> String {
+        let (_parts, body) = router_req.router_request.into_parts();
+        let body = hyper::body::to_bytes(body).await.unwrap();
+        String::from_utf8(body.to_vec()).unwrap()
+    }
+
+    /// Creates a mocked router service that reflects the incoming body
+    /// back to the client.
+    /// We are using this mocked router in order to make sure that the Persisted Documents layer 
+    /// is able to resolve, fetch and pass the document to the next layer.
+    fn create_reflecting_mocked_router() -> MockRouterService {
+        let mut mocked_execution: MockRouterService = MockRouterService::new();
+
+        mocked_execution
+            .expect_call()
+            .times(1)
+            .returning(move |req| {
+                let incoming_body = block_on(get_body(req));
+                Ok(router::Response::fake_builder()
+                    .data(json!({
+                        "incomingBody": incoming_body,
+                    }))
+                    .build()
+                    .unwrap())
+            });
+
+        mocked_execution
+    }
+
+    /// Creates a mocked router service that returns a fake GraphQL response.
+    fn create_dummy_mocked_router() -> MockRouterService {
+        let mut mocked_execution = MockRouterService::new();
+
+        mocked_execution.expect_call().times(1).returning(move |_| {
+            Ok(router::Response::fake_builder()
+                .data(json!({
+                    "__typename": "Query"
+                }))
+                .build()
+                .unwrap())
+        });
+
+        mocked_execution
+    }
+
+    #[tokio::test]
+    async fn should_allow_arbitrary_when_regular_req_is_sent() {
+        let service = create_reflecting_mocked_router();
+        let service_stack = PersistedDocumentsPlugin::new(Config {
+            enabled: Some(true),
+            endpoint: Some("https://cdn.example.com".into()),
+            key: Some("123".into()),
+            allow_arbitrary_documents: Some(true),
+            ..Default::default()
+        })
+        .router_service(service.boxed());
+
+        let request = create_regular_request();
+        let mut response = service_stack.oneshot(request).await.unwrap();
+        let response_inner = response.next_response().await.unwrap().unwrap();
+
+        assert_eq!(response.response.status(), StatusCode::OK);
+        assert_eq!(
+            response_inner,
+            json!({
+                "data": {
+                    "incomingBody": "{\"query\":\"query { __typename }\"}"
+                }
+            })
+            .to_string()
+            .as_bytes()
+        );
+    }
+
+    #[tokio::test]
+    async fn should_disallow_arbitrary_when_regular_req_sent() {
+        let service_stack = PersistedDocumentsPlugin::new(Config {
+            enabled: Some(true),
+            endpoint: Some("https://cdn.example.com".into()),
+            key: Some("123".into()),
+            allow_arbitrary_documents: Some(false),
+            ..Default::default()
+        })
+        .router_service(MockRouterService::new().boxed());
+
+        let request = create_regular_request();
+        let mut response = service_stack.oneshot(request).await.unwrap();
+        let response_inner = response.next_response().await.unwrap().unwrap();
+
+        assert_eq!(response.response.status(), StatusCode::OK);
+        assert_eq!(
+            response_inner,
+            json!({
+                "errors": [
+                    {
+                        "message": "No persisted document provided, or document id cannot be resolved.",
+                        "extensions": {
+                            "code": "PERSISTED_DOCUMENT_REQUIRED"
+                        }
+                    }
+                ]
+            })
+            .to_string()
+            .as_bytes()
+        );
+    }
+
+    #[tokio::test]
+    async fn returns_not_found_error_for_missing_persisted_query() {
+        let cdn_mock = PersistedDocumentsCDNMock::new();
+        let service_stack = PersistedDocumentsPlugin::new(Config {
+            enabled: Some(true),
+            endpoint: Some(cdn_mock.endpoint()),
+            key: Some("123".into()),
+            allow_arbitrary_documents: Some(true),
+            ..Default::default()
+        })
+        .router_service(MockRouterService::new().boxed());
+
+        let request = create_persisted_request("123", None);
+        let mut response = service_stack.oneshot(request).await.unwrap();
+        let response_inner = response.next_response().await.unwrap().unwrap();
+
+        assert_eq!(response.response.status(), StatusCode::OK);
+        assert_eq!(
+            response_inner,
+            json!({
+                "errors": [
+                    {
+                        "message": "Persisted document not found.",
+                        "extensions": {
+                            "code": "PERSISTED_DOCUMENT_NOT_FOUND"
+                        }
+                    }
+                ]
+            })
+            .to_string()
+            .as_bytes()
+        );
+    }
+
+    #[tokio::test]
+    async fn returns_key_not_found_error_for_missing_input() {
+        let service_stack = PersistedDocumentsPlugin::new(Config {
+            enabled: Some(true),
+            endpoint: Some("https://cdn.example.com".into()),
+            key: Some("123".into()),
+            allow_arbitrary_documents: Some(true),
+            ..Default::default()
+        })
+        .router_service(MockRouterService::new().boxed());
+
+        let request = create_invalid_req();
+        let mut response = service_stack.oneshot(request).await.unwrap();
+        let response_inner = response.next_response().await.unwrap().unwrap();
+
+        assert_eq!(response.response.status(), StatusCode::OK);
+        assert_eq!(
+            response_inner,
+            json!({
+                "errors": [
+                    {
+                        "message": "Failed to locate the persisted document key in request.",
+                        "extensions": {
+                            "code": "PERSISTED_DOCUMENT_KEY_NOT_FOUND"
+                        }
+                    }
+                ]
+            })
+            .to_string()
+            .as_bytes()
+        );
+    }
+
+    #[tokio::test]
+    async fn rejects_req_when_cdn_not_available() {
+        let service_stack = PersistedDocumentsPlugin::new(Config {
+            enabled: Some(true),
+            endpoint: Some("https://127.0.0.1:9999".into()), // Invalid endpoint
+            key: Some("123".into()),
+            allow_arbitrary_documents: Some(false),
+            ..Default::default()
+        })
+        .router_service(MockRouterService::new().boxed());
+
+        let request = create_persisted_request("123", None);
+        let mut response = service_stack.oneshot(request).await.unwrap();
+        let response_inner = response.next_response().await.unwrap().unwrap();
+
+        assert_eq!(response.response.status(), StatusCode::OK);
+        assert_eq!(
+            response_inner,
+            json!({
+                "errors": [
+                    {
+                        "message": "Failed to validate persisted document",
+                        "extensions": {
+                            "code": "FAILED_TO_FETCH_FROM_CDN"
+                        }
+                    }
+                ]
+            })
+            .to_string()
+            .as_bytes()
+        );
+    }
+
+    #[tokio::test]
+    async fn should_return_valid_response() {
+        let cdn_mock = PersistedDocumentsCDNMock::new();
+        cdn_mock.add_valid("my-app~cacb95c69ba4684aec972777a38cd106740c6453~04bfa72dfb83b297dd8a5b6fed9bafac2b395a0f");
+        let upstream = create_dummy_mocked_router();
+        let service_stack = PersistedDocumentsPlugin::new(Config {
+            enabled: Some(true),
+            endpoint: Some(cdn_mock.endpoint()),
+            key: Some("123".into()),           
+            allow_arbitrary_documents: Some(false),
+            ..Default::default()
+        })
+        .router_service(upstream.boxed());
+
+        let request = create_persisted_request("my-app~cacb95c69ba4684aec972777a38cd106740c6453~04bfa72dfb83b297dd8a5b6fed9bafac2b395a0f", None);
+        let mut response = service_stack.oneshot(request).await.unwrap();
+        let response_inner = response.next_response().await.unwrap().unwrap();
+
+        assert_eq!(response.response.status(), StatusCode::OK);
+        assert_eq!(
+            response_inner,
+            json!({
+                "data": {
+                    "__typename": "Query"
+                }
+            })
+            .to_string()
+            .as_bytes()
+        );
+    }
+
+    #[tokio::test]
+    async fn should_passthrough_additional_req_params() {
+        let cdn_mock = PersistedDocumentsCDNMock::new();
+        cdn_mock.add_valid("my-app~cacb95c69ba4684aec972777a38cd106740c6453~04bfa72dfb83b297dd8a5b6fed9bafac2b395a0f");
+        let upstream = create_reflecting_mocked_router();
+        let service_stack = PersistedDocumentsPlugin::new(Config {
+            enabled: Some(true),
+            endpoint: Some(cdn_mock.endpoint()),
+            key: Some("123".into()),           
+            allow_arbitrary_documents: Some(false),
+            ..Default::default()
+        })
+        .router_service(upstream.boxed());
+
+        let request = create_persisted_request(
+            "my-app~cacb95c69ba4684aec972777a38cd106740c6453~04bfa72dfb83b297dd8a5b6fed9bafac2b395a0f", 
+            Some(json!({"var": "value"}))
+        );
+        let mut response = service_stack.oneshot(request).await.unwrap();
+        let response_inner = response.next_response().await.unwrap().unwrap();
+
+        assert_eq!(response.response.status(), StatusCode::OK);
+        assert_eq!(
+            response_inner,
+            "{\"data\":{\"incomingBody\":\"{\\\"query\\\":\\\"query { __typename }\\\",\\\"variables\\\":{\\\"var\\\":\\\"value\\\"}}\"}}"
+        );
+    }
+
+    #[tokio::test]
+    async fn should_use_caching_for_documents() {
+        let cdn_mock = PersistedDocumentsCDNMock::new();
+        let cdn_req_mock = cdn_mock.add_valid("my-app~cacb95c69ba4684aec972777a38cd106740c6453~04bfa72dfb83b297dd8a5b6fed9bafac2b395a0f");
+
+        let p = PersistedDocumentsPlugin::new(Config {
+            enabled: Some(true),
+            endpoint: Some(cdn_mock.endpoint()),
+            key: Some("123".into()),            allow_arbitrary_documents: Some(false),
+            ..Default::default()
+        });
+        let s1 = p.router_service(create_dummy_mocked_router().boxed());
+        let s2 = p.router_service(create_dummy_mocked_router().boxed());
+
+        // first call
+        let request = create_persisted_request("my-app~cacb95c69ba4684aec972777a38cd106740c6453~04bfa72dfb83b297dd8a5b6fed9bafac2b395a0f", None);
+
+        let mut response = s1.oneshot(request).await.unwrap();
+        let response_inner = response.next_response().await.unwrap().unwrap();
+        assert_eq!(response.response.status(), StatusCode::OK);
+        assert_eq!(
+            response_inner,
+            json!({
+                "data": {
+                    "__typename": "Query"
+                }
+            })
+            .to_string()
+            .as_bytes()
+        );
+
+        // second call
+        let request = create_persisted_request("my-app~cacb95c69ba4684aec972777a38cd106740c6453~04bfa72dfb83b297dd8a5b6fed9bafac2b395a0f", None);
+        let mut response = s2.oneshot(request).await.unwrap();
+        let response_inner = response.next_response().await.unwrap().unwrap();
+        assert_eq!(response.response.status(), StatusCode::OK);
+        assert_eq!(
+            response_inner,
+            json!({
+                "data": {
+                    "__typename": "Query"
+                }
+            })
+            .to_string()
+            .as_bytes()
+        );
+
+        // makes sure cdn called only once. If called more than once, it will fail with 404 -> leading to error (and the above assertion will fail...)
+        cdn_req_mock.assert();
+    }
+}

packages/libraries/router/src/registry.rs

@@ -93,7 +93,6 @@ impl HiveRegistry {
             None => 10,
         };
         let accept_invalid_certs = config.accept_invalid_certs.unwrap_or_else(|| false);
-
         let logger = Logger::new();
 
         // In case of an endpoint and an key being empty, we don't start the polling and skip the registry

packages/libraries/router/src/usage.rs

@@ -1,8 +1,8 @@
 use crate::agent::{AgentError, ExecutionReport, UsageAgent};
+use crate::persisted_documents::PERSISTED_DOCUMENT_HASH_KEY;
 use apollo_router::layers::ServiceBuilderExt;
 use apollo_router::plugin::Plugin;
 use apollo_router::plugin::PluginInit;
-use apollo_router::register_plugin;
 use apollo_router::services::*;
 use apollo_router::Context;
 use core::ops::Drop;
@@ -15,7 +15,7 @@ use std::collections::HashSet;
 use std::env;
 use std::sync::Arc;
 use std::sync::Mutex;
-use std::time::Instant;
+use std::time::{Duration, Instant};
 use std::time::{SystemTime, UNIX_EPOCH};
 use tower::BoxError;
 use tower::ServiceBuilder;
@@ -23,7 +23,7 @@ use tower::ServiceExt;
 
 pub(crate) static OPERATION_CONTEXT: &str = "hive::operation_context";
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Debug)]
 struct OperationContext {
     pub(crate) client_name: Option<String>,
     pub(crate) client_version: Option<String>,
@@ -41,15 +41,20 @@ struct OperationConfig {
     client_version_header: String,
 }
 
-struct UsagePlugin {
+pub struct UsagePlugin {
     config: OperationConfig,
     agent: Option<Arc<Mutex<UsageAgent>>>,
 }
 
 #[derive(Clone, Debug, Deserialize, JsonSchema)]
-struct Config {
+pub struct Config {
     /// Default: true
     enabled: Option<bool>,
+    /// Hive token, can also be set using the HIVE_TOKEN environment variable
+    registry_token: Option<String>,
+    /// Hive registry token. Set to your `/usage` endpoint if you are self-hosting.
+    /// Default: https://app.graphql-hive.com/usage
+    registry_usage_endpoint: Option<String>,
     /// Sample rate to determine sampling.
     /// 0.0 = 0% chance of being sent
     /// 1.0 = 100% chance of being sent.
@@ -73,12 +78,17 @@ struct Config {
     /// Accept invalid SSL certificates
     /// Default: false
     accept_invalid_certs: Option<bool>,
+    /// Frequency of flushing the buffer to the server
+    /// Default: 5 seconds
+    flush_interval: Option<u64>,
 }
 
 impl Default for Config {
     fn default() -> Self {
         Self {
             enabled: Some(true),
+            registry_token: None,
+            registry_usage_endpoint: Some(DEFAULT_HIVE_USAGE_ENDPOINT.into()),
             sample_rate: Some(1.0),
             exclude: None,
             client_name_header: Some(String::from("graphql-client-name")),
@@ -87,6 +97,7 @@ impl Default for Config {
             buffer_size: Some(1000),
             connect_timeout: Some(5),
             request_timeout: Some(15),
+            flush_interval: Some(5),
         }
     }
 }
@@ -158,18 +169,30 @@ impl UsagePlugin {
     }
 }
 
+static DEFAULT_HIVE_USAGE_ENDPOINT: &str = "https://app.graphql-hive.com/usage";
+
 #[async_trait::async_trait]
 impl Plugin for UsagePlugin {
     type Config = Config;
 
     async fn new(init: PluginInit<Config>) -> Result<Self, BoxError> {
-        let token =
-            env::var("HIVE_TOKEN").map_err(|_| "environment variable HIVE_TOKEN not found")?;
-        let endpoint = env::var("HIVE_ENDPOINT");
-        let endpoint = match endpoint {
-            Ok(endpoint) => endpoint,
-            Err(_) => "https://app.graphql-hive.com/usage".to_string(),
-        };
+        let token = init
+            .config
+            .registry_token
+            .clone()
+            .or_else(|| env::var("HIVE_TOKEN").ok());
+
+        if let None = token {
+            return Err("Hive token is required".into());
+        }
+
+        let endpoint = init
+            .config
+            .registry_usage_endpoint
+            .clone()
+            .unwrap_or_else(|| {
+                env::var("HIVE_ENDPOINT").unwrap_or(DEFAULT_HIVE_USAGE_ENDPOINT.to_string())
+            });
 
         let default_config = Config::default();
         let user_config = init.config;
@@ -193,6 +216,10 @@ impl Plugin for UsagePlugin {
             .request_timeout
             .or(default_config.request_timeout)
             .expect("request_timeout has no default value");
+        let flush_interval = user_config
+            .flush_interval
+            .or(default_config.flush_interval)
+            .expect("request_timeout has no default value");
 
         if enabled {
             tracing::info!("Starting GraphQL Hive Usage plugin");
@@ -217,12 +244,13 @@ impl Plugin for UsagePlugin {
             agent: match enabled {
                 true => Some(Arc::new(Mutex::new(UsageAgent::new(
                     init.supergraph_sdl.to_string(),
-                    token,
+                    token.unwrap(),
                     endpoint,
                     buffer_size,
                     connect_timeout,
                     request_timeout,
                     accept_invalid_certs,
+                    Duration::from_secs(flush_interval),
                 )))),
                 false => None,
             },
@@ -243,7 +271,7 @@ impl Plugin for UsagePlugin {
                         move |ctx: Context, fut| {
                             let agent_clone = agent.clone();
                             async move {
-                                let start = Instant::now();
+                                let start: Instant = Instant::now();
 
                                 // nested async block, bc async is unstable with closures that receive arguments
                                 let operation_context = ctx
@@ -251,6 +279,13 @@ impl Plugin for UsagePlugin {
                                     .unwrap_or_default()
                                     .unwrap();
 
+                                // Injected by the persisted document plugin, if it was activated
+                                // and discovered document id
+                                let persisted_document_hash = ctx
+                                    .get::<_, String>(PERSISTED_DOCUMENT_HASH_KEY)
+                                    .ok()
+                                    .unwrap();
+
                                 let result: supergraph::ServiceResult = fut.await;
 
                                 if operation_context.dropped {
@@ -289,6 +324,7 @@ impl Plugin for UsagePlugin {
                                                 errors: 1,
                                                 operation_body,
                                                 operation_name,
+                                                persisted_document_hash,
                                             },
                                         );
                                         Err(e)
@@ -318,6 +354,8 @@ impl Plugin for UsagePlugin {
                                                             errors: response.errors.len(),
                                                             operation_body: operation_body.clone(),
                                                             operation_name: operation_name.clone(),
+                                                            persisted_document_hash:
+                                                                persisted_document_hash.clone(),
                                                         },
                                                     );
 
@@ -342,9 +380,7 @@ impl Plugin for UsagePlugin {
 fn try_add_report(agent: Arc<Mutex<UsageAgent>>, execution_report: ExecutionReport) {
     agent
         .lock()
-        .map_err(|e| {
-AgentError::Lock(e.to_string())
-        })
+        .map_err(|e| AgentError::Lock(e.to_string()))
         .and_then(|a| a.add_report(execution_report))
         .unwrap_or_else(|e| {
             tracing::error!("Error adding report: {}", e);
@@ -358,7 +394,171 @@ impl Drop for UsagePlugin {
     }
 }
 
-// Register the hive.usage plugin
-pub fn register() {
-    register_plugin!("hive", "usage", UsagePlugin);
+#[cfg(test)]
+mod hive_usage_tests {
+    use apollo_router::{
+        plugin::{test::MockSupergraphService, Plugin, PluginInit},
+        services::supergraph,
+    };
+    use httpmock::{Method::POST, Mock, MockServer};
+    use jsonschema::Validator;
+    use serde_json::json;
+    use tower::ServiceExt;
+
+    use super::{Config, UsagePlugin};
+
+    lazy_static::lazy_static! {
+        static ref SCHEMA_VALIDATOR: Validator =
+                jsonschema::validator_for(&serde_json::from_str(&std::fs::read_to_string("../../services/usage/usage-report-v2.schema.json").expect("can't load json schema file")).expect("failed to parse json schema")).expect("failed to parse schema");
+    }
+
+    struct UsageTestHelper {
+        mocked_upstream: MockServer,
+        plugin: UsagePlugin,
+    }
+
+    impl UsageTestHelper {
+        async fn new() -> Self {
+            let server: MockServer = MockServer::start();
+            let usage_endpoint = server.url("/usage");
+            let mut config = Config::default();
+            config.enabled = Some(true);
+            config.registry_usage_endpoint = Some(usage_endpoint.to_string());
+            config.registry_token = Some("123".into());
+            config.buffer_size = Some(1);
+            config.flush_interval = Some(1);
+
+            let plugin_service = UsagePlugin::new(
+                PluginInit::fake_builder()
+                    .config(config)
+                    .supergraph_sdl("type Query { dummy: String! }".to_string().into())
+                    .build(),
+            )
+            .await
+            .expect("failed to init plugin");
+
+            UsageTestHelper {
+                mocked_upstream: server,
+                plugin: plugin_service,
+            }
+        }
+
+        fn wait_for_processing(&self) -> tokio::time::Sleep {
+            tokio::time::sleep(tokio::time::Duration::from_secs(1))
+        }
+
+        fn activate_usage_mock(&self) -> Mock {
+            self.mocked_upstream.mock(|when, then| {
+                when.method(POST).path("/usage").matches(|r| {
+                    // This mock also validates that the content of the reported usage is valid
+                    // when it comes to the JSON schema validation.
+                    // if it does not match, the request matching will fail and this will lead
+                    // to a failed assertion
+                    let body = r.body.as_ref().unwrap();
+                    let body = String::from_utf8(body.to_vec()).unwrap();
+                    let body = serde_json::from_str(&body).unwrap();
+
+                    SCHEMA_VALIDATOR.is_valid(&body)
+                });
+                then.status(200);
+            })
+        }
+
+        async fn execute_operation(&self, req: supergraph::Request) -> supergraph::Response {
+            let mut supergraph_service_mock = MockSupergraphService::new();
+
+            supergraph_service_mock
+                .expect_call()
+                .times(1)
+                .returning(move |_| {
+                    Ok(supergraph::Response::fake_builder()
+                        .data(json!({
+                            "data": { "hello": "world" },
+                        }))
+                        .build()
+                        .unwrap())
+                });
+
+            let tower_service = self
+                .plugin
+                .supergraph_service(supergraph_service_mock.boxed());
+
+            let response = tower_service
+                .oneshot(req)
+                .await
+                .expect("failed to execute operation");
+
+            response
+        }
+    }
+
+    #[tokio::test]
+    async fn should_work_correctly_for_simple_query() {
+        let instance = UsageTestHelper::new().await;
+        let req = supergraph::Request::fake_builder()
+            .query("query test { hello }")
+            .operation_name("test")
+            .build()
+            .unwrap();
+        let mock = instance.activate_usage_mock();
+
+        instance.execute_operation(req).await.next_response().await;
+
+        instance.wait_for_processing().await;
+
+        mock.assert();
+        mock.assert_hits(1);
+    }
+
+    #[tokio::test]
+    async fn without_operation_name() {
+        let instance = UsageTestHelper::new().await;
+        let req = supergraph::Request::fake_builder()
+            .query("query { hello }")
+            .build()
+            .unwrap();
+        let mock = instance.activate_usage_mock();
+
+        instance.execute_operation(req).await.next_response().await;
+
+        instance.wait_for_processing().await;
+
+        mock.assert();
+        mock.assert_hits(1);
+    }
+
+    #[tokio::test]
+    async fn multiple_operations() {
+        let instance = UsageTestHelper::new().await;
+        let req = supergraph::Request::fake_builder()
+            .query("query test { hello } query test2 { hello }")
+            .operation_name("test")
+            .build()
+            .unwrap();
+        let mock = instance.activate_usage_mock();
+
+        instance.execute_operation(req).await.next_response().await;
+
+        instance.wait_for_processing().await;
+
+        mock.assert();
+        mock.assert_hits(1);
+    }
+
+    #[tokio::test]
+    async fn invalid_query_reported() {
+        let instance = UsageTestHelper::new().await;
+        let req = supergraph::Request::fake_builder()
+            .query("query {")
+            .build()
+            .unwrap();
+        let mock = instance.activate_usage_mock();
+
+        instance.execute_operation(req).await.next_response().await;
+
+        instance.wait_for_processing().await;
+
+        mock.assert();
+        mock.assert_hits(1);
+    }
 }

packages/services/usage/usage-report-v2.schema.json

@@ -0,0 +1,136 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "title": "Report",
+  "additionalProperties": false,
+  "type": "object",
+  "properties": {
+    "size": {
+      "type": "integer"
+    },
+    "map": {
+      "type": "object",
+      "patternProperties": {
+        "^(.*)$": {
+          "title": "OperationMapRecord",
+          "additionalProperties": false,
+          "type": "object",
+          "properties": {
+            "operation": {
+              "type": "string"
+            },
+            "operationName": {
+              "type": "string"
+            },
+            "fields": {
+              "minItems": 1,
+              "type": "array",
+              "items": {
+                "type": "string"
+              }
+            }
+          },
+          "required": ["operation", "fields"]
+        }
+      }
+    },
+    "operations": {
+      "type": "array",
+      "items": {
+        "title": "RequestOperation",
+        "additionalProperties": false,
+        "type": "object",
+        "properties": {
+          "timestamp": {
+            "type": "integer"
+          },
+          "operationMapKey": {
+            "type": "string"
+          },
+          "execution": {
+            "title": "Execution",
+            "additionalProperties": false,
+            "type": "object",
+            "properties": {
+              "ok": {
+                "type": "boolean"
+              },
+              "duration": {
+                "type": "integer"
+              },
+              "errorsTotal": {
+                "type": "integer"
+              }
+            },
+            "required": ["ok", "duration", "errorsTotal"]
+          },
+          "metadata": {
+            "title": "Metadata",
+            "additionalProperties": false,
+            "type": "object",
+            "properties": {
+              "client": {
+                "title": "Client",
+                "additionalProperties": false,
+                "type": "object",
+                "properties": {
+                  "name": {
+                    "type": "string"
+                  },
+                  "version": {
+                    "type": "string"
+                  }
+                },
+                "required": ["name", "version"]
+              }
+            }
+          },
+          "persistedDocumentHash": {
+            "type": "string",
+            "title": "PersistedDocumentHash",
+            "pattern": "^[a-zA-Z0-9_-]{1,64}~[a-zA-Z0-9._-]{1,64}~([A-Za-z]|[0-9]|_){1,128}$"
+          }
+        },
+        "required": ["timestamp", "operationMapKey", "execution"]
+      }
+    },
+    "subscriptionOperations": {
+      "type": "array",
+      "items": {
+        "title": "SubscriptionOperation",
+        "additionalProperties": false,
+        "type": "object",
+        "properties": {
+          "timestamp": {
+            "type": "integer"
+          },
+          "operationMapKey": {
+            "type": "string"
+          },
+          "metadata": {
+            "title": "Metadata",
+            "additionalProperties": false,
+            "type": "object",
+            "properties": {
+              "client": {
+                "title": "Client",
+                "additionalProperties": false,
+                "type": "object",
+                "properties": {
+                  "name": {
+                    "type": "string"
+                  },
+                  "version": {
+                    "type": "string"
+                  }
+                },
+                "required": ["name", "version"]
+              }
+            }
+          }
+        },
+        "required": ["timestamp", "operationMapKey"]
+      }
+    }
+  },
+  "required": ["size", "map"]
+}

packages/web/docs/src/components/usage-reports-json-schema.tsx

@@ -0,0 +1,32 @@
+import { readFileSync } from 'node:fs';
+import type { GetStaticProps } from 'next';
+// eslint-disable-next-line import/no-extraneous-dependencies
+import { buildDynamicMDX } from 'nextra/remote';
+import { RemoteContent } from '@theguild/components';
+
+export function UsageReportsJSONSchema() {
+  return <RemoteContent />;
+}
+
+export const getStaticProps: GetStaticProps = async () => {
+  const data = [
+    '```json',
+    readFileSync('../../services/usage/usage-report-v2.schema.json', 'utf-8'),
+    '```',
+  ].join('\n');
+  const dynamicMdx = await buildDynamicMDX(data, {
+    defaultShowCopyCode: true,
+  });
+
+  return {
+    props: {
+      ...dynamicMdx,
+      __nextra_dynamic_opts: {
+        title: 'Usage Report JSON Schema / Specification',
+        frontMatter: {
+          description: 'Hive Usage Report JSON Schema / Specification',
+        },
+      },
+    },
+  };
+};

packages/web/docs/src/pages/docs/other-integrations/apollo-router.mdx

@@ -178,8 +178,10 @@ You can send usage reporting to Hive registry by enabling `hive.usage` plugin in
 
 ### Configuration
 
-- `HIVE_TOKEN` - Your 'Registry Access Token' as configured in hive For self hosted Hive:
-- `HIVE_ENDPOINT` - The usage endpoint (defaults to https://app.graphql-hive.com/usage)
+- `HIVE_TOKEN` (**required**) - Your
+  [Registry Access Token](/docs/management/targets#registry-access-tokens) with write permission.
+- `HIVE_ENDPOINT` (**optional**) - For self-hosting, you can override `/usage` endpoint (defaults to
+  `https://app.graphql-hive.com/usage`)
 
 <Tabs items={['Binary', 'Docker', 'Custom']}>
   <Tabs.Tab>
@@ -296,6 +298,71 @@ plugins:
     # accept_invalid_certs: true
 ```
 
+## Persisted Documents (App Deployments)
+
+To activate [App Deployments](/docs/schema-registry/app-deployments) in your Apollo Router, you need
+to enable the `hive.persisted_documents` plugin. The plugin uses the `HIVE_CDN_ENDPOINT` and
+`HIVE_CDN_KEY` environment variables to resolve the persisted documents IDs.
+
+```yaml filename="router.yaml"
+# ... the rest of your configuration
+plugins:
+  hive.usage: {}
+  hive.persisted_documents:
+    enabled: true
+```
+
+Once enabled, follow the
+[App Deployments guide](/docs/schema-registry/app-deployments#sending-persisted-document-requests-from-your-app)
+to send persisted document requests from your app, using the Hive CDN for resolving the document
+IDs.
+
+### Configuration
+
+The following environment variables are required:
+
+- `HIVE_CDN_ENDPOINT` - the endpoint Hive generated for you in the previous step (for example:
+  `https://cdn.graphql-hive.com/artifacts/v1/TARGET_ID/supergraph`)
+- `HIVE_CDN_KEY` - the access key to Hive CDN
+
+You may configure the plugin with the following options:
+
+```yaml filename="router.yaml"
+# ... the rest of your configuration
+plugins:
+  hive.usage: {}
+  hive.persisted_documents:
+    # Enables/disables the plugin
+    # Default: true
+    enabled: true
+    # The endpoint of the Hive CDN. You can either specify it here, or use the existing HIVE_CDN_ENDPOINT environment variable.
+    # Required
+    endpoint: 'https://cdn.graphql-hive.com/artifacts/v1/TARGET_ID/supergraph'
+    # The access key to the Hive CDN. You can either specify it here, or use the existing HIVE_CDN_KEY environment variable.
+    # Required
+    key: '...'
+    # Enables/disables the option to allow arbitrary documents to be executed.
+    # Optional
+    # Default: false
+    allow_arbitrary_documents: false
+    # HTTP connect timeout in seconds for the Hive CDN requests.
+    # Optional
+    # Default: 5
+    connect_timeout: 5
+    # HTTP request timeout in seconds for the Hive CDN requests.
+    # Optional
+    # Default: 15
+    request_timeout: 15
+    # Accepts invalid SSL certificates for the Hive CDN requests.
+    # Optional
+    # Default: false
+    accept_invalid_certs: false
+    # The maximum number of documents to hold in a LRU cache.
+    # Optional
+    # Default: 1000
+    cache_size: 1000
+```
+
 ## Additional Resources
 
 - [Get started with Apollo Federation and Hive guide](/docs/get-started/apollo-federation)

packages/web/docs/src/pages/docs/schema-registry/app-deployments.mdx

@@ -336,9 +336,27 @@ For further configuration options, please refer to the
 
 <Tabs.Tab>
 
-Using the Hive Schema Registry for persisted documents with Apollo Router is currently not
-supported. Progress of the support is tracked in
-[this GitHub issue](https://github.com/graphql-hive/platform/issues/5498).
+To use App Deployments with Apollo-Router, you can use the Hive CDN for resolving persisted
+documents.
+
+Use the [Apollo Router custom build for Hive](/docs/other-integrations/apollo-router) as an
+alternative to the official Apollo Router.
+
+Enable the Persisted Document plugin by adding the following to your `router.yaml` configuration.
+
+```yaml filename="router.yaml"
+# ... the rest of your configuration
+plugins:
+  hive.usage: {}
+  hive.persisted_documents:
+    enabled: true
+```
+
+The plugin uses the `HIVE_CDN_ENDPOINT` and `HIVE_CDN_KEY` environment variables to resolve the
+persisted documents IDs.
+
+For additional information and configuration options, please refer to the
+[Apollo-Router integration page](/docs/other-integrations/apollo-router).
 
 </Tabs.Tab>
 

packages/web/docs/src/pages/docs/specs/usage-reports.mdx

@@ -1,3 +1,7 @@
+import { UsageReportsJSONSchema } from '../../../components/usage-reports-json-schema'
+
+export { getStaticProps } from '../../../components/usage-reports-json-schema'
+
 # Usage Reporting
 
 The official JavaScript Hive Client (`@graphql-hive/core`) collects executed operations and sends
@@ -71,144 +75,7 @@ export interface Metadata {
 <details>
   <summary>JSON Schema</summary>
 
-```json
-{
-  "$schema": "http://json-schema.org/draft-07/schema#",
-  "title": "Report",
-  "additionalProperties": false,
-  "type": "object",
-  "properties": {
-    "size": {
-      "type": "integer"
-    },
-    "map": {
-      "type": "object",
-      "patternProperties": {
-        "^(.*)$": {
-          "title": "OperationMapRecord",
-          "additionalProperties": false,
-          "type": "object",
-          "properties": {
-            "operation": {
-              "type": "string"
-            },
-            "operationName": {
-              "type": "string"
-            },
-            "fields": {
-              "minItems": 1,
-              "type": "array",
-              "items": {
-                "type": "string"
-              }
-            }
-          },
-          "required": ["operation", "fields"]
-        }
-      }
-    },
-    "operations": {
-      "type": "array",
-      "items": {
-        "title": "RequestOperation",
-        "additionalProperties": false,
-        "type": "object",
-        "properties": {
-          "timestamp": {
-            "type": "integer"
-          },
-          "operationMapKey": {
-            "type": "string"
-          },
-          "execution": {
-            "title": "Execution",
-            "additionalProperties": false,
-            "type": "object",
-            "properties": {
-              "ok": {
-                "type": "boolean"
-              },
-              "duration": {
-                "type": "integer"
-              },
-              "errorsTotal": {
-                "type": "integer"
-              }
-            },
-            "required": ["ok", "duration", "errorsTotal"]
-          },
-          "metadata": {
-            "title": "Metadata",
-            "additionalProperties": false,
-            "type": "object",
-            "properties": {
-              "client": {
-                "title": "Client",
-                "additionalProperties": false,
-                "type": "object",
-                "properties": {
-                  "name": {
-                    "type": "string"
-                  },
-                  "version": {
-                    "type": "string"
-                  }
-                },
-                "required": ["name", "version"]
-              }
-            }
-          },
-          "persistedDocumentHash": {
-            "type": "string",
-            "title": "PersistedDocumentHash",
-            "pattern": "^[a-zA-Z0-9_-]{1,64}~[a-zA-Z0-9._-]{1,64}~([A-Za-z]|[0-9]|_){1,128}$"
-          }
-        },
-        "required": ["timestamp", "operationMapKey", "execution"]
-      }
-    },
-    "subscriptionOperations": {
-      "type": "array",
-      "items": {
-        "title": "SubscriptionOperation",
-        "additionalProperties": false,
-        "type": "object",
-        "properties": {
-          "timestamp": {
-            "type": "integer"
-          },
-          "operationMapKey": {
-            "type": "string"
-          },
-          "metadata": {
-            "title": "Metadata",
-            "additionalProperties": false,
-            "type": "object",
-            "properties": {
-              "client": {
-                "title": "Client",
-                "additionalProperties": false,
-                "type": "object",
-                "properties": {
-                  "name": {
-                    "type": "string"
-                  },
-                  "version": {
-                    "type": "string"
-                  }
-                },
-                "required": ["name", "version"]
-              }
-            }
-          }
-        },
-        "required": ["timestamp", "operationMapKey"]
-      }
-    }
-  },
-  "required": ["size", "map"]
-}
-```
+<UsageReportsJSONSchema />
 
 </details>
 
@@ -303,7 +170,7 @@ export interface Metadata {
 }
 ```
 
-## Curl example request
+## `curl` example request
 
 ```bash
 curl -X POST \

packages/web/docs/src/pages/product-updates/2024-11-20-hive-plugin-apollo-router-crate.mdx

@@ -3,7 +3,7 @@ title: Hive Plugin for Apollo-Router now available on Crates.io
 description:
   We've published the Hive Plugin for Apollo-Router on Crates.io. Learn how to use it in your custom
   Apollo-Router projects.
-date: 2024-11-13
+date: 2024-11-20
 authors: [dotan]
 ---
 

packages/web/docs/src/pages/product-updates/2024-12-16-apollo-router-app-deployments/index.mdx

@@ -0,0 +1,66 @@
+---
+title: Persisted Documents Support for Apollo Router
+description:
+  App Deployments feature is now available for Apollo Router. Learn how to use and how it can
+  improve your GraphQL API.
+date: 2024-12-16
+authors: [dotan]
+---
+
+import NextImage from 'next/image'
+import { Callout } from '@theguild/components'
+
+We're excited to announce that the [**App Deployments**](/docs/other-integrations/apollo-router)
+feature is now available for Apollo Router!
+
+## App Deployments / Persisted Documents
+
+App Deployments (persisted documents) are a way to group and publish your GraphQL operations as a
+single app version to the Hive Registry. This allows you to keep track of your different app
+versions, their operations usage, and performance.
+
+import pendingAppImage from '../../../../public/docs/pages/features/app-deployments/pending-app.png'
+
+<NextImage
+  alt="Pending App Deployment"
+  src={pendingAppImage}
+  className="mt-10 max-w-2xl rounded-lg drop-shadow-md"
+/>
+
+## How to use App Deployments with Apollo Router
+
+```mermaid
+flowchart LR
+    C["Apollo Router"]
+    B["Hive CDN"]
+    D["App"]
+
+    B-- "Load persisted
+ documents" -->C
+    C-- "Report persisted
+ document usage" -->B
+    D-- "Send persisted
+ document request" -->C
+```
+
+To use App Deployments with Apollo Router, follow the
+[installation instructions](/docs/other-integrations/apollo-router), and use the latest image
+version (`TODO_IMAGE_VERSION`) of the
+[custom build of Apollo Router](/docs/other-integrations/apollo-router).
+
+Push your GraphQL operations to the App Deployments store, and then configure Apollo-Router with the
+`hive.persisted_documents` plugin:
+
+```yaml filename="router.yaml"
+# ... the rest of your configuration
+plugins:
+  hive.usage: {}
+  hive.persisted_documents:
+    enabled: true
+```
+
+To learn more about App Deployments and how to use them with Apollo Router, check out the following
+resources:
+
+- [Hive custom build of Apollo-Router](/docs/other-integrations/apollo-router#app-deployments-persisted-documents)
+- [App Deployments documentation](/docs/schema-registry/app-deployments)