//! `QueueRepo` — CRUD over the `queue_messages` table.
//!
//! Producer path: `enqueue(...)` writes one row. Dispatcher hot path:
//! `claim(...)` does the single-round-trip atomic claim
//! (`UPDATE ... WHERE id = (SELECT ... FOR UPDATE SKIP LOCKED)`).
//! Outcome: `ack` deletes the row, `nack` clears the claim with a
//! `deliver_after` retry delay, `dead_letter` moves the payload to
//! `dead_letters` and deletes the queue row.
//!
//! `reclaim_visibility_timeouts()` is the periodic safety net — a
//! crashed dispatcher (or one whose handler hung past the visibility
//! timeout) loses its claim and the message becomes claimable again.
//!
//! All methods take `app_id` explicitly; the service layer derives it
//! from `cx.app_id` and passes it in — repo never reads cx.

use async_trait::async_trait;
use chrono::{DateTime, Utc};
use picloud_shared::{AdminUserId, AppId, QueueMessageId, TriggerId};
use sqlx::PgPool;
use uuid::Uuid;

use crate::dead_letter_repo::{DeadLetterRepoError, NewDeadLetter};

/// F-P-006: maximum rows scanned by `depth()` / `depth_pending()`.
/// Scripts calling `queue::depth(name)` per invocation paid an unbounded
/// COUNT(*); capping the scan at 10k lets the SDK report "10000" as a
/// floor for very-deep queues without ever touching more than that many
/// index rows.
const QUEUE_DEPTH_SCAN_CAP: i64 = 10_000;

#[derive(Debug, thiserror::Error)]
pub enum QueueRepoError {
    #[error("database error: {0}")]
    Db(#[from] sqlx::Error),

    /// JSON serialization failure on the payload (extremely rare; only
    /// surfaces if a non-serializable Value somehow reached the repo).
    #[error("payload serialization: {0}")]
    Payload(String),

    /// Surfaces from the dead_letter() helper if the dead-letter insert
    /// fails.
    #[error("dead-letter write: {0}")]
    DeadLetter(String),
}

/// Insert payload — what `QueueService::enqueue` hands the repo.
#[derive(Debug, Clone)]
pub struct NewQueueMessage {
    pub app_id: AppId,
    pub queue_name: String,
    pub payload: serde_json::Value,
    /// `Some` → defer until `NOW() + delay`; `None` → immediate.
    pub deliver_after: Option<DateTime<Utc>>,
    pub max_attempts: u32,
    pub enqueued_by_principal: Option<AdminUserId>,
}

/// One claimed message ready for handler dispatch.
#[derive(Debug, Clone)]
pub struct ClaimedMessage {
    pub id: QueueMessageId,
    pub app_id: AppId,
    pub queue_name: String,
    pub payload: serde_json::Value,
    pub enqueued_at: DateTime<Utc>,
    pub attempt: u32,
    pub max_attempts: u32,
    pub claim_token: Uuid,
}

/// Aggregate counts for the dashboard's read-only queue overview.
#[derive(Debug, Clone, Copy, Default)]
pub struct QueueStats {
    /// `COUNT(*)` for `(app_id, queue_name)`.
    pub total: u64,
    /// Claimable now — `claim_token IS NULL AND (deliver_after IS NULL OR <= NOW())`.
    pub pending: u64,
    /// Currently leased to a dispatcher — `claim_token IS NOT NULL`.
    pub claimed: u64,
}

#[async_trait]
pub trait QueueRepo: Send + Sync {
    /// Insert one row. Returns the new message id.
    async fn enqueue(&self, msg: NewQueueMessage) -> Result<QueueMessageId, QueueRepoError>;

    /// Atomic claim of a single ready message from `(app_id, queue_name)`.
    /// Returns `Ok(None)` if nothing is claimable.
    async fn claim(
        &self,
        app_id: AppId,
        queue_name: &str,
    ) -> Result<Option<ClaimedMessage>, QueueRepoError>;

