//! Error types for the Horizon SDK.

use std::fmt;
use std::result::Result as StdResult;

use config::ConfigError as ConfigLoadError;
use rdkafka::error::KafkaError as RdKafkaError;
use uuid::Uuid;

/// Crate-level Result alias using [`HorizonError`].
pub type Result<T> = StdResult<T, HorizonError>;

/// Top-level error type for all Horizon SDK operations.
///
/// Each variant corresponds to a subsystem that can produce errors.
#[non_exhaustive]
#[derive(Debug, thiserror::Error)]
pub enum HorizonError {
    /// An Arrow IPC serialization or deserialization error.
    #[error(transparent)]
    ArrowIpc(#[from] arrow_schema::ArrowError),

    /// An error from the SDK configuration subsystem.
    #[error(transparent)]
    Config(#[from] ConfigError),

    /// An error from the HTTP subsystem.
    #[error(transparent)]
    Http(#[from] HttpError),

    /// An error from the Iceberg subsystem.
    #[error(transparent)]
    Iceberg(#[from] IcebergError),

    /// An error from the Kafka subsystem.
    #[error(transparent)]
    Kafka(#[from] KafkaError),

    /// A dual-write operation committed to Postgres but failed on the
    /// Iceberg/Kafka leg, leaving the row persisted only in Postgres.
    ///
    /// Raised by `create_data_row`, `create_data_row_batch`,
    /// `create_metadata_row`, and `create_metadata_row_batch` when the
    /// Postgres insert succeeds and the subsequent Iceberg/Kafka publish
    /// fails. The Postgres row is NOT rolled back and the Iceberg write
    /// is NOT retried — callers are responsible for reconciliation.
    ///
    /// Typical responses: log and continue (accept eventual consistency
    /// drift), enqueue a compensating delete, or reject the operation
    /// and require the caller to retry the whole dual-write.
    #[error(
        "dual-write partial failure: Postgres committed but Iceberg/Kafka leg failed: {source}"
    )]
    PartialDualWrite {
        #[source]
        source: Box<Self>,
    },

    /// A `PostgreSQL` operation failed.
    #[error(transparent)]
    Postgres(#[from] PostgresError),
}

/// Errors produced by SDK configuration (builder validation, URL parsing,
/// and loading from environment or file sources).
///
/// These are distinct from the errors a subsystem produces at runtime —
/// `ConfigError` is raised before the subsystem is initialized.
#[non_exhaustive]
#[derive(Debug, thiserror::Error)]
pub enum ConfigError {
    /// A `PostgreSQL` role name failed validation.
    ///
    /// Raised when the configured role name does not satisfy the required
    /// character policy (lowercase alphanumeric plus underscore, starting
    /// with a letter or underscore).
    #[error("invalid role name '{role}': {reason}")]
    InvalidRoleName { role: String, reason: &'static str },

    /// A URL string in the configuration could not be parsed.
    ///
    /// Raised when the SDK attempts to parse a connection URL (Postgres
    /// connection string, coverage service base URL, etc.) that is
    /// malformed.
    #[error("invalid {field} URL '{url}': {source}")]
    InvalidUrl {
        field: &'static str,
        url: String,
        #[source]
        source: url::ParseError,
    },

    /// Failed to load configuration from environment or file sources.
    ///
    /// Wraps errors from the `config` crate's layered deserialization.
    #[error(transparent)]
    Load(#[from] ConfigLoadError),

    /// A required subsystem configuration block was not provided.
    ///
    /// Raised when a builder method needs a subsystem that has not been
    /// configured — e.g. calling `build` without calling
    /// `with_postgres_config` or `with_postgres_url` first.
    #[error("{subsystem} configuration is required: {hint}")]
    MissingSubsystem {
        subsystem: &'static str,
        hint: &'static str,
    },
}

/// Errors produced by the HTTP subsystem (reqwest-based requests to the
/// external `BentoML` coverage service).
///
/// Each variant carries the structured context needed for the FFI layer to
/// build a Python-equivalent exception without parsing strings.
#[non_exhaustive]
#[derive(Debug, thiserror::Error)]
pub enum HttpError {
    /// Failed to build the underlying `reqwest` HTTP client.
    ///
    /// Raised when the client cannot be instantiated — usually because the
    /// TLS backend rejects the configuration or the timeout value is
    /// invalid.
    #[error("failed to create HTTP client: {source}")]
    ClientBuild {
        #[source]
        source: reqwest::Error,
    },

    /// Failed to decode an HTTP response body into JSON.
    ///
    /// Raised when the downstream service returned a successful status
    /// but the payload could not be parsed as valid JSON.
    #[error("failed to decode response from {endpoint}: {source}")]
    DecodeFailed {
        endpoint: String,
        #[source]
        source: reqwest::Error,
    },

    /// A generic HTTP request failure that is not a connect failure or
    /// timeout.
    ///
    /// Raised when the request was attempted but failed before receiving
    /// a status code — e.g. TLS handshake failure or protocol error.
    #[error("{service} HTTP request to {endpoint} failed: {source}")]
    RequestFailed {
        service: &'static str,
        endpoint: String,
        #[source]
        source: reqwest::Error,
    },

    /// The downstream HTTP service could not be reached at all.
    ///
    /// Raised when `reqwest::Error::is_connect()` is true — usually
    /// because the service is not listening on the configured URL.
    #[error("{service} at {url} is unreachable: {source}")]
    ServiceUnreachable {
        service: &'static str,
        url: String,
        #[source]
        source: reqwest::Error,
    },

    /// The downstream HTTP service timed out on a request.
    ///
    /// Raised when `reqwest::Error::is_timeout()` is true — usually
    /// because the service is overloaded or the configured timeout is
    /// too short for the requested operation.
    #[error("{service} at {url} timed out on {endpoint}: {source}")]
    Timeout {
        service: &'static str,
        url: String,
        endpoint: String,
        #[source]
        source: reqwest::Error,
    },

    /// The downstream HTTP service returned a non-success status code.
    ///
    /// Raised after a response was received but the HTTP status was not
    /// in the 2xx range. No `reqwest::Error` is captured because the
    /// request technically succeeded at the transport layer.
    #[error("{service} returned {status} for {endpoint}")]
    UnexpectedStatus {
        service: &'static str,
        status: u16,
        endpoint: String,
    },
}

/// Errors produced by the Iceberg subsystem (schema discovery, partition
/// computation, and Arrow IPC serialization for the Kafka write path).
///
/// Each variant carries the structured context needed for the FFI layer to
/// build a Python-equivalent exception without parsing strings.
#[non_exhaustive]
#[derive(Debug, thiserror::Error)]
pub enum IcebergError {
    /// An Arrow column type did not match what the Iceberg schema requires.
    ///
    /// Raised during Arrow IPC serialization when a column's runtime
    /// `DataType` is incompatible with the expected Iceberg field type.
    #[error("column {column_name} has type {actual} but expected {expected}")]
    ArrowTypeMismatch {
        column_name: String,
        expected: String,
        actual: String,
    },

    /// Failed to connect to the Iceberg REST catalog.
    ///
    /// Raised by `build_catalog` when the REST catalog client cannot be
    /// constructed — usually because the catalog URI is unreachable, the
    /// warehouse property is malformed, or authentication fails.
    #[error("failed to connect to Iceberg catalog at {catalog_uri}: {source}")]
    CatalogConnection {
        catalog_uri: String,
        #[source]
        source: Box<iceberg::Error>,
    },

    /// A column was not found in the Iceberg field ID mapping.
    ///
    /// Raised by `FieldIdMap::get_field_id` when the SDK looks up a column
    /// that the catalog-discovered schema does not contain.
    #[error("column {column_name} not found in table {table_name}")]
    ColumnNotFound {
        column_name: String,
        table_name: String,
    },

    /// An empty batch list was passed to an operation requiring at least
    /// one batch.
    ///
    /// Raised when the SDK is asked to concatenate or serialize an empty
    /// slice of `RecordBatch` values.
    #[error("cannot {operation} an empty batch list")]
    EmptyBatchList { operation: String },

    /// An Iceberg table name could not be parsed into a `TableIdent`.
    ///
    /// Raised by `parse_table_ident` when the dot-separated identifier is
    /// malformed (empty, contains invalid characters, or has the wrong
    /// number of segments).
    #[error("invalid Iceberg table name '{table_name}': {source}")]
    InvalidTableName {
        table_name: String,
        #[source]
        source: Box<iceberg::Error>,
    },

    /// Required columns are missing from an Iceberg table schema.
    ///
    /// Raised by `FieldIdMap::validate_columns` when the SDK expects columns
    /// that the catalog-discovered schema does not contain.
    #[error("columns {missing_columns:?} required by SDK but not found in table '{table_name}'")]
    MissingColumns {
        missing_columns: Vec<String>,
        table_name: String,
    },

    /// A partition source field id was not found in the catalog schema.
    ///
    /// Raised during Iceberg schema discovery when a partition spec
    /// references a field id that the schema does not declare.
    #[error("partition source field id {field_id} not found in schema for table {table_name}")]
    PartitionFieldNotFound { field_id: i32, table_name: String },

    /// A partition value count mismatch.
    ///
    /// Raised by `PartitionSpec::compute_key` when the caller passes a
    /// number of partition values that does not match the spec.
    #[error("expected {expected} partition values but got {actual}")]
    PartitionValueCountMismatch { expected: usize, actual: usize },

    /// Failed to load an Iceberg table from the REST catalog.
    ///
    /// Raised when `Catalog::load_table` returns an error — usually because
    /// the table does not exist, the caller lacks permission, or the
    /// catalog is in an inconsistent state.
    #[error("failed to load Iceberg table '{table_name}': {source}")]
    TableLoad {
        table_name: String,
        #[source]
        source: Box<iceberg::Error>,
    },

    /// An unsupported Iceberg partition transform was requested.
    ///
    /// Raised by `PartitionValue::from_datetime` when the caller passes a
    /// transform name that is not one of the supported Iceberg transforms.
    #[error("unsupported datetime partition transform: {transform}")]
    UnsupportedPartitionTransform { transform: String },
}

/// Errors produced by the Kafka subsystem (producer initialization and
/// message delivery for the Iceberg write path).
///
/// Each variant carries the original `rdkafka` error so downstream code and
/// the FFI layer can distinguish producer-setup failures from delivery
/// failures without parsing strings.
#[non_exhaustive]
#[derive(Debug, thiserror::Error)]
pub enum KafkaError {
    /// Failed to create or configure the underlying `rdkafka` producer.
    ///
    /// Raised by `KafkaProducer::new` when the rdkafka client cannot be
    /// instantiated — usually because the bootstrap servers are unreachable,
    /// TLS configuration is invalid, or SASL credentials are rejected.
    #[error("failed to create Kafka producer: {source}")]
    Client {
        #[source]
        source: RdKafkaError,
    },

    /// A message was accepted by the producer but could not be delivered
    /// to the broker.
    ///
    /// Raised by `KafkaProducer::produce` when the broker returns an error
    /// during the delivery handshake. The owned message is dropped; only
    /// the underlying rdkafka error is retained.
    #[error("Kafka message delivery failed: {source}")]
    Delivery {
        #[source]
        source: RdKafkaError,
    },

    /// The producer failed to flush pending messages within the timeout.
    ///
    /// Raised by `KafkaProducer::flush` when the underlying rdkafka flush
    /// call returns an error (broker disconnect, timeout, or client
    /// shutdown). The in-flight count cannot be trusted after this error
    /// because rdkafka may have given up tracking queued messages.
    #[error("Kafka producer flush failed: {source}")]
    Flush {
        #[source]
        source: RdKafkaError,
    },
}

/// Errors produced by the postgres subsystem.
///
/// Each variant carries the structured context needed for the FFI layer to
/// build a Python-equivalent exception without parsing strings.
#[non_exhaustive]
#[derive(Debug, thiserror::Error)]
pub enum PostgresError {
    /// A row was expected to exist for an update but did not.
    ///
    /// Raised by `update_*` methods when an `UPDATE ... WHERE id = $1`
    /// query affects zero rows, indicating the target id is not present
    /// in the database.
    #[error("{entity} with id {id} not found")]
    NotFound { entity: String, id: EntityId },

    /// A raw `sqlx` query or connection error.
    #[error(transparent)]
    Query(#[from] sqlx::Error),
}

impl From<sqlx::Error> for HorizonError {
    fn from(err: sqlx::Error) -> Self {
        Self::Postgres(PostgresError::Query(err))
    }
}

/// An opaque entity identifier for use in error messages.
///
/// Wraps either a UUID or a string key so `NotFound` errors can reference
/// the lookup value without forcing callers to convert types.
#[non_exhaustive]
#[derive(Debug)]
pub enum EntityId {
    String(String),
    Uuid(Uuid),
}

impl fmt::Display for EntityId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::String(id) => write!(f, "{id}"),
            Self::Uuid(id) => write!(f, "{id}"),
        }
    }
}

impl From<String> for EntityId {
    fn from(id: String) -> Self {
        Self::String(id)
    }
}

impl From<Uuid> for EntityId {
    fn from(id: Uuid) -> Self {
        Self::Uuid(id)
    }
}

impl From<Option<Uuid>> for EntityId {
    /// Converts `Some(uuid)` to `EntityId::Uuid` and `None` to
    /// `EntityId::String("None")`. The `"None"` literal matches Python's
    /// behavior when `f"...{model.id}..."` formats a `None` id, so error
    /// messages stay character-for-character equivalent across SDKs.
    fn from(id: Option<Uuid>) -> Self {
        id.map_or_else(|| Self::String("None".to_owned()), Self::Uuid)
    }
}

#[cfg(test)]
#[allow(
    clippy::expect_used,
    clippy::panic,
    reason = "tests use expect/panic for brevity"
)]
mod tests {
    use super::*;

    #[test]
    fn entity_id_from_some_uuid_preserves_uuid() {
        let uuid = Uuid::new_v4();
        let id: EntityId = Some(uuid).into();
        match id {
            EntityId::Uuid(inner) => assert_eq!(inner, uuid),
            EntityId::String(_) => panic!("expected Uuid variant"),
        }
    }

    #[test]
    fn entity_id_from_none_uses_python_str_repr() {
        let id: EntityId = Option::<Uuid>::None.into();
        // Python's `str(None)` produces "None"; the FFI parity message
        // depends on this exact form, so pin it in a test.
        assert_eq!(id.to_string(), "None");
    }

    #[test]
    fn not_found_message_matches_python_format_for_none_id() {
        let err = PostgresError::NotFound {
            entity: "platform".to_owned(),
            id: Option::<Uuid>::None.into(),
        };
        assert_eq!(err.to_string(), "platform with id None not found");
    }

    #[test]
    fn not_found_message_matches_python_format_for_uuid() {
        let uuid =
            Uuid::parse_str("00000000-0000-0000-0000-000000000001").expect("valid uuid literal");
        let err = PostgresError::NotFound {
            entity: "data_stream".to_owned(),
            id: Some(uuid).into(),
        };
        assert_eq!(
            err.to_string(),
            "data_stream with id 00000000-0000-0000-0000-000000000001 not found"
        );
    }

    #[test]
    fn sqlx_error_converts_to_horizon_error_via_postgres_query() {
        // Verifies the manual `From<sqlx::Error> for HorizonError` impl
        // routes through `PostgresError::Query` so existing `?` flows in
        // repository methods produce the new wrapped variant rather than
        // the previous flat sqlx-only form.
        let sqlx_err = sqlx::Error::RowNotFound;
        let horizon_err: HorizonError = sqlx_err.into();
        let HorizonError::Postgres(PostgresError::Query(_)) = horizon_err else {
            panic!("expected HorizonError::Postgres(PostgresError::Query(_))");
        };
    }
}
