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

//! Repository 模式的 CRUD 操作
//!
//! 为各表提供 Repo 结构体,统一实现 insert / get_by_id / list_all / query / update_field / delete。
//!
//! 按表域拆分(SMELL-P1-9,对齐 SMELL-P0-2 拆分思路降 2212 行单文件复杂度):
//! - [`mod@settings`]:SettingsRepo + 列白名单(allowed_columns_for/validate_column_name/is_allowed_column)
//! - [`mod@project_repo`]:ProjectRepo/BranchRepo/ReleaseRepo/WorkflowRepo/NodeExecutionRepo
//! - [`mod@task_repo`]:TaskRepo(含 advance_status_atomic 状态机收口)
//! - [`mod@task_link_repo`]:TaskLinkRepo(任务横向关联 task_links,V29,知识图谱 Phase 1)
//! - [`mod@project_event_repo`]:ProjectEventRepo(统一事件流 project_events,V30,知识图谱 Phase 2)
//! - [`mod@project_service_repo`]:ProjectServiceRepo(基础设施配置 project_services,V31,知识图谱 Phase 3)
//! - [`mod@conversation_repo`]:AiProviderRepo/AiConversationRepo/AiToolExecutionRepo
//! - [`mod@idea_repo`]:IdeaRepo/KnowledgeRepo/KnowledgeEventsRepo + 向量工具
//! - [`mod@idea_eval_repo`]:IdeaEvalRepo(灵感评估历史追加型审计表 idea_evaluations,V22)
//! - [`mod@message_repo`]:AiMessageRepo(F-260619-03 消息拆分存储,全专用方法不走宏)
//!
//! re-export(`pub use ...::*`)保持 `df_storage::crud::XxxRepo` /
//! `df_storage::crud::is_allowed_column` 路径不变,**调用方零改动**。

mod conversation_repo;
mod idea_eval_repo;
mod idea_repo;
mod message_repo;
mod project_event_repo;
mod project_module_repo;
mod project_repo;
mod project_service_repo;
mod settings;
mod task_link_repo;
mod task_repo;

pub use conversation_repo::*;
pub use idea_eval_repo::*;
pub use idea_repo::*;
pub use message_repo::*;
pub use project_event_repo::*;
pub use project_module_repo::*;
pub use project_repo::*;
pub use project_service_repo::*;
pub use settings::*;
pub use task_link_repo::*;
pub use task_repo::*;

// ============================================================
// 辅助宏 — 消除多个 Repo 的重复样板
// ============================================================

