//! PostgreSQL storage backend for durable message queues.
//!
//! Persists queue messages to a PostgreSQL database via `sqlx`. Suitable for
//! cloud/HA deployments where queue state must survive node replacement and
//! be accessible from multiple relay instances.
//!
//! Enable with: `cargo build --features queue-postgres`
//!
//! # Schema
//!
//! This backend expects two tables: `pulse_queue_messages` and `pulse_queue_dlq`.
//! Call [`PostgresQueueBackend::migrate`] on startup to create them automatically.

use pulse_core::queue::{MessageState, QueueMessage};
use pulse_core::queue_storage::{QueueStorageBackend, QueueStorageError};
use sqlx::PgPool;
use tokio::runtime::Handle;

// ═══════════════════════════════════════════════════════════════════════════
// State mapping
// ═══════════════════════════════════════════════════════════════════════════

fn state_to_i16(state: MessageState) -> i16 {
    match state {
        MessageState::Pending => 0,
        MessageState::InFlight => 1,
        MessageState::Acked => 2,
        MessageState::DeadLettered => 3,
    }
}

fn i16_to_state(v: i16) -> MessageState {
    match v {
        0 => MessageState::Pending,
        1 => MessageState::InFlight,
        2 => MessageState::Acked,
        3 => MessageState::DeadLettered,
        _ => MessageState::Pending, // fallback
    }
}

// ═══════════════════════════════════════════════════════════════════════════
// PostgreSQL Backend
// ═══════════════════════════════════════════════════════════════════════════

/// PostgreSQL-backed storage for durable message queues.
///
/// Uses a shared `PgPool` for all operations. The backend uses
/// `tokio::runtime::Handle::current().block_on()` to bridge from the
/// synchronous [`QueueStorageBackend`] trait to async sqlx operations.
pub struct PostgresQueueBackend {
    pool: PgPool,
}

impl PostgresQueueBackend {
    /// Create a new backend with an existing connection pool.
    pub fn new(pool: PgPool) -> Self {
        Self { pool }
    }

    /// Connect to PostgreSQL and create a new backend.
    pub async fn connect(url: &str, max_connections: u32) -> Result<Self, QueueStorageError> {
        let pool = sqlx::postgres::PgPoolOptions::new()
            .max_connections(max_connections)
            .connect(url)
            .await
            .map_err(|e| QueueStorageError::Io(format!("connecting to PostgreSQL: {e}")))?;
        Ok(Self { pool })
    }

    /// Run database migrations to create the required tables.
    pub async fn migrate(&self) -> Result<(), QueueStorageError> {
        // Create tables and indexes individually (PostgreSQL doesn't allow
        // multiple commands in a single prepared statement).

        sqlx::query(
            r#"
            CREATE TABLE IF NOT EXISTS pulse_queue_messages (
                queue_name      TEXT        NOT NULL,
                sequence        BIGINT      NOT NULL,
                payload         BYTEA       NOT NULL,
                state           SMALLINT    NOT NULL DEFAULT 0,
                delivery_count  INT         NOT NULL DEFAULT 0,
                ttl_secs        BIGINT,
                enqueued_at     BIGINT      NOT NULL,
                idempotency_key TEXT,
                updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                PRIMARY KEY (queue_name, sequence)
            )
            "#,
        )
        .execute(&self.pool)
        .await
        .map_err(|e| QueueStorageError::Io(format!("create pulse_queue_messages failed: {e}")))?;

        sqlx::query(
            r#"
            CREATE INDEX IF NOT EXISTS idx_pulse_queue_pending
                ON pulse_queue_messages (queue_name, state, sequence)
                WHERE state IN (0, 1)
            "#,
        )
        .execute(&self.pool)
        .await
        .map_err(|e| QueueStorageError::Io(format!("create idx_pulse_queue_pending failed: {e}")))?;

        sqlx::query(
            r#"
            CREATE UNIQUE INDEX IF NOT EXISTS idx_pulse_queue_dedup
                ON pulse_queue_messages (queue_name, idempotency_key)
                WHERE idempotency_key IS NOT NULL
            "#,
        )
        .execute(&self.pool)
        .await
        .map_err(|e| QueueStorageError::Io(format!("create idx_pulse_queue_dedup failed: {e}")))?;

        sqlx::query(
            r#"
            CREATE TABLE IF NOT EXISTS pulse_queue_dlq (
                queue_name          TEXT     NOT NULL,
                sequence            BIGINT   NOT NULL,
                payload             BYTEA    NOT NULL,
                delivery_count      INT      NOT NULL,
                dead_lettered_at    TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                original_enqueued_at BIGINT  NOT NULL,
                PRIMARY KEY (queue_name, sequence)
            )
            "#,
        )
        .execute(&self.pool)
        .await
        .map_err(|e| QueueStorageError::Io(format!("create pulse_queue_dlq failed: {e}")))?;

        tracing::info!("[postgres-queue] migrations applied");
        Ok(())
    }