    /// Handler succeeded: delete the row IFF the claim_token matches
    /// (so a stale dispatcher whose lease expired doesn't accidentally
    /// delete a re-claimed message). Returns `true` if 1 row deleted.
    async fn ack(
        &self,
        message_id: QueueMessageId,
        claim_token: Uuid,
    ) -> Result<bool, QueueRepoError>;

    /// Handler threw: clear the claim, defer redelivery by `retry_delay`.
    /// IFF the claim_token matches. Returns `true` on success.
    async fn nack(
        &self,
        message_id: QueueMessageId,
        claim_token: Uuid,
        retry_delay: chrono::Duration,
    ) -> Result<bool, QueueRepoError>;

    /// TRANSIENT release: the handler never ran (gate saturated, or the script
    /// was disabled at fire time), so re-queue AND undo `claim`'s pre-increment
    /// of `attempt` — a non-execution must not burn the retry budget (else a
    /// message could be dead-lettered under sustained overload having executed
    /// zero times). Floors `attempt` at 0. IFF the claim_token matches.
    async fn release(
        &self,
        message_id: QueueMessageId,
        claim_token: Uuid,
        retry_delay: chrono::Duration,
    ) -> Result<bool, QueueRepoError>;

    /// Periodic safety net: clear claims whose `claimed_at` exceeds the
    /// per-queue `visibility_timeout_secs` (joined from
    /// `queue_trigger_details`). Returns the number of rows reclaimed.
    async fn reclaim_visibility_timeouts(&self) -> Result<u64, QueueRepoError>;

    /// `queue::depth(name)` — total rows in the queue.
    async fn depth(&self, app_id: AppId, queue_name: &str) -> Result<u64, QueueRepoError>;

    /// `queue::depth_pending(name)` — currently claimable rows.
    async fn depth_pending(&self, app_id: AppId, queue_name: &str) -> Result<u64, QueueRepoError>;

    /// Dashboard read-only view: every distinct queue name in `app_id`
    /// with its aggregate counts.
    async fn list_for_app(
        &self,
        app_id: AppId,
    ) -> Result<Vec<(String, QueueStats)>, QueueRepoError>;

    /// Dispatcher dead-letter helper: move a claimed message to
    /// `dead_letters` and delete it from `queue_messages`, in one
    /// transaction. The dispatcher's existing `fan_out_dead_letter`
    /// fires registered `dead_letter` triggers off the new row.
    #[allow(clippy::too_many_arguments)]
    async fn dead_letter(
        &self,
        message_id: QueueMessageId,
        claim_token: Uuid,
        app_id: AppId,
        queue_name: &str,
        trigger_id: Option<TriggerId>,
        script_id: Option<picloud_shared::ScriptId>,
        attempt: u32,
        first_attempt_at: DateTime<Utc>,
        last_error: &str,
    ) -> Result<picloud_shared::DeadLetterId, QueueRepoError>;
}

pub struct PostgresQueueRepo {
    pool: PgPool,
}

impl PostgresQueueRepo {
    #[must_use]
    pub fn new(pool: PgPool) -> Self {
        Self { pool }
    }
}

#[async_trait]
impl QueueRepo for PostgresQueueRepo {
    async fn enqueue(&self, msg: NewQueueMessage) -> Result<QueueMessageId, QueueRepoError> {
        let (id,): (Uuid,) = sqlx::query_as(
            "INSERT INTO queue_messages ( \
                app_id, queue_name, payload, deliver_after, \
                max_attempts, enqueued_by_principal \
             ) VALUES ($1, $2, $3, $4, $5, $6) \
             RETURNING id",
        )
        .bind(msg.app_id.into_inner())
        .bind(&msg.queue_name)
        .bind(&msg.payload)
        .bind(msg.deliver_after)
        .bind(i32::try_from(msg.max_attempts).unwrap_or(3))
        .bind(msg.enqueued_by_principal.map(AdminUserId::into_inner))
        .fetch_one(&self.pool)
        .await?;
        Ok(id.into())
    }

