//! Approval queue. Requests are submitted by the manager (`RequestInitConfig`
//! / `RequestUpdateMetaInputs`), the config-PR webhook (`MergeConfigPr`), or
//! the operator (`Spawn`); the user approves/denies via the host admin CLI;
//! on approval the host runs the corresponding action.

use std::path::Path;
use std::sync::Mutex;

use anyhow::{Context, Result, bail};
use chrono::Utc;
use hive_sh4re::{Approval, ApprovalKind, ApprovalStatus};
use rusqlite::{Connection, OptionalExtension, params};

use crate::db::Migration;

const SCHEMA: &str = r"
CREATE TABLE IF NOT EXISTS approvals (
    id           INTEGER PRIMARY KEY AUTOINCREMENT,
    agent        TEXT    NOT NULL,
    commit_ref   TEXT    NOT NULL,
    requested_at INTEGER NOT NULL,
    status       TEXT    NOT NULL,
    resolved_at  INTEGER,
    note         TEXT
);
CREATE INDEX IF NOT EXISTS idx_approvals_pending
    ON approvals (id) WHERE status = 'pending';
";

/// Ordered schema migrations tracked in `schema_versions` (key `"approvals"`).
/// Each migration declares the column it adds, so a legacy DB (fully or
/// partially migrated before versioning) converges by skipping the migrations
/// whose column already exists. New columns go here as v5, v6, …
const MIGRATIONS: &[Migration] = &[
    // v1: `kind` (pre-Phase-8 dbs): legacy rows default to `apply_commit`.
    Migration {
        sql: "ALTER TABLE approvals ADD COLUMN \
     kind TEXT NOT NULL DEFAULT 'apply_commit'",
        adds_column: Some(("approvals", "kind")),
    },
    // v2: `description`: manager-supplied note on the dashboard card.
    Migration {
        sql: "ALTER TABLE approvals ADD COLUMN description TEXT",
        adds_column: Some(("approvals", "description")),
    },
    // v3: `fetched_sha`: canonical sha hive-c0re resolved at submit time.
    Migration {
        sql: "ALTER TABLE approvals ADD COLUMN fetched_sha TEXT",
        adds_column: Some(("approvals", "fetched_sha")),
    },
    // v4: `submitter`: authenticated agent that submitted the approval.
    // Legacy rows are NULL → callers fall back to the root agent.
    Migration {
        sql: "ALTER TABLE approvals ADD COLUMN submitter TEXT",
        adds_column: Some(("approvals", "submitter")),
    },
];

pub struct Approvals {
    conn: Mutex<Connection>,
}

impl Approvals {
    pub fn open(path: &Path) -> Result<Self> {
        let conn = crate::db::open(path, "approvals")?;
        conn.execute_batch(SCHEMA)
            .context("apply approvals schema")?;
        crate::db::apply_versioned_migrations(&conn, "approvals", MIGRATIONS)?;
        Ok(Self {
            conn: Mutex::new(conn),
        })
    }

    /// Insert a new pending approval row.  `fetched_sha` may be supplied
    /// when the sha is already known at submission time (e.g. `MergeConfigPr`
    /// fetches the PR head before inserting), making the insert + sha-set
    /// atomic.  Pass `None` when the kind carries no sha (e.g. `Spawn` /
    /// `InitConfig`).
    pub fn submit_kind(
        &self,
        agent: &str,
        kind: ApprovalKind,
        commit_ref: &str,
        description: Option<&str>,
        submitter: &str,
        fetched_sha: Option<&str>,
    ) -> Result<i64> {
        let conn = self.conn.lock().unwrap();
        conn.execute(
            "INSERT INTO approvals
               (agent, kind, commit_ref, requested_at, status, description, submitter,
                fetched_sha)
             VALUES (?1, ?2, ?3, ?4, 'pending', ?5, ?6, ?7)",
            params![
                agent,
                kind.as_str(),
                commit_ref,
                Utc::now().timestamp(),
                description,
                submitter,
                fetched_sha,
            ],
        )?;
        Ok(conn.last_insert_rowid())
    }

    /// The agent that submitted approval `id` (the authenticated socket
    /// caller at submit time). `None` for legacy rows predating the
    /// `submitter` column — callers route those to the root agent.
    ///
    /// # Errors
    ///
    /// Returns an error if the sqlite prepare/query fails. A missing row
    /// or a `NULL` submitter is not an error — both yield `Ok(None)`.
    pub fn submitter_of(&self, id: i64) -> Result<Option<String>> {
        let conn = self.conn.lock().unwrap();
        let submitter: Option<String> = conn
            .query_row(
                "SELECT submitter FROM approvals WHERE id = ?1",
                params![id],
                |row| row.get(0),
            )
            .optional()?
            .flatten();
        Ok(submitter)
    }

