use serde::{Deserialize, Serialize};
use sqlx::Row;

use crate::scheduler::Schedule;
use crate::storage::StorageError;

/// A scheduled job stored in the database.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ScheduledJob {
    pub id: String,
    pub name: String,
    /// JSON-serialized `Schedule` stored as TEXT in SQLite.
    pub schedule: Schedule,
    pub prompt: String,
    pub channel: String,
    pub chat_id: String,
    pub model: Option<String>,
    pub enabled: bool,
    pub delete_after_run: bool,
    pub next_run_at: i64,
    pub last_run_at: Option<i64>,
    pub last_status: Option<String>,
    pub last_error: Option<String>,
    pub created_at: i64,
    pub updated_at: i64,
}

/// A single execution record for a job.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct JobRun {
    pub id: i64,
    pub job_id: String,
    pub started_at: i64,
    pub finished_at: i64,
    pub status: String,
    pub output: Option<String>,
    pub error: Option<String>,
    pub duration_ms: i64,
}

impl crate::storage::Storage {
    /// Insert a new scheduled job.
    pub async fn add_scheduled_job(&self, job: &ScheduledJob) -> Result<(), StorageError> {
        let schedule_json = serialize_schedule(&job.schedule)?;
        sqlx::query(
            r#"
            INSERT INTO scheduled_jobs
                (id, name, schedule, prompt, channel, chat_id, model,
                 enabled, delete_after_run, next_run_at, last_run_at,
                 last_status, last_error, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            "#,
        )
        .bind(&job.id)
        .bind(&job.name)
        .bind(&schedule_json)
        .bind(&job.prompt)
        .bind(&job.channel)
        .bind(&job.chat_id)
        .bind(&job.model)
        .bind(job.enabled as i32)
        .bind(job.delete_after_run as i32)
        .bind(job.next_run_at)
        .bind(job.last_run_at)
        .bind(&job.last_status)
        .bind(&job.last_error)
        .bind(job.created_at)
        .bind(job.updated_at)
        .execute(self.pool())
        .await?;
        Ok(())
    }

    /// Fetch a single scheduled job by ID.
    pub async fn get_scheduled_job(&self, id: &str) -> Result<ScheduledJob, StorageError> {
        let row = sqlx::query("SELECT * FROM scheduled_jobs WHERE id = ?")
            .bind(id)
            .fetch_optional(self.pool())
            .await?
            .ok_or_else(|| StorageError::NotFound(format!("scheduled job {id}")))?;
        row_to_job(&row)
    }

    /// List all scheduled jobs, ordered by next_run_at ascending.
    pub async fn list_scheduled_jobs(&self) -> Result<Vec<ScheduledJob>, StorageError> {
        let rows = sqlx::query("SELECT * FROM scheduled_jobs ORDER BY next_run_at ASC")
            .fetch_all(self.pool())
            .await?;
        rows.iter().map(row_to_job).collect()
    }

    /// Delete a scheduled job (cascades to job_runs).
    pub async fn remove_scheduled_job(&self, id: &str) -> Result<(), StorageError> {
        sqlx::query("DELETE FROM scheduled_jobs WHERE id = ?")
            .bind(id)
            .execute(self.pool())
            .await?;
        Ok(())
    }

    /// Enable or disable a scheduled job.
    pub async fn set_scheduled_job_enabled(
        &self,
        id: &str,
        enabled: bool,
    ) -> Result<(), StorageError> {
        sqlx::query("UPDATE scheduled_jobs SET enabled = ?, updated_at = ? WHERE id = ?")
            .bind(enabled as i32)
            .bind(now_ms())
            .bind(id)
            .execute(self.pool())
            .await?;
        Ok(())
    }

    /// Update selective fields on a scheduled job.
    pub async fn update_scheduled_job(
        &self,
        id: &str,
        prompt: Option<String>,
        schedule: Option<Schedule>,
        channel: Option<String>,
        chat_id: Option<String>,
        model: Option<String>,
    ) -> Result<(), StorageError> {
        let now = now_ms();

        if let Some(p) = prompt {
            sqlx::query("UPDATE scheduled_jobs SET prompt = ?, updated_at = ? WHERE id = ?")
                .bind(&p)
                .bind(now)
                .bind(id)
                .execute(self.pool())
                .await?;
        }
        if let Some(s) = schedule {
            let json = serialize_schedule(&s)?;
            sqlx::query("UPDATE scheduled_jobs SET schedule = ?, updated_at = ? WHERE id = ?")
                .bind(&json)
                .bind(now)
                .bind(id)
                .execute(self.pool())
                .await?;
        }
        if let Some(c) = channel {
            sqlx::query("UPDATE scheduled_jobs SET channel = ?, updated_at = ? WHERE id = ?")
                .bind(&c)
                .bind(now)
                .bind(id)
                .execute(self.pool())
                .await?;
        }
        if let Some(c) = chat_id {
            sqlx::query("UPDATE scheduled_jobs SET chat_id = ?, updated_at = ? WHERE id = ?")
                .bind(&c)
                .bind(now)
                .bind(id)
                .execute(self.pool())
                .await?;
        }
        if let Some(m) = model {
            sqlx::query("UPDATE scheduled_jobs SET model = ?, updated_at = ? WHERE id = ?")
                .bind(&m)
                .bind(now)
                .bind(id)
                .execute(self.pool())
                .await?;
        }
        Ok(())
    }

    /// Update next_run_at and last_run_at for a job.
    pub async fn set_scheduled_job_next_run(
        &self,
        id: &str,
        next_run_at: i64,
    ) -> Result<(), StorageError> {
        let now = now_ms();
        sqlx::query(
            "UPDATE scheduled_jobs SET next_run_at = ?, last_run_at = ?, updated_at = ? WHERE id = ?",
        )
        .bind(next_run_at)
        .bind(now)
        .bind(now)
        .bind(id)
        .execute(self.pool())
        .await?;
        Ok(())
    }

    /// Atomically claim due jobs for one scheduler instance. A crashed worker's
    /// claims become eligible again after `lease_until`.
    pub async fn claim_due_scheduled_jobs(
        &self,
        now: i64,
        lease_until: i64,
        owner: &str,
        limit: usize,
    ) -> Result<Vec<ScheduledJob>, StorageError> {
        if limit == 0 {
            return Ok(Vec::new());
        }
        let rows = sqlx::query(
            r#"
            UPDATE scheduled_jobs
            SET locked_at = ?, lock_owner = ?, lease_until = ?, last_run_at = ?, updated_at = ?
            WHERE id IN (
                SELECT id FROM scheduled_jobs
                WHERE enabled = 1
                  AND next_run_at <= ?
                  AND (lease_until IS NULL OR lease_until <= ?)
                ORDER BY next_run_at ASC
                LIMIT ?
            )
              AND (lease_until IS NULL OR lease_until <= ?)
            RETURNING *
            "#,
        )
        .bind(now)
        .bind(owner)
        .bind(lease_until)
        .bind(now)
        .bind(now)
        .bind(now)
        .bind(now)
        .bind(limit as i64)
        .bind(now)
        .fetch_all(self.pool())
        .await?;
        rows.iter().map(row_to_job).collect()
    }

    /// Persist the run result, reschedule/disable the job, and release its
    /// lease in one transaction. The owner check prevents a stale worker from
    /// completing a claim that has already been recovered elsewhere.
    pub async fn complete_scheduled_job(
        &self,
        run: &JobRun,
        owner: &str,
        next_run_at: Option<i64>,
        disable: bool,
        delete: bool,
    ) -> Result<(), StorageError> {
        let mut tx = self.pool().begin().await?;

        if delete {
            let result = sqlx::query("DELETE FROM scheduled_jobs WHERE id = ? AND lock_owner = ?")
                .bind(&run.job_id)
                .bind(owner)
                .execute(&mut *tx)
                .await?;
            if result.rows_affected() != 1 {
                return Err(StorageError::Conflict(format!(
                    "scheduled job lease lost before delete: {}",
                    run.job_id
                )));
            }
        } else {
            sqlx::query(
                r#"
                INSERT INTO job_runs (job_id, started_at, finished_at, status, output, error, duration_ms)
                VALUES (?, ?, ?, ?, ?, ?, ?)
                "#,
            )
            .bind(&run.job_id)
            .bind(run.started_at)
            .bind(run.finished_at)
            .bind(&run.status)
            .bind(&run.output)
            .bind(&run.error)
            .bind(run.duration_ms)
            .execute(&mut *tx)
            .await?;

            let result = sqlx::query(
                r#"
                UPDATE scheduled_jobs
                SET next_run_at = COALESCE(?, next_run_at),
                    enabled = CASE WHEN ? THEN 0 ELSE enabled END,
                    last_status = ?, last_error = ?,
                    locked_at = NULL, lock_owner = NULL, lease_until = NULL,
                    updated_at = ?
                WHERE id = ? AND lock_owner = ?
                "#,
            )
            .bind(next_run_at)
            .bind(disable)
            .bind(&run.status)
            .bind(&run.error)
            .bind(run.finished_at)
            .bind(&run.job_id)
            .bind(owner)
            .execute(&mut *tx)
            .await?;
            if result.rows_affected() != 1 {
                return Err(StorageError::Conflict(format!(
                    "scheduled job lease lost before completion: {}",
                    run.job_id
                )));
            }
        }

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

    pub async fn release_scheduled_job_lease(
        &self,
        job_id: &str,
        owner: &str,
    ) -> Result<(), StorageError> {
        sqlx::query(
            "UPDATE scheduled_jobs SET locked_at = NULL, lock_owner = NULL, lease_until = NULL WHERE id = ? AND lock_owner = ?",
        )
        .bind(job_id)
        .bind(owner)
        .execute(self.pool())
        .await?;
        Ok(())
    }

    /// List recent runs for a job, newest first.
    pub async fn list_scheduled_job_runs(
        &self,
        job_id: &str,
        limit: usize,
    ) -> Result<Vec<JobRun>, StorageError> {
        let rows = sqlx::query(
            "SELECT * FROM job_runs WHERE job_id = ? ORDER BY finished_at DESC LIMIT ?",
        )
        .bind(job_id)
        .bind(limit as i64)
        .fetch_all(self.pool())
        .await?;
        rows.iter()
            .map(|r| {
                Ok(JobRun {
                    id: r.try_get("id")?,
                    job_id: r.try_get("job_id")?,
                    started_at: r.try_get("started_at")?,
                    finished_at: r.try_get("finished_at")?,
                    status: r.try_get("status")?,
                    output: r.try_get("output")?,
                    error: r.try_get("error")?,
                    duration_ms: r.try_get("duration_ms")?,
                })
            })
            .collect()
    }

    /// Delete disabled jobs whose updated_at is before `before`.
    pub async fn cleanup_disabled_scheduled_jobs(&self, before: i64) -> Result<(), StorageError> {
        sqlx::query("DELETE FROM scheduled_jobs WHERE enabled = 0 AND updated_at < ?")
            .bind(before)
            .execute(self.pool())
            .await?;
        Ok(())
    }
}

fn now_ms() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as i64
}

fn serialize_schedule(schedule: &Schedule) -> Result<String, StorageError> {
    serde_json::to_string(schedule).map_err(|error| StorageError::Serialization(error.to_string()))
}

fn row_to_job(row: &sqlx::sqlite::SqliteRow) -> Result<ScheduledJob, StorageError> {
    let schedule_json: String = row.try_get("schedule")?;
    let schedule: Schedule = serde_json::from_str(&schedule_json)
        .map_err(|error| StorageError::Serialization(error.to_string()))?;
    Ok(ScheduledJob {
        id: row.try_get("id")?,
        name: row.try_get("name")?,
        schedule,
        prompt: row.try_get("prompt")?,
        channel: row.try_get("channel")?,
        chat_id: row.try_get("chat_id")?,
        model: row.try_get("model")?,
        enabled: row.try_get::<i32, _>("enabled")? != 0,
        delete_after_run: row.try_get::<i32, _>("delete_after_run")? != 0,
        next_run_at: row.try_get("next_run_at")?,
        last_run_at: row.try_get("last_run_at")?,
        last_status: row.try_get("last_status")?,
        last_error: row.try_get("last_error")?,
        created_at: row.try_get("created_at")?,
        updated_at: row.try_get("updated_at")?,
    })
}

#[cfg(test)]
mod tests {
    use super::ScheduledJob;
    use crate::scheduler::Schedule;
    use crate::storage::Storage;
    use sqlx::SqlitePool;