    async fn claim(
        &self,
        app_id: AppId,
        queue_name: &str,
    ) -> Result<Option<ClaimedMessage>, QueueRepoError> {
        let token = Uuid::new_v4();
        let row: Option<ClaimedRow> = sqlx::query_as(
            "UPDATE queue_messages \
             SET claim_token = $3, claimed_at = NOW(), attempt = attempt + 1 \
             WHERE id = ( \
                SELECT id FROM queue_messages \
                WHERE app_id = $1 AND queue_name = $2 \
                  AND claim_token IS NULL \
                  AND (deliver_after IS NULL OR deliver_after <= NOW()) \
                ORDER BY enqueued_at \
                FOR UPDATE SKIP LOCKED \
                LIMIT 1 \
             ) \
             RETURNING id, app_id, queue_name, payload, enqueued_at, attempt, max_attempts",
        )
        .bind(app_id.into_inner())
        .bind(queue_name)
        .bind(token)
        .fetch_optional(&self.pool)
        .await?;
        Ok(row.map(|r| ClaimedMessage {
            id: r.id.into(),
            app_id: r.app_id.into(),
            queue_name: r.queue_name,
            payload: r.payload,
            enqueued_at: r.enqueued_at,
            attempt: u32::try_from(r.attempt).unwrap_or(1),
            max_attempts: u32::try_from(r.max_attempts).unwrap_or(3),
            claim_token: token,
        }))
    }

    async fn ack(
        &self,
        message_id: QueueMessageId,
        claim_token: Uuid,
    ) -> Result<bool, QueueRepoError> {
        let res = sqlx::query("DELETE FROM queue_messages WHERE id = $1 AND claim_token = $2")
            .bind(message_id.into_inner())
            .bind(claim_token)
            .execute(&self.pool)
            .await?;
        Ok(res.rows_affected() == 1)
    }

    async fn nack(
        &self,
        message_id: QueueMessageId,
        claim_token: Uuid,
        retry_delay: chrono::Duration,
    ) -> Result<bool, QueueRepoError> {
        let next = Utc::now() + retry_delay;
        let res = sqlx::query(
            "UPDATE queue_messages \
             SET claim_token = NULL, claimed_at = NULL, deliver_after = $3 \
             WHERE id = $1 AND claim_token = $2",
        )
        .bind(message_id.into_inner())
        .bind(claim_token)
        .bind(next)
        .execute(&self.pool)
        .await?;
        Ok(res.rows_affected() == 1)
    }

    async fn release(
        &self,
        message_id: QueueMessageId,
        claim_token: Uuid,
        retry_delay: chrono::Duration,
    ) -> Result<bool, QueueRepoError> {
        let next = Utc::now() + retry_delay;
        let res = sqlx::query(
            "UPDATE queue_messages \
             SET claim_token = NULL, claimed_at = NULL, deliver_after = $3, \
                 attempt = GREATEST(attempt - 1, 0) \
             WHERE id = $1 AND claim_token = $2",
        )
        .bind(message_id.into_inner())
        .bind(claim_token)
        .bind(next)
        .execute(&self.pool)
        .await?;
        Ok(res.rows_affected() == 1)
    }

    async fn reclaim_visibility_timeouts(&self) -> Result<u64, QueueRepoError> {
        let res = sqlx::query(
            "UPDATE queue_messages m \
             SET claim_token = NULL, claimed_at = NULL \
             FROM triggers t \
             JOIN queue_trigger_details d ON d.trigger_id = t.id \
             WHERE m.app_id = t.app_id \
               AND m.queue_name = d.queue_name \
               AND m.claim_token IS NOT NULL \
               AND m.claimed_at < NOW() - (d.visibility_timeout_secs || ' seconds')::INTERVAL \
               AND t.kind = 'queue' AND t.enabled = TRUE",
        )
        .execute(&self.pool)
        .await?;
        Ok(res.rows_affected())
    }

