horizon-sdk 16.0.0

//! Horizon SDK: canonical Rust data access layer for the Horizon platform.
//!
//! # Quick Start
//!
//! ```rust,no_run
//! use horizon_sdk::HorizonSdk;
//!
//! # async fn example() -> Result<(), horizon_sdk::types::error::HorizonError> {
//! let sdk = HorizonSdk::builder()
//!     .with_postgres_url("postgresql://user:pass@localhost/horizon")
//!     .build()
//!     .await?;
//! # Ok(())
//! # }
//! ```
//!
//! # Offline builds
//!
//! The SDK uses `sqlx` compile-time-checked queries. The published crate bundles
//! a `.sqlx` query cache so it compiles without a database: when `DATABASE_URL`
//! is unset, `sqlx` uses the bundled cache. Set `SQLX_OFFLINE=true` to force
//! offline compilation even when `DATABASE_URL` is present.

#![allow(
    clippy::multiple_inherent_impl,
    reason = "builder methods are grouped by subsystem for readability"
)]

pub mod bench_support;
pub mod coverage;
pub mod horizon_api;
pub mod iceberg;
pub mod kafka;
pub mod postgres;
pub mod types;

use std::fmt;

use tracing::{info, warn};

use crate::coverage::client::CoverageClient;
use crate::horizon_api::client::HorizonApiClient;
use crate::iceberg::discovery::discover_table_metadata;
use crate::iceberg::partition::{PartitionSpec, PartitionValue};
use crate::iceberg::repository::IcebergRepository;
use crate::kafka::producer::KafkaProducer;
use crate::postgres::pool::create_pool;
use crate::postgres::repository::PostgresRepository;
use crate::types::config::{
    CoverageConfig, HorizonApiConfig, IcebergConfig, KafkaConfig, PostgresConfig, SdkConfig,
};
use crate::types::error::{ConfigError, HorizonError, IcebergError, Result};
use crate::types::model::{
    AudioSpecification, BeamgramSpecification, BearingTimeRecordSpecification, DataCollection,
    DataCollectionInterval, DataRow, DataStream, DirectionalSpectrogramSpecification, MetadataRow,
    Mission, MissionPlatform, Ontology, OntologyClass, Platform, PlatformAudioSpecification,
    PlatformBeamgramSpecification, PlatformBearingTimeRecordSpecification, PlatformInformation,
    PlatformKind, PlatformSpectrogramSpecification, SpectrogramSpecification, VariantDataRow,
};

pub use crate::types::write_backend::WriteBackend;
/// Source columns [`metadata_row_partition_values`] can resolve.
const METADATA_ROW_PARTITION_COLUMNS: &[&str] = &["data_stream_id", "datetime"];

/// Source columns [`variant_data_row_partition_values`] can resolve.
const VARIANT_DATA_ROW_PARTITION_COLUMNS: &[&str] = &[
    "data_stream_id",
    "data_type",
    "datetime",
    "specification_id",
];

/// Top-level SDK facade composing all Horizon subsystems.
///
/// Constructed via [`HorizonSdk::builder()`]. Postgres is required;
/// Kafka/Iceberg, Coverage, and the Horizon API are optional subsystems.
///
/// When Kafka and Iceberg are both configured, data row and metadata row
/// writes are dual-written: first to Postgres, then to Kafka (as Arrow IPC)
/// for eventual Iceberg ingestion via `horizon-iceberg-sink`.
pub struct HorizonSdk {
    /// Coverage HTTP client, when the service is configured.
    possible_coverage_client: Option<CoverageClient>,
    /// Iceberg repository for data row dual-write via Kafka.
    possible_data_row_iceberg: Option<IcebergRepository>,
    /// Horizon GraphQL API client, when the API is configured.
    possible_horizon_api_client: Option<HorizonApiClient>,
    /// Iceberg repository for metadata row dual-write via Kafka.
    possible_metadata_row_iceberg: Option<IcebergRepository>,
    /// Required `PostgreSQL` repository for all CRUD operations.
    postgres_repo: PostgresRepository,
}

impl HorizonSdk {
    /// Start building a new SDK instance.
    #[must_use]
    pub fn builder() -> HorizonSdkBuilder {
        HorizonSdkBuilder::default()
    }

    /// Access the coverage client, if configured.
    #[must_use]
    pub const fn coverage(&self) -> Option<&CoverageClient> {
        self.possible_coverage_client.as_ref()
    }

    /// Insert a batch of audio specifications into Postgres in a single
    /// transaction. Returns the inserted rows with database-assigned
    /// fields populated.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails. The
    /// entire batch rolls back on any per-row failure.
    pub async fn create_audio_specification_batch(
        &self,
        specs: &[AudioSpecification],
    ) -> Result<Vec<AudioSpecification>> {
        self.postgres_repo
            .insert_audio_specification_batch(specs)
            .await
    }

    /// Insert a batch of beamgram specifications into Postgres in a
    /// single transaction. Returns the inserted rows with
    /// database-assigned fields populated.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails. The
    /// entire batch rolls back on any per-row failure.
    pub async fn create_beamgram_specification_batch(
        &self,
        specs: &[BeamgramSpecification],
    ) -> Result<Vec<BeamgramSpecification>> {
        self.postgres_repo
            .insert_beamgram_specification_batch(specs)
            .await
    }

    /// Insert a batch of bearing-time-record specifications into
    /// Postgres in a single transaction. Returns the inserted rows with
    /// database-assigned fields populated.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails. The
    /// entire batch rolls back on any per-row failure.
    pub async fn create_bearing_time_record_specification_batch(
        &self,
        specs: &[BearingTimeRecordSpecification],
    ) -> Result<Vec<BearingTimeRecordSpecification>> {
        self.postgres_repo
            .insert_bearing_time_record_specification_batch(specs)
            .await
    }

