DevFlow/crates/df-storage/src/crud/conversation_repo.rs

//! AI 对话域 Repo:AiProviderRepo / AiConversationRepo / AiToolExecutionRepo

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::{AiConversationRecord, AiProviderRecord, AiToolExecutionRecord};

use super::impl_repo;
use super::{now_millis_str, storage_err, validate_column_name};

// ============================================================
// from_row 辅助函数
// ============================================================

fn ai_provider_from_row(row: &Row<'_>) -> std::result::Result<AiProviderRecord, rusqlite::Error> {
    // model_configs:DB TEXT 列存 JSON 字符串。读 Option<String> 兼容老库 NULL,
    // 再经 deserialize_model_configs 解析(老字符串数组/新对象数组/空 → Vec<ModelConfig>)。
    // 解析失败不致命:降级为空 Vec(防单行坏数据拖垮 list_all)。
    let model_configs: Vec<df_ai_core::model::ModelConfig> = {
        let raw: Option<String> = row.get("model_configs").ok();
        match raw {
            None => Vec::new(),
            Some(s) => serde_json::from_str::<ModelConfigsWrap>(&format!(
                r#"{{"v":{}}}"#,
                if s.trim().is_empty() {
                    "null".to_string()
                } else if s.trim_start().starts_with('[') || s.trim_start().starts_with('{') {
                    s
                } else {
                    // 非 JSON 字面文本(理论不会出现)→ 包装为 JSON 字符串让 deserialize 兜底
                    serde_json::to_string(&s).unwrap_or_else(|_| "null".into())
                }
            ))
            .map(|w| w.v)
            .unwrap_or_default(),
        }
    };
    Ok(AiProviderRecord {
        id: row.get("id")?,
        name: row.get("name")?,
        provider_type: row.get("provider_type")?,
        api_key: row.get("api_key")?,
        base_url: row.get("base_url")?,
        default_model: row.get("default_model")?,
        models: row.get("models")?,
        model_configs,
        is_default: row.get::<_, i32>("is_default")? != 0,
        config: row.get("config")?,
        created_at: row.get("created_at")?,
        updated_at: row.get("updated_at")?,
        // F-260614-04: enabled/weight 列老库经 v19 迁移补建,DEFAULT 1 / DEFAULT 50。
        // from_row 按 i32 取列值兼容(SQLite 无真 BOOLEAN),0→false/非0→true。
        enabled: row.get::<_, i32>("enabled").unwrap_or(1) != 0,
        // weight 读侧 clamp [0,100]:与 insert/update_full 落库的 `.min(100)` 对齐,
        // 防老库(clamp 落地前写入的)或外部直改 DB 产生的越界值污染路由权重语义。
        weight: row.get::<_, i32>("weight")
            .unwrap_or(50)
            .clamp(0, 100) as u32,
    })
}

/// from_row 内部辅助:复用 deserialize_model_configs 解析 DB TEXT 列 JSON。
/// 包一层 { "v": <原始值> } 把任意 JSON 值送进 deserialize_model_configs。
#[derive(serde::Deserialize)]
struct ModelConfigsWrap {
    #[serde(default, deserialize_with = "df_ai_core::model::deserialize_model_configs")]
    v: Vec<df_ai_core::model::ModelConfig>,
}

fn ai_conversation_from_row(row: &Row<'_>) -> std::result::Result<AiConversationRecord, rusqlite::Error> {
    Ok(AiConversationRecord {
        id: row.get("id")?,
        title: row.get("title")?,
        messages: row.get("messages")?,
        provider_id: row.get("provider_id")?,
        model: row.get("model")?,
        models: row.get("models")?,
        archived: row.get::<_, i32>("archived")? != 0,
        pinned: row.get::<_, i32>("pinned")? != 0,
        prompt_tokens: row.get("prompt_tokens")?,
        completion_tokens: row.get("completion_tokens")?,
        pinned_goals: row.get("pinned_goals")?,
        created_at: row.get("created_at")?,
        updated_at: row.get("updated_at")?,
    })
}

fn ai_tool_execution_from_row(row: &Row<'_>) -> std::result::Result<AiToolExecutionRecord, rusqlite::Error> {
    Ok(AiToolExecutionRecord {
        id: row.get("id")?,
        conversation_id: row.get("conversation_id")?,
        // F-260619-04:message_id 列老库经 v21 迁移补建。unwrap_or(None) 兜底:
        // 新库空表直接有列;老库行 ALTER 后 NULL;极端情况(迁移未跑/手工删列)防御。
        message_id: row.get("message_id").unwrap_or(None),
        tool_call_id: row.get("tool_call_id")?,
        tool_name: row.get("tool_name")?,
        arguments: row.get("arguments")?,
        result: row.get("result")?,
        status: row.get("status")?,
        risk_level: row.get("risk_level")?,
        requested_at: row.get("requested_at")?,
        executed_at: row.get("executed_at")?,
        decided_by: row.get("decided_by")?,
    })
}