    /// Return the `(id, fetched_sha)` of the pending `merge_config_pr`
    /// approval for `(agent, pr_number)`, if one exists. Drives
    /// `submit_merge_config_pr`'s idempotency + PR-drift handling: same
    /// `fetched_sha` → no new request (the webhook + poll both call submit,
    /// so re-submits of an unchanged PR must be no-ops); a drifted head →
    /// cancel this stale row and queue a fresh approval.
    pub fn pending_merge_config_pr(
        &self,
        agent: &str,
        pr_number: u64,
    ) -> Result<Option<(i64, Option<String>)>> {
        let conn = self.conn.lock().unwrap();
        let row = conn
            .query_row(
                "SELECT id, fetched_sha FROM approvals \
                  WHERE agent = ?1 AND kind = 'merge_config_pr' \
                    AND commit_ref = ?2 AND status = 'pending' \
                  ORDER BY id DESC LIMIT 1",
                params![agent, pr_number.to_string()],
                |row| Ok((row.get(0)?, row.get(1)?)),
            )
            .optional()?;
        Ok(row)
    }

    /// Last `limit` resolved approvals (approved / denied / failed),
    /// newest-first. Drives the history tab on the dashboard.
    pub fn recent_resolved(&self, limit: u64) -> Result<Vec<Approval>> {
        let conn = self.conn.lock().unwrap();
        let mut stmt = conn.prepare(
            "SELECT id, agent, kind, commit_ref, requested_at, status, resolved_at, note, fetched_sha, description
               FROM approvals
              WHERE status IN ('approved', 'denied', 'failed', 'cancelled')
              ORDER BY resolved_at DESC, id DESC
              LIMIT ?1",
        )?;
        let rows = stmt.query_map([limit], row_to_approval)?;
        Ok(collect_lenient(rows))
    }

    pub fn pending(&self) -> Result<Vec<Approval>> {
        let conn = self.conn.lock().unwrap();
        let mut stmt = conn.prepare(
            "SELECT id, agent, kind, commit_ref, requested_at, status, resolved_at, note, fetched_sha, description
               FROM approvals
              WHERE status = 'pending'
              ORDER BY id ASC",
        )?;
        let rows = stmt.query_map([], row_to_approval)?;
        Ok(collect_lenient(rows))
    }

    pub fn get(&self, id: i64) -> Result<Option<Approval>> {
        let conn = self.conn.lock().unwrap();
        conn.query_row(
            "SELECT id, agent, kind, commit_ref, requested_at, status, resolved_at, note, fetched_sha, description
               FROM approvals WHERE id = ?1",
            params![id],
            row_to_approval,
        )
        .optional()
        .map_err(Into::into)
    }

    /// Mark pending -> approved (or fail if not pending). Returns the (now-updated)
    /// approval so the caller can run the action and pass the agent name.
    pub fn mark_approved(&self, id: i64) -> Result<Approval> {
        let conn = self.conn.lock().unwrap();
        let row: Option<ApprovalLookup> = conn
            .query_row(
                ApprovalLookup::SELECT,
                params![id],
                ApprovalLookup::from_row,
            )
            .optional()?;
        let Some(row) = row else {
            bail!("approval {id} not found");
        };
        if row.status != "pending" {
            bail!("approval {id} is {}, not pending", row.status);
        }
        let resolved_at = Utc::now().timestamp();
        conn.execute(
            "UPDATE approvals SET status = 'approved', resolved_at = ?1 WHERE id = ?2",
            params![resolved_at, id],
        )?;
        Ok(Approval {
            id,
            agent: row.agent,
            kind: kind_from_str(&row.kind)?,
            commit_ref: row.commit_ref,
            requested_at: hive_sh4re::wire_time::from_secs(row.requested_at),
            status: ApprovalStatus::Approved,
            resolved_at: Some(hive_sh4re::wire_time::from_secs(resolved_at)),
            note: None,
            fetched_sha: row.fetched_sha,
            description: row.description,
        })
    }

