horizon-sdk 6.9.0

use chrono::{DateTime, Utc};
use tracing::debug;
use uuid::Uuid;

use crate::postgres::retry::{RetryPolicy, retry};
use crate::types::error::{HorizonError, PostgresError, Result};
use crate::types::model::Platform;

use super::PostgresRepository;

impl PostgresRepository {
    /// Upsert with datetime min/max merging for `start_datetime`/`end_datetime`.
    ///
    /// Keeps the earlier `start_datetime` and the later `end_datetime`.
    pub async fn create_or_update_platform(&self, platform: &Platform) -> Result<Platform> {
        let now = Utc::now();
        let org = self.organization_id.or(platform.organization_id);
        let pos_x = platform.position.map(|pos| pos.0);
        let pos_y = platform.position.map(|pos| pos.1);
        let result = sqlx::query_as!(
            Platform,
            r#"
            INSERT INTO horizon_public.platform
                (id, name, kind_id, free_text, organization_id,
                 start_datetime, end_datetime, position, modified_datetime)
            VALUES (COALESCE($1, gen_random_uuid()), $2, $3, $4, $5, $6, $7,
                    CASE WHEN $8::float8 IS NOT NULL THEN point($8, $9::float8) END, $10)
            ON CONFLICT (id) DO UPDATE SET
                name = EXCLUDED.name,
                kind_id = EXCLUDED.kind_id,
                free_text = EXCLUDED.free_text,
                organization_id = EXCLUDED.organization_id,
                start_datetime = LEAST(
                    COALESCE(EXCLUDED.start_datetime, horizon_public.platform.start_datetime),
                    COALESCE(horizon_public.platform.start_datetime, EXCLUDED.start_datetime)
                ),
                end_datetime = GREATEST(
                    COALESCE(EXCLUDED.end_datetime, horizon_public.platform.end_datetime),
                    COALESCE(horizon_public.platform.end_datetime, EXCLUDED.end_datetime)
                ),
                position = EXCLUDED.position,
                modified_datetime = EXCLUDED.modified_datetime
            RETURNING
                id, created_datetime, modified_datetime, name, kind_id, free_text,
                position as "position: _", organization_id, start_datetime, end_datetime
            "#,
            platform.id,
            platform.name,
            platform.kind_id,
            platform.free_text,
            org,
            platform.start_datetime,
            platform.end_datetime,
            pos_x,
            pos_y,
            now,
        )
        .fetch_one(&self.pool)
        .await?;
        debug!(
            platform_id = ?result.id,
            platform_name = ?result.name,
            "create_or_update_platform"
        );
        Ok(result)
    }

    /// Delete a platform by ID. Returns the number of rows removed.
    pub async fn delete_platform(&self, id: Uuid) -> Result<u64> {
        let result = sqlx::query!("DELETE FROM horizon_public.platform WHERE id = $1", id)
            .execute(&self.pool)
            .await?;
        Ok(result.rows_affected())
    }

    /// Create a new platform.
    pub async fn insert_platform(&self, platform: &Platform) -> Result<Platform> {
        let now = Utc::now();
        let org = self.organization_id.or(platform.organization_id);
        let pos_x = platform.position.map(|pos| pos.0);
        let pos_y = platform.position.map(|pos| pos.1);
        Ok(sqlx::query_as!(
            Platform,
            r#"
            INSERT INTO horizon_public.platform
                (id, name, kind_id, free_text, organization_id,
                 start_datetime, end_datetime, position, modified_datetime)
            VALUES (COALESCE($1, gen_random_uuid()), $2, $3, $4, $5, $6, $7,
                    CASE WHEN $8::float8 IS NOT NULL THEN point($8, $9::float8) END, $10)
            RETURNING
                id, created_datetime, modified_datetime, name, kind_id, free_text,
                position as "position: _", organization_id, start_datetime, end_datetime
            "#,
            platform.id,
            platform.name,
            platform.kind_id,
            platform.free_text,
            org,
            platform.start_datetime,
            platform.end_datetime,
            pos_x,
            pos_y,
            now,
        )
        .fetch_one(&self.pool)
        .await?)
    }

