//! The durable work queue (oxygenhub#276, docs/roles.md): items any worker
//! process claims for a lease, retried with backoff, dead after the last
//! attempt. On PostgreSQL a claim skips rows another worker has locked
//! (`FOR UPDATE SKIP LOCKED`); on SQLite the one writer serialises claims.

use super::{now_ms, Pool};
use sqlx::FromRow;

/// Attempts before an item is kept as dead.
pub const MAX_ATTEMPTS: i64 = 8;

#[derive(Clone, Debug, FromRow)]
pub struct Item {
    pub id: i64,
    pub kind: String,
    pub payload: String,
    pub dedupe: String,
    pub run_after: i64,
    pub attempts: i64,
    pub last_error: String,
    pub claimed_by: String,
    pub claimed_until: i64,
    pub dead: bool,
    pub created_ms: i64,
}

/// Queue `kind` with its payload, due after `delay_ms`. With a `dedupe`
/// key, nothing is added while a live item of that kind and key waits.
pub async fn enqueue(pool: &Pool, kind: &str, payload: &serde_json::Value, dedupe: &str, delay_ms: i64) -> sqlx::Result<bool> {
    let now = now_ms();
    let mut tx = super::begin_write(pool).await?;
    if !dedupe.is_empty() {
        let live: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM work_queue WHERE kind = $1 AND dedupe = $2 AND dead = $3")
            .bind(kind)
            .bind(dedupe)
            .bind(false)
            .fetch_one(&mut *tx)
            .await?;
        if live > 0 {
            return Ok(false);
        }
    }
    sqlx::query("INSERT INTO work_queue (kind, payload, dedupe, run_after, created_ms) VALUES ($1, $2, $3, $4, $5)")
        .bind(kind)
        .bind(payload.to_string())
        .bind(dedupe)
        .bind(now + delay_ms.max(0))
        .bind(now)
        .execute(&mut *tx)
        .await?;
    tx.commit().await?;
    Ok(true)
}

/// Claim up to `limit` due items for `worker`, each for `lease_ms`: an item
/// whose worker died is due again when its lease ends.
pub async fn claim(pool: &Pool, worker: &str, limit: i64, lease_ms: i64) -> sqlx::Result<Vec<Item>> {
    let now = now_ms();
    #[cfg(feature = "postgres")]
    let pick = "SELECT id FROM work_queue WHERE dead = $3 AND run_after <= $4 AND claimed_until <= $4 ORDER BY run_after, id LIMIT $5 FOR UPDATE SKIP LOCKED";
    #[cfg(not(feature = "postgres"))]
    let pick = "SELECT id FROM work_queue WHERE dead = $3 AND run_after <= $4 AND claimed_until <= $4 ORDER BY run_after, id LIMIT $5";
    let sql =
        format!("UPDATE work_queue SET claimed_by = $1, claimed_until = $2, attempts = attempts + 1 WHERE id IN ({pick}) RETURNING *");
    let mut tx = pool.begin().await?;
    let items: Vec<Item> =
        sqlx::query_as(&sql).bind(worker).bind(now + lease_ms).bind(false).bind(now).bind(limit).fetch_all(&mut *tx).await?;
    tx.commit().await?;
    Ok(items)
}

/// The item is done: it leaves the queue.
pub async fn done(pool: &Pool, id: i64) -> sqlx::Result<()> {
    sqlx::query("DELETE FROM work_queue WHERE id = $1").bind(id).execute(pool).await?;
    Ok(())
}

/// The item failed: due again after a backoff that doubles each attempt
/// (30 s, 1 min, 2 min, … at most an hour), or dead after the last.
pub async fn failed(pool: &Pool, item: &Item, error: &str) -> sqlx::Result<()> {
    let dead = item.attempts >= MAX_ATTEMPTS;
    let backoff = (30_000i64 << (item.attempts.clamp(1, 8) - 1)).min(3_600_000);
    sqlx::query("UPDATE work_queue SET last_error = $1, dead = $2, run_after = $3, claimed_by = '', claimed_until = 0 WHERE id = $4")
        .bind(error)
        .bind(dead)
        .bind(now_ms() + backoff)
        .bind(item.id)
        .execute(pool)
        .await?;
    Ok(())
}

/// Items waiting (claimed or not) and items dead.
pub async fn counts(pool: &Pool) -> sqlx::Result<(i64, i64)> {
    let waiting: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM work_queue WHERE dead = $1").bind(false).fetch_one(pool).await?;
    let dead: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM work_queue WHERE dead = $1").bind(true).fetch_one(pool).await?;
    Ok((waiting, dead))
}