// ============================================================
// Repo 实现
// ============================================================

impl_repo!(
    /// AI 提供商配置表 CRUD
    AiProviderRepo,
    AiProviderRecord,
    "ai_providers",
    from_row => |row| ai_provider_from_row(row),
    insert  => |conn, rec| {
        let is_default = if rec.is_default { 1i32 } else { 0i32 };
        // model_configs:Vec<ModelConfig> → JSON 字符串落 TEXT 列
        let model_configs_json = serde_json::to_string(&rec.model_configs).unwrap_or_else(|_| "[]".into());
        // F-260614-04: enabled/weight 落库(SQLite 无 BOOLEAN,i32 承载)。
        let enabled_i = if rec.enabled { 1i32 } else { 0i32 };
        let weight_i = rec.weight.min(100) as i32;
        conn.execute(
            "INSERT OR REPLACE INTO ai_providers (id, name, provider_type, api_key, base_url, default_model, models, model_configs, is_default, config, created_at, updated_at, enabled, weight)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)",
            params![
                rec.id, rec.name, rec.provider_type, rec.api_key, rec.base_url,
                rec.default_model, rec.models, model_configs_json, is_default, rec.config, rec.created_at, rec.updated_at,
                enabled_i, weight_i
            ],
        )
    },
    update  => |conn, rec| {
        let is_default = if rec.is_default { 1i32 } else { 0i32 };
        let model_configs_json = serde_json::to_string(&rec.model_configs).unwrap_or_else(|_| "[]".into());
        let enabled_i = if rec.enabled { 1i32 } else { 0i32 };
        let weight_i = rec.weight.min(100) as i32;
        conn.execute(
            "UPDATE ai_providers SET name = ?1, provider_type = ?2, api_key = ?3, base_url = ?4, default_model = ?5, models = ?6, model_configs = ?7, is_default = ?8, config = ?9, updated_at = ?10, enabled = ?11, weight = ?12 WHERE id = ?13",
            params![
                rec.name, rec.provider_type, rec.api_key, rec.base_url,
                rec.default_model, rec.models, model_configs_json, is_default, rec.config, rec.updated_at,
                enabled_i, weight_i, rec.id
            ],
        )
    }
);

impl_repo!(
    /// AI 对话历史表 CRUD
    AiConversationRepo,
    AiConversationRecord,
    "ai_conversations",
    from_row => |row| ai_conversation_from_row(row),
    insert  => |conn, rec| {
        conn.execute(
            "INSERT INTO ai_conversations (id, title, messages, provider_id, model, models, archived, pinned, prompt_tokens, completion_tokens, pinned_goals, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
            params![
                rec.id, rec.title, rec.messages, rec.provider_id, rec.model, rec.models, rec.archived,
                if rec.pinned { 1i32 } else { 0i32 },
                rec.prompt_tokens, rec.completion_tokens,
                rec.pinned_goals, rec.created_at, rec.updated_at
            ],
        )
    },
    update  => |conn, rec| {
        conn.execute(
            "UPDATE ai_conversations SET title = ?1, messages = ?2, provider_id = ?3, model = ?4, models = ?5, archived = ?6, pinned = ?7, prompt_tokens = ?8, completion_tokens = ?9, pinned_goals = ?10, updated_at = ?11 WHERE id = ?12",
            params![
                rec.title, rec.messages, rec.provider_id, rec.model, rec.models, rec.archived,
                if rec.pinned { 1i32 } else { 0i32 },
                rec.prompt_tokens, rec.completion_tokens,
                rec.pinned_goals, rec.updated_at, rec.id
            ],
        )
    }
);

impl_repo!(
    /// AI 工具执行审计表 CRUD
    AiToolExecutionRepo,
    AiToolExecutionRecord,
    "ai_tool_executions",
    from_row => |row| ai_tool_execution_from_row(row),
    insert  => |conn, rec| {
        conn.execute(
            "INSERT INTO ai_tool_executions (id, conversation_id, message_id, tool_call_id, tool_name, arguments, result, status, risk_level, requested_at, executed_at, decided_by)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
            params![
                rec.id, rec.conversation_id, rec.message_id, rec.tool_call_id, rec.tool_name,
                rec.arguments, rec.result, rec.status, rec.risk_level,
                rec.requested_at, rec.executed_at, rec.decided_by
            ],
        )
    },
    update  => |conn, rec| {
        conn.execute(
            "UPDATE ai_tool_executions SET conversation_id = ?1, message_id = ?2, tool_call_id = ?3, tool_name = ?4, arguments = ?5, result = ?6, status = ?7, risk_level = ?8, requested_at = ?9, executed_at = ?10, decided_by = ?11 WHERE id = ?12",
            params![
                rec.conversation_id, rec.message_id, rec.tool_call_id, rec.tool_name,
                rec.arguments, rec.result, rec.status, rec.risk_level,
                rec.requested_at, rec.executed_at, rec.decided_by, rec.id
            ],
        )
    }
);