    /// Create a data collection through the Horizon GraphQL API. Returns the
    /// created row with server-assigned fields populated.
    ///
    /// # Errors
    ///
    /// Returns [`ConfigError::MissingSubsystem`] if the Horizon API is not
    /// configured, or the underlying `HttpError` if the request or mutation
    /// fails.
    pub async fn create_data_collection(
        &self,
        data_collection: &DataCollection,
    ) -> Result<DataCollection> {
        self.require_horizon_api()?
            .create_data_collection(data_collection)
            .await
    }

    /// Create a data collection interval through the Horizon GraphQL API.
    /// Returns the created row with server-assigned fields populated.
    ///
    /// # Errors
    ///
    /// Returns [`ConfigError::MissingSubsystem`] if the Horizon API is not
    /// configured, or the underlying `HttpError` if the request or mutation
    /// fails.
    pub async fn create_data_collection_interval(
        &self,
        interval: &DataCollectionInterval,
    ) -> Result<DataCollectionInterval> {
        self.require_horizon_api()?
            .create_data_collection_interval(interval)
            .await
    }

    /// Insert a data row into Postgres.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the insert fails.
    #[deprecated(note = "use `create_data_row_with` and pass a `WriteBackend`")]
    pub async fn create_data_row(&self, row: &DataRow) -> Result<()> {
        self.create_data_row_with(row, WriteBackend::Postgres).await
    }

    /// Insert a batch of data rows into Postgres.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the insert fails.
    #[deprecated(note = "use `create_data_row_batch_with` and pass a `WriteBackend`")]
    pub async fn create_data_row_batch(&self, rows: &[DataRow]) -> Result<()> {
        self.create_data_row_batch_with(rows, WriteBackend::Postgres)
            .await
    }

    /// Insert a batch of data rows into Iceberg ONLY, bypassing Postgres.
    ///
    /// Batch counterpart to
    /// [`create_data_row_iceberg_only`](Self::create_data_row_iceberg_only).
    /// Returns after every message is acknowledged.
    ///
    /// # Errors
    ///
    /// Returns [`ConfigError::MissingSubsystem`] if the Iceberg write path is
    /// not configured, or the underlying Iceberg/Kafka error if serialization
    /// or delivery of any partition's message fails.
    pub async fn create_data_row_batch_iceberg_only(&self, rows: &[DataRow]) -> Result<()> {
        self.create_data_row_batch_with(rows, WriteBackend::Iceberg)
            .await
    }

    /// Insert a batch of data rows into the selected write backend(s).
    ///
    /// # Consistency
    ///
    /// Same as [`create_data_row_with`](Self::create_data_row_with): for
    /// [`WriteBackend::Both`], Postgres commits first; if the Iceberg batch
    /// publish fails, a [`HorizonError::PartialDualWrite`] is returned and
    /// the Postgres rows are NOT rolled back.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails,
    /// [`ConfigError::MissingSubsystem`] if Iceberg is required but not
    /// configured, `HorizonError::PartialDualWrite` wrapping the
    /// Iceberg/Kafka failure if the dual-write second leg fails, or the
    /// underlying Iceberg/Kafka error if only Iceberg is selected and the
    /// publish fails.
    pub async fn create_data_row_batch_with(
        &self,
        rows: &[DataRow],
        write_backend: WriteBackend,
    ) -> Result<()> {
        let variant_data_row_list: Vec<VariantDataRow> =
            rows.iter().map(VariantDataRow::from).collect();
        self.create_variant_data_row_batch_with(&variant_data_row_list, write_backend)
            .await
    }

    /// Insert a data row into Iceberg ONLY, bypassing Postgres entirely.
    ///
    /// Unlike [`create_data_row_with`](Self::create_data_row_with), this does
    /// not write to Postgres. The row is serialized to Arrow IPC and produced
    /// to Kafka for eventual ingestion by `horizon-iceberg-sink`. The call
    /// returns after Kafka acknowledgement (acks=all).
    ///
    /// Because Postgres — the system of record — is skipped, a missing Iceberg
    /// write path is treated as an error here rather than silently ignored
    /// (the dual-write methods skip Iceberg silently when it is unconfigured).
    ///
    /// # Errors
    ///
    /// Returns [`ConfigError::MissingSubsystem`] if the Iceberg write path is
    /// not configured (call [`with_iceberg`](HorizonSdkBuilder::with_iceberg)
    /// and [`with_kafka`](HorizonSdkBuilder::with_kafka) on the builder), or
    /// the underlying Iceberg/Kafka error if serialization or delivery fails.
    pub async fn create_data_row_iceberg_only(&self, row: &DataRow) -> Result<()> {
        self.create_data_row_with(row, WriteBackend::Iceberg).await
    }

    /// Insert a data row into the selected write backend(s).
    ///
    /// [`WriteBackend::Postgres`] writes only to Postgres.
    /// [`WriteBackend::Iceberg`] writes only to Kafka for Iceberg ingestion.
    /// [`WriteBackend::Both`] writes to Postgres first, then to Kafka when
    /// Iceberg is configured.
    ///
    /// # Consistency
    ///
    /// For [`WriteBackend::Both`], the Postgres insert commits before the
    /// Iceberg publish is attempted. If the Iceberg leg fails, the Postgres
    /// row is NOT rolled back — the error returns as
    /// [`HorizonError::PartialDualWrite`] so callers can distinguish
    /// "nothing happened" (Postgres leg failed) from "half applied"
    /// (Postgres committed, Iceberg did not).
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails,
    /// [`ConfigError::MissingSubsystem`] if Iceberg is required but not
    /// configured, `HorizonError::PartialDualWrite` wrapping the
    /// Iceberg/Kafka failure if the dual-write second leg fails, or the
    /// underlying Iceberg/Kafka error if only Iceberg is selected and the
    /// publish fails.
    pub async fn create_data_row_with(
        &self,
        row: &DataRow,
        write_backend: WriteBackend,
    ) -> Result<()> {
        self.create_variant_data_row_with(&VariantDataRow::from(row), write_backend)
            .await
    }