    pub fn mark_denied(&self, id: i64, note: Option<&str>) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        let affected = conn.execute(
            "UPDATE approvals SET status = 'denied', resolved_at = ?1, note = ?2
               WHERE id = ?3 AND status = 'pending'",
            params![Utc::now().timestamp(), note, id],
        )?;
        if affected == 0 {
            bail!("approval {id} not pending");
        }
        Ok(())
    }

    pub fn mark_failed(&self, id: i64, note: &str) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        conn.execute(
            "UPDATE approvals SET status = 'failed', resolved_at = ?1, note = ?2 WHERE id = ?3",
            params![Utc::now().timestamp(), note, id],
        )?;
        Ok(())
    }

    /// Withdraw a pending approval. Returns the now-updated
    /// row so the caller can emit `ApprovalResolved` with the right
    /// kind / agent / sha. Errors if the approval isn't pending — once
    /// it's approved/denied/failed/cancelled, the resolution is final.
    pub fn mark_cancelled(&self, id: i64, canceller: &str) -> Result<Approval> {
        let mut conn = self.conn.lock().unwrap();
        let tx = conn.transaction()?;
        let row: Option<ApprovalLookup> = tx
            .query_row(
                ApprovalLookup::SELECT,
                params![id],
                ApprovalLookup::from_row,
            )
            .optional()?;
        let Some(row) = row else {
            bail!("approval {id} not found");
        };
        if row.status != "pending" {
            bail!("approval {id} is {}, not pending", row.status);
        }
        let resolved_at = Utc::now().timestamp();
        let note = format!("cancelled by {canceller}");
        tx.execute(
            "UPDATE approvals SET status = 'cancelled', resolved_at = ?1, note = ?2 WHERE id = ?3",
            params![resolved_at, note, id],
        )?;
        tx.commit()?;
        Ok(Approval {
            id,
            agent: row.agent,
            kind: kind_from_str(&row.kind)?,
            commit_ref: row.commit_ref,
            requested_at: hive_sh4re::wire_time::from_secs(row.requested_at),
            status: ApprovalStatus::Cancelled,
            resolved_at: Some(hive_sh4re::wire_time::from_secs(resolved_at)),
            note: Some(note),
            fetched_sha: row.fetched_sha,
            description: row.description,
        })
    }

    /// Mark every pending approval for `agent` as failed (returns rows affected).
    /// Used by `destroy` to clear the queue of an agent that no longer exists.
    pub fn fail_pending_for_agent(&self, agent: &str, note: &str) -> Result<usize> {
        let conn = self.conn.lock().unwrap();
        let n = conn.execute(
            "UPDATE approvals SET status = 'failed', resolved_at = ?1, note = ?2
               WHERE agent = ?3 AND status = 'pending'",
            params![Utc::now().timestamp(), note, agent],
        )?;
        Ok(n)
    }
}

/// Columns needed to rebuild an [`Approval`] after a status transition,
/// shared by `mark_approved` / `mark_cancelled`. Replaces a 7-field
/// tuple that tripped `clippy::type_complexity` and was duplicated
/// across both callers (one suppressed the lint, the other aliased the
/// tuple) — one named projection + mapper now backs both.
struct ApprovalLookup {
    agent: hive_types::Ident,
    kind: String,
    commit_ref: String,
    requested_at: i64,
    status: String,
    fetched_sha: Option<String>,
    description: Option<String>,
}

impl ApprovalLookup {
    /// The single-row lookup by id (`?1`). Column order matches
    /// [`ApprovalLookup::from_row`].
    const SELECT: &str = "SELECT agent, kind, commit_ref, requested_at, status, fetched_sha, \
         description FROM approvals WHERE id = ?1";

    fn from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<Self> {
        let agent: String = row.get(0)?;
        let agent = hive_types::Ident::parse(&agent).map_err(|e| {
            rusqlite::Error::FromSqlConversionFailure(
                0,
                rusqlite::types::Type::Text,
                format!("invalid approval agent {agent:?}: {e}").into(),
            )
        })?;
        Ok(Self {
            agent,
            kind: row.get(1)?,
            commit_ref: row.get(2)?,
            requested_at: row.get(3)?,
            status: row.get(4)?,
            fetched_sha: row.get(5)?,
            description: row.get(6)?,
        })
    }
}

