horizon-sdk 16.0.0

//! HTTP client wrapper for the Horizon GraphQL API.

use std::time::Duration;

use graphql_client::{GraphQLQuery, Response};
use reqwest::Client;
use reqwest::header::{AUTHORIZATION, HeaderMap, HeaderValue};
use tracing::{debug, error};
use uuid::Uuid;

use crate::horizon_api::node_id::{decode_node_id, encode_node_id};
use crate::horizon_api::operation::{
    CreateDataCollection, CreateDataCollectionInterval, create_data_collection,
    create_data_collection_interval,
};
use crate::types::config::HorizonApiConfig;
use crate::types::error::{ConfigError, HorizonError, HttpError, Result};
use crate::types::model::{DataCollection, DataCollectionInterval};

/// Path of the GraphQL endpoint relative to the Horizon API base URL.
const GRAPHQL_ENDPOINT: &str = "/api/graphql";

/// Service name reported in HTTP errors.
const SERVICE: &str = "horizon-app";

/// Async HTTP client for mutations against the Horizon GraphQL API.
///
/// Authenticates with the configured API key, so writes run under that key's
/// organization and permissions rather than a direct Postgres role.
#[derive(Debug, Clone)]
pub struct HorizonApiClient {
    /// Base URL of the Horizon API (without trailing slash).
    base_url: String,
    /// HTTP client configured with the API key and timeout.
    client: Client,
}

impl HorizonApiClient {
    /// The base URL this client connects to.
    #[must_use]
    pub fn base_url(&self) -> &str {
        &self.base_url
    }

    /// Create a data collection through the `createDataCollection` mutation.
    /// Returns the created row with server-assigned fields populated.
    ///
    /// # Errors
    ///
    /// Returns `HttpError::ServiceUnreachable` / `HttpError::Timeout` /
    /// `HttpError::RequestFailed` on transport failures,
    /// `HttpError::UnexpectedStatus` / `HttpError::DecodeFailed` on an
    /// unusable HTTP response, `HttpError::GraphQl` if the API rejects the
    /// mutation, or `HttpError::InvalidResponse` if the payload is malformed.
    pub async fn create_data_collection(
        &self,
        data_collection: &DataCollection,
    ) -> Result<DataCollection> {
        let response_data = self
            .mutate::<CreateDataCollection>(data_collection.into())
            .await?;
        response_data.try_into()
    }

    /// Create a data collection interval through the
    /// `createDataCollectionInterval` mutation. Returns the created row with
    /// server-assigned fields populated.
    ///
    /// # Errors
    ///
    /// Returns `HttpError::ServiceUnreachable` / `HttpError::Timeout` /
    /// `HttpError::RequestFailed` on transport failures,
    /// `HttpError::UnexpectedStatus` / `HttpError::DecodeFailed` on an
    /// unusable HTTP response, `HttpError::GraphQl` if the API rejects the
    /// mutation, or `HttpError::InvalidResponse` if the payload is malformed.
    pub async fn create_data_collection_interval(
        &self,
        interval: &DataCollectionInterval,
    ) -> Result<DataCollectionInterval> {
        let response_data = self
            .mutate::<CreateDataCollectionInterval>(interval.into())
            .await?;
        response_data.try_into()
    }

    /// Send a schema-checked mutation and return its response data.
    async fn mutate<Q: GraphQLQuery>(&self, variables: Q::Variables) -> Result<Q::ResponseData> {
        let url = format!("{}{GRAPHQL_ENDPOINT}", self.base_url);
        let body = Q::build_query(variables);
        let operation = body.operation_name;

        let response = self
            .client
            .post(&url)
            .json(&body)
            .send()
            .await
            .map_err(|source| {
                HttpError::from_send_error(SERVICE, &self.base_url, GRAPHQL_ENDPOINT, source)
            })?;

        let status = response.status();
        if !status.is_success() {
            error!(
                status = status.as_u16(),
                operation, "Horizon API returned error"
            );
            return Err(HttpError::UnexpectedStatus {
                service: SERVICE,
                status: status.as_u16(),
                endpoint: GRAPHQL_ENDPOINT.to_owned(),
            }
            .into());
        }

        let graphql_response: Response<Q::ResponseData> =
            response.json().await.map_err(|source| {
                error!(operation, error = %source, "failed to decode Horizon API response");
                HttpError::DecodeFailed {
                    endpoint: GRAPHQL_ENDPOINT.to_owned(),
                    source,
                }
            })?;

        response_data(operation, graphql_response)
    }

