//! 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))

}