/// Collect approval rows, dropping any that fail to deserialize. A single
/// malformed / unknown-kind row must never blank the whole list:
/// `collect::<Result<Vec>>()` is all-or-nothing, so one bad row used to make
/// `pending()` / `recent_resolved()` error out wholesale — the dashboard then
/// rendered an empty approvals queue.
///
/// Drops are aggregated by error message and logged **once per call** rather
/// than one line per row: a batch of legacy unknown-kind rows (e.g. the
/// retired `apply_commit` approvals from the removed non-PR config flow) sit
/// `pending` forever and would otherwise flood the journal with an identical
/// warning on every dashboard render. Aggregating keeps the signal (how many,
/// which error) without the flood.
fn collect_lenient(rows: impl Iterator<Item = rusqlite::Result<Approval>>) -> Vec<Approval> {
    let mut out = Vec::new();
    let mut dropped: std::collections::HashMap<String, usize> = std::collections::HashMap::new();
    for r in rows {
        match r {
            Ok(a) => out.push(a),
            Err(e) => *dropped.entry(e.to_string()).or_default() += 1,
        }
    }
    if !dropped.is_empty() {
        let total: usize = dropped.values().sum();
        tracing::warn!(
            dropped = total,
            by_error = ?dropped,
            "skipped unparseable approval rows"
        );
    }
    out
}

fn row_to_approval(row: &rusqlite::Row<'_>) -> rusqlite::Result<Approval> {
    // Column order: id, agent, kind, commit_ref, requested_at, status, resolved_at, note, fetched_sha, description.
    let kind: String = row.get(2)?;
    let kind = match kind.as_str() {
        "spawn" => ApprovalKind::Spawn,
        "init_config" => ApprovalKind::InitConfig,
        "update_meta_inputs" => ApprovalKind::UpdateMetaInputs,
        "schedule_prompt" => ApprovalKind::SchedulePrompt,
        "merge_config_pr" => ApprovalKind::MergeConfigPr,
        other => {
            return Err(rusqlite::Error::FromSqlConversionFailure(
                2,
                rusqlite::types::Type::Text,
                format!("unknown approval kind '{other}'").into(),
            ));
        }
    };
    let status: String = row.get(5)?;
    let status = match status.as_str() {
        "pending" => ApprovalStatus::Pending,
        "approved" => ApprovalStatus::Approved,
        "denied" => ApprovalStatus::Denied,
        "failed" => ApprovalStatus::Failed,
        "cancelled" => ApprovalStatus::Cancelled,
        other => {
            return Err(rusqlite::Error::FromSqlConversionFailure(
                5,
                rusqlite::types::Type::Text,
                format!("unknown approval status '{other}'").into(),
            ));
        }
    };
    let agent: String = row.get(1)?;
    let agent = hive_types::Ident::parse(&agent).map_err(|e| {
        rusqlite::Error::FromSqlConversionFailure(
            1,
            rusqlite::types::Type::Text,
            format!("invalid approval agent {agent:?}: {e}").into(),
        )
    })?;
    Ok(Approval {
        id: row.get(0)?,
        agent,
        kind,
        commit_ref: row.get(3)?,
        requested_at: hive_sh4re::wire_time::from_secs(row.get(4)?),
        status,
        resolved_at: row
            .get::<_, Option<i64>>(6)?
            .map(hive_sh4re::wire_time::from_secs),
        note: row.get(7)?,
        fetched_sha: row.get(8)?,
        description: row.get(9)?,
    })
}