    /// Create a batch of platforms in a single multi-row INSERT.
    ///
    /// Per-column slices are passed to `PostgreSQL` `UNNEST(...)` so the
    /// entire batch becomes one SQL statement and one network round-trip.
    /// Missing IDs are filled by `gen_random_uuid()` server-side, and
    /// `modified_datetime` is set to the same `now()` value for every
    /// row.
    ///
    /// Returns the inserted rows in input order. Wraps the statement in
    /// a transaction so a per-row failure rolls back the entire batch.
    #[allow(
        clippy::as_conversions,
        reason = "sqlx::query_as! requires `&Vec<T> as &[T]` ascription to bind a nullable Postgres array; conversion direction is unambiguous (same element type)"
    )]
    pub async fn insert_platform_batch(&self, platforms: &[Platform]) -> Result<Vec<Platform>> {
        if platforms.is_empty() {
            return Ok(Vec::new());
        }
        let now = Utc::now();
        let ids: Vec<Option<Uuid>> = platforms.iter().map(|platform| platform.id).collect();
        let names: Vec<Option<String>> = platforms
            .iter()
            .map(|platform| platform.name.clone())
            .collect();
        let kind_ids: Vec<Option<Uuid>> =
            platforms.iter().map(|platform| platform.kind_id).collect();
        let free_texts: Vec<Option<String>> = platforms
            .iter()
            .map(|platform| platform.free_text.clone())
            .collect();
        let orgs: Vec<Option<Uuid>> = platforms
            .iter()
            .map(|platform| self.organization_id.or(platform.organization_id))
            .collect();
        let start_datetimes: Vec<Option<DateTime<Utc>>> = platforms
            .iter()
            .map(|platform| platform.start_datetime)
            .collect();
        let end_datetimes: Vec<Option<DateTime<Utc>>> = platforms
            .iter()
            .map(|platform| platform.end_datetime)
            .collect();
        let longitudes: Vec<Option<f64>> = platforms
            .iter()
            .map(|platform| platform.position.map(|pos| pos.0))
            .collect();
        let latitudes: Vec<Option<f64>> = platforms
            .iter()
            .map(|platform| platform.position.map(|pos| pos.1))
            .collect();
        let mut tx = self.pool.begin().await?;
        let rows = sqlx::query_as!(
            Platform,
            r#"
            WITH source AS (
                SELECT
                    COALESCE(id, gen_random_uuid()) AS id,
                    name, kind_id, free_text, organization_id,
                    start_datetime, end_datetime, pos_x, pos_y, ord
                FROM UNNEST(
                    $1::uuid[], $2::text[], $3::uuid[], $4::text[], $5::uuid[],
                    $6::timestamptz[], $7::timestamptz[], $8::float8[], $9::float8[]
                ) WITH ORDINALITY
                    AS batch(id, name, kind_id, free_text, organization_id,
                             start_datetime, end_datetime, pos_x, pos_y, ord)
            ),
            inserted AS (
                INSERT INTO horizon_public.platform
                    (id, name, kind_id, free_text, organization_id,
                     start_datetime, end_datetime, position, modified_datetime)
                SELECT
                    id, name, kind_id, free_text, organization_id,
                    start_datetime, end_datetime,
                    CASE WHEN pos_x IS NOT NULL AND pos_y IS NOT NULL THEN point(pos_x, pos_y) END,
                    $10
                FROM source
                RETURNING
                    id, created_datetime, modified_datetime, name, kind_id, free_text,
                    position, organization_id, start_datetime, end_datetime
            )
            SELECT
                inserted.id, inserted.created_datetime, inserted.modified_datetime,
                inserted.name, inserted.kind_id, inserted.free_text,
                inserted.position as "position: _", inserted.organization_id,
                inserted.start_datetime, inserted.end_datetime
            FROM inserted
            JOIN source ON source.id = inserted.id
            ORDER BY source.ord
            "#,
            &ids as &[Option<Uuid>],
            &names as &[Option<String>],
            &kind_ids as &[Option<Uuid>],
            &free_texts as &[Option<String>],
            &orgs as &[Option<Uuid>],
            &start_datetimes as &[Option<DateTime<Utc>>],
            &end_datetimes as &[Option<DateTime<Utc>>],
            &longitudes as &[Option<f64>],
            &latitudes as &[Option<f64>],
            now,
        )
        .fetch_all(&mut *tx)
        .await?;
        tx.commit().await?;
        debug!(batch_size = rows.len(), "insert_platform_batch");
        Ok(rows)
    }

    /// List all platforms. Retried under the default [`RetryPolicy`] on transient
    /// Postgres errors; reads are naturally idempotent.
    pub async fn list_platforms(&self) -> Result<Vec<Platform>> {
        retry(RetryPolicy::default(), || async move {
            Ok(sqlx::query_as!(
                Platform,
                r#"
                SELECT
                    id, created_datetime, modified_datetime, name, kind_id, free_text,
                    position as "position: _", organization_id, start_datetime, end_datetime
                FROM horizon_public.platform
                "#,
            )
            .fetch_all(&self.pool)
            .await?)
        })
        .await
    }

    /// Read a platform by ID. Retried under the default [`RetryPolicy`] on transient
    /// Postgres errors; reads are naturally idempotent.
    pub async fn read_platform(&self, id: Uuid) -> Result<Option<Platform>> {
        retry(RetryPolicy::default(), || async move {
            Ok(sqlx::query_as!(
                Platform,
                r#"
                SELECT
                    id, created_datetime, modified_datetime, name, kind_id, free_text,
                    position as "position: _", organization_id, start_datetime, end_datetime
                FROM horizon_public.platform
                WHERE id = $1
                "#,
                id,
            )
            .fetch_optional(&self.pool)
            .await?)
        })
        .await
    }

    /// Update an existing platform by ID.
    ///
    /// Returns `HorizonError::Postgres(PostgresError::NotFound)` if no row
    /// with the given ID exists. Matches the Python SDK's
    /// `BaseRepository.update()` semantics, which raises
    /// `ValueError("Record with ID <id> not found")`.
    pub async fn update_platform(&self, platform: &Platform) -> Result<Platform> {
        let now = Utc::now();
        let org = self.organization_id.or(platform.organization_id);
        let pos_x = platform.position.map(|pos| pos.0);
        let pos_y = platform.position.map(|pos| pos.1);
        let result = sqlx::query_as!(
            Platform,
            r#"
            UPDATE horizon_public.platform
            SET
                name = $2,
                kind_id = $3,
                free_text = $4,
                organization_id = $5,
                start_datetime = $6,
                end_datetime = $7,
                position = CASE WHEN $8::float8 IS NOT NULL THEN point($8, $9::float8) END,
                modified_datetime = $10
            WHERE id = $1
            RETURNING
                id, created_datetime, modified_datetime, name, kind_id, free_text,
                position as "position: _", organization_id, start_datetime, end_datetime
            "#,
            platform.id,
            platform.name,
            platform.kind_id,
            platform.free_text,
            org,
            platform.start_datetime,
            platform.end_datetime,
            pos_x,
            pos_y,
            now,
        )
        .fetch_optional(&self.pool)
        .await?;

        result.ok_or_else(|| {
            HorizonError::Postgres(PostgresError::NotFound {
                entity: "platform".to_owned(),
                id: platform.id.into(),
            })
        })
    }

    /// Create a new platform or update if it already exists.
    pub async fn upsert_platform(&self, platform: &Platform) -> Result<Platform> {
        let now = Utc::now();
        let org = self.organization_id.or(platform.organization_id);
        let pos_x = platform.position.map(|pos| pos.0);
        let pos_y = platform.position.map(|pos| pos.1);
        Ok(sqlx::query_as!(
            Platform,
            r#"
            INSERT INTO horizon_public.platform
                (id, name, kind_id, free_text, organization_id,
                 start_datetime, end_datetime, position, modified_datetime)
            VALUES (COALESCE($1, gen_random_uuid()), $2, $3, $4, $5, $6, $7,
                    CASE WHEN $8::float8 IS NOT NULL THEN point($8, $9::float8) END, $10)
            ON CONFLICT (id) DO UPDATE SET
                name = EXCLUDED.name,
                kind_id = EXCLUDED.kind_id,
                free_text = EXCLUDED.free_text,
                organization_id = EXCLUDED.organization_id,
                start_datetime = EXCLUDED.start_datetime,
                end_datetime = EXCLUDED.end_datetime,
                position = EXCLUDED.position,
                modified_datetime = EXCLUDED.modified_datetime
            RETURNING
                id, created_datetime, modified_datetime, name, kind_id, free_text,
                position as "position: _", organization_id, start_datetime, end_datetime
            "#,
            platform.id,
            platform.name,
            platform.kind_id,
            platform.free_text,
            org,
            platform.start_datetime,
            platform.end_datetime,
            pos_x,
            pos_y,
            now,
        )
        .fetch_one(&self.pool)
        .await?)
    }
}