    /// Helper: run an async operation synchronously, safe to call from within
    /// the tokio runtime (uses `block_in_place` to avoid the "cannot start a
    /// runtime from within a runtime" panic).
    fn block_on<F: std::future::Future>(&self, f: F) -> F::Output {
        tokio::task::block_in_place(|| Handle::current().block_on(f))
    }
}

impl QueueStorageBackend for PostgresQueueBackend {
    fn append(&self, queue: &str, msg: &QueueMessage) -> Result<(), QueueStorageError> {
        self.block_on(async {
            sqlx::query(
                r#"
                INSERT INTO pulse_queue_messages
                    (queue_name, sequence, payload, state, delivery_count, ttl_secs, enqueued_at, idempotency_key)
                VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
                ON CONFLICT (queue_name, sequence) DO NOTHING
                "#,
            )
            .bind(queue)
            .bind(msg.sequence as i64)
            .bind(&msg.payload)
            .bind(state_to_i16(msg.state))
            .bind(msg.delivery_count as i32)
            .bind(msg.ttl_secs.map(|t| t as i64))
            .bind(msg.enqueued_at as i64)
            .bind(&msg.idempotency_key)
            .execute(&self.pool)
            .await
            .map_err(|e| QueueStorageError::Io(format!("append failed: {e}")))?;
            Ok(())
        })
    }

    fn load_pending(&self, queue: &str) -> Result<Vec<QueueMessage>, QueueStorageError> {
        self.block_on(async {
            let rows: Vec<(i64, Vec<u8>, i16, i32, Option<i64>, i64, Option<String>)> =
                sqlx::query_as(
                    r#"
                    SELECT sequence, payload, state, delivery_count, ttl_secs, enqueued_at, idempotency_key
                    FROM pulse_queue_messages
                    WHERE queue_name = $1 AND state IN (0, 1)
                    ORDER BY sequence ASC
                    "#,
                )
                .bind(queue)
                .fetch_all(&self.pool)
                .await
                .map_err(|e| QueueStorageError::Io(format!("load_pending failed: {e}")))?;

            Ok(rows
                .into_iter()
                .map(
                    |(sequence, payload, state, delivery_count, ttl_secs, enqueued_at, idempotency_key)| {
                        QueueMessage {
                            sequence: sequence as u64,
                            payload,
                            state: i16_to_state(state),
                            delivery_count: delivery_count as u32,
                            ttl_secs: ttl_secs.map(|t| t as u64),
                            enqueued_at: enqueued_at as u64,
                            idempotency_key,
                        }
                    },
                )
                .collect())
        })
    }

    fn update_state(
        &self,
        queue: &str,
        sequence: u64,
        new_state: MessageState,
        delivery_count: u32,
    ) -> Result<(), QueueStorageError> {
        self.block_on(async {
            let result = sqlx::query(
                r#"
                UPDATE pulse_queue_messages
                SET state = $3, delivery_count = $4, updated_at = NOW()
                WHERE queue_name = $1 AND sequence = $2
                "#,
            )
            .bind(queue)
            .bind(sequence as i64)
            .bind(state_to_i16(new_state))
            .bind(delivery_count as i32)
            .execute(&self.pool)
            .await
            .map_err(|e| QueueStorageError::Io(format!("update_state failed: {e}")))?;

            if result.rows_affected() == 0 {
                return Err(QueueStorageError::NotFound);
            }
            Ok(())
        })
    }

    fn dead_letter(&self, queue: &str, msg: &QueueMessage) -> Result<(), QueueStorageError> {
        self.block_on(async {
            sqlx::query(
                r#"
                INSERT INTO pulse_queue_dlq
                    (queue_name, sequence, payload, delivery_count, original_enqueued_at)
                VALUES ($1, $2, $3, $4, $5)
                ON CONFLICT (queue_name, sequence) DO NOTHING
                "#,
            )
            .bind(queue)
            .bind(msg.sequence as i64)
            .bind(&msg.payload)
            .bind(msg.delivery_count as i32)
            .bind(msg.enqueued_at as i64)
            .execute(&self.pool)
            .await
            .map_err(|e| QueueStorageError::Io(format!("dead_letter failed: {e}")))?;
            Ok(())
        })
    }

    fn compact(&self, queue: &str, before_seq: u64) -> Result<u64, QueueStorageError> {
        self.block_on(async {
            let result = sqlx::query(
                r#"
                DELETE FROM pulse_queue_messages
                WHERE queue_name = $1
                  AND sequence < $2
                  AND state IN (2, 3)
                "#,
            )
            .bind(queue)
            .bind(before_seq as i64)
            .execute(&self.pool)
            .await
            .map_err(|e| QueueStorageError::Io(format!("compact failed: {e}")))?;
            Ok(result.rows_affected())
        })
    }