fn kind_from_str(s: &str) -> Result<ApprovalKind> {
    Ok(match s {
        "spawn" => ApprovalKind::Spawn,
        "init_config" => ApprovalKind::InitConfig,
        "update_meta_inputs" => ApprovalKind::UpdateMetaInputs,
        "schedule_prompt" => ApprovalKind::SchedulePrompt,
        "merge_config_pr" => ApprovalKind::MergeConfigPr,
        other => bail!("unknown approval kind '{other}'"),
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use hive_sh4re::ApprovalKind;

    fn open_temp() -> (tempfile::TempDir, std::path::PathBuf, Approvals) {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("approvals.sqlite");
        let db = Approvals::open(&path).expect("open approvals db");
        (dir, path, db)
    }

    #[test]
    fn init_config_approval_round_trips() {
        // Regression test: an `init_config` row used to fail
        // deserialization (row_to_approval matched only apply_commit +
        // spawn), erroring out the whole `pending()` query — every
        // approval then vanished from the dashboard.
        let (_dir, _path, db) = open_temp();
        let id = db
            .submit_kind(
                "bitburner",
                ApprovalKind::InitConfig,
                "",
                Some("scaffold"),
                "bitburner",
                None,
            )
            .expect("submit init_config");
        let pending = db
            .pending()
            .expect("pending() must not error on an init_config row");
        assert_eq!(pending.len(), 1);
        assert_eq!(pending[0].id, id);
        assert!(matches!(pending[0].kind, ApprovalKind::InitConfig));
    }

    #[test]
    fn mixed_kinds_all_listed() {
        let (_dir, _path, db) = open_temp();
        db.submit_kind(
            "a",
            ApprovalKind::MergeConfigPr,
            "deadbeef",
            None,
            "a",
            None,
        )
        .unwrap();
        db.submit_kind("b", ApprovalKind::Spawn, "", None, "b", None)
            .unwrap();
        db.submit_kind("c", ApprovalKind::InitConfig, "", None, "c", None)
            .unwrap();
        let pending = db.pending().expect("pending");
        assert_eq!(pending.len(), 3, "all three kinds must be visible");
    }

    #[test]
    fn mark_cancelled_transitions_pending_row() {
        // Manager withdraws a pending approval. Row leaves pending(),
        // shows up in recent_resolved() with the cancelled status + a
        // "cancelled by <who>" note.
        let (_dir, _path, db) = open_temp();
        let id = db
            .submit_kind(
                "bitburner",
                ApprovalKind::MergeConfigPr,
                "cafef00d",
                Some("test"),
                "bitburner",
                None,
            )
            .unwrap();
        let row = db.mark_cancelled(id, "manager").expect("cancel");
        assert_eq!(row.id, id);
        assert!(matches!(row.status, ApprovalStatus::Cancelled));
        assert_eq!(row.note.as_deref(), Some("cancelled by manager"));
        assert!(row.resolved_at.is_some());
        assert!(db.pending().unwrap().is_empty(), "row leaves pending");
        let resolved = db.recent_resolved(10).unwrap();
        assert_eq!(resolved.len(), 1);
        assert!(matches!(resolved[0].status, ApprovalStatus::Cancelled));
    }

    #[test]
    fn mark_cancelled_refuses_already_resolved_row() {
        // Once approved/denied/failed/cancelled the resolution is
        // final — re-cancelling errors instead of silently overwriting.
        let (_dir, _path, db) = open_temp();
        let id = db
            .submit_kind("a", ApprovalKind::Spawn, "deadbeef", None, "a", None)
            .unwrap();
        db.mark_cancelled(id, "manager").expect("first cancel");
        let err = db
            .mark_cancelled(id, "manager")
            .expect_err("second cancel must fail");
        assert!(err.to_string().contains("not pending"), "{err}");
    }

    #[test]
    fn unknown_kind_row_is_skipped_not_fatal() {
        // A single malformed / future-kind row must not blank the
        // whole list — collect_lenient skips it instead of failing.
        let (_dir, path, db) = open_temp();
        let good = db
            .submit_kind(
                "good",
                ApprovalKind::MergeConfigPr,
                "cafe",
                None,
                "good",
                None,
            )
            .unwrap();
        let raw = Connection::open(&path).unwrap();
        raw.execute(
            "INSERT INTO approvals (agent, kind, commit_ref, requested_at, status)
             VALUES ('weird', 'from_the_future', '', 0, 'pending')",
            [],
        )
        .unwrap();
        let pending = db
            .pending()
            .expect("pending() must survive an unparseable row");
        assert_eq!(pending.len(), 1);
        assert_eq!(pending[0].id, good);
    }

    #[test]
    fn submitter_round_trips_and_legacy_reads_none() {
        // A submitted approval records its submitter; a legacy row
        // (inserted without the column) reads back as None so callers
        // fall back to the root agent.
        let (_dir, path, db) = open_temp();
        let id = db
            .submit_kind(
                "child",
                ApprovalKind::MergeConfigPr,
                "cafe",
                None,
                "parent",
                None,
            )
            .unwrap();
        assert_eq!(db.submitter_of(id).unwrap().as_deref(), Some("parent"));

        let raw = Connection::open(&path).unwrap();
        raw.execute(
            "INSERT INTO approvals (agent, kind, commit_ref, requested_at, status)
             VALUES ('old', 'spawn', '', 0, 'pending')",
            [],
        )
        .unwrap();
        let legacy_id = raw.last_insert_rowid();
        assert_eq!(db.submitter_of(legacy_id).unwrap(), None);
    }

    #[test]
    fn fetched_sha_in_insert_is_readable_via_get() {
        // `submit_kind` with `Some(sha)` must store it atomically in the
        // INSERT — the `get()` row must reflect it without a separate
        // sha-set step. This is the MergeConfigPr path.
        let (_dir, _path, db) = open_temp();
        let sha = "abc1234567890abc1234567890abc1234567890ab";
        let id = db
            .submit_kind(
                "janet",
                ApprovalKind::MergeConfigPr,
                "42",
                None,
                "ruth",
                Some(sha),
            )
            .unwrap();
        let row = db.get(id).unwrap().expect("row must exist");
        assert_eq!(row.fetched_sha.as_deref(), Some(sha));
    }
}