/// 一次性为一张表生成完整的 Repo 结构体及其 CRUD 实现。
///
/// 用法: `impl_repo!(RepoName, ModelType, "table_name", from_row_body, insert_body, update_body);`
///
/// 约束:`$record` 须实现 `Clone`(`update_full` 内部 clone 后丢给 spawn_blocking)。
///
/// 宏通过文件末 `pub(crate) use impl_repo;`(Rust 2e textual scope 重导出,非
/// `#[macro_use]`)对各子模块可见,子模块 `use super::impl_repo;` 后直接调用。
/// `from_row`/`insert`/`update` 体内引用的 helper
/// (storage_err/now_millis_str/validate_column_name 及各 from_row)须在调用处可见。
macro_rules! impl_repo {
    (
        $(#[$meta:meta])*
        $name:ident, $record:ident, $table:literal,
        from_row => |$row:ident| $from_body:expr,
        insert  => |$conn:ident, $rec:ident| $insert_body:expr,
        update  => |$u_conn:ident, $u_rec:ident| $update_body:expr
    ) => {
        $(#[$meta])*
        pub struct $name {
            conn: Arc<Mutex<Connection>>,
        }

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

            pub async fn insert(&self, record: $record) -> Result<String> {
                let conn = self.conn.clone();
                tokio::task::spawn_blocking(move || {
                    let guard = conn.blocking_lock();
                    let id = record.id.clone();
                    let $conn = &*guard;
                    let $rec = &record;
                    $insert_body.map_err(storage_err)?;
                    Ok(id)
                })
                .await
                .map_err(storage_err)?
            }

            pub async fn get_by_id(&self, id: &str) -> Result<Option<$record>> {
                let conn = self.conn.clone();
                let id = id.to_owned();
                tokio::task::spawn_blocking(move || {
                    let guard = conn.blocking_lock();
                    let mut stmt = guard
                        .prepare(&format!("SELECT * FROM {} WHERE id = ?1", $table))
                        .map_err(storage_err)?;
                    let row = stmt
                        .query_row(params![id], |$row| $from_body)
                        .optional()
                        .map_err(storage_err)?;
                    Ok(row)
                })
                .await
                .map_err(storage_err)?
            }

            pub async fn list_all(&self) -> Result<Vec<$record>> {
                let conn = self.conn.clone();
                tokio::task::spawn_blocking(move || {
                    let guard = conn.blocking_lock();
                    let mut stmt = guard
                        .prepare(&format!("SELECT * FROM {} ORDER BY created_at DESC", $table))
                        .map_err(storage_err)?;
                    let rows = stmt
                        .query_map([], |$row| $from_body)
                        .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)?
            }

            pub async fn query(&self, field: &str, value: &str) -> Result<Vec<$record>> {
                // 白名单校验,防止 SQL 注入
                validate_column_name(field, $table)?;
                let conn = self.conn.clone();
                let sql = format!("SELECT * FROM {} WHERE {} = ?1 ORDER BY created_at DESC", $table, field);
                let value = value.to_owned();
                tokio::task::spawn_blocking(move || {
                    let guard = conn.blocking_lock();
                    let mut stmt = guard
                        .prepare(&sql)
                        .map_err(storage_err)?;
                    let rows = stmt
                        .query_map(params![value], |$row| $from_body)
                        .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)?
            }

            pub async fn update_field(&self, id: &str, field: &str, value: &str) -> Result<bool> {
                validate_column_name(field, $table)?;
                let conn = self.conn.clone();
                let sql = format!("UPDATE {} SET {} = ?1, updated_at = ?2 WHERE id = ?3", $table, field);
                let id = id.to_owned();
                let value = value.to_owned();
                let now = now_millis_str();
                tokio::task::spawn_blocking(move || {
                    let guard = conn.blocking_lock();
                    let affected = guard
                        .execute(&sql, params![value, now, id])
                        .map_err(storage_err)?;
                    Ok(affected > 0)
                })
                .await
                .map_err(storage_err)?
            }

            /// 整体更新记录(全可变字段,保留 id 与 created_at),单次原子写
            pub async fn update_full(&self, record: &$record) -> Result<bool> {
                let conn = self.conn.clone();
                let rec = record.clone();
                tokio::task::spawn_blocking(move || {
                    let guard = conn.blocking_lock();
                    let $u_conn = &*guard;
                    let $u_rec = &rec;
                    let affected =
                        $update_body.map_err(storage_err)?;
                    Ok(affected > 0)
                })
                .await
                .map_err(storage_err)?
            }

            pub async fn delete(&self, id: &str) -> 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(&format!("DELETE FROM {} WHERE id = ?1", $table), params![id])
                        .map_err(storage_err)?;
                    Ok(affected > 0)
                })
                .await
                .map_err(storage_err)?
            }
        }
    };
}

// 让 impl_repo! 宏对下方声明的子模块可见(textual scope 注入)。
// 必须放在宏定义之后、`mod` 声明之前——顺序错了子模块拿不到宏。
pub(crate) use impl_repo;

// ============================================================
// 公共工具(各子模块 use super::* 复用)
// ============================================================

/// rusqlite 错误 → df-types `Error::Storage` 统一包装(R-PD-10 DRY,
/// 替代散落的 `.map_err(storage_err)`)。
pub(crate) fn storage_err<E: std::string::ToString>(e: E) -> df_types::error::Error {
    df_types::error::Error::Storage(e.to_string())
}

/// 判定扁平化后的存储错误是否为 SQLite 唯一约束冲突(SQLite extended code 2067 /
/// SQLITE_CONSTRAINT_UNIQUE)。
///
/// 设计取舍:存储层在 [`storage_err`] 已将原始 `rusqlite::Error` 扁平化为 `String`
/// (Error::Storage(String)),调用方拿不到结构化 `rusqlite::Error::SqliteFailure` 的
/// `ErrorCode`,无法直接按 code 判定。故在此 flatten 边界集中收口检测逻辑,避免每个
/// 调用方各自 `e.to_string().contains("UNIQUE constraint failed")` 散落脆弱匹配。
///
/// `"UNIQUE constraint failed"` 是 SQLite C 库对 2067 固定输出的文案(大写为 C 库常量,
/// 不随 SQLite 版本/locale 变化),大小写不敏感匹配兜底未来可能的微小差异。
pub fn is_unique_constraint_err(err: &df_types::error::Error) -> bool {
    let df_types::error::Error::Storage(msg) = err else { return false };
    msg.to_ascii_lowercase().contains("unique constraint failed")
}