    /// Insert a batch of data streams into Postgres in a single
    /// transaction. Returns the inserted rows with database-assigned
    /// fields populated.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails. The
    /// entire batch rolls back on any per-row failure.
    pub async fn create_data_stream_batch(
        &self,
        data_streams: &[DataStream],
    ) -> Result<Vec<DataStream>> {
        self.postgres_repo
            .insert_data_stream_batch(data_streams)
            .await
    }

    /// Insert a batch of directional spectrogram specifications into
    /// Postgres in a single transaction. Returns the inserted rows with
    /// database-assigned fields populated.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails. The
    /// entire batch rolls back on any per-row failure.
    pub async fn create_directional_spectrogram_specification_batch(
        &self,
        specs: &[DirectionalSpectrogramSpecification],
    ) -> Result<Vec<DirectionalSpectrogramSpecification>> {
        self.postgres_repo
            .insert_directional_spectrogram_specification_batch(specs)
            .await
    }

    /// Insert a metadata row into Postgres.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the insert fails.
    #[deprecated(note = "use `create_metadata_row_with` and pass a `WriteBackend`")]
    pub async fn create_metadata_row(&self, row: &MetadataRow) -> Result<()> {
        self.create_metadata_row_with(row, WriteBackend::Postgres)
            .await
    }

    /// Insert a batch of metadata rows into Postgres.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the insert fails.
    #[deprecated(note = "use `create_metadata_row_batch_with` and pass a `WriteBackend`")]
    pub async fn create_metadata_row_batch(&self, rows: &[MetadataRow]) -> Result<()> {
        self.create_metadata_row_batch_with(rows, WriteBackend::Postgres)
            .await
    }

    /// Insert a batch of metadata rows into the selected write backend(s).
    ///
    /// # Consistency
    ///
    /// Same as [`create_data_row_with`](Self::create_data_row_with): for
    /// [`WriteBackend::Both`], Postgres commits first; Iceberg batch
    /// failure surfaces as [`HorizonError::PartialDualWrite`] without
    /// rolling back.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails,
    /// [`ConfigError::MissingSubsystem`] if Iceberg is required but not
    /// configured, `HorizonError::PartialDualWrite` wrapping the
    /// Iceberg/Kafka failure if the dual-write second leg fails, or the
    /// underlying Iceberg/Kafka error if only Iceberg is selected and the
    /// publish fails.
    pub async fn create_metadata_row_batch_with(
        &self,
        rows: &[MetadataRow],
        write_backend: WriteBackend,
    ) -> Result<()> {
        match write_backend {
            WriteBackend::Both => {
                self.postgres_repo.insert_metadata_row_batch(rows).await?;

                if let Some(iceberg_repo) = &self.possible_metadata_row_iceberg {
                    iceberg_repo
                        .insert_batch(rows, metadata_row_partition_values)
                        .await
                        .map_err(|source| HorizonError::PartialDualWrite {
                            source: Box::new(source),
                        })?;
                }
            }
            WriteBackend::Iceberg => {
                let iceberg_repo = self.possible_metadata_row_iceberg.as_ref().ok_or(
                    ConfigError::MissingSubsystem {
                        subsystem: "Iceberg",
                        hint: "call with_iceberg() and with_kafka() on the builder",
                    },
                )?;

                iceberg_repo
                    .insert_batch(rows, metadata_row_partition_values)
                    .await?;
            }
            WriteBackend::Postgres => {
                self.postgres_repo.insert_metadata_row_batch(rows).await?;
            }
        }

        Ok(())
    }

    /// Insert a metadata row into the selected write backend(s).
    ///
    /// # Consistency
    ///
    /// Same as [`create_data_row_with`](Self::create_data_row_with): for
    /// [`WriteBackend::Both`], Postgres commits first; Iceberg failure
    /// surfaces as [`HorizonError::PartialDualWrite`] without rolling back.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails,
    /// [`ConfigError::MissingSubsystem`] if Iceberg is required but not
    /// configured, `HorizonError::PartialDualWrite` wrapping the
    /// Iceberg/Kafka failure if the dual-write second leg fails, or the
    /// underlying Iceberg/Kafka error if only Iceberg is selected and the
    /// publish fails.
    pub async fn create_metadata_row_with(
        &self,
        row: &MetadataRow,
        write_backend: WriteBackend,
    ) -> Result<()> {
        match write_backend {
            WriteBackend::Both => {
                self.postgres_repo.insert_metadata_row(row).await?;

                if let Some(iceberg_repo) = &self.possible_metadata_row_iceberg {
                    iceberg_repo
                        .insert(row, metadata_row_partition_values)
                        .await
                        .map_err(|source| HorizonError::PartialDualWrite {
                            source: Box::new(source),
                        })?;
                }
            }
            WriteBackend::Iceberg => {
                let iceberg_repo = self.possible_metadata_row_iceberg.as_ref().ok_or(
                    ConfigError::MissingSubsystem {
                        subsystem: "Iceberg",
                        hint: "call with_iceberg() and with_kafka() on the builder",
                    },
                )?;

                iceberg_repo
                    .insert(row, metadata_row_partition_values)
                    .await?;
            }
            WriteBackend::Postgres => {
                self.postgres_repo.insert_metadata_row(row).await?;
            }
        }

        Ok(())
    }

    /// Insert a batch of missions into Postgres in a single
    /// transaction. Returns the inserted rows with database-assigned
    /// fields populated.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails. The
    /// entire batch rolls back on any per-row failure.
    pub async fn create_mission_batch(&self, missions: &[Mission]) -> Result<Vec<Mission>> {
        self.postgres_repo.insert_mission_batch(missions).await
    }