// ai_tool_executions 无 created_at 列(用 requested_at/executed_at 计时),
// 通用 query 宏硬编码 ORDER BY created_at 会触发 "no such column" → 调用方 unwrap_or_default 吞错。
// 故为此表提供专用查询,绕过通用 query。详见 ai.rs audit_finalize。
impl AiToolExecutionRepo {
    /// 按 tool_call_id 查最新一条审计记录(审批回填定位用)。
    pub async fn find_by_tool_call_id(
        &self,
        tool_call_id: &str,
    ) -> Result<Option<AiToolExecutionRecord>> {
        let conn = self.conn.clone();
        let tid = tool_call_id.to_owned();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let mut stmt = guard
                .prepare(
                    "SELECT * FROM ai_tool_executions WHERE tool_call_id = ?1 ORDER BY requested_at DESC LIMIT 1",
                )
                .map_err(storage_err)?;
            let row = stmt
                .query_row(params![tid], |row| ai_tool_execution_from_row(row))
                .optional()
                .map_err(storage_err)?;
            Ok(row)
        })
        .await
        .map_err(storage_err)?
    }

    /// 列出所有 status=pending 的审计行(启动重建 pending_approvals 用)
    ///
    /// 专用 SELECT(非 query 宏——后者硬编码 ORDER BY created_at,而本表无该列)。
    pub async fn list_pending(&self) -> Result<Vec<AiToolExecutionRecord>> {
        let conn = self.conn.clone();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let mut stmt = guard
                .prepare("SELECT * FROM ai_tool_executions WHERE status = 'pending' ORDER BY requested_at ASC")
                .map_err(storage_err)?;
            let rows = stmt
                .query_map([], |row| ai_tool_execution_from_row(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)?
    }

    /// 审批历史面板分页查询:按 requested_at 倒序(最新在前),limit 默认 50。
    ///
    /// 与 list_pending 同理走专用 SELECT,绕过通用 query 宏(后者硬编码
    /// ORDER BY created_at,本表无该列)。limit/offset 上限钳制(limit ≤ 200),
    /// 防前端恶意/失误传超大值。
    pub async fn list_recent(
        &self,
        limit: u32,
        offset: u32,
    ) -> Result<Vec<AiToolExecutionRecord>> {
        let conn = self.conn.clone();
        // 钳制 limit 防滥用(默认 50,最大 200)
        let safe_limit = limit.min(200) as i64;
        let safe_offset = offset as i64;
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let mut stmt = guard
                .prepare(
                    "SELECT * FROM ai_tool_executions ORDER BY requested_at DESC LIMIT ?1 OFFSET ?2",
                )
                .map_err(storage_err)?;
            let rows = stmt
                .query_map(params![safe_limit, safe_offset], |row| {
                    ai_tool_execution_from_row(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)?
    }
}

// AiConversationRepo 的整体更新已由 impl_repo! 宏统一生成的 update_full 提供。

impl AiConversationRepo {
    /// 清空对话消息内容(保留 conversation 记录本身,只清 messages JSON + 清零 token 计数)
    ///
    /// "清空对话"语义:对话壳保留(侧栏仍可见,可继续在该对话内聊),仅清空历史消息。
    /// messages 是 ai_conversations 表内的 JSON 列而非独立行,故"删 messages"= 置空该列。
    pub async fn clear_messages(&self, id: &str) -> Result<bool> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        let now = now_millis_str();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE ai_conversations SET messages = '[]', prompt_tokens = 0, completion_tokens = 0, updated_at = ?1 WHERE id = ?2",
                    params![now, id],
                )
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }

    /// 设置归档标记(仅改 archived,不动 updated_at)
    ///
    /// 区别于 update_field(后者强制 SET updated_at=now,会把归档/取消归档误判为内容更新,
    /// 导致侧栏相对时间跳变为"刚刚")。归档是纯元数据标记,应保持时间不变。
    pub async fn set_archived(&self, id: &str, archived: bool) -> Result<bool> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE ai_conversations SET archived = ?1 WHERE id = ?2",
                    params![if archived { 1 } else { 0 }, id],
                )
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }

    /// 设置标题(仅改 title,不动 updated_at)
    ///
    /// 区别于 update_field(强制 SET updated_at=now,会把标题生成误判为内容更新,
    /// 导致侧栏时间分组/排序跳变)。标题生成是系统后台操作,应保持会话相对时间不变。
    pub async fn set_title(&self, id: &str, title: &str) -> Result<bool> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        let title = title.to_owned();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE ai_conversations SET title = ?1 WHERE id = ?2",
                    params![title, id],
                )
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }

    /// 设置置顶标记(仅改 pinned,不动 updated_at) — UX-17
    ///
    /// 同 set_archived:置顶是纯元数据标记,不应改变相对时间。前端排序读 pinned DESC, updated_at DESC。
    pub async fn set_pinned(&self, id: &str, pinned: bool) -> Result<bool> {
        let conn = self.conn.clone();
        let id = id.to_owned();
        tokio::task::spawn_blocking(move || {
            let guard = conn.blocking_lock();
            let affected = guard
                .execute(
                    "UPDATE ai_conversations SET pinned = ?1 WHERE id = ?2",
                    params![if pinned { 1 } else { 0 }, id],
                )
                .map_err(storage_err)?;
            Ok(affected > 0)
        })
        .await
        .map_err(storage_err)?
    }
}