    async fn depth(&self, app_id: AppId, queue_name: &str) -> Result<u64, QueueRepoError> {
        // F-P-006: cap the scan at QUEUE_DEPTH_SCAN_CAP rows. Scripts
        // calling queue::depth(name) per invocation were doing
        // unbounded COUNT(*) over the queue_messages partial index;
        // on a backed-up queue with millions of rows that meant a
        // full-partition scan every call. The capped subquery means
        // the worst-case cost is bounded; callers needing exact
        // depth on huge queues should use the admin dashboard which
        // tolerates the slower COUNT for its read frequency.
        let (n,): (i64,) = sqlx::query_as(
            "SELECT COUNT(*) FROM ( \
                 SELECT 1 FROM queue_messages \
                 WHERE app_id = $1 AND queue_name = $2 \
                 LIMIT $3 \
             ) sub",
        )
        .bind(app_id.into_inner())
        .bind(queue_name)
        .bind(QUEUE_DEPTH_SCAN_CAP)
        .fetch_one(&self.pool)
        .await?;
        Ok(u64::try_from(n).unwrap_or(0))
    }

    async fn depth_pending(&self, app_id: AppId, queue_name: &str) -> Result<u64, QueueRepoError> {
        let (n,): (i64,) = sqlx::query_as(
            "SELECT COUNT(*) FROM ( \
                 SELECT 1 FROM queue_messages \
                 WHERE app_id = $1 AND queue_name = $2 \
                   AND claim_token IS NULL \
                   AND (deliver_after IS NULL OR deliver_after <= NOW()) \
                 LIMIT $3 \
             ) sub",
        )
        .bind(app_id.into_inner())
        .bind(queue_name)
        .bind(QUEUE_DEPTH_SCAN_CAP)
        .fetch_one(&self.pool)
        .await?;
        Ok(u64::try_from(n).unwrap_or(0))
    }

    async fn list_for_app(
        &self,
        app_id: AppId,
    ) -> Result<Vec<(String, QueueStats)>, QueueRepoError> {
        let rows: Vec<QueueAggRow> = sqlx::query_as(
            "SELECT queue_name, \
                    COUNT(*) AS total, \
                    COUNT(*) FILTER ( \
                        WHERE claim_token IS NULL \
                          AND (deliver_after IS NULL OR deliver_after <= NOW()) \
                    ) AS pending, \
                    COUNT(*) FILTER (WHERE claim_token IS NOT NULL) AS claimed \
             FROM queue_messages \
             WHERE app_id = $1 \
             GROUP BY queue_name \
             ORDER BY queue_name",
        )
        .bind(app_id.into_inner())
        .fetch_all(&self.pool)
        .await?;
        Ok(rows
            .into_iter()
            .map(|r| {
                (
                    r.queue_name,
                    QueueStats {
                        total: u64::try_from(r.total).unwrap_or(0),
                        pending: u64::try_from(r.pending).unwrap_or(0),
                        claimed: u64::try_from(r.claimed).unwrap_or(0),
                    },
                )
            })
            .collect())
    }