    fn now() -> i64 {
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_millis() as i64
    }

    async fn setup_storage() -> Storage {
        let pool = SqlitePool::connect("sqlite::memory:").await.unwrap();
        let storage = Storage { pool };
        Storage::init_scheduler_schema(storage.pool())
            .await
            .unwrap();
        storage
    }

    #[tokio::test]
    async fn test_init_creates_tables() {
        let storage = setup_storage().await;
        let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM scheduled_jobs")
            .fetch_one(storage.pool())
            .await
            .unwrap();
        assert_eq!(row.0, 0);
    }

    #[tokio::test]
    async fn test_add_and_get_job() {
        let storage = setup_storage().await;
        let t = now();
        let job = ScheduledJob {
            id: "job-1".into(),
            name: "test job".into(),
            schedule: Schedule::Every { every_ms: 3600000 },
            prompt: "say hello".into(),
            channel: "cli_chat".into(),
            chat_id: "conn-1".into(),
            model: None,
            enabled: true,
            delete_after_run: false,
            next_run_at: t + 3600000,
            last_run_at: None,
            last_status: None,
            last_error: None,
            created_at: t,
            updated_at: t,
        };
        storage.add_scheduled_job(&job).await.unwrap();
        let got = storage.get_scheduled_job("job-1").await.unwrap();
        assert_eq!(got.id, "job-1");
        assert_eq!(got.name, "test job");
        assert_eq!(got.prompt, "say hello");
    }