// ============================================================
// 单元测试 — AiProviderRepo model_configs DB roundtrip + 老库兼容
// ============================================================

#[cfg(test)]
mod tests {
    use super::*;
    use crate::db::Database;
    use crate::models::AiProviderRecord;
    use df_ai_core::model::{Capability, IntelligenceTier, Modality, ModelConfig};

    /// model_configs DB roundtrip + 老库空兼容(F-01 阶段1)
    #[tokio::test]
    async fn ai_provider_model_configs_roundtrip_and_old_db_compat() {
        let db = Database::open_in_memory().await.expect("open_in_memory");
        let repo = AiProviderRepo::new(&db);

        // 新格式:带多模型 + 多维度配置
        let configs = vec![
            ModelConfig::with_defaults("glm-4-flash"),
            ModelConfig {
                model_id: "glm-4v".into(),
                modalities: vec![Modality::Text, Modality::Vision],
                capabilities: vec![Capability::ToolUse],
                intelligence: IntelligenceTier::Plus,
                weight: 70,
                context_window: 128_000,
                ..ModelConfig::with_defaults("glm-4v")
            },
        ];
        let rec = AiProviderRecord {
            id: "p1".into(),
            name: "测试".into(),
            provider_type: "openai_compat".into(),
            api_key: String::new(),
            base_url: "https://x".into(),
            default_model: "glm-4-flash".into(),
            models: None,
            model_configs: configs.clone(),
            is_default: false,
            config: None,
            created_at: "0".into(),
            updated_at: "0".into(),
            enabled: true,
            weight: 50,
        };
        repo.insert(rec).await.expect("insert");

        let got = repo.get_by_id("p1").await.expect("get").expect("row exists");
        assert_eq!(got.model_configs.len(), 2);
        assert_eq!(got.model_configs[0].model_id, "glm-4-flash");
        assert_eq!(got.model_configs[1].model_id, "glm-4v");
        assert_eq!(got.model_configs[1].intelligence, IntelligenceTier::Plus);
        assert_eq!(got.model_configs[1].context_window, 128_000);

        // 老库空兼容:直接写 model_configs=NULL 的行(模拟 V18 之前的老库行)
        // 然后 from_row 应得空 Vec
        {
            let conn = db.conn();
            let g = conn.lock().await;
            g.execute(
                "INSERT OR REPLACE INTO ai_providers \
                 (id,name,provider_type,api_key,base_url,default_model,models,model_configs,is_default,config,created_at,updated_at) \
                 VALUES ('old','','openai_compat','','','','{}',NULL,0,NULL,'0','0')",
                [],
            )
            .expect("raw insert old row");
        }
        let old = repo.get_by_id("old").await.expect("get").expect("old row");
        assert!(old.model_configs.is_empty(), "NULL 列应得空 Vec");

        // 老格式字符串数组 JSON(向后兼容 deserialize_model_configs)
        {
            let conn = db.conn();
            let g = conn.lock().await;
            g.execute(
                "INSERT OR REPLACE INTO ai_providers \
                 (id,name,provider_type,api_key,base_url,default_model,models,model_configs,is_default,config,created_at,updated_at) \
                 VALUES ('legacy','','openai_compat','','','','{}','[\"glm-4-flash\",\"glm-4v\"]',0,NULL,'0','0')",
                [],
            )
            .expect("raw insert legacy row");
        }
        let legacy = repo.get_by_id("legacy").await.expect("get").expect("legacy row");
        assert_eq!(legacy.model_configs.len(), 2, "老字符串数组应转 2 个默认 ModelConfig");
        assert_eq!(legacy.model_configs[0].model_id, "glm-4-flash");
    }
}