use std::collections::HashMap;

use rusqlite::Connection;

use crate::core::error::{LoreError, Result};
use crate::core::time::now_ms;

#[derive(Debug)]
pub struct PendingJob {
    pub id: i64,
    pub project_id: i64,
    pub entity_type: String,
    pub entity_iid: i64,
    pub entity_local_id: i64,
    pub job_type: String,
    pub payload_json: Option<String>,
    pub attempts: i32,
}

pub fn enqueue_job(
    conn: &Connection,
    project_id: i64,
    entity_type: &str,
    entity_iid: i64,
    entity_local_id: i64,
    job_type: &str,
    payload_json: Option<&str>,
) -> Result<bool> {
    let now = now_ms();
    let changes = conn.execute(
        "INSERT OR IGNORE INTO pending_dependent_fetches
         (project_id, entity_type, entity_iid, entity_local_id, job_type, payload_json, enqueued_at)
         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
        rusqlite::params![
            project_id,
            entity_type,
            entity_iid,
            entity_local_id,
            job_type,
            payload_json,
            now
        ],
    )?;

    Ok(changes > 0)
}

pub fn claim_jobs(
    conn: &Connection,
    job_type: &str,
    project_id: i64,
    batch_size: usize,
) -> Result<Vec<PendingJob>> {
    if batch_size == 0 {
        return Ok(Vec::new());
    }

    let now = now_ms();

    let mut stmt = conn.prepare_cached(
        "UPDATE pending_dependent_fetches
         SET locked_at = ?1
         WHERE id IN (
             SELECT id FROM pending_dependent_fetches
             WHERE job_type = ?2
               AND project_id = ?4
               AND locked_at IS NULL
               AND (next_retry_at IS NULL OR next_retry_at <= ?1)
             ORDER BY enqueued_at ASC
             LIMIT ?3
         )
         RETURNING id, project_id, entity_type, entity_iid, entity_local_id,
                   job_type, payload_json, attempts",
    )?;

    let jobs = stmt
        .query_map(
            rusqlite::params![now, job_type, batch_size as i64, project_id],
            |row| {
                Ok(PendingJob {
                    id: row.get(0)?,
                    project_id: row.get(1)?,
                    entity_type: row.get(2)?,
                    entity_iid: row.get(3)?,
                    entity_local_id: row.get(4)?,
                    job_type: row.get(5)?,
                    payload_json: row.get(6)?,
                    attempts: row.get(7)?,
                })
            },
        )?
        .collect::<std::result::Result<Vec<_>, _>>()?;

    Ok(jobs)
}

pub fn complete_job(conn: &Connection, job_id: i64) -> Result<()> {
    conn.execute(
        "DELETE FROM pending_dependent_fetches WHERE id = ?1",
        rusqlite::params![job_id],
    )?;

    Ok(())
}

/// Same DELETE as `complete_job`, but on an existing transaction so the caller
/// can bundle it atomically with a watermark update.
pub fn complete_job_tx(tx: &rusqlite::Transaction<'_>, job_id: i64) -> Result<()> {
    tx.execute(
        "DELETE FROM pending_dependent_fetches WHERE id = ?1",
        rusqlite::params![job_id],
    )?;

    Ok(())
}

/// Release all currently locked jobs (set `locked_at = NULL`).
/// Used during graceful shutdown so the next sync doesn't wait for stale locks.
pub fn release_all_locked_jobs(conn: &Connection) -> Result<usize> {
    let changes = conn.execute(
        "UPDATE pending_dependent_fetches SET locked_at = NULL WHERE locked_at IS NOT NULL",
        [],
    )?;

    Ok(changes)
}

pub fn fail_job(conn: &Connection, job_id: i64, error: &str) -> Result<()> {
    let now = now_ms();

    let changes = conn.execute(
        "UPDATE pending_dependent_fetches
         SET attempts = attempts + 1,
             next_retry_at = ?1 + MIN(30000 * (1 << MIN(attempts, 4)), 480000),
             locked_at = NULL,
             last_error = ?2
         WHERE id = ?3",
        rusqlite::params![now, error, job_id],
    )?;

    if changes == 0 {
        return Err(LoreError::Other(
            "fail_job: job not found (may have been reclaimed or completed)".into(),
        ));
    }

    Ok(())
}

pub fn reclaim_stale_locks(conn: &Connection, stale_threshold_minutes: u32) -> Result<usize> {
    let threshold_ms = now_ms() - (i64::from(stale_threshold_minutes) * 60 * 1000);

    let changes = conn.execute(
        "UPDATE pending_dependent_fetches SET locked_at = NULL WHERE locked_at < ?1",
        rusqlite::params![threshold_ms],
    )?;

    Ok(changes)
}