    /// Insert a batch of mission-platform links into Postgres in a
    /// single transaction. Returns the inserted rows with
    /// database-assigned fields populated.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails. The
    /// entire batch rolls back on any per-row failure.
    pub async fn create_mission_platform_batch(
        &self,
        mission_platforms: &[MissionPlatform],
    ) -> Result<Vec<MissionPlatform>> {
        self.postgres_repo
            .insert_mission_platform_batch(mission_platforms)
            .await
    }

    /// Insert a batch of ontologies into Postgres in a single
    /// transaction. Returns the inserted rows with database-assigned
    /// fields populated.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails. The
    /// entire batch rolls back on any per-row failure.
    pub async fn create_ontology_batch(&self, ontologies: &[Ontology]) -> Result<Vec<Ontology>> {
        self.postgres_repo.insert_ontology_batch(ontologies).await
    }

    /// Insert a batch of ontology classes into Postgres in a single
    /// transaction. Returns the inserted rows with database-assigned
    /// fields populated.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails. The
    /// entire batch rolls back on any per-row failure.
    pub async fn create_ontology_class_batch(
        &self,
        ontology_classes: &[OntologyClass],
    ) -> Result<Vec<OntologyClass>> {
        self.postgres_repo
            .insert_ontology_class_batch(ontology_classes)
            .await
    }

    /// Insert a batch of platform-audio specification links into
    /// Postgres in a single transaction. Returns the inserted rows with
    /// database-assigned fields populated.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails. The
    /// entire batch rolls back on any per-row failure.
    pub async fn create_platform_audio_specification_batch(
        &self,
        links: &[PlatformAudioSpecification],
    ) -> Result<Vec<PlatformAudioSpecification>> {
        self.postgres_repo
            .insert_platform_audio_specification_batch(links)
            .await
    }

    /// Insert a batch of platforms into Postgres in a single
    /// transaction. Returns the inserted rows with database-assigned
    /// fields populated.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails. The
    /// entire batch rolls back on any per-row failure.
    pub async fn create_platform_batch(&self, platforms: &[Platform]) -> Result<Vec<Platform>> {
        self.postgres_repo.insert_platform_batch(platforms).await
    }

    /// Insert a batch of platform-beamgram specification links into
    /// Postgres in a single transaction. Returns the inserted rows with
    /// database-assigned fields populated.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails. The
    /// entire batch rolls back on any per-row failure.
    pub async fn create_platform_beamgram_specification_batch(
        &self,
        links: &[PlatformBeamgramSpecification],
    ) -> Result<Vec<PlatformBeamgramSpecification>> {
        self.postgres_repo
            .insert_platform_beamgram_specification_batch(links)
            .await
    }

    /// Insert a batch of platform-bearing-time-record specification
    /// links into Postgres in a single transaction. Returns the
    /// inserted rows with database-assigned fields populated.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails. The
    /// entire batch rolls back on any per-row failure.
    pub async fn create_platform_bearing_time_record_specification_batch(
        &self,
        links: &[PlatformBearingTimeRecordSpecification],
    ) -> Result<Vec<PlatformBearingTimeRecordSpecification>> {
        self.postgres_repo
            .insert_platform_bearing_time_record_specification_batch(links)
            .await
    }

    /// Insert a batch of platform-information rows into Postgres in a
    /// single transaction. Returns the inserted rows with
    /// database-assigned fields populated.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails. The
    /// entire batch rolls back on any per-row failure.
    pub async fn create_platform_information_batch(
        &self,
        records: &[PlatformInformation],
    ) -> Result<Vec<PlatformInformation>> {
        self.postgres_repo
            .insert_platform_information_batch(records)
            .await
    }

    /// Insert a batch of platform kinds into Postgres in a single
    /// transaction. Returns the inserted rows with database-assigned
    /// fields populated.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails. The
    /// entire batch rolls back on any per-row failure.
    pub async fn create_platform_kind_batch(
        &self,
        platform_kinds: &[PlatformKind],
    ) -> Result<Vec<PlatformKind>> {
        self.postgres_repo
            .insert_platform_kind_batch(platform_kinds)
            .await
    }

    /// Insert a batch of platform-spectrogram specification links into
    /// Postgres in a single transaction. Returns the inserted rows with
    /// database-assigned fields populated.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails. The
    /// entire batch rolls back on any per-row failure.
    pub async fn create_platform_spectrogram_specification_batch(
        &self,
        links: &[PlatformSpectrogramSpecification],
    ) -> Result<Vec<PlatformSpectrogramSpecification>> {
        self.postgres_repo
            .insert_platform_spectrogram_specification_batch(links)
            .await
    }

    /// Insert a batch of spectrogram specifications into Postgres in a
    /// single transaction. Returns the inserted rows with
    /// database-assigned fields populated.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails. The
    /// entire batch rolls back on any per-row failure.
    pub async fn create_spectrogram_specification_batch(
        &self,
        specs: &[SpectrogramSpecification],
    ) -> Result<Vec<SpectrogramSpecification>> {
        self.postgres_repo
            .insert_spectrogram_specification_batch(specs)
            .await
    }

