//! 待发箱 Repo:OutboxRepo(ai_outbox 表,V51)
//!
//! 队列与定时消息统一实体的持久层。因需按 id 逐条 CAS 更新(claim)与状态条件扫描,
//! 不走 `impl_repo!` 宏,手写(同 `SettingsRepo` 范式)。
//!
//! 设计见 docs/02-架构设计/专项设计/定时消息与待发箱持久化-设计-2026-10-10.md §3.2。
//!
//! 状态落点纪律(V21 教训):每项状态由调度器按项逐条 CAS 写,不依赖全局内存推算;
//! `reset_stale_sending` 在启动时把崩溃残留的 `sending` 复位为 `pending`,镜像无人值守引擎的
//! `in_flight` 泄漏处置。

use std::sync::Arc;

use rusqlite::{params, Connection, OptionalExtension, Row};
use tokio::sync::Mutex;

use df_types::error::Result;

use crate::db::Database;
use crate::models::AiOutboxRecord;

use super::storage_err;

// ============================================================
// 状态 / 类型字面量(单一真值来源,避免调度器与 Repo 各写一份裸串漂移)
// ============================================================

/// 待派发(初始态)。
pub const STATUS_PENDING: &str = "pending";
/// 已被调度器 CAS 占取、正在派发。
pub const STATUS_SENDING: &str = "sending";
/// 受理成功(已交给正典发送链)。
pub const STATUS_SENT: &str = "sent";
/// 受理失败且不再自动重试。
pub const STATUS_FAILED: &str = "failed";
/// 用户取消 / 清空。
pub const STATUS_CANCELED: &str = "canceled";

/// 尽快发(kind=queue,due_at 为空)。
pub const KIND_QUEUE: &str = "queue";
/// 定时发(kind=scheduled,due_at 为 epoch 毫秒)。
pub const KIND_SCHEDULED: &str = "scheduled";

/// 终态集合(不再被调度器扫描)。
const DONE_STATUSES: &str = "'sent','failed','canceled'";

fn outbox_from_row(row: &Row<'_>) -> std::result::Result<AiOutboxRecord, rusqlite::Error> {
    Ok(AiOutboxRecord {
        id: row.get("id")?,
        conversation_id: row.get("conversation_id")?,
        kind: row.get("kind")?,
        text: row.get("text")?,
        skill: row.get("skill")?,
        model_override: row.get("model_override")?,
        parts_json: row.get("parts_json")?,
        spans_json: row.get("spans_json")?,
        due_at: row.get("due_at")?,
        status: row.get("status")?,
        attempts: row.get("attempts")?,
        last_error: row.get("last_error")?,
        created_at: row.get("created_at")?,
        updated_at: row.get("updated_at")?,
        sent_at: row.get("sent_at")?,
    })
}

/// 待发箱 Repo(表 `ai_outbox`)。
pub struct OutboxRepo {
    conn: Arc<Mutex<Connection>>,
}

impl OutboxRepo {
    pub fn new(db: &Database) -> Self {
        Self { conn: db.conn() }
    }