    /// Create a new Horizon API client from SDK configuration.
    ///
    /// # Errors
    ///
    /// Returns `ConfigError::InvalidUrl` if the base URL is malformed, or
    /// `HttpError::ClientBuild` if the HTTP client cannot be created (an API key
    /// that is not a valid header value counts as a build failure).
    pub fn new(config: &HorizonApiConfig) -> Result<Self> {
        let base_url = config.base_url.trim_end_matches('/').to_owned();
        url::Url::parse(&base_url).map_err(|source| ConfigError::InvalidUrl {
            field: "Horizon API base",
            url: base_url.clone(),
            source,
        })?;

        let mut authorization = HeaderValue::from_str(&format!("Bearer {}", config.api_key))
            .map_err(|_invalid_header_value| HttpError::InvalidApiKey)?;
        authorization.set_sensitive(true);
        let mut default_headers = HeaderMap::new();
        default_headers.insert(AUTHORIZATION, authorization);

        let client = Client::builder()
            .default_headers(default_headers)
            .timeout(Duration::from_secs(config.timeout_secs))
            .build()
            .map_err(|source| HttpError::ClientBuild { source })?;

        debug!(base_url = base_url.as_str(), "initialized HorizonApiClient");

        Ok(Self { base_url, client })
    }
}

impl From<&DataCollection> for create_data_collection::Variables {
    fn from(data_collection: &DataCollection) -> Self {
        Self {
            input: create_data_collection::CreateDataCollectionInput {
                client_mutation_id: None,
                data_collection: create_data_collection::DataCollectionInput {
                    collected_secs: data_collection.collected_secs.map(i64::from),
                    collected_size: data_collection.collected_size.map(|size| size.to_string()),
                    ingest_pipeline_version: data_collection.ingest_pipeline_version.clone(),
                    ingested_datetime: data_collection.ingested_datetime,
                    name: data_collection.name.clone(),
                    platform: encode_node_id("Platform", data_collection.platform_id),
                    platform_software_version: data_collection.platform_software_version.clone(),
                    row_id: data_collection.id,
                    s3_prefix: data_collection.s3_prefix.clone(),
                    status: data_collection.status,
                    upload_complete_datetime: data_collection.upload_complete_datetime,
                    xml_sensor_config: data_collection.xml_sensor_config.clone(),
                },
            },
        }
    }
}

impl From<&DataCollectionInterval> for create_data_collection_interval::Variables {
    fn from(interval: &DataCollectionInterval) -> Self {
        Self {
            input: create_data_collection_interval::CreateDataCollectionIntervalInput {
                client_mutation_id: None,
                data_collection_interval:
                    create_data_collection_interval::DataCollectionIntervalInput {
                        collected_secs: interval.collected_secs.map(i64::from),
                        collected_size: interval.collected_size.map(|size| size.to_string()),
                        data_collection: encode_node_id(
                            "DataCollection",
                            interval.data_collection_id,
                        ),
                        data_stream: encode_node_id("DataStream", interval.data_stream_id),
                        end_datetime: interval.end_datetime,
                        group_hash_id: interval.group_hash_id.clone(),
                        has_classification_override: interval.has_classification_override,
                        has_detection_override: interval.has_detection_override,
                        has_track_override: interval.has_track_override,
                        processing_end_datetime: interval.processing_end_datetime,
                        processing_start_datetime: interval.processing_start_datetime,
                        row_id: interval.id,
                        start_datetime: interval.start_datetime,
                        status: interval.status,
                    },
            },
        }
    }
}

impl TryFrom<create_data_collection::ResponseData> for DataCollection {
    type Error = HorizonError;

    fn try_from(response_data: create_data_collection::ResponseData) -> Result<Self> {
        const OPERATION: &str = create_data_collection::OPERATION_NAME;

        let node = response_data
            .create_data_collection
            .and_then(|payload| payload.data_collection)
            .ok_or_else(|| missing_payload(OPERATION))?;

        Ok(Self {
            collected_secs: parse_int(OPERATION, node.collected_secs)?,
            collected_size: parse_big_int(OPERATION, node.collected_size)?,
            created_datetime: Some(node.created_datetime),
            id: Some(parse_node_id(OPERATION, "DataCollection", &node.id)?),
            ingest_pipeline_version: node.ingest_pipeline_version,
            ingested_datetime: node.ingested_datetime,
            modified_datetime: Some(node.modified_datetime),
            name: node.name,
            organization_id: node
                .organization
                .map(|organization| parse_node_id(OPERATION, "Organization", &organization.id))
                .transpose()?,
            platform_id: parse_node_id(
                OPERATION,
                "Platform",
                &node.platform.ok_or_else(|| missing_payload(OPERATION))?.id,
            )?,
            platform_software_version: node.platform_software_version,
            s3_prefix: node.s3_prefix,
            status: Some(node.status),
            upload_complete_datetime: node.upload_complete_datetime,
            xml_sensor_config: node.xml_sensor_config,
        })
    }
}