    async fn dead_letter(
        &self,
        message_id: QueueMessageId,
        claim_token: Uuid,
        app_id: AppId,
        queue_name: &str,
        trigger_id: Option<TriggerId>,
        script_id: Option<picloud_shared::ScriptId>,
        attempt: u32,
        first_attempt_at: DateTime<Utc>,
        last_error: &str,
    ) -> Result<picloud_shared::DeadLetterId, QueueRepoError> {
        let mut tx = self.pool.begin().await?;
        // Pull the row inside the transaction; we need the original
        // payload for the dead_letters row. Filtered by claim_token so
        // a lost lease can't dead-letter a re-claimed message.
        let row: QueueRowMinimal = sqlx::query_as(
            "SELECT id, payload, enqueued_at FROM queue_messages \
             WHERE id = $1 AND claim_token = $2",
        )
        .bind(message_id.into_inner())
        .bind(claim_token)
        .fetch_one(&mut *tx)
        .await?;

        let payload = serde_json::json!({
            "source": "queue",
            "queue_name": queue_name,
            "message": row.payload,
            "enqueued_at": row.enqueued_at,
            "attempt": attempt,
            "message_id": row.id.to_string(),
        });

        let dl_id = picloud_shared::DeadLetterId::new();
        sqlx::query(
            "INSERT INTO dead_letters ( \
                id, app_id, original_event_id, source, op, trigger_id, script_id, \
                payload, attempt_count, first_attempt_at, last_attempt_at, last_error \
             ) VALUES ($1, $2, $3, 'queue', 'receive', $4, $5, $6, $7, $8, NOW(), $9)",
        )
        .bind(dl_id.into_inner())
        .bind(app_id.into_inner())
        .bind(row.id)
        .bind(trigger_id.map(TriggerId::into_inner))
        .bind(script_id.map(picloud_shared::ScriptId::into_inner))
        .bind(&payload)
        .bind(i32::try_from(attempt).unwrap_or(0))
        .bind(first_attempt_at)
        .bind(last_error)
        .execute(&mut *tx)
        .await?;

        sqlx::query("DELETE FROM queue_messages WHERE id = $1 AND claim_token = $2")
            .bind(message_id.into_inner())
            .bind(claim_token)
            .execute(&mut *tx)
            .await?;

        tx.commit().await?;
        Ok(dl_id)
    }
}

/// Bridge between this repo and the existing `DeadLetterRepo`. Not used
/// by `dead_letter()` directly (that path inlines its own INSERT for one-
/// transaction atomicity), but exposed for callers that already have a
/// `DeadLetterRepo` and want to keep one writer.
#[must_use]
pub fn build_new_dead_letter_from_claimed(
    msg: &ClaimedMessage,
    trigger_id: Option<TriggerId>,
    script_id: Option<picloud_shared::ScriptId>,
    last_error: String,
    first_attempt_at: DateTime<Utc>,
) -> NewDeadLetter {
    let payload = serde_json::json!({
        "source": "queue",
        "queue_name": msg.queue_name,
        "message": msg.payload,
        "enqueued_at": msg.enqueued_at,
        "attempt": msg.attempt,
        "message_id": msg.id.to_string(),
    });
    NewDeadLetter {
        app_id: msg.app_id,
        original_event_id: msg.id.into_inner(),
        source: "queue".into(),
        op: "receive".into(),
        trigger_id,
        script_id,
        payload,
        attempt_count: msg.attempt,
        first_attempt_at,
        last_attempt_at: Utc::now(),
        last_error,
    }
}

/// Surface to bridge DeadLetterRepoError → QueueRepoError when callers
/// route the dead-letter write through the standard repo instead of the
/// inline `dead_letter()` path.
impl From<DeadLetterRepoError> for QueueRepoError {
    fn from(e: DeadLetterRepoError) -> Self {
        Self::DeadLetter(e.to_string())
    }
}

#[derive(sqlx::FromRow)]
struct ClaimedRow {
    id: Uuid,
    app_id: Uuid,
    queue_name: String,
    payload: serde_json::Value,
    enqueued_at: DateTime<Utc>,
    attempt: i32,
    max_attempts: i32,
}

#[derive(sqlx::FromRow)]
struct QueueAggRow {
    queue_name: String,
    total: i64,
    pending: i64,
    claimed: i64,
}

#[derive(sqlx::FromRow)]
struct QueueRowMinimal {
    id: Uuid,
    payload: serde_json::Value,
    enqueued_at: DateTime<Utc>,
}