    /// 插入一条待发项,返回 id。created_at/updated_at 由本方法取当前毫秒覆盖
    /// (写入时机即落库时机,调用方传入的不可信)。
    pub async fn insert(&self, record: AiOutboxRecord) -> Result<String> {
        let conn = self.conn.clone();
        let now = df_types::now_millis();
        let mut rec = record;
        rec.created_at = now;
        rec.updated_at = now;
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let id = rec.id.clone();
            guard
                .execute(
                    "INSERT INTO ai_outbox
                     (id, conversation_id, kind, text, skill, model_override, parts_json,
                      spans_json, due_at, status, attempts, last_error, created_at, updated_at, sent_at)
                     VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15)",
                    params![
                        rec.id, rec.conversation_id, rec.kind, rec.text, rec.skill,
                        rec.model_override, rec.parts_json, rec.spans_json, rec.due_at,
                        rec.status, rec.attempts, rec.last_error, rec.created_at, rec.updated_at,
                        rec.sent_at
                    ],
                )
                .map_err(storage_err)?;
            Ok(id)
        })
        .await
        .map_err(storage_err)?
    }

    /// 按 id 取单条(IPC 回显 / 测试核验用)。
    pub async fn get_by_id(&self, id: &str) -> Result<Option<AiOutboxRecord>> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let rec = guard
                .query_row("SELECT * FROM ai_outbox WHERE id = ?1", params![id], outbox_from_row)
                .optional()
                .map_err(storage_err)?;
            Ok(rec)
        })
        .await
        .map_err(storage_err)?
    }

    /// 按会话取待发视图(前端 `state.queue` 的来源)。`include_done=false` 时排除终态,
    /// 按 created_at 升序(先入先出,保序)。
    pub async fn list_by_conv(
        &self,
        conversation_id: &str,
        include_done: bool,
    ) -> Result<Vec<AiOutboxRecord>> {
        let conn = self.conn.clone();
        let conversation_id = conversation_id.to_owned();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let sql = if include_done {
                "SELECT * FROM ai_outbox WHERE conversation_id = ?1 ORDER BY created_at ASC"
                    .to_string()
            } else {
                format!(
                    "SELECT * FROM ai_outbox WHERE conversation_id = ?1 \
                     AND status NOT IN ({DONE_STATUSES}) ORDER BY created_at ASC"
                )
            };
            let mut stmt = guard.prepare(&sql).map_err(storage_err)?;
            let rows = stmt
                .query_map(params![conversation_id], outbox_from_row)
                .map_err(storage_err)?;
            let mut results = Vec::new();
            for r in rows {
                results.push(r.map_err(storage_err)?);
            }
            Ok(results)
        })
        .await
        .map_err(storage_err)?
    }

    /// 全量列表(跨会话,「定时列表」抽屉数据源)。`include_done=false` 时排除终态,
    /// 按 created_at 升序(与 list_by_conv 同序语义)。
    pub async fn list_all(&self, include_done: bool) -> Result<Vec<AiOutboxRecord>> {
        let conn = self.conn.clone();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let sql = if include_done {
                "SELECT * FROM ai_outbox ORDER BY created_at ASC".to_string()
            } else {
                format!(
                    "SELECT * FROM ai_outbox \
                     WHERE status NOT IN ({DONE_STATUSES}) ORDER BY created_at ASC"
                )
            };
            let mut stmt = guard.prepare(&sql).map_err(storage_err)?;
            let rows = stmt.query_map([], outbox_from_row).map_err(storage_err)?;
            let mut results = Vec::new();
            for r in rows {
                results.push(r.map_err(storage_err)?);
            }
            Ok(results)
        })
        .await
        .map_err(storage_err)?
    }

    /// 取到点的待派发项(status='pending' 且 due_at 为空或已到),按 created_at 升序,至多 limit 条。
    /// 命中 idx_ai_outbox_due。
    pub async fn list_due(&self, now_ms: i64, limit: i64) -> Result<Vec<AiOutboxRecord>> {
        let conn = self.conn.clone();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let mut stmt = guard
                .prepare(
                    "SELECT * FROM ai_outbox \
                     WHERE status = 'pending' AND (due_at IS NULL OR due_at <= ?1) \
                     ORDER BY created_at ASC LIMIT ?2",
                )
                .map_err(storage_err)?;
            let rows = stmt
                .query_map(params![now_ms, limit], outbox_from_row)
                .map_err(storage_err)?;
            let mut results = Vec::new();
            for r in rows {
                results.push(r.map_err(storage_err)?);
            }
            Ok(results)
        })
        .await
        .map_err(storage_err)?
    }

    /// CAS 占取:仅当项仍为 pending 时置 sending。changes==1 才返回 true,
    /// 防并发双取(多轮/多消费方同取同一条时只有一个成功)。
    pub async fn claim(&self, id: &str) -> Result<bool> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        let now = df_types::now_millis();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE ai_outbox SET status = 'sending', updated_at = ?1 \
                     WHERE id = ?2 AND status = 'pending'",
                    params![now, id],
                )
                .map_err(storage_err)?;
            Ok(affected == 1)
        })
        .await
        .map_err(storage_err)?
    }

    /// 标记受理成功(终态)。清空 last_error。
    pub async fn mark_sent(&self, id: &str) -> Result<bool> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        let now = df_types::now_millis();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE ai_outbox SET status = 'sent', sent_at = ?1, updated_at = ?1, \
                     last_error = NULL WHERE id = ?2",
                    params![now, id],
                )
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }

    /// 标记永久失败(终态,不再自动重试)。
    pub async fn mark_failed(&self, id: &str, err: &str) -> Result<bool> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        let err = err.to_owned();
        let now = df_types::now_millis();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE ai_outbox SET status = 'failed', last_error = ?1, updated_at = ?2 \
                     WHERE id = ?3",
                    params![err, now, id],
                )
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }

    /// 瞬时失败回退:attempts+1、状态回 pending、记 last_error,返回新的 attempts。
    /// 保留下轮再试(网络超时/会话忙场景天然重试),不删项——对齐「错误不清队列」纪律。
    pub async fn revert_pending(&self, id: &str, err: Option<&str>) -> Result<i64> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        let err = err.map(|s| s.to_owned());
        let now = df_types::now_millis();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            guard
                .execute(
                    "UPDATE ai_outbox SET status = 'pending', attempts = attempts + 1, \
                     last_error = ?1, updated_at = ?2 WHERE id = ?3",
                    params![err, now, id],
                )
                .map_err(storage_err)?;
            let attempts: i64 = guard
                .query_row(
                    "SELECT attempts FROM ai_outbox WHERE id = ?1",
                    params![id],
                    |r| r.get(0),
                )
                .map_err(storage_err)?;
            Ok(attempts)
        })
        .await
        .map_err(storage_err)?
    }

    /// 改期:更新 due_at(NULL=改为尽快)。
    pub async fn reschedule(&self, id: &str, due_at: Option<i64>) -> Result<bool> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        let now = df_types::now_millis();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE ai_outbox SET due_at = ?1, updated_at = ?2 WHERE id = ?3",
                    params![due_at, now, id],
                )
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }

    /// 手动重试入口:仅 failed → pending(attempts 归零=手动重试给全新预算,清 last_error)。
    /// 其余状态不动返回 false——调度器自动重试走 revert_pending(瞬时失败保序),两者互补。
    pub async fn requeue(&self, id: &str) -> Result<bool> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        let now = df_types::now_millis();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE ai_outbox SET status = 'pending', attempts = 0, last_error = NULL, \
                     updated_at = ?1 WHERE id = ?2 AND status = 'failed'",
                    params![now, id],
                )
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }

    /// 取消单项。
    pub async fn cancel(&self, id: &str) -> Result<bool> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        let now = df_types::now_millis();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE ai_outbox SET status = 'canceled', updated_at = ?1 WHERE id = ?2",
                    params![now, id],
                )
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }

    /// 编辑待发文本。
    pub async fn update_text(&self, id: &str, text: &str) -> Result<bool> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        let text = text.to_owned();
        let now = df_types::now_millis();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE ai_outbox SET text = ?1, updated_at = ?2 WHERE id = ?3",
                    params![text, now, id],
                )
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }

    /// 清空待发(pending→canceled,不物理删——保留行便于审计)。`conv_id=None` 时清全部会话。
    /// 返回被取消的行数。
    pub async fn clear(&self, conv_id: Option<&str>) -> Result<usize> {
        let conn = self.conn.clone();
        let conv_id = conv_id.map(|s| s.to_owned());
        let now = df_types::now_millis();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = match conv_id {
                Some(id) => guard.execute(
                    "UPDATE ai_outbox SET status = 'canceled', updated_at = ?1 \
                     WHERE status = 'pending' AND conversation_id = ?2",
                    params![now, id],
                ),
                None => guard.execute(
                    "UPDATE ai_outbox SET status = 'canceled', updated_at = ?1 \
                     WHERE status = 'pending'",
                    params![now],
                ),
            }
            .map_err(storage_err)?;
            Ok(affected)
        })
        .await
        .map_err(storage_err)?
    }

    /// 启动复位:把崩溃半途残留的 `sending` 全部复位为 `pending`(进程重启时前者必为陈旧值)。
    /// 返回复位的行数。镜像无人值守引擎的 in_flight 泄漏处置。
    pub async fn reset_stale_sending(&self) -> Result<usize> {
        let conn = self.conn.clone();
        let now = df_types::now_millis();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE ai_outbox SET status = 'pending', updated_at = ?1 WHERE status = 'sending'",
                    params![now],
                )
                .map_err(storage_err)?;
            Ok(affected)
        })
        .await
        .map_err(storage_err)?
    }
}