    /// Ensure a data stream exists, creating it on first sight. Reads before
    /// inserting so a later operator `query_source` change survives (unlike an
    /// upsert, which would clobber it); a lost creation race surfaces as an
    /// insert error and self-resolves on the caller's retry. A `platform_id`
    /// with no matching platform row is dropped rather than failing the FK, so
    /// device streams flow before their platform is registered.
    ///
    /// A no-op when the stream carries no id. Callers that see the same stream
    /// repeatedly should cache the ids they have ensured to skip the read.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if a read or the insert fails.
    pub async fn ensure_data_stream(&self, data_stream: &DataStream) -> Result<()> {
        let Some(id) = data_stream.id else {
            return Ok(());
        };
        if self.postgres_repo.read_data_stream(id).await?.is_some() {
            return Ok(());
        }

        let mut linked_stream = data_stream.clone();
        if let Some(platform_id) = linked_stream.platform_id
            && self
                .postgres_repo
                .read_platform(platform_id)
                .await?
                .is_none()
        {
            warn!(%platform_id, "Platform row missing; creating data stream unlinked");
            linked_stream.platform_id = None;
        }

        self.postgres_repo
            .insert_data_stream(&linked_stream)
            .await?;
        info!(%id, name = linked_stream.name.as_deref().unwrap_or(""), "Created data stream");
        Ok(())
    }

    /// Whether the Iceberg write path is enabled.
    ///
    /// The Iceberg repository owns the Kafka producer internally, so this
    /// single predicate gates the Postgres-plus-Iceberg dual-write path —
    /// there is no Kafka-only mode to branch on separately.
    #[must_use]
    pub const fn has_iceberg(&self) -> bool {
        self.possible_data_row_iceberg.is_some()
    }

    /// Access the Horizon GraphQL API client, if configured.
    #[must_use]
    pub const fn horizon_api(&self) -> Option<&HorizonApiClient> {
        self.possible_horizon_api_client.as_ref()
    }

    /// Access the `PostgreSQL` repository for direct CRUD operations.
    #[must_use]
    pub const fn postgres(&self) -> &PostgresRepository {
        &self.postgres_repo
    }

    /// The Horizon API client, or a missing-subsystem error.
    fn require_horizon_api(&self) -> Result<&HorizonApiClient> {
        self.possible_horizon_api_client.as_ref().ok_or_else(|| {
            ConfigError::MissingSubsystem {
                subsystem: "Horizon API",
                hint: "call with_horizon_api() on the builder",
            }
            .into()
        })
    }
}

impl HorizonSdk {
    /// Insert a variant data row into Postgres.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the insert fails.
    #[deprecated(note = "use `create_variant_data_row_with` and pass a `WriteBackend`")]
    pub async fn create_variant_data_row(&self, row: &VariantDataRow) -> Result<()> {
        self.create_variant_data_row_with(row, WriteBackend::Postgres)
            .await
    }

    /// Insert variant data rows into Postgres.
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the insert fails.
    #[deprecated(note = "use `create_variant_data_row_batch_with` and pass a `WriteBackend`")]
    pub async fn create_variant_data_row_batch(&self, rows: &[VariantDataRow]) -> Result<()> {
        self.create_variant_data_row_batch_with(rows, WriteBackend::Postgres)
            .await
    }

    /// Insert variant data rows into Iceberg only, bypassing Postgres.
    ///
    /// # Errors
    ///
    /// Returns [`ConfigError::MissingSubsystem`] if the Iceberg write path is
    /// not configured, or the underlying Iceberg/Kafka error if serialization
    /// or delivery fails.
    pub async fn create_variant_data_row_batch_iceberg_only(
        &self,
        rows: &[VariantDataRow],
    ) -> Result<()> {
        self.create_variant_data_row_batch_with(rows, WriteBackend::Iceberg)
            .await
    }

    /// Insert variant data rows into the selected write backend(s).
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails,
    /// [`ConfigError::MissingSubsystem`] if Iceberg is required but not
    /// configured, `HorizonError::PartialDualWrite` when the Iceberg/Kafka
    /// write fails after the Postgres insert commits, or the underlying
    /// Iceberg/Kafka error if only Iceberg is selected and the publish fails.
    pub async fn create_variant_data_row_batch_with(
        &self,
        rows: &[VariantDataRow],
        write_backend: WriteBackend,
    ) -> Result<()> {
        match write_backend {
            WriteBackend::Both => {
                self.postgres_repo
                    .insert_variant_data_row_batch(rows)
                    .await?;

                if let Some(iceberg_repo) = &self.possible_data_row_iceberg {
                    iceberg_repo
                        .insert_batch(rows, variant_data_row_partition_values)
                        .await
                        .map_err(|source| HorizonError::PartialDualWrite {
                            source: Box::new(source),
                        })?;
                }
            }
            WriteBackend::Iceberg => {
                let iceberg_repo = self.possible_data_row_iceberg.as_ref().ok_or(
                    ConfigError::MissingSubsystem {
                        subsystem: "Iceberg",
                        hint: "call with_iceberg() and with_kafka() on the builder",
                    },
                )?;

                iceberg_repo
                    .insert_batch(rows, variant_data_row_partition_values)
                    .await?;
            }
            WriteBackend::Postgres => {
                self.postgres_repo
                    .insert_variant_data_row_batch(rows)
                    .await?;
            }
        }

        Ok(())
    }

    /// Insert a variant data row into Iceberg only, bypassing Postgres.
    ///
    /// # Errors
    ///
    /// Returns [`ConfigError::MissingSubsystem`] if the Iceberg write path is
    /// not configured, or the underlying Iceberg/Kafka error if serialization
    /// or delivery fails.
    pub async fn create_variant_data_row_iceberg_only(&self, row: &VariantDataRow) -> Result<()> {
        self.create_variant_data_row_with(row, WriteBackend::Iceberg)
            .await
    }

