//! Scheduled prompts: persistent sqlite queue of `(fire_at, targets,
//! body)` rows that the worker fans out as broker `Message`s to each
//! target's inbox at fire time. Recurring schedules carry
//! `interval_seconds` and re-arm `next_fire_at` on delivery;
//! one-shots are reaped.
//!
//! Schema + retention: `docs/persistence.md::/var/lib/hyperhive/broker.sqlite`
//! (the `scheduled_prompts` / `scheduled_prompt_targets` table bullets).
//! Submit paths (operator-direct vs `ApprovalKind::SchedulePrompt`,
//! plus why even agent-self schedules go through approval):
//! `docs/approvals.md::Scheduled prompts (submit paths)`.
//! Catch-up clamp on resume + per-target tombstoning:
//! `docs/approvals.md::Scheduled prompt worker (catch-up clamp)`.

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

use anyhow::{Context, Result, bail};
use rusqlite::{Connection, OptionalExtension, params};
use serde::{Deserialize, Serialize};

const SCHEMA: &str = r"
CREATE TABLE IF NOT EXISTS scheduled_prompts (
    id                INTEGER PRIMARY KEY AUTOINCREMENT,
    owner             TEXT    NOT NULL,
    body              TEXT    NOT NULL,
    interval_seconds  INTEGER,
    next_fire_at_unix INTEGER NOT NULL,
    created_at_unix   INTEGER NOT NULL,
    source            TEXT    NOT NULL,
    cancelled_at_unix INTEGER,
    description       TEXT
);
CREATE INDEX IF NOT EXISTS idx_scheduled_due
    ON scheduled_prompts (next_fire_at_unix)
    WHERE cancelled_at_unix IS NULL;

CREATE TABLE IF NOT EXISTS scheduled_prompt_targets (
    schedule_id        INTEGER NOT NULL,
    target             TEXT    NOT NULL,
    cancelled_at_unix  INTEGER,
    last_fired_at_unix INTEGER,
    last_result        TEXT,
    PRIMARY KEY (schedule_id, target),
    FOREIGN KEY (schedule_id) REFERENCES scheduled_prompts(id) ON DELETE CASCADE
);
";

/// One scheduled-prompt row + its current target set. Returned by
/// `list` / `get`; the per-target last-result blob is suitable for
/// rendering on the dashboard without a second query.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Schedule {
    pub id: i64,
    /// `"operator"` or an agent name — drives cancel-permission
    /// checks. For approval-sourced rows this is the requesting
    /// agent (the `Approval.agent` at submit time).
    pub owner: String,
    pub body: String,
    /// `None` = one-shot, deleted after first fire.
    /// `Some(n)` = recurring every `n` seconds.
    pub interval_seconds: Option<u64>,
    pub next_fire_at_unix: i64,
    pub created_at_unix: i64,
    pub source: ScheduleSource,
    /// Set when the *entire* schedule was cancelled (all targets
    /// flipped, or operator cancel-all). Worker reaps these on the
    /// next pass.
    pub cancelled_at_unix: Option<i64>,
    pub description: Option<String>,
    pub targets: Vec<ScheduleTarget>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ScheduleTarget {
    pub target: String,
    pub cancelled_at_unix: Option<i64>,
    pub last_fired_at_unix: Option<i64>,
    pub last_result: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum ScheduleSource {
    Operator,
    Approval { id: i64 },
}

impl ScheduleSource {
    fn to_db_string(&self) -> String {
        match self {
            ScheduleSource::Operator => "operator".to_owned(),
            ScheduleSource::Approval { id } => format!("approval:{id}"),
        }
    }

    fn from_db_string(s: &str) -> Self {
        if let Some(rest) = s.strip_prefix("approval:")
            && let Ok(id) = rest.parse::<i64>()
        {
            return ScheduleSource::Approval { id };
        }
        // Unknown / "operator" / corrupt → operator (best the row
        // can do without an unparseable-source variant).
        ScheduleSource::Operator
    }
}

/// Submission payload — everything the caller knows at insert time.
/// Used by both the operator-direct path and the approval-flow
/// path; the latter sets `source = Approval { id }`.
#[derive(Debug, Clone)]
pub struct NewSchedule {
    pub owner: String,
    pub targets: Vec<String>,
    pub body: String,
    pub first_fire_at_unix: i64,
    pub interval_seconds: Option<u64>,
    pub description: Option<String>,
    pub source: ScheduleSource,
}

/// Partial-update payload for `ScheduledPrompts::update`.
/// Every field is `Option<_>`; `None` keeps the existing value.
/// The doubly-wrapped `Option<Option<u64>>` on `interval_seconds`
/// is intentional: outer `None` = "don't touch", outer
/// `Some(None)` = "set to NULL (toggle to one-shot)", outer
/// `Some(Some(n))` = "set to n seconds." Same encoding as JSON
/// "missing key" vs "explicit null" — the dashboard surface
/// preserves the distinction.
///
/// Target add/remove: `targets_add` / `targets_remove`
/// run inside the same transaction as the scalar field updates so
/// "save my changes" is atomic. Remove delegates to the same
/// cancel-targets path used by `cancel_targets` (tombstoning, preserves
/// audit, auto-cancels parent when no active targets remain). Add uses
/// `INSERT OR REPLACE` so re-adding a previously-cancelled target
/// drops the tombstone and starts fresh (operator intent on re-add =
/// "this target is active again"; prior history was already visible
/// at cancel time).
#[derive(Debug, Clone, Default)]
#[allow(
    clippy::option_option,
    reason = "double-Option carries three-state PATCH semantics: outer None = \
              leave alone, Some(None) = clear, Some(Some(v)) = set. \
              collapsing to a single Option would lose the 'clear' state"
)]
pub struct UpdateSchedule {
    pub body: Option<String>,
    pub description: Option<Option<String>>,
    pub interval_seconds: Option<Option<u64>>,
    pub next_fire_at_unix: Option<i64>,
    pub targets_add: Option<Vec<String>>,
    pub targets_remove: Option<Vec<String>>,
}

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