    #[tokio::test]
    async fn test_list_jobs() {
        let storage = setup_storage().await;
        let t = now();
        for i in 0..3 {
            let job = ScheduledJob {
                id: format!("job-{}", i),
                name: format!("job {}", i),
                schedule: Schedule::Every { every_ms: 3600000 },
                prompt: "ping".into(),
                channel: "cli_chat".into(),
                chat_id: "conn-1".into(),
                model: None,
                enabled: true,
                delete_after_run: false,
                next_run_at: t + 1000,
                last_run_at: None,
                last_status: None,
                last_error: None,
                created_at: t,
                updated_at: t,
            };
            storage.add_scheduled_job(&job).await.unwrap();
        }
        let jobs = storage.list_scheduled_jobs().await.unwrap();
        assert_eq!(jobs.len(), 3);
    }

    #[tokio::test]
    async fn test_remove_job() {
        let storage = setup_storage().await;
        let t = now();
        let job = ScheduledJob {
            id: "job-rm".into(),
            name: "remove me".into(),
            schedule: Schedule::Every { every_ms: 1000 },
            prompt: "hi".into(),
            channel: "cli_chat".into(),
            chat_id: "c".into(),
            model: None,
            enabled: true,
            delete_after_run: false,
            next_run_at: t,
            last_run_at: None,
            last_status: None,
            last_error: None,
            created_at: t,
            updated_at: t,
        };
        storage.add_scheduled_job(&job).await.unwrap();
        storage.remove_scheduled_job("job-rm").await.unwrap();
        let result = storage.get_scheduled_job("job-rm").await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_set_enabled() {
        let storage = setup_storage().await;
        let t = now();
        let job = ScheduledJob {
            id: "job-toggle".into(),
            name: "toggle".into(),
            schedule: Schedule::Every { every_ms: 1000 },
            prompt: "hi".into(),
            channel: "cli_chat".into(),
            chat_id: "c".into(),
            model: None,
            enabled: true,
            delete_after_run: false,
            next_run_at: t,
            last_run_at: None,
            last_status: None,
            last_error: None,
            created_at: t,
            updated_at: t,
        };
        storage.add_scheduled_job(&job).await.unwrap();
        storage
            .set_scheduled_job_enabled("job-toggle", false)
            .await
            .unwrap();
        let got = storage.get_scheduled_job("job-toggle").await.unwrap();
        assert!(!got.enabled);
    }

    #[tokio::test]
    async fn test_update_job() {
        let storage = setup_storage().await;
        let t = now();
        let job = ScheduledJob {
            id: "job-update".into(),
            name: "old name".into(),
            schedule: Schedule::Every { every_ms: 1000 },
            prompt: "old prompt".into(),
            channel: "feishu".into(),
            chat_id: "oc_1".into(),
            model: None,
            enabled: true,
            delete_after_run: false,
            next_run_at: t,
            last_run_at: None,
            last_status: None,
            last_error: None,
            created_at: t,
            updated_at: t,
        };
        storage.add_scheduled_job(&job).await.unwrap();
        storage
            .update_scheduled_job(
                "job-update",
                Some("new prompt".into()),
                Some(Schedule::Every { every_ms: 60000 }),
                None,
                None,
                None,
            )
            .await
            .unwrap();
        let got = storage.get_scheduled_job("job-update").await.unwrap();
        assert_eq!(got.prompt, "new prompt");
    }

    #[tokio::test]
    async fn claim_is_exclusive_until_lease_expires() {
        let storage = setup_storage().await;
        let t = now();
        let job = ScheduledJob {
            id: "leased-job".into(),
            name: "leased".into(),
            schedule: Schedule::Every { every_ms: 1000 },
            prompt: "run".into(),
            channel: "cli_chat".into(),
            chat_id: "c".into(),
            model: None,
            enabled: true,
            delete_after_run: false,
            next_run_at: t,
            last_run_at: None,
            last_status: None,
            last_error: None,
            created_at: t,
            updated_at: t,
        };
        storage.add_scheduled_job(&job).await.unwrap();

        let first = storage
            .claim_due_scheduled_jobs(t, t + 100, "owner-1", 1)
            .await
            .unwrap();
        let duplicate = storage
            .claim_due_scheduled_jobs(t, t + 100, "owner-2", 1)
            .await
            .unwrap();
        let recovered = storage
            .claim_due_scheduled_jobs(t + 101, t + 201, "owner-2", 1)
            .await
            .unwrap();

        assert_eq!(first.len(), 1);
        assert!(duplicate.is_empty());
        assert_eq!(recovered.len(), 1);
    }

    #[tokio::test]
    async fn completion_is_atomic_and_releases_lease() {
        let storage = setup_storage().await;
        let t = now();
        let job = ScheduledJob {
            id: "complete-job".into(),
            name: "complete".into(),
            schedule: Schedule::Every { every_ms: 1000 },
            prompt: "run".into(),
            channel: "cli_chat".into(),
            chat_id: "c".into(),
            model: None,
            enabled: true,
            delete_after_run: false,
            next_run_at: t,
            last_run_at: None,
            last_status: None,
            last_error: None,
            created_at: t,
            updated_at: t,
        };
        storage.add_scheduled_job(&job).await.unwrap();
        storage
            .claim_due_scheduled_jobs(t, t + 1000, "owner", 1)
            .await
            .unwrap();
        let run = super::JobRun {
            id: 0,
            job_id: job.id.clone(),
            started_at: t,
            finished_at: t + 10,
            status: "ok".into(),
            output: Some("done".into()),
            error: None,
            duration_ms: 10,
        };
        storage
            .complete_scheduled_job(&run, "owner", Some(t + 2000), false, false)
            .await
            .unwrap();

        let completed = storage.get_scheduled_job(&job.id).await.unwrap();
        let runs = storage.list_scheduled_job_runs(&job.id, 10).await.unwrap();
        let lease: (Option<String>, Option<i64>) =
            sqlx::query_as("SELECT lock_owner, lease_until FROM scheduled_jobs WHERE id = ?")
                .bind(&job.id)
                .fetch_one(storage.pool())
                .await
                .unwrap();
        assert_eq!(completed.next_run_at, t + 2000);
        assert_eq!(completed.last_status.as_deref(), Some("ok"));
        assert_eq!(runs.len(), 1);
        assert_eq!(lease, (None, None));
    }
}