// ============================================================
// 单元测试 — 迁移可重入 / CAS 互斥 / sender 残留复位(内存 DB)
// ============================================================

#[cfg(test)]
mod tests {
    use super::*;
    use crate::db::Database;

    /// 构造一条 pending 待发项(due_at=None=尽快)。
    fn rec(id: &str, conv: &str) -> AiOutboxRecord {
        AiOutboxRecord {
            id: id.to_string(),
            conversation_id: conv.to_string(),
            kind: KIND_QUEUE.to_string(),
            text: format!("msg-{id}"),
            skill: None,
            model_override: None,
            parts_json: None,
            spans_json: None,
            due_at: None,
            status: STATUS_PENDING.to_string(),
            attempts: 0,
            last_error: None,
            created_at: 0,
            updated_at: 0,
            sent_at: None,
        }
    }

    async fn setup() -> (Database, OutboxRepo) {
        let db = Database::open_in_memory().await.expect("open_in_memory");
        let repo = OutboxRepo::new(&db);
        (db, repo)
    }

    /// V51 迁移可重入:全链在已建表库上重跑 again 不报错,ai_outbox 表与两索引仍在。
    #[tokio::test]
    async fn migration_v51_is_reentrant() {
        let db = Database::open_in_memory().await.expect("open_in_memory");
        let conn = db.conn();
        let guard = conn.lock().await;
        // 首次已由 open_in_memory 跑过;再显式重跑一遍全链(模拟版本号回退/崩溃重跑)。
        crate::migrations::run(&guard).expect("重跑迁移不应失败(幂等)");
        crate::migrations::run(&guard).expect("二次重跑仍应幂等");
        // 表存在
        let has_table: bool = guard
            .query_row(
                "SELECT COUNT(*) > 0 FROM sqlite_master WHERE type='table' AND name='ai_outbox'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert!(has_table, "ai_outbox 表应存在");
        // 两索引存在
        for idx in ["idx_ai_outbox_due", "idx_ai_outbox_conv"] {
            let has_idx: bool = guard
                .query_row(
                    "SELECT COUNT(*) > 0 FROM sqlite_master WHERE type='index' AND name=?1",
                    params![idx],
                    |r| r.get(0),
                )
                .unwrap();
            assert!(has_idx, "索引 {idx} 应存在");
        }
    }

    /// claim CAS 互斥:同一条只被占取一次,二次 claim 返回 false。
    #[tokio::test]
    async fn claim_is_mutually_exclusive() {
        let (_db, repo) = setup().await;
        repo.insert(rec("o1", "conv-1")).await.unwrap();

        assert!(repo.claim("o1").await.unwrap(), "首次 claim 应成功");
        assert!(!repo.claim("o1").await.unwrap(), "二次 claim 应失败(已非 pending)");
        let got = repo.get_by_id("o1").await.unwrap().unwrap();
        assert_eq!(got.status, STATUS_SENDING);
    }

    /// reset_stale_sending 复位崩溃残留:sending → pending,计数正确。
    #[tokio::test]
    async fn reset_stale_sending_recovers_crash_residue() {
        let (_db, repo) = setup().await;
        repo.insert(rec("o1", "conv-1")).await.unwrap();
        repo.insert(rec("o2", "conv-1")).await.unwrap();
        // 模拟两条被占取后进程崩溃(残留 sending)
        assert!(repo.claim("o1").await.unwrap());
        assert!(repo.claim("o2").await.unwrap());
        assert_eq!(repo.get_by_id("o1").await.unwrap().unwrap().status, STATUS_SENDING);

        let recovered = repo.reset_stale_sending().await.unwrap();
        assert_eq!(recovered, 2, "两条残留 sending 应被复位");
        assert_eq!(repo.get_by_id("o1").await.unwrap().unwrap().status, STATUS_PENDING);
        assert_eq!(repo.get_by_id("o2").await.unwrap().unwrap().status, STATUS_PENDING);
    }

    /// list_due:只取到点的 pending(尽快项恒命中,未到点定时项不取),按 created_at 升序。
    #[tokio::test]
    async fn list_due_respects_due_and_order() {
        let (_db, repo) = setup().await;
        repo.insert(rec("o1", "conv-1")).await.unwrap();
        // 未到点定时项(远未来)
        let mut future = rec("o2", "conv-1");
        future.kind = KIND_SCHEDULED.to_string();
        future.due_at = Some(9_999_999_999_999);
        repo.insert(future).await.unwrap();
        // 已到点定时项
        let mut due = rec("o3", "conv-1");
        due.kind = KIND_SCHEDULED.to_string();
        due.due_at = Some(0);
        repo.insert(due).await.unwrap();

        let got = repo.list_due(df_types::now_millis(), 32).await.unwrap();
        let ids: Vec<&str> = got.iter().map(|r| r.id.as_str()).collect();
        assert_eq!(ids, vec!["o1", "o3"], "仅到点的 pending,按 created_at 升序");
    }

    /// revert_pending:瞬时失败回退 pending + attempts 自增 + 记 last_error,项不丢。
    #[tokio::test]
    async fn revert_pending_keeps_item_and_bumps_attempts() {
        let (_db, repo) = setup().await;
        repo.insert(rec("o1", "conv-1")).await.unwrap();
        repo.claim("o1").await.unwrap();

        let n = repo.revert_pending("o1", Some("boom")).await.unwrap();
        assert_eq!(n, 1);
        let got = repo.get_by_id("o1").await.unwrap().unwrap();
        assert_eq!(got.status, STATUS_PENDING, "回退后应可再次被调度");
        assert_eq!(got.attempts, 1);
        assert_eq!(got.last_error.as_deref(), Some("boom"));
    }

    /// cancel/clear 只动 pending:已终态项不受影响,clear 按会话范围限定。
    #[tokio::test]
    async fn clear_cancels_only_pending_in_scope() {
        let (_db, repo) = setup().await;
        repo.insert(rec("o1", "conv-1")).await.unwrap();
        repo.insert(rec("o2", "conv-2")).await.unwrap();
        repo.mark_sent("o1").await.unwrap();

        let n = repo.clear(Some("conv-1")).await.unwrap();
        assert_eq!(n, 0, "conv-1 已无 pending(sent 不算),不误改");
        assert_eq!(repo.get_by_id("o1").await.unwrap().unwrap().status, STATUS_SENT);

        let n = repo.clear(Some("conv-2")).await.unwrap();
        assert_eq!(n, 1);
        assert_eq!(repo.get_by_id("o2").await.unwrap().unwrap().status, STATUS_CANCELED);
    }

    /// list_all:跨会话全量返回;include_done=false 排除终态,created_at 升序。
    #[tokio::test]
    async fn list_all_covers_convs_and_filters_done() {
        let (_db, repo) = setup().await;
        repo.insert(rec("o1", "conv-1")).await.unwrap();
        repo.insert(rec("o2", "conv-2")).await.unwrap();
        let mut done = rec("o3", "conv-1");
        done.status = STATUS_SENT.to_string();
        repo.insert(done).await.unwrap();

        let all = repo.list_all(true).await.unwrap();
        assert_eq!(all.len(), 3, "include_done=true 全量(含跨会话)");

        let active = repo.list_all(false).await.unwrap();
        let ids: Vec<&str> = active.iter().map(|r| r.id.as_str()).collect();
        assert_eq!(ids, vec!["o1", "o2"], "排除终态后仅两条 pending,升序");
    }

    /// requeue:仅 failed → pending 且 attempts 归零清错;pending/sent 不动返 false。
    #[tokio::test]
    async fn requeue_only_resets_failed() {
        let (_db, repo) = setup().await;
        // o1:pending → claim(占取)→ revert(attempts=1)→ mark_failed(failed,attempts 保留 1)
        repo.insert(rec("o1", "conv-1")).await.unwrap();
        repo.claim("o1").await.unwrap();
        let attempts = repo.revert_pending("o1", Some("boom")).await.unwrap();
        assert_eq!(attempts, 1);
        repo.mark_failed("o1", "boom").await.unwrap();

        assert!(repo.requeue("o1").await.unwrap(), "failed 应可重入队");
        let got = repo.get_by_id("o1").await.unwrap().unwrap();
        assert_eq!(got.status, STATUS_PENDING);
        assert_eq!(got.attempts, 0, "手动重试给全新预算");
        assert_eq!(got.last_error, None, "重入队清上次错误");

        // pending 项:requeue 不动(返回 false)
        repo.insert(rec("o2", "conv-1")).await.unwrap();
        assert!(!repo.requeue("o2").await.unwrap(), "pending 非 failed,requeue 不动");
        // sent 终态:requeue 不动
        repo.insert(rec("o3", "conv-1")).await.unwrap();
        repo.mark_sent("o3").await.unwrap();
        assert!(!repo.requeue("o3").await.unwrap(), "sent 终态不可重入队");
    }
}