impl ScheduledPrompts {
    pub fn open(path: &Path) -> Result<Self> {
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent).with_context(|| {
                format!("create scheduled_prompts db parent {}", parent.display())
            })?;
        }
        let conn = Connection::open(path)
            .with_context(|| format!("open scheduled_prompts db {}", path.display()))?;
        // Required for ON DELETE CASCADE to actually fire — sqlite
        // ships with FKs disabled per connection by default.
        conn.execute_batch("PRAGMA foreign_keys = ON;")
            .context("enable foreign keys")?;
        conn.execute_batch(SCHEMA)
            .context("apply scheduled_prompts schema")?;
        Ok(Self {
            conn: Mutex::new(conn),
        })
    }

    /// Insert a new schedule. Returns the new id. Empty `targets` is
    /// rejected — a schedule with no recipients would silently
    /// never fan out, masking caller bugs.
    pub fn submit(&self, new: &NewSchedule) -> Result<i64> {
        if new.targets.is_empty() {
            bail!("schedule must have at least one target");
        }
        let mut conn = self.conn.lock().unwrap();
        let tx = conn.transaction()?;
        tx.execute(
            "INSERT INTO scheduled_prompts
                (owner, body, interval_seconds, next_fire_at_unix,
                 created_at_unix, source, description)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
            params![
                &new.owner,
                &new.body,
                new.interval_seconds.map(i64::try_from).and_then(Result::ok),
                new.first_fire_at_unix,
                now_unix(),
                new.source.to_db_string(),
                &new.description,
            ],
        )?;
        let id = tx.last_insert_rowid();
        for target in &new.targets {
            tx.execute(
                "INSERT INTO scheduled_prompt_targets (schedule_id, target)
                 VALUES (?1, ?2)",
                params![id, target],
            )?;
        }
        tx.commit()?;
        Ok(id)
    }

    /// Fetch a single schedule by id (with its target rows).
    /// `Ok(None)` for a non-existent / already-reaped id.
    pub fn get(&self, id: i64) -> Result<Option<Schedule>> {
        let conn = self.conn.lock().unwrap();
        let row = conn
            .query_row(
                "SELECT id, owner, body, interval_seconds, next_fire_at_unix,
                        created_at_unix, source, cancelled_at_unix, description
                   FROM scheduled_prompts WHERE id = ?1",
                params![id],
                row_to_schedule_header,
            )
            .optional()?;
        let Some(mut s) = row else {
            return Ok(None);
        };
        s.targets = load_targets(&conn, id)?;
        Ok(Some(s))
    }

    /// Every active (non-globally-cancelled) schedule in insert
    /// order. Used by the dashboard list view + the cancel-auth
    /// check (the latter only needs the header but `list()` is the
    /// shared hot path).
    pub fn list(&self) -> Result<Vec<Schedule>> {
        let conn = self.conn.lock().unwrap();
        let mut stmt = conn.prepare(
            "SELECT id, owner, body, interval_seconds, next_fire_at_unix,
                    created_at_unix, source, cancelled_at_unix, description
               FROM scheduled_prompts
              ORDER BY next_fire_at_unix ASC, id ASC",
        )?;
        let rows = stmt.query_map([], row_to_schedule_header)?;
        let mut out = Vec::new();
        for row in rows {
            let mut s = row?;
            s.targets = load_targets(&conn, s.id)?;
            out.push(s);
        }
        Ok(out)
    }

    /// Pop the set of active rows that are due (`next_fire_at_unix <= now`)
    /// up to `limit`. Read-only — the worker calls `mark_fired`
    /// after each successful fan-out so the rows reappear on the
    /// next tick when re-armed.
    pub fn due(&self, now: i64, limit: u64) -> Result<Vec<Schedule>> {
        let conn = self.conn.lock().unwrap();
        let mut stmt = conn.prepare(
            "SELECT id, owner, body, interval_seconds, next_fire_at_unix,
                    created_at_unix, source, cancelled_at_unix, description
               FROM scheduled_prompts
              WHERE cancelled_at_unix IS NULL AND next_fire_at_unix <= ?1
              ORDER BY next_fire_at_unix ASC, id ASC
              LIMIT ?2",
        )?;
        let rows = stmt.query_map(params![now, limit], row_to_schedule_header)?;
        let mut out = Vec::new();
        for row in rows {
            let mut s = row?;
            s.targets = load_targets(&conn, s.id)?;
            out.push(s);
        }
        Ok(out)
    }

    /// Record a per-target fan-out result. `result` is "ok" or a
    /// short error string; surfaces on the dashboard last-result
    /// column. No-op for a target row that's already cancelled.
    pub fn record_target_result(
        &self,
        schedule_id: i64,
        target: &str,
        fired_at_unix: i64,
        result: &str,
    ) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        conn.execute(
            "UPDATE scheduled_prompt_targets
                SET last_fired_at_unix = ?1, last_result = ?2
              WHERE schedule_id = ?3 AND target = ?4 AND cancelled_at_unix IS NULL",
            params![fired_at_unix, result, schedule_id, target],
        )?;
        Ok(())
    }

    /// Advance a recurring schedule's `next_fire_at` to the smallest
    /// multiple-of-interval > `from`. Returns the count of skipped
    /// cycles (≥ 0); the worker stamps that into the per-row
    /// `last_result` so operators see "caught up from N missed".
    ///
    /// For one-shots (`interval_seconds IS NULL`) this is a no-op
    /// at the SQL level; callers should `delete` them after fan-out
    /// instead.
    pub fn rearm(&self, id: i64, from_unix: i64) -> Result<u64> {
        let conn = self.conn.lock().unwrap();
        let row: Option<(Option<i64>, i64)> = conn
            .query_row(
                "SELECT interval_seconds, next_fire_at_unix
                   FROM scheduled_prompts WHERE id = ?1",
                params![id],
                |row| Ok((row.get(0)?, row.get(1)?)),
            )
            .optional()?;
        let Some((interval, current_next)) = row else {
            return Ok(0);
        };
        let Some(interval) = interval.filter(|&i| i > 0) else {
            return Ok(0);
        };
        // Smallest multiple-of-interval > `from_unix`. Catch-up
        // semantics: when from_unix > current_next, every step in
        // between gets counted as a skipped cycle.
        let mut next = current_next + interval;
        let mut skipped: u64 = 0;
        while next <= from_unix {
            next += interval;
            skipped += 1;
        }
        conn.execute(
            "UPDATE scheduled_prompts SET next_fire_at_unix = ?1 WHERE id = ?2",
            params![next, id],
        )?;
        Ok(skipped)
    }

    /// Partial-update an existing schedule's mutable fields.
    /// Every scalar field is `Option<_>`; `None` means "leave the
    /// existing value alone", `Some(_)` means "set it to this."
    /// Refuses cancelled rows (no point editing a tombstone —
    /// operator can just submit a new schedule). Refuses
    /// `interval_seconds = Some(0)` — same rule as submit-time
    /// validation.
    ///
    /// Targets are mutable via `targets_remove` + `targets_add`.
    /// Both are processed in the same
    /// transaction, with **removes before adds** so a single PATCH
    /// can swap a target without ever leaving the schedule
    /// target-less mid-tx. Remove writes a tombstone via
    /// `COALESCE(cancelled_at_unix, ?1)` so it's idempotent. Add
    /// uses `INSERT OR REPLACE` so re-adding a previously-removed
    /// target drops the tombstone and resets per-target history
    /// (fresh start — option C from the iris design thread). If
    /// the post-tx active-target count is zero the parent schedule
    /// is auto-cancelled — the worker would otherwise spin forever
    /// firing nothing.
    pub fn update(&self, id: i64, patch: UpdateSchedule) -> Result<()> {
        // Pre-flight: row must exist and not be cancelled.
        let mut conn = self.conn.lock().unwrap();
        let cancelled: Option<i64> = conn
            .query_row(
                "SELECT cancelled_at_unix FROM scheduled_prompts WHERE id = ?1",
                params![id],
                |row| row.get(0),
            )
            .optional()?
            .ok_or_else(|| anyhow::anyhow!("schedule {id} not found"))?;
        if cancelled.is_some() {
            bail!("schedule {id} is cancelled — submit a new one");
        }
        if let Some(Some(0)) = patch.interval_seconds {
            bail!("interval_seconds must be > 0 (use None for one-shot)");
        }
        // Single transaction so a partial failure can't leave a
        // half-updated row.
        let tx = conn.transaction()?;
        if let Some(body) = patch.body {
            tx.execute(
                "UPDATE scheduled_prompts SET body = ?1 WHERE id = ?2",
                params![body, id],
            )?;
        }
        if let Some(description) = patch.description {
            tx.execute(
                "UPDATE scheduled_prompts SET description = ?1 WHERE id = ?2",
                params![description, id],
            )?;
        }
        if let Some(interval) = patch.interval_seconds {
            // `Some(Some(n))` = set to n. `Some(None)` = clear (one-shot).
            // `None` is filtered out by the outer `if let`.
            let value: Option<i64> = interval.map(i64::try_from).and_then(Result::ok);
            tx.execute(
                "UPDATE scheduled_prompts SET interval_seconds = ?1 WHERE id = ?2",
                params![value, id],
            )?;
        }
        if let Some(next) = patch.next_fire_at_unix {
            tx.execute(
                "UPDATE scheduled_prompts SET next_fire_at_unix = ?1 WHERE id = ?2",
                params![next, id],
            )?;
        }
        // Target removals first, then adds. Order matters: an
        // operator who sends `{add: ["alice"], remove: ["alice"]}`
        // probably typo'd, but treating add-after-remove as the
        // surviving intent means a clean re-arm rather than an
        // orphaned tombstone. Plus removing-then-adding is the
        // natural way to "restart history" on one target without
        // a separate flow.
        let now = now_unix();
        if let Some(remove) = patch.targets_remove.as_deref() {
            for target in remove {
                tx.execute(
                    "UPDATE scheduled_prompt_targets
                        SET cancelled_at_unix = COALESCE(cancelled_at_unix, ?1)
                      WHERE schedule_id = ?2 AND target = ?3",
                    params![now, id, target],
                )?;
            }
        }
        if let Some(add) = patch.targets_add.as_deref() {
            for target in add {
                // INSERT OR REPLACE drops any prior row (cancelled
                // tombstone or otherwise) and inserts a fresh one
                // with empty history. Operator intent on re-add =
                // "this target is active again, fresh start."
                tx.execute(
                    "INSERT OR REPLACE INTO scheduled_prompt_targets
                        (schedule_id, target, cancelled_at_unix,
                         last_fired_at_unix, last_result)
                     VALUES (?1, ?2, NULL, NULL, NULL)",
                    params![id, target],
                )?;
            }
        }
        // If the remove set drained the last active target (and
        // the add set didn't bring any back), auto-cancel the
        // parent so the worker stops scanning it. Mirrors
        // `cancel_targets`'s behaviour for the same edge case.
        if patch.targets_remove.is_some() || patch.targets_add.is_some() {
            let active_targets: i64 = tx.query_row(
                "SELECT COUNT(*) FROM scheduled_prompt_targets
                  WHERE schedule_id = ?1 AND cancelled_at_unix IS NULL",
                params![id],
                |row| row.get(0),
            )?;
            if active_targets == 0 {
                tx.execute(
                    "UPDATE scheduled_prompts
                        SET cancelled_at_unix = COALESCE(cancelled_at_unix, ?1)
                      WHERE id = ?2",
                    params![now, id],
                )?;
            }
        }
        tx.commit()?;
        Ok(())
    }

    /// Delete a one-shot row after its single fire. Cascades the
    /// target rows via the FK constraint. Idempotent on a missing
    /// id.
    pub fn delete(&self, id: i64) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        conn.execute("DELETE FROM scheduled_prompts WHERE id = ?1", params![id])?;
        Ok(())
    }

    /// Cancel the entire schedule (all targets flipped + the parent
    /// row marked cancelled). Idempotent; safe to call on an
    /// already-cancelled row.
    pub fn cancel_all(&self, id: i64) -> Result<()> {
        let mut conn = self.conn.lock().unwrap();
        let now = now_unix();
        let tx = conn.transaction()?;
        tx.execute(
            "UPDATE scheduled_prompts
                SET cancelled_at_unix = COALESCE(cancelled_at_unix, ?1)
              WHERE id = ?2",
            params![now, id],
        )?;
        tx.execute(
            "UPDATE scheduled_prompt_targets
                SET cancelled_at_unix = COALESCE(cancelled_at_unix, ?1)
              WHERE schedule_id = ?2",
            params![now, id],
        )?;
        tx.commit()?;
        Ok(())
    }

    /// Cancel a subset of a schedule's targets. When the last
    /// active target is cancelled, the parent row is auto-cancelled
    /// too (so the worker reaps it). Unknown targets in the list
    /// are silently skipped.
    pub fn cancel_targets(&self, id: i64, targets: &[String]) -> Result<()> {
        let mut conn = self.conn.lock().unwrap();
        let now = now_unix();
        let tx = conn.transaction()?;
        for target in targets {
            tx.execute(
                "UPDATE scheduled_prompt_targets
                    SET cancelled_at_unix = COALESCE(cancelled_at_unix, ?1)
                  WHERE schedule_id = ?2 AND target = ?3",
                params![now, id, target],
            )?;
        }
        // If every target on this schedule is now cancelled, flip
        // the parent so the worker stops scanning it.
        let active_targets: i64 = tx.query_row(
            "SELECT COUNT(*) FROM scheduled_prompt_targets
              WHERE schedule_id = ?1 AND cancelled_at_unix IS NULL",
            params![id],
            |row| row.get(0),
        )?;
        if active_targets == 0 {
            tx.execute(
                "UPDATE scheduled_prompts
                    SET cancelled_at_unix = COALESCE(cancelled_at_unix, ?1)
                  WHERE id = ?2",
                params![now, id],
            )?;
        }
        tx.commit()?;
        Ok(())
    }

    /// Reap cancelled rows older than `older_than_unix`. Returns
    /// the number of rows deleted. Called from the worker on each
    /// tick so cancellations clear out of the dashboard without an
    /// extra periodic vacuum task.
    pub fn reap_cancelled(&self, older_than_unix: i64) -> Result<usize> {
        let conn = self.conn.lock().unwrap();
        let n = conn.execute(
            "DELETE FROM scheduled_prompts
              WHERE cancelled_at_unix IS NOT NULL
                AND cancelled_at_unix <= ?1",
            params![older_than_unix],
        )?;
        Ok(n)
    }
}