    fn has_idempotency_key(&self, queue: &str, key: &str) -> Result<bool, QueueStorageError> {
        self.block_on(async {
            let row: Option<(i64,)> = sqlx::query_as(
                r#"
                SELECT 1 FROM pulse_queue_messages
                WHERE queue_name = $1 AND idempotency_key = $2
                LIMIT 1
                "#,
            )
            .bind(queue)
            .bind(key)
            .fetch_optional(&self.pool)
            .await
            .map_err(|e| QueueStorageError::Io(format!("has_idempotency_key failed: {e}")))?;
            Ok(row.is_some())
        })
    }

    fn record_idempotency_key(&self, _queue: &str, _key: &str) -> Result<(), QueueStorageError> {
        // In PostgreSQL mode, the idempotency key is stored as part of the
        // message row itself (via the UNIQUE index). No separate recording needed.
        Ok(())
    }

    fn evict_idempotency_keys(&self, _queue: &str, _keep: usize) -> Result<(), QueueStorageError> {
        // Idempotency keys are evicted when messages are compacted/deleted.
        // No separate eviction needed.
        Ok(())
    }

    fn purge(&self, queue: &str) -> Result<usize, QueueStorageError> {
        self.block_on(async {
            let result = sqlx::query(
                "DELETE FROM pulse_queue_messages WHERE queue_name = $1",
            )
            .bind(queue)
            .execute(&self.pool)
            .await
            .map_err(|e| QueueStorageError::Io(format!("purge failed: {e}")))?;
            Ok(result.rows_affected() as usize)
        })
    }

    fn load_dead_letters(&self, queue: &str) -> Result<Vec<QueueMessage>, QueueStorageError> {
        self.block_on(async {
            let rows: Vec<(i64, Vec<u8>, i32, i64)> = sqlx::query_as(
                r#"
                SELECT sequence, payload, delivery_count, original_enqueued_at
                FROM pulse_queue_dlq
                WHERE queue_name = $1
                ORDER BY sequence ASC
                "#,
            )
            .bind(queue)
            .fetch_all(&self.pool)
            .await
            .map_err(|e| QueueStorageError::Io(format!("load_dead_letters failed: {e}")))?;

            Ok(rows
                .into_iter()
                .map(|(sequence, payload, delivery_count, enqueued_at)| QueueMessage {
                    sequence: sequence as u64,
                    payload,
                    state: MessageState::DeadLettered,
                    delivery_count: delivery_count as u32,
                    ttl_secs: None,
                    enqueued_at: enqueued_at as u64,
                    idempotency_key: None,
                })
                .collect())
        })
    }
}

// ═══════════════════════════════════════════════════════════════════════════
// Tests (require a running PostgreSQL instance, so marked #[ignore])
// ═══════════════════════════════════════════════════════════════════════════

#[cfg(test)]
mod tests {
    use super::*;

    /// These tests require a PostgreSQL instance. Run them with:
    /// ```shell
    /// DATABASE_URL=postgres://user:pass@localhost/pulse_test \
    ///   cargo test -p sansavision-pulse --features queue-postgres -- postgres_queue --ignored
    /// ```
    fn get_test_url() -> Option<String> {
        std::env::var("DATABASE_URL").ok()
    }

    #[tokio::test]
    #[ignore = "requires PostgreSQL — set DATABASE_URL to run"]
    async fn postgres_queue_lifecycle() {
        let url = get_test_url().expect("DATABASE_URL not set");
        let backend = PostgresQueueBackend::connect(&url, 5).await.unwrap();
        backend.migrate().await.unwrap();

        let q = "pg-test-lifecycle";

        // Clean up from previous runs
        let _ = backend.purge(q);

        // Append
        let msg = QueueMessage {
            sequence: 1,
            payload: b"hello-postgres".to_vec(),
            idempotency_key: Some("pg-key-1".to_string()),
            ttl_secs: Some(3600),
            enqueued_at: 1000,
            delivery_count: 0,
            state: MessageState::Pending,
        };
        backend.append(q, &msg).unwrap();

        // Load pending
        let pending = backend.load_pending(q).unwrap();
        assert_eq!(pending.len(), 1);
        assert_eq!(pending[0].payload, b"hello-postgres");

        // Idempotency check
        assert!(backend.has_idempotency_key(q, "pg-key-1").unwrap());
        assert!(!backend.has_idempotency_key(q, "pg-key-2").unwrap());

        // Update state
        backend
            .update_state(q, 1, MessageState::Acked, 1)
            .unwrap();
        let pending = backend.load_pending(q).unwrap();
        assert!(pending.is_empty());

        // Compact
        let removed = backend.compact(q, 2).unwrap();
        assert_eq!(removed, 1);

        // Purge
        let _ = backend.purge(q);
    }
}