    /// Insert a variant data row into the selected write backend(s).
    ///
    /// # Errors
    ///
    /// Returns `PostgresError::Query` if the Postgres insert fails,
    /// [`ConfigError::MissingSubsystem`] if Iceberg is required but not
    /// configured, `HorizonError::PartialDualWrite` when the Iceberg/Kafka
    /// write fails after the Postgres insert commits, or the underlying
    /// Iceberg/Kafka error if only Iceberg is selected and the publish fails.
    pub async fn create_variant_data_row_with(
        &self,
        row: &VariantDataRow,
        write_backend: WriteBackend,
    ) -> Result<()> {
        match write_backend {
            WriteBackend::Both => {
                self.postgres_repo.insert_variant_data_row(row).await?;

                if let Some(iceberg_repo) = &self.possible_data_row_iceberg {
                    iceberg_repo
                        .insert(row, variant_data_row_partition_values)
                        .await
                        .map_err(|source| HorizonError::PartialDualWrite {
                            source: Box::new(source),
                        })?;
                }
            }
            WriteBackend::Iceberg => {
                let iceberg_repo = self.possible_data_row_iceberg.as_ref().ok_or(
                    ConfigError::MissingSubsystem {
                        subsystem: "Iceberg",
                        hint: "call with_iceberg() and with_kafka() on the builder",
                    },
                )?;

                iceberg_repo
                    .insert(row, variant_data_row_partition_values)
                    .await?;
            }
            WriteBackend::Postgres => {
                self.postgres_repo.insert_variant_data_row(row).await?;
            }
        }

        Ok(())
    }
}

impl fmt::Debug for HorizonSdk {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("HorizonSdk")
            .field("coverage_enabled", &self.possible_coverage_client.is_some())
            .field(
                "horizon_api_enabled",
                &self.possible_horizon_api_client.is_some(),
            )
            .field("iceberg_enabled", &self.has_iceberg())
            .field("postgres", &self.postgres_repo)
            .finish_non_exhaustive()
    }
}

/// Builder for constructing an [`HorizonSdk`] instance.
///
/// Postgres is required — call either [`with_postgres_config`](Self::with_postgres_config)
/// or [`with_postgres_url`](Self::with_postgres_url) before [`build`](Self::build).
/// All other subsystems are optional.
#[derive(Default)]
pub struct HorizonSdkBuilder {
    /// Coverage (inference service) configuration.
    coverage_config: Option<CoverageConfig>,
    /// Horizon API configuration for GraphQL writes.
    horizon_api_config: Option<HorizonApiConfig>,
    /// Iceberg catalog configuration for schema discovery.
    iceberg_config: Option<IcebergConfig>,
    /// Kafka producer configuration for dual-write.
    kafka_config: Option<KafkaConfig>,
    /// Organization ID for multi-tenant row-level security.
    organization_id: Option<uuid::Uuid>,
    /// Structured `PostgreSQL` connection configuration.
    postgres_config: Option<PostgresConfig>,
    /// `PostgreSQL` connection URL (alternative to structured config).
    postgres_url: Option<String>,
}

impl HorizonSdkBuilder {
    /// Build the SDK, connecting to all configured subsystems.
    ///
    /// # Errors
    ///
    /// Returns `ConfigError::MissingSubsystem` if Postgres is not configured,
    /// or subsystem-specific errors if connections fail.
    pub async fn build(self) -> Result<HorizonSdk> {
        if self.iceberg_config.is_some() && self.kafka_config.is_none() {
            return Err(ConfigError::MissingSubsystem {
                subsystem: "Kafka",
                hint: "required when Iceberg is configured for the write path",
            }
            .into());
        }

        let pg_config = self.resolve_postgres_config()?;
        let pool = create_pool(&pg_config).await?;
        let org_id = self.organization_id.or(pg_config.possible_organization_id);
        let postgres_repo = PostgresRepository::new(pool, org_id);

        info!("initialized Postgres repository");

        let (possible_data_row_iceberg, possible_metadata_row_iceberg) =
            if let Some(iceberg_config) = &self.iceberg_config {
                let kafka_config = self.kafka_config.as_ref().ok_or({
                    ConfigError::MissingSubsystem {
                        subsystem: "Kafka",
                        hint: "required when Iceberg is configured for the write path",
                    }
                })?;

                let (dr_field_ids, dr_partition) =
                    discover_table_metadata(iceberg_config, &iceberg_config.data_row_table).await?;
                let dr_validated =
                    dr_partition.validate_source_columns(VARIANT_DATA_ROW_PARTITION_COLUMNS)?;
                let dr_producer = KafkaProducer::new(kafka_config, &kafka_config.data_row_topic)?;
                let dr_repo = IcebergRepository::new(dr_producer, dr_field_ids, dr_validated);

                let (mr_field_ids, mr_partition) =
                    discover_table_metadata(iceberg_config, &iceberg_config.metadata_row_table)
                        .await?;
                let mr_validated =
                    mr_partition.validate_source_columns(METADATA_ROW_PARTITION_COLUMNS)?;
                let mr_producer =
                    KafkaProducer::new(kafka_config, &kafka_config.metadata_row_topic)?;
                let mr_repo = IcebergRepository::new(mr_producer, mr_field_ids, mr_validated);

                (Some(dr_repo), Some(mr_repo))
            } else {
                (None, None)
            };

        let possible_coverage_client = self
            .coverage_config
            .as_ref()
            .map(CoverageClient::new)
            .transpose()?;

        let possible_horizon_api_client = self
            .horizon_api_config
            .as_ref()
            .map(HorizonApiClient::new)
            .transpose()?;

        info!(
            dual_write = possible_data_row_iceberg.is_some(),
            coverage = possible_coverage_client.is_some(),
            horizon_api = possible_horizon_api_client.is_some(),
            "HorizonSdk initialized"
        );

        Ok(HorizonSdk {
            possible_coverage_client,
            possible_data_row_iceberg,
            possible_horizon_api_client,
            possible_metadata_row_iceberg,
            postgres_repo,
        })
    }