fn row_to_schedule_header(row: &rusqlite::Row) -> rusqlite::Result<Schedule> {
    let interval: Option<i64> = row.get(3)?;
    let source_str: String = row.get(6)?;
    Ok(Schedule {
        id: row.get(0)?,
        owner: row.get(1)?,
        body: row.get(2)?,
        interval_seconds: interval.and_then(|i| u64::try_from(i).ok()),
        next_fire_at_unix: row.get(4)?,
        created_at_unix: row.get(5)?,
        source: ScheduleSource::from_db_string(&source_str),
        cancelled_at_unix: row.get(7)?,
        description: row.get(8)?,
        targets: Vec::new(),
    })
}

fn load_targets(conn: &Connection, schedule_id: i64) -> Result<Vec<ScheduleTarget>> {
    let mut stmt = conn.prepare(
        "SELECT target, cancelled_at_unix, last_fired_at_unix, last_result
           FROM scheduled_prompt_targets
          WHERE schedule_id = ?1
          ORDER BY target ASC",
    )?;
    let rows = stmt.query_map(params![schedule_id], |row| {
        Ok(ScheduleTarget {
            target: row.get(0)?,
            cancelled_at_unix: row.get(1)?,
            last_fired_at_unix: row.get(2)?,
            last_result: row.get(3)?,
        })
    })?;
    let mut out = Vec::new();
    for row in rows {
        out.push(row?);
    }
    Ok(out)
}