pub fn count_pending_jobs(
    conn: &Connection,
    project_id: Option<i64>,
) -> Result<HashMap<String, usize>> {
    let mut counts = HashMap::new();

    match project_id {
        Some(pid) => {
            let mut stmt = conn.prepare_cached(
                "SELECT job_type, COUNT(*) FROM pending_dependent_fetches \
                 WHERE project_id = ?1 GROUP BY job_type",
            )?;
            let rows = stmt.query_map(rusqlite::params![pid], |row| {
                let job_type: String = row.get(0)?;
                let count: i64 = row.get(1)?;
                Ok((job_type, count as usize))
            })?;
            for row in rows {
                let (job_type, count) = row?;
                counts.insert(job_type, count);
            }
        }
        None => {
            let mut stmt = conn.prepare_cached(
                "SELECT job_type, COUNT(*) FROM pending_dependent_fetches GROUP BY job_type",
            )?;
            let rows = stmt.query_map([], |row| {
                let job_type: String = row.get(0)?;
                let count: i64 = row.get(1)?;
                Ok((job_type, count as usize))
            })?;
            for row in rows {
                let (job_type, count) = row?;
                counts.insert(job_type, count);
            }
        }
    }

    Ok(counts)
}

pub fn count_claimable_jobs(conn: &Connection, project_id: i64) -> Result<HashMap<String, usize>> {
    let now = now_ms();
    let mut counts = HashMap::new();

    let mut stmt = conn.prepare_cached(
        "SELECT job_type, COUNT(*) FROM pending_dependent_fetches \
         WHERE project_id = ?1 \
           AND locked_at IS NULL \
           AND (next_retry_at IS NULL OR next_retry_at <= ?2) \
         GROUP BY job_type",
    )?;
    let rows = stmt.query_map(rusqlite::params![project_id, now], |row| {
        let job_type: String = row.get(0)?;
        let count: i64 = row.get(1)?;
        Ok((job_type, count as usize))
    })?;
    for row in rows {
        let (job_type, count) = row?;
        counts.insert(job_type, count);
    }

    Ok(counts)
}

#[cfg(test)]
mod tests {
    use std::path::Path;

    use super::*;
    use crate::core::db::{create_connection, run_migrations};

    fn setup_db_with_job() -> (Connection, i64) {
        let conn = create_connection(Path::new(":memory:")).unwrap();
        run_migrations(&conn).unwrap();

        conn.execute(
            "INSERT INTO projects (gitlab_project_id, path_with_namespace, web_url) \
             VALUES (1, 'group/repo', 'https://gitlab.com/group/repo')",
            [],
        )
        .unwrap();

        let project_id: i64 = conn
            .query_row("SELECT id FROM projects ORDER BY id LIMIT 1", [], |row| {
                row.get(0)
            })
            .unwrap();

        enqueue_job(&conn, project_id, "issue", 42, 100, "resource_events", None).unwrap();

        let job_id: i64 = conn
            .query_row(
                "SELECT id FROM pending_dependent_fetches ORDER BY id LIMIT 1",
                [],
                |row| row.get(0),
            )
            .unwrap();

        (conn, job_id)
    }

    #[test]
    fn complete_job_tx_commits() {
        let (conn, job_id) = setup_db_with_job();

        let tx = conn.unchecked_transaction().unwrap();
        complete_job_tx(&tx, job_id).unwrap();
        tx.commit().unwrap();

        let count: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM pending_dependent_fetches WHERE id = ?1",
                [job_id],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(count, 0, "job should be deleted after commit");
    }

    #[test]
    fn complete_job_tx_rollback() {
        let (conn, job_id) = setup_db_with_job();

        {
            let tx = conn.unchecked_transaction().unwrap();
            complete_job_tx(&tx, job_id).unwrap();
            // drop tx without commit = rollback
        }

        let count: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM pending_dependent_fetches WHERE id = ?1",
                [job_id],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(count, 1, "job should survive dropped (rolled-back) tx");
    }

    #[test]
    fn release_all_locked_jobs_clears_locks() {
        let (conn, _job_id) = setup_db_with_job();

        let project_id: i64 = conn
            .query_row("SELECT id FROM projects ORDER BY id LIMIT 1", [], |row| {
                row.get(0)
            })
            .unwrap();
        let jobs = claim_jobs(&conn, "resource_events", project_id, 10).unwrap();
        assert_eq!(jobs.len(), 1);

        let locked: bool = conn
            .query_row(
                "SELECT locked_at IS NOT NULL FROM pending_dependent_fetches WHERE id = ?1",
                [jobs[0].id],
                |row| row.get(0),
            )
            .unwrap();
        assert!(locked, "job should be locked after claim");

        let released = release_all_locked_jobs(&conn).unwrap();
        assert_eq!(released, 1);

        let locked: bool = conn
            .query_row(
                "SELECT locked_at IS NOT NULL FROM pending_dependent_fetches WHERE id = ?1",
                [jobs[0].id],
                |row| row.get(0),
            )
            .unwrap();
        assert!(!locked, "job should be unlocked after release_all");
    }
}