    /// Resolve `PostgreSQL` configuration from structured config or URL.
    fn resolve_postgres_config(&self) -> Result<PostgresConfig> {
        if let Some(config) = &self.postgres_config {
            return Ok(config.clone());
        }

        if let Some(url) = &self.postgres_url {
            let parsed = url::Url::parse(url).map_err(|source| ConfigError::InvalidUrl {
                field: "Postgres",
                url: url.clone(),
                source,
            })?;

            let host = parsed.host_str().unwrap_or("localhost").to_owned();
            let port = parsed.port().unwrap_or(5432);
            let user = if parsed.username().is_empty() {
                "postgres".to_owned()
            } else {
                parsed.username().to_owned()
            };
            let password = parsed.password().unwrap_or_default().to_owned();
            let database = parsed.path().trim_start_matches('/').to_owned();
            let sslmode = parsed
                .query_pairs()
                .find(|(k, _)| k == "sslmode")
                .map_or_else(|| "require".to_owned(), |(_, v)| v.into_owned());

            return Ok(PostgresConfig {
                database,
                host,
                password,
                port,
                possible_organization_id: None,
                role: None,
                sslmode,
                user,
            });
        }

        Err(ConfigError::MissingSubsystem {
            subsystem: "Postgres",
            hint: "call with_postgres_config() or with_postgres_url()",
        }
        .into())
    }

    /// Load configuration from an [`SdkConfig`] struct.
    ///
    /// This sets all subsystems at once. Individual `with_*` calls
    /// after this will override the corresponding subsystem.
    #[must_use]
    pub fn with_config(mut self, config: SdkConfig) -> Self {
        self.horizon_api_config = config.horizon_api;
        self.coverage_config = config.coverage;
        self.iceberg_config = config.iceberg;
        self.kafka_config = config.kafka;
        self.postgres_config = Some(config.postgres);
        self
    }

    /// Set coverage service configuration.
    #[must_use]
    pub fn with_coverage(mut self, config: CoverageConfig) -> Self {
        self.coverage_config = Some(config);
        self
    }

    /// Set the Horizon API credentials, endpoint, and timeout used for GraphQL writes.
    #[must_use]
    pub fn with_horizon_api(mut self, config: HorizonApiConfig) -> Self {
        self.horizon_api_config = Some(config);
        self
    }

    /// Set Iceberg configuration for schema discovery.
    #[must_use]
    pub fn with_iceberg(mut self, config: IcebergConfig) -> Self {
        self.iceberg_config = Some(config);
        self
    }

    /// Set Kafka configuration for dual-write.
    #[must_use]
    pub fn with_kafka(mut self, config: KafkaConfig) -> Self {
        self.kafka_config = Some(config);
        self
    }

    /// Set the organization ID for multi-tenant row-level security.
    #[must_use]
    pub const fn with_organization_id(mut self, id: uuid::Uuid) -> Self {
        self.organization_id = Some(id);
        self
    }

    /// Set Postgres configuration from a structured config.
    #[must_use]
    pub fn with_postgres_config(mut self, config: PostgresConfig) -> Self {
        self.postgres_config = Some(config);
        self
    }

    /// Set Postgres connection from a URL string.
    ///
    /// Format: `postgresql://user:password@host:port/database?sslmode=require`.
    #[must_use]
    pub fn with_postgres_url(mut self, url: &str) -> Self {
        self.postgres_url = Some(url.to_owned());
        self
    }
}

/// Load SDK configuration from layered sources.
///
/// Priority (highest to lowest):
/// 1. Environment variables (prefixed with `HORIZON_`)
/// 2. TOML configuration file (if `config_path` is provided)
/// 3. Struct defaults
///
/// # Errors
///
/// Returns `ConfigError::Load` if the configuration cannot be loaded.
pub fn load_config(config_path: Option<&str>) -> Result<SdkConfig> {
    let mut builder = config::Config::builder();

    if let Some(path) = config_path {
        builder = builder.add_source(config::File::with_name(path));
    }

    builder = builder.add_source(
        config::Environment::with_prefix("HORIZON")
            .separator("__")
            .try_parsing(true),
    );

    builder
        .build()
        .and_then(config::Config::try_deserialize)
        .map_err(|source| ConfigError::from(source).into())
}

/// Partition values for the Iceberg `data_row` table, in `spec`'s field order
/// and with its transforms.
fn variant_data_row_partition_values(
    spec: &PartitionSpec,
    row: &VariantDataRow,
) -> Result<Vec<Option<PartitionValue>>> {
    spec.fields()
        .iter()
        .map(|field| match field.source_field_name.as_str() {
            "data_stream_id" => Ok(Some(PartitionValue::from(row.data_stream_id))),
            "data_type" => Ok(Some(PartitionValue::from(row.data_type.as_str()))),
            "datetime" => PartitionValue::from_datetime(row.datetime, &field.transform).map(Some),
            "specification_id" => Ok(Some(PartitionValue::from(row.specification_id))),
            column_name => Err(IcebergError::PartitionColumnUnmapped {
                column_name: column_name.to_owned(),
                table_name: spec.table_name().to_owned(),
            }
            .into()),
        })
        .collect()
}

/// Partition values for the Iceberg `metadata_row` table, in `spec`'s field
/// order and with its transforms.
fn metadata_row_partition_values(
    spec: &PartitionSpec,
    row: &MetadataRow,
) -> Result<Vec<Option<PartitionValue>>> {
    spec.fields()
        .iter()
        .map(|field| match field.source_field_name.as_str() {
            "data_stream_id" => Ok(Some(PartitionValue::from(row.data_stream_id))),
            "datetime" => PartitionValue::from_datetime(row.datetime, &field.transform).map(Some),
            column_name => Err(IcebergError::PartitionColumnUnmapped {
                column_name: column_name.to_owned(),
                table_name: spec.table_name().to_owned(),
            }
            .into()),
        })
        .collect()
}

#[cfg(test)]
#[allow(
    clippy::absolute_paths,
    clippy::assertions_on_result_states,
    clippy::panic_in_result_fn,
    clippy::single_call_fn,
    clippy::unwrap_used,
    reason = "unit tests use assertions, unwrap, and test helpers freely"
)]
mod tests {
    use super::*;