fn now_unix() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .ok()
        .and_then(|d| i64::try_from(d.as_secs()).ok())
        .unwrap_or(0)
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::TempDir;

    fn open() -> (TempDir, ScheduledPrompts) {
        let dir = TempDir::new().expect("tempdir");
        let path = dir.path().join("schedules.sqlite");
        let db = ScheduledPrompts::open(&path).expect("open");
        (dir, db)
    }

    fn submit_one_shot(db: &ScheduledPrompts, fire_at: i64, targets: &[&str]) -> i64 {
        db.submit(&NewSchedule {
            owner: "operator".into(),
            targets: targets.iter().map(|t| (*t).to_owned()).collect(),
            body: "wake".into(),
            first_fire_at_unix: fire_at,
            interval_seconds: None,
            description: None,
            source: ScheduleSource::Operator,
        })
        .expect("submit")
    }

    #[test]
    fn submit_and_get_round_trips_targets() {
        let (_dir, db) = open();
        let id = submit_one_shot(&db, 100, &["alice", "bob"]);
        let s = db.get(id).expect("get").expect("present");
        assert_eq!(s.id, id);
        assert_eq!(s.targets.len(), 2);
        assert_eq!(s.targets[0].target, "alice");
        assert_eq!(s.targets[1].target, "bob");
        assert!(s.targets.iter().all(|t| t.cancelled_at_unix.is_none()));
    }

    #[test]
    fn submit_rejects_empty_targets() {
        let (_dir, db) = open();
        let err = db
            .submit(&NewSchedule {
                owner: "operator".into(),
                targets: Vec::new(),
                body: "wake".into(),
                first_fire_at_unix: 100,
                interval_seconds: None,
                description: None,
                source: ScheduleSource::Operator,
            })
            .unwrap_err();
        assert!(format!("{err:#}").contains("at least one target"));
    }

    #[test]
    fn due_returns_only_past_active_rows() {
        let (_dir, db) = open();
        let _future = submit_one_shot(&db, 1000, &["alice"]);
        let past = submit_one_shot(&db, 50, &["alice"]);
        let cancelled_past = submit_one_shot(&db, 50, &["bob"]);
        db.cancel_all(cancelled_past).expect("cancel");
        let due = db.due(100, 10).expect("due");
        assert_eq!(due.len(), 1);
        assert_eq!(due[0].id, past);
    }

    #[test]
    fn rearm_handles_catch_up() {
        let (_dir, db) = open();
        // Recurring every 60s, last fire at t=100.
        let id = db
            .submit(&NewSchedule {
                owner: "operator".into(),
                targets: vec!["alice".into()],
                body: "wake".into(),
                first_fire_at_unix: 100,
                interval_seconds: Some(60),
                description: None,
                source: ScheduleSource::Operator,
            })
            .expect("submit");
        // Worker comes back at t=400 — 5 missed cycles (160, 220,
        // 280, 340, 400) → next should be 460, skipped = 5.
        let skipped = db.rearm(id, 400).expect("rearm");
        assert_eq!(skipped, 5);
        let s = db.get(id).expect("get").expect("present");
        assert_eq!(s.next_fire_at_unix, 460);
    }

    #[test]
    fn rearm_advances_one_step_when_caught_up() {
        let (_dir, db) = open();
        let id = db
            .submit(&NewSchedule {
                owner: "operator".into(),
                targets: vec!["alice".into()],
                body: "wake".into(),
                first_fire_at_unix: 100,
                interval_seconds: Some(60),
                description: None,
                source: ScheduleSource::Operator,
            })
            .expect("submit");
        // Worker fires right at t=100 — single advance to t=160.
        let skipped = db.rearm(id, 100).expect("rearm");
        assert_eq!(skipped, 0);
        let s = db.get(id).expect("get").expect("present");
        assert_eq!(s.next_fire_at_unix, 160);
    }

    #[test]
    fn rearm_is_a_no_op_for_one_shots() {
        let (_dir, db) = open();
        let id = submit_one_shot(&db, 100, &["alice"]);
        let skipped = db.rearm(id, 1000).expect("rearm");
        assert_eq!(skipped, 0);
        // next_fire_at unchanged — one-shots are reaped via delete().
        let s = db.get(id).expect("get").expect("present");
        assert_eq!(s.next_fire_at_unix, 100);
    }

    #[test]
    fn cancel_targets_auto_cancels_parent_when_last_drops() {
        let (_dir, db) = open();
        let id = submit_one_shot(&db, 100, &["alice", "bob"]);
        db.cancel_targets(id, &["alice".to_owned()])
            .expect("cancel");
        let s = db.get(id).expect("get").expect("present");
        // Parent still active (bob remains).
        assert!(s.cancelled_at_unix.is_none());
        db.cancel_targets(id, &["bob".to_owned()]).expect("cancel");
        let s = db.get(id).expect("get").expect("present");
        // Parent auto-cancels once every target is gone.
        assert!(s.cancelled_at_unix.is_some());
        assert!(s.targets.iter().all(|t| t.cancelled_at_unix.is_some()));
    }

    #[test]
    fn cancel_all_flips_parent_and_targets() {
        let (_dir, db) = open();
        let id = submit_one_shot(&db, 100, &["alice", "bob"]);
        db.cancel_all(id).expect("cancel");
        let s = db.get(id).expect("get").expect("present");
        assert!(s.cancelled_at_unix.is_some());
        assert!(s.targets.iter().all(|t| t.cancelled_at_unix.is_some()));
    }

    #[test]
    fn delete_cascades_targets() {
        let (_dir, db) = open();
        let id = submit_one_shot(&db, 100, &["alice", "bob"]);
        db.delete(id).expect("delete");
        assert!(db.get(id).expect("get").is_none());
    }

    #[test]
    fn record_target_result_skips_cancelled_targets() {
        let (_dir, db) = open();
        let id = submit_one_shot(&db, 100, &["alice", "bob"]);
        db.cancel_targets(id, &["alice".to_owned()])
            .expect("cancel");
        db.record_target_result(id, "alice", 200, "ok")
            .expect("record alice");
        db.record_target_result(id, "bob", 200, "ok")
            .expect("record bob");
        let s = db.get(id).expect("get").expect("present");
        let alice = s.targets.iter().find(|t| t.target == "alice").unwrap();
        let bob = s.targets.iter().find(|t| t.target == "bob").unwrap();
        // Cancelled targets do NOT get last-result writes.
        assert!(alice.last_fired_at_unix.is_none());
        assert!(alice.last_result.is_none());
        assert_eq!(bob.last_fired_at_unix, Some(200));
        assert_eq!(bob.last_result.as_deref(), Some("ok"));
    }

    #[test]
    fn update_partial_only_touches_set_fields() {
        let (_dir, db) = open();
        let id = db
            .submit(&NewSchedule {
                owner: "operator".into(),
                targets: vec!["alice".into()],
                body: "old body".into(),
                first_fire_at_unix: 100,
                interval_seconds: Some(60),
                description: Some("old desc".into()),
                source: ScheduleSource::Operator,
            })
            .expect("submit");
        // Only change body — everything else must stay.
        db.update(
            id,
            UpdateSchedule {
                body: Some("new body".into()),
                ..Default::default()
            },
        )
        .expect("update body");
        let s = db.get(id).expect("get").expect("present");
        assert_eq!(s.body, "new body");
        assert_eq!(s.description.as_deref(), Some("old desc"));
        assert_eq!(s.interval_seconds, Some(60));
        assert_eq!(s.next_fire_at_unix, 100);
    }

    #[test]
    fn update_interval_toggle_recurring_to_one_shot() {
        let (_dir, db) = open();
        let id = db
            .submit(&NewSchedule {
                owner: "operator".into(),
                targets: vec!["alice".into()],
                body: "x".into(),
                first_fire_at_unix: 100,
                interval_seconds: Some(60),
                description: None,
                source: ScheduleSource::Operator,
            })
            .expect("submit");
        db.update(
            id,
            UpdateSchedule {
                interval_seconds: Some(None),
                ..Default::default()
            },
        )
        .expect("update");
        let s = db.get(id).expect("get").expect("present");
        assert_eq!(s.interval_seconds, None);
    }

    #[test]
    fn update_interval_toggle_one_shot_to_recurring() {
        let (_dir, db) = open();
        let id = submit_one_shot(&db, 100, &["alice"]);
        db.update(
            id,
            UpdateSchedule {
                interval_seconds: Some(Some(3600)),
                ..Default::default()
            },
        )
        .expect("update");
        let s = db.get(id).expect("get").expect("present");
        assert_eq!(s.interval_seconds, Some(3600));
    }

    #[test]
    fn update_rejects_cancelled_row() {
        let (_dir, db) = open();
        let id = submit_one_shot(&db, 100, &["alice"]);
        db.cancel_all(id).expect("cancel");
        let err = db
            .update(
                id,
                UpdateSchedule {
                    body: Some("nope".into()),
                    ..Default::default()
                },
            )
            .unwrap_err();
        assert!(format!("{err:#}").contains("cancelled"));
    }

    #[test]
    fn update_rejects_zero_interval() {
        let (_dir, db) = open();
        let id = submit_one_shot(&db, 100, &["alice"]);
        let err = db
            .update(
                id,
                UpdateSchedule {
                    interval_seconds: Some(Some(0)),
                    ..Default::default()
                },
            )
            .unwrap_err();
        assert!(format!("{err:#}").contains("interval_seconds"));
    }

    #[test]
    fn update_clears_description() {
        let (_dir, db) = open();
        let id = db
            .submit(&NewSchedule {
                owner: "operator".into(),
                targets: vec!["alice".into()],
                body: "x".into(),
                first_fire_at_unix: 100,
                interval_seconds: None,
                description: Some("first".into()),
                source: ScheduleSource::Operator,
            })
            .expect("submit");
        db.update(
            id,
            UpdateSchedule {
                description: Some(None),
                ..Default::default()
            },
        )
        .expect("update");
        let s = db.get(id).expect("get").expect("present");
        assert!(s.description.is_none());
    }

    #[test]
    fn update_adds_new_targets() {
        let (_dir, db) = open();
        let id = submit_one_shot(&db, 100, &["alice"]);
        db.update(
            id,
            UpdateSchedule {
                targets_add: Some(vec!["bob".into(), "carol".into()]),
                ..Default::default()
            },
        )
        .expect("update");
        let s = db.get(id).expect("get").expect("present");
        let names: Vec<_> = s.targets.iter().map(|t| t.target.as_str()).collect();
        assert!(names.contains(&"alice"));
        assert!(names.contains(&"bob"));
        assert!(names.contains(&"carol"));
    }

    #[test]
    fn update_removes_targets_via_cancel_tombstone() {
        let (_dir, db) = open();
        let id = submit_one_shot(&db, 100, &["alice", "bob"]);
        db.update(
            id,
            UpdateSchedule {
                targets_remove: Some(vec!["alice".into()]),
                ..Default::default()
            },
        )
        .expect("update");
        let s = db.get(id).expect("get").expect("present");
        let alice = s.targets.iter().find(|t| t.target == "alice").unwrap();
        let bob = s.targets.iter().find(|t| t.target == "bob").unwrap();
        // alice tombstoned, bob still active, parent still active.
        assert!(alice.cancelled_at_unix.is_some());
        assert!(bob.cancelled_at_unix.is_none());
        assert!(s.cancelled_at_unix.is_none());
    }

    #[test]
    fn update_removing_last_target_auto_cancels_parent() {
        let (_dir, db) = open();
        let id = submit_one_shot(&db, 100, &["alice"]);
        db.update(
            id,
            UpdateSchedule {
                targets_remove: Some(vec!["alice".into()]),
                ..Default::default()
            },
        )
        .expect("update");
        let s = db.get(id).expect("get").expect("present");
        assert!(s.cancelled_at_unix.is_some());
    }

    #[test]
    fn update_re_add_resurrects_cancelled_target_fresh() {
        let (_dir, db) = open();
        let id = submit_one_shot(&db, 100, &["alice", "bob"]);
        // Record some history on alice, then cancel her.
        db.record_target_result(id, "alice", 50, "ok")
            .expect("record");
        db.update(
            id,
            UpdateSchedule {
                targets_remove: Some(vec!["alice".into()]),
                ..Default::default()
            },
        )
        .expect("cancel alice");
        // Re-add alice — should drop the tombstone + prior history.
        db.update(
            id,
            UpdateSchedule {
                targets_add: Some(vec!["alice".into()]),
                ..Default::default()
            },
        )
        .expect("re-add alice");
        let s = db.get(id).expect("get").expect("present");
        let alice = s.targets.iter().find(|t| t.target == "alice").unwrap();
        assert!(alice.cancelled_at_unix.is_none(), "fresh row, no tombstone");
        assert!(alice.last_fired_at_unix.is_none(), "history reset");
        assert!(alice.last_result.is_none(), "history reset");
    }

    #[test]
    fn update_add_and_remove_in_one_transaction() {
        let (_dir, db) = open();
        let id = submit_one_shot(&db, 100, &["alice"]);
        db.update(
            id,
            UpdateSchedule {
                targets_remove: Some(vec!["alice".into()]),
                targets_add: Some(vec!["bob".into()]),
                ..Default::default()
            },
        )
        .expect("update");
        let s = db.get(id).expect("get").expect("present");
        let alice = s.targets.iter().find(|t| t.target == "alice").unwrap();
        let bob = s.targets.iter().find(|t| t.target == "bob").unwrap();
        assert!(alice.cancelled_at_unix.is_some());
        assert!(bob.cancelled_at_unix.is_none());
        // Bob arrived AFTER alice was tombstoned in the same tx —
        // active-target count > 0, so parent stays alive.
        assert!(s.cancelled_at_unix.is_none());
    }

    #[test]
    fn approval_source_round_trips() {
        let (_dir, db) = open();
        let id = db
            .submit(&NewSchedule {
                owner: "manager".into(),
                targets: vec!["alice".into()],
                body: "wake".into(),
                first_fire_at_unix: 100,
                interval_seconds: None,
                description: None,
                source: ScheduleSource::Approval { id: 42 },
            })
            .expect("submit");
        let s = db.get(id).expect("get").expect("present");
        assert_eq!(s.source, ScheduleSource::Approval { id: 42 });
    }

    #[test]
    fn reap_cancelled_removes_old_rows_only() {
        let (_dir, db) = open();
        let stale = submit_one_shot(&db, 100, &["alice"]);
        let recent = submit_one_shot(&db, 100, &["bob"]);
        db.cancel_all(stale).expect("cancel stale");
        // Manually backdate the stale row.
        {
            let conn = db.conn.lock().unwrap();
            conn.execute(
                "UPDATE scheduled_prompts SET cancelled_at_unix = ?1 WHERE id = ?2",
                params![10_i64, stale],
            )
            .expect("backdate");
        }
        db.cancel_all(recent).expect("cancel recent");
        let n = db.reap_cancelled(100).expect("reap");
        assert_eq!(n, 1);
        assert!(db.get(stale).expect("get stale").is_none());
        assert!(db.get(recent).expect("get recent").is_some());
    }
}