impl TryFrom<create_data_collection_interval::ResponseData> for DataCollectionInterval {
    type Error = HorizonError;

    fn try_from(response_data: create_data_collection_interval::ResponseData) -> Result<Self> {
        const OPERATION: &str = create_data_collection_interval::OPERATION_NAME;

        let node = response_data
            .create_data_collection_interval
            .and_then(|payload| payload.data_collection_interval)
            .ok_or_else(|| missing_payload(OPERATION))?;

        Ok(Self {
            collected_secs: parse_int(OPERATION, node.collected_secs)?,
            collected_size: parse_big_int(OPERATION, node.collected_size)?,
            created_datetime: Some(node.created_datetime),
            data_collection_id: parse_node_id(
                OPERATION,
                "DataCollection",
                &node
                    .data_collection
                    .ok_or_else(|| missing_payload(OPERATION))?
                    .id,
            )?,
            data_stream_id: parse_node_id(
                OPERATION,
                "DataStream",
                &node
                    .data_stream
                    .ok_or_else(|| missing_payload(OPERATION))?
                    .id,
            )?,
            end_datetime: node.end_datetime,
            group_hash_id: node.group_hash_id,
            has_classification_override: node.has_classification_override,
            has_detection_override: node.has_detection_override,
            has_track_override: node.has_track_override,
            id: Some(parse_node_id(
                OPERATION,
                "DataCollectionInterval",
                &node.id,
            )?),
            modified_datetime: Some(node.modified_datetime),
            organization_id: node
                .organization
                .map(|organization| parse_node_id(OPERATION, "Organization", &organization.id))
                .transpose()?,
            processing_end_datetime: node.processing_end_datetime,
            processing_start_datetime: node.processing_start_datetime,
            start_datetime: node.start_datetime,
            status: Some(node.status),
        })
    }
}

/// An error for a mutation payload or relation the API returned as null.
fn missing_payload(operation: &'static str) -> HorizonError {
    HttpError::InvalidResponse {
        operation,
        reason: "mutation payload or a selected relation is null".to_owned(),
    }
    .into()
}

/// Parse a `BigInt` scalar, which the API serializes as a string.
fn parse_big_int(operation: &'static str, value: Option<String>) -> Result<Option<i64>> {
    value
        .map(|text| {
            text.parse().map_err(|_parse_error| {
                HttpError::InvalidResponse {
                    operation,
                    reason: format!("BigInt '{text}' is not an i64"),
                }
                .into()
            })
        })
        .transpose()
}

/// Narrow a GraphQL `Int` to the `integer` column width.
fn parse_int(operation: &'static str, value: Option<i64>) -> Result<Option<i32>> {
    value
        .map(|number| {
            i32::try_from(number).map_err(|_overflow| {
                HttpError::InvalidResponse {
                    operation,
                    reason: format!("Int {number} is not an i32"),
                }
                .into()
            })
        })
        .transpose()
}

/// Decode a Relay global node id of `type_name` into its UUID.
fn parse_node_id(operation: &'static str, type_name: &str, node_id: &str) -> Result<Uuid> {
    decode_node_id(type_name, node_id).ok_or_else(|| {
        HttpError::InvalidResponse {
            operation,
            reason: format!("'{node_id}' is not a {type_name} node id"),
        }
        .into()
    })
}

/// Unwrap a GraphQL response, turning reported errors or missing data into
/// [`HttpError`]s.
#[allow(
    clippy::single_call_fn,
    reason = "split out of `mutate` so response handling is unit-testable without HTTP"
)]
fn response_data<T>(operation: &'static str, graphql_response: Response<T>) -> Result<T> {
    if let Some(error_list) = graphql_response.errors
        && !error_list.is_empty()
    {
        let message_list: Vec<String> = error_list.into_iter().map(|error| error.message).collect();
        error!(operation, ?message_list, "Horizon API rejected mutation");
        return Err(HttpError::GraphQl {
            operation,
            message_list,
        }
        .into());
    }

    graphql_response
        .data
        .ok_or_else(|| missing_payload(operation))
}

#[cfg(test)]
#[allow(
    clippy::default_numeric_fallback,
    clippy::indexing_slicing,
    clippy::unwrap_used,
    reason = "unit tests use numeric literals, JSON indexing, and unwrap for brevity"
)]
mod tests {
    use super::*;
    use crate::types::model::DataCollectionStatus;

    fn config(api_key: &str, base_url: &str) -> HorizonApiConfig {
        HorizonApiConfig {
            api_key: api_key.to_owned(),
            base_url: base_url.to_owned(),
            timeout_secs: 30,
        }
    }