/// 规范化路径用于比较:canonicalize 解析绝对规范路径(失败降级),
/// 统一正斜杠 + 小写。与 `df_project::scan::normalize_path` **同算法镜像**(df-storage
/// 不依赖 df-project,故独立实现;改动须同步)。防 `C:\a\b` vs `C:/a/b/` 绕过。
pub(crate) fn normalize_stored_path(p: &str) -> String {
    match std::path::Path::new(p).canonicalize() {
        Ok(abs) => abs.to_string_lossy().replace('\\', "/").to_lowercase(),
        Err(_) => p
            .trim_end_matches(['\\', '/'])
            .replace('\\', "/")
            .to_lowercase(),
    }
}

// ============================================================
// 时间工具
// ============================================================

pub(crate) fn now_millis_str() -> String {
    // 转发 df_types::now_millis(DRY:时间获取统一入口在 df-types,避免本 crate 与 commands 层各写一份 SystemTime 调用)
    df_types::now_millis().to_string()
}

// ============================================================
// 基线测试 — 锁 Repo 数/表名防回归(SMELL-P1-9 对齐 SMELL-P0-2)
// ============================================================
//
// 拆分后,各 Repo 散落 5 个子模块。此基线测试确保:
// ① 所有 Repo 类型经 mod.rs `pub use` re-export 后在 `df_storage::crud::` 路径下仍可统一访问
//    (调用方 `df_storage::crud::XxxRepo::new(&db)` 路径不破,降回归)。
// ② 各 Repo::new 在内存 DB 上构造不 panic + 列白名单覆盖每张表
//    (拆分时误删 Repo / 漏 re-export / 表名漂移会被此测试立即捕获)。

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

    /// 每张表都登记了列白名单(拆分后白名单仍在 settings.rs 单一来源,无表遗漏)。
    /// 新增表须同步登记白名单,否则此测试失败(防遗漏)。
    #[test]
    fn all_known_tables_have_column_whitelist() {
        let tables = [
            "ideas", "projects", "tasks", "releases", "branches", "workflow_executions",
            "node_executions", "ai_providers", "ai_conversations", "ai_tool_executions",
            "knowledges", "knowledge_events", "project_services",
        ];
        for t in tables {
            assert!(
                allowed_columns_for(t).is_some(),
                "表 {t} 应登记列白名单(防 SQL 注入 + 按表隔离);拆分后白名单单一来源在 settings.rs"
            );
        }
        // ai_messages 表走全专用 AiMessageRepo(无标准 created_at / 批量语义特殊),
        // 不走通用 query/update_field 路径,故不登记白名单(allowed_columns_for 返回 None)。
        // 此处显式断言其为 None,防止后人误登记破坏"专用 Repo 不进白名单"约定。
        assert!(
            allowed_columns_for("ai_messages").is_none(),
            "ai_messages 走专用 AiMessageRepo,不应登记通用列白名单"
        );
    }

    /// re-export 完整性:所有 Repo 类型经 `df_storage::crud::XxxRepo` 路径可访问 + new 不 panic。
    /// 若拆分时漏 `pub use` 某子模块,对应 Repo 名在此处编译失败(立即暴露)。
    #[tokio::test]
    async fn all_repos_constructible_in_memory() {
        let db = Database::open_in_memory().await.expect("open_in_memory");
        // 各子模块 Repo:new 仅取 conn 句柄,不触 DB IO,构造成功即证明类型可见 + Database 路径通。
        let _ = SettingsRepo::new(&db);
        let _ = IdeaRepo::new(&db);
        let _ = ProjectRepo::new(&db);
        let _ = TaskRepo::new(&db);
        let _ = TaskLinkRepo::new(&db);
        let _ = ProjectEventRepo::new(&db);
        let _ = ProjectServiceRepo::new(&db);
        let _ = ProjectModuleRepo::new(&db);
        let _ = BranchRepo::new(&db);
        let _ = ReleaseRepo::new(&db);
        let _ = WorkflowRepo::new(&db);
        let _ = NodeExecutionRepo::new(&db);
        let _ = AiProviderRepo::new(&db);
        let _ = AiConversationRepo::new(&db);
        let _ = AiToolExecutionRepo::new(&db);
        let _ = KnowledgeRepo::new(&db);
        let _ = KnowledgeEventsRepo::new(&db);
        let _ = AiMessageRepo::new(&db);
        let _ = IdeaEvalRepo::new(&db);
    }
}