    use crate::iceberg::partition::PartitionFieldInfo;
    use crate::types::model::DataRowPayload;

    fn parse_postgres_url(url: &str) -> Result<PostgresConfig> {
        HorizonSdkBuilder::default()
            .with_postgres_url(url)
            .resolve_postgres_config()
    }

    /// A spec partitioning `table` on `column` alone, with the given transform.
    fn single_column_spec(table: &str, column: &str, transform: &str) -> PartitionSpec {
        PartitionSpec::new(
            table,
            vec![PartitionFieldInfo::new(column, column, transform)],
        )
    }

    /// The transform used for `column` in production; `datetime` partitions on `day`.
    fn production_transform(column: &str) -> &'static str {
        if column == "datetime" {
            "day"
        } else {
            "identity"
        }
    }

    #[test]
    fn every_supported_metadata_row_partition_column_resolves() {
        let row = MetadataRow {
            altitude: None,
            created_datetime: None,
            data_stream_id: uuid::Uuid::new_v4(),
            datetime: chrono::Utc::now(),
            heading: None,
            latitude: None,
            longitude: None,
            modified_datetime: None,
            pitch: None,
            roll: None,
            speed: None,
            speed_over_ground: None,
        };

        for column in METADATA_ROW_PARTITION_COLUMNS {
            let spec =
                single_column_spec("horizon.metadata_row", column, production_transform(column));
            let result = metadata_row_partition_values(&spec, &row);
            assert!(result.is_ok(), "column '{column}' is listed but unmapped");
            assert_eq!(result.unwrap().len(), 1);
        }
    }

    #[test]
    fn every_supported_variant_data_row_partition_column_resolves() {
        let row = VariantDataRow::from_payload(
            uuid::Uuid::new_v4(),
            chrono::Utc::now(),
            "beamgram".to_owned(),
            uuid::Uuid::new_v4(),
            DataRowPayload::Float32(1.0),
        );

        for column in VARIANT_DATA_ROW_PARTITION_COLUMNS {
            let spec = single_column_spec("horizon.data_row", column, production_transform(column));
            let result = variant_data_row_partition_values(&spec, &row);
            assert!(result.is_ok(), "column '{column}' is listed but unmapped");
            assert_eq!(result.unwrap().len(), 1);
        }
    }

    #[test]
    fn unsupported_partition_column_fails_validation() {
        let spec = single_column_spec("horizon.data_row", "vector_start_bound", "identity");
        let err = spec
            .validate_source_columns(VARIANT_DATA_ROW_PARTITION_COLUMNS)
            .unwrap_err()
            .to_string();
        assert!(
            err.contains("vector_start_bound"),
            "unexpected error: {err}"
        );
    }

    #[test]
    fn builder_organization_id_propagates() {
        let id = uuid::Uuid::new_v4();
        let builder = HorizonSdk::builder().with_organization_id(id);
        assert_eq!(builder.organization_id, Some(id));
    }

    #[test]
    fn builder_requires_postgres() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let result = rt.block_on(HorizonSdk::builder().build());
        assert!(result.is_err());
        let err = result.unwrap_err().to_string();
        assert!(err.contains("Postgres configuration is required"));
    }

    #[test]
    fn builder_with_config_sets_all_subsystems() {
        let config = SdkConfig {
            horizon_api: Some(HorizonApiConfig {
                api_key: "test_api_key".to_owned(),
                base_url: "http://localhost:3000".to_owned(),
                timeout_secs: 30,
            }),
            coverage: Some(CoverageConfig {
                api_key: None,
                base_url: "http://localhost:3000".to_owned(),
                timeout_secs: 30,
            }),
            iceberg: None,
            kafka: Some(KafkaConfig {
                bootstrap_servers: "localhost:9092".to_owned(),
                compression_type: "lz4".to_owned(),
                data_row_topic: "horizon.data_row".to_owned(),
                metadata_row_topic: "horizon.metadata_row".to_owned(),
                sasl_mechanism: "PLAIN".to_owned(),
                sasl_password: None,
                sasl_username: None,
                security_protocol: "PLAINTEXT".to_owned(),
            }),
            postgres: PostgresConfig {
                database: "horizon".to_owned(),
                host: "localhost".to_owned(),
                password: "test".to_owned(),
                port: 5432,
                possible_organization_id: None,
                role: None,
                sslmode: "disable".to_owned(),
                user: "test".to_owned(),
            },
        };
        let builder = HorizonSdk::builder().with_config(config);
        assert!(builder.postgres_config.is_some());
        assert!(builder.kafka_config.is_some());
        assert!(builder.iceberg_config.is_none());
        assert!(builder.coverage_config.is_some());
    }

    #[test]
    fn parse_postgres_url_defaults() {
        let config = parse_postgres_url("postgresql://localhost/horizon").unwrap();
        assert_eq!(config.host, "localhost");
        assert_eq!(config.port, 5432);
        assert_eq!(config.user, "postgres");
        assert_eq!(config.password, "");
        assert_eq!(config.database, "horizon");
        assert_eq!(config.sslmode, "require");
    }

    #[test]
    fn parse_postgres_url_full() {
        let config =
            parse_postgres_url("postgresql://user:pass@db.example.com:5433/mydb?sslmode=disable")
                .unwrap();
        assert_eq!(config.host, "db.example.com");
        assert_eq!(config.port, 5433);
        assert_eq!(config.user, "user");
        assert_eq!(config.password, "pass");
        assert_eq!(config.database, "mydb");
        assert_eq!(config.sslmode, "disable");
    }

    #[test]
    fn parse_postgres_url_invalid() {
        let result = parse_postgres_url("not a url");
        assert!(result.is_err());
        let err = result.unwrap_err().to_string();
        assert!(err.contains("invalid Postgres URL"));
    }
}