    #[test]
    fn debug_impl_hides_api_key() {
        let client =
            HorizonApiClient::new(&config("secret-token", "http://localhost:3000")).unwrap();
        let debug = format!("{client:?}");
        assert!(debug.contains("http://localhost:3000"));
        assert!(!debug.contains("secret-token"));
    }

    #[test]
    fn horizon_api_client_rejects_api_key_with_invalid_header_characters() {
        let error =
            HorizonApiClient::new(&config("bad\nkey", "http://localhost:3000")).unwrap_err();
        assert!(matches!(
            error,
            HorizonError::Http(HttpError::InvalidApiKey)
        ));
    }

    #[test]
    fn horizon_api_client_trims_trailing_slash() {
        let client = HorizonApiClient::new(&config("key", "http://localhost:3000/")).unwrap();
        assert_eq!(client.base_url(), "http://localhost:3000");
    }

    #[test]
    fn parse_big_int_rejects_non_integer() {
        let error = parse_big_int("Test", Some("1.5".to_owned())).unwrap_err();
        assert!(matches!(
            error,
            HorizonError::Http(HttpError::InvalidResponse { .. })
        ));
    }

    #[test]
    fn create_data_collection_request_omits_unset_fields() {
        let platform_id = Uuid::new_v4();
        let data_collection = DataCollection {
            collected_secs: Some(90),
            collected_size: Some(5_000_000_000),
            created_datetime: None,
            id: None,
            ingest_pipeline_version: None,
            ingested_datetime: None,
            modified_datetime: None,
            name: "Sortie".to_owned(),
            organization_id: None,
            platform_id,
            platform_software_version: None,
            s3_prefix: None,
            status: None,
            upload_complete_datetime: None,
            xml_sensor_config: None,
        };

        let body = CreateDataCollection::build_query((&data_collection).into());
        let json = serde_json::to_value(&body).unwrap();

        assert_eq!(json["operationName"], "CreateDataCollection");
        assert_eq!(
            json["variables"]["input"]["dataCollection"],
            serde_json::json!({
                "collectedSecs": 90,
                "collectedSize": "5000000000",
                "name": "Sortie",
                "platform": encode_node_id("Platform", platform_id),
            })
        );
    }

    #[test]
    fn create_data_collection_interval_response_converts_to_model() {
        let data_collection_id = Uuid::new_v4();
        let data_stream_id = Uuid::new_v4();
        let id = Uuid::new_v4();
        let organization_id = Uuid::new_v4();
        let response: Response<create_data_collection_interval::ResponseData> =
            serde_json::from_value(serde_json::json!({
                "data": {
                    "createDataCollectionInterval": {
                        "dataCollectionInterval": {
                            "collectedSecs": 60,
                            "collectedSize": "1024",
                            "createdDatetime": "2026-09-25T12:00:00Z",
                            "dataCollection": { "id": encode_node_id("DataCollection", data_collection_id) },
                            "dataStream": { "id": encode_node_id("DataStream", data_stream_id) },
                            "endDatetime": "2026-09-25T11:01:00Z",
                            "groupHashId": null,
                            "hasClassificationOverride": null,
                            "hasDetectionOverride": true,
                            "hasTrackOverride": null,
                            "id": encode_node_id("DataCollectionInterval", id),
                            "modifiedDatetime": "2026-09-25T12:00:00Z",
                            "organization": { "id": encode_node_id("Organization", organization_id) },
                            "processingEndDatetime": null,
                            "processingStartDatetime": null,
                            "startDatetime": "2026-09-25T11:00:00Z",
                            "status": "ingested"
                        }
                    }
                }
            }))
            .unwrap();

        let interval =
            DataCollectionInterval::try_from(response_data("Test", response).unwrap()).unwrap();

        assert_eq!(interval.collected_secs, Some(60));
        assert_eq!(interval.collected_size, Some(1024));
        assert_eq!(interval.data_collection_id, data_collection_id);
        assert_eq!(interval.data_stream_id, data_stream_id);
        assert_eq!(interval.has_detection_override, Some(true));
        assert_eq!(interval.id, Some(id));
        assert_eq!(interval.organization_id, Some(organization_id));
        assert_eq!(interval.status, Some(DataCollectionStatus::Ingested));
    }

    #[test]
    fn response_data_surfaces_graphql_errors() {
        let response: Response<create_data_collection::ResponseData> =
            serde_json::from_value(serde_json::json!({
                "data": { "createDataCollection": null },
                "errors": [{ "message": "permission denied for table data_collection" }]
            }))
            .unwrap();

        let error = response_data("CreateDataCollection", response).unwrap_err();

        assert!(matches!(
            error,
            HorizonError::Http(HttpError::GraphQl { message_list, .. })
                if message_list == ["permission denied for table data_collection"]
        ));
    }
}