//! 项目域 Repo:ProjectRepo / BranchRepo / ReleaseRepo / WorkflowRepo / NodeExecutionRepo use std::sync::Arc; use rusqlite::{params, Connection, OptionalExtension, Row}; use tokio::sync::Mutex; use df_types::error::Result; use df_types::types::ProjectStatus; use crate::db::Database; use crate::models::{ BranchRecord, NodeExecutionRecord, ProjectRecord, ReleaseRecord, WorkflowRecord, }; use super::impl_repo; use super::{normalize_stored_path, now_millis_str, storage_err, validate_column_name}; // ============================================================ // 项目查询入参(F-260621-02 P2/P3 查询维度补全) // ============================================================ /// 项目列表查询条件(可选字段,全 None = 当前 list_active 全量行为)。 /// /// 对齐方案 docs/02-架构设计/专项设计/查询能力补全方案-2026-06-21.md §4.1 统一查询结构。 /// 项目表无 status 生命周期(P1 跳过),仅软删 deleted_at(由 list_by_query 固定排除)。 /// 故本结构不含 status,只含 keyword/order_by/limit/offset 四维。 /// /// 序列化 derive:命令层经 serde 从前端 IPC 接收(Option 字段缺省 None,旧调用方不传等价全量)。 #[derive(Debug, Default, Clone, serde::Serialize, serde::Deserialize)] pub struct ProjectQuery { /// 关键词:name/description LIKE %kw%(trim 后空字符串视为无关键词) #[serde(default)] pub keyword: Option, /// 排序字段(白名单校验,格式 "col" / "col asc" / "col desc",默认 created_at desc) #[serde(default)] pub order_by: Option, /// 返回上限(钳制 [0,200],None = 不加 LIMIT 全量) #[serde(default)] pub limit: Option, /// 偏移量(None = 0,配合 limit 分页) #[serde(default)] pub offset: Option, } /// 解析 order_by 入参为 "col DIR" SQL 片段(列名走白名单校验防注入)。 /// /// 接受 "col" / "col asc" / "col desc"(DIR 大小写不敏感)。col 走 `validate_column_name` /// 校验(列名不可参数化,必须拼字符串,白名单是唯一防注入手段,对齐 impl_repo! 宏)。 /// 非法列名返回 Err;合法但无 DIR 默认 DESC(与 list_active 一致)。 fn build_order_clause(order_by: Option<&str>) -> Result { let Some(raw) = order_by else { return Ok("created_at DESC".to_string()); }; let raw = raw.trim(); if raw.is_empty() { return Ok("created_at DESC".to_string()); } // 拆 "col [dir]",最多 2 段。 let parts: Vec<&str> = raw.split_whitespace().collect(); let col = parts[0]; let dir = parts.get(1).map(|s| s.to_ascii_uppercase()); // 白名单校验列名(防 SQL 注入:列名拼字符串前必须校验)。 validate_column_name(col, "projects")?; match dir.as_deref() { None | Some("DESC") => Ok(format!("{col} DESC")), Some("ASC") => Ok(format!("{col} ASC")), // 其它非法方向词:拒绝(防 "; DROP TABLE" 之类拼串绕过)。 Some(other) => Err(df_types::error::Error::Storage(format!( "非法排序方向: {other}(仅支持 asc/desc)" ))), } } // ============================================================ // from_row 辅助函数 // ============================================================ fn project_from_row(row: &Row<'_>) -> std::result::Result { Ok(ProjectRecord { id: row.get("id")?, name: row.get("name")?, description: row.get("description")?, status: { let s: String = row.get("status")?; ProjectStatus::from_db_str(&s).unwrap_or_default() }, idea_id: row.get("idea_id")?, path: row.get("path")?, stack: row.get("stack")?, created_at: row.get("created_at")?, updated_at: row.get("updated_at")?, }) } fn branch_from_row(row: &Row<'_>) -> std::result::Result { Ok(BranchRecord { id: row.get("id")?, project_id: row.get("project_id")?, task_id: row.get("task_id")?, name: row.get("name")?, base: row.get("base")?, status: row.get("status")?, created_at: row.get("created_at")?, updated_at: row.get("updated_at")?, merged_at: row.get("merged_at")?, }) } fn release_from_row(row: &Row<'_>) -> std::result::Result { Ok(ReleaseRecord { id: row.get("id")?, project_id: row.get("project_id")?, version: row.get("version")?, status: row.get("status")?, task_ids: row.get("task_ids")?, changelog: row.get("changelog")?, created_at: row.get("created_at")?, released_at: row.get("released_at")?, }) } fn workflow_from_row(row: &Row<'_>) -> std::result::Result { Ok(WorkflowRecord { id: row.get("id")?, name: row.get("name")?, dag_json: row.get("dag_json")?, status: row.get("status")?, triggered_by: row.get("triggered_by")?, project_id: row.get("project_id")?, task_id: row.get("task_id")?, created_at: row.get("created_at")?, completed_at: row.get("completed_at")?, }) } fn node_execution_from_row( row: &Row<'_>, ) -> std::result::Result { Ok(NodeExecutionRecord { id: row.get("id")?, workflow_id: row.get("workflow_id")?, node_id: row.get("node_id")?, node_type: row.get("node_type")?, status: row.get("status")?, input_json: row.get("input_json")?, output_json: row.get("output_json")?, error_message: row.get("error_message")?, started_at: row.get("started_at")?, completed_at: row.get("completed_at")?, }) } // ============================================================ // Repo 实现 // ============================================================ impl_repo!( /// 项目表 CRUD ProjectRepo, ProjectRecord, "projects", from_row => |row| project_from_row(row), insert => |conn, rec| { conn.execute( "INSERT INTO projects (id, name, description, status, idea_id, path, stack, created_at, updated_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)", params![ rec.id, rec.name, rec.description, rec.status.as_str(), rec.idea_id, rec.path, rec.stack, rec.created_at, rec.updated_at ], ) }, update => |conn, rec| { conn.execute( "UPDATE projects SET name = ?1, description = ?2, status = ?3, idea_id = ?4, path = ?5, stack = ?6, updated_at = ?7 WHERE id = ?8", params![ rec.name, rec.description, rec.status.as_str(), rec.idea_id, rec.path, rec.stack, rec.updated_at, rec.id ], ) } ); impl ProjectRepo { /// 列出未删除项目(deleted_at IS NULL) pub async fn list_active(&self) -> Result> { let conn = self.conn.clone(); tokio::task::spawn_blocking(move || { let guard = conn.blocking_lock(); let mut stmt = guard .prepare("SELECT id, name, description, status, idea_id, path, stack, created_at, updated_at FROM projects WHERE deleted_at IS NULL ORDER BY created_at DESC") .map_err(storage_err)?; let rows = stmt .query_map([], |row| project_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)? } /// 按条件查询未删除项目(F-260621-02 P2/P3:关键词搜索 + 排序 + 分页)。 /// /// 复用 `KnowledgeRepo::search` 动态 WHERE 拼接模式:按可选字段 if-let 拼 SQL 子句 + /// 分支化参数绑定。始终排除 `deleted_at IS NOT NULL`(软删),与 `list_active` 行为对齐。 /// /// - `keyword`:name/description `LIKE %kw%`(对齐知识库 title/content LIKE)。 /// - `order_by`:走白名单 `validate_column_name` 防 SQL 注入(列名不可参数化,必须拼字符串, /// 白名单是唯一防注入手段)。默认 `created_at DESC`(与 list_active 等价)。 /// - `limit`:钳制上限 200(对齐 `KnowledgeEventsRepo::list_recent`),防恶意/失误传超大值。 /// - `offset`:偏移量,配合 limit 做分页。 /// /// 向后兼容:全字段 None 时 SQL 退化为 /// `SELECT ... WHERE deleted_at IS NULL ORDER BY created_at DESC` /// 与 `list_active` 完全等价,零破坏。 pub async fn list_by_query(&self, q: ProjectQuery) -> Result> { // order_by 白名单校验(列名拼字符串前必须校验,防注入)。 // 校验后再拼方向:允许 "col" / "col asc" / "col desc"(大小写不敏感)。 let order_clause = build_order_clause(q.order_by.as_deref())?; let conn = self.conn.clone(); // limit 钳制:None 不加 LIMIT(全量,对齐 list_active);Some 钳到 [0,200]。 let limit_i: Option = q.limit.map(|n| n.min(200) as i64); let offset_i: i64 = q.offset.unwrap_or(0) as i64; tokio::task::spawn_blocking(move || { let guard = conn.blocking_lock(); // 动态拼 WHERE + 参数。WHERE 1=1 起点 + deleted_at IS NULL 固定基础条件, // 后续可选条件 AND 拼接(对齐 KnowledgeRepo::search 的 if-let 分支模式)。 let mut sql = String::from( "SELECT id, name, description, status, idea_id, path, stack, created_at, updated_at \ FROM projects WHERE deleted_at IS NULL", ); // keyword 用 ?N 占位两次(name LIKE ? AND description LIKE ? 同 pattern), // 对齐 KnowledgeRepo::search 的 `params![pattern, pattern, ...]`。 let mut params_vec: Vec> = Vec::new(); if let Some(kw) = &q.keyword { let trimmed = kw.trim(); if !trimmed.is_empty() { sql.push_str(" AND (name LIKE ? OR description LIKE ?)"); let pattern = format!("%{trimmed}%"); params_vec.push(Box::new(pattern.clone())); params_vec.push(Box::new(pattern)); } } sql.push_str(" ORDER BY "); sql.push_str(&order_clause); // 分页:有 limit 才加 LIMIT/OFFSET(全量场景不加,等价 list_active)。 if let Some(lim) = limit_i { sql.push_str(" LIMIT ? OFFSET ?"); params_vec.push(Box::new(lim)); params_vec.push(Box::new(offset_i)); } let mut stmt = guard.prepare(&sql).map_err(storage_err)?; let param_refs: Vec<&dyn rusqlite::ToSql> = params_vec.iter().map(|p| p.as_ref()).collect(); let rows = stmt .query_map(param_refs.as_slice(), |row| project_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)? } /// 软删:标记 deleted_at(进回收站,可恢复)。仅作用于未删项目,返回是否命中。 pub async fn soft_delete(&self, id: &str) -> Result { 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 projects SET deleted_at = ?1, updated_at = ?1 WHERE id = ?2 AND deleted_at IS NULL", params![now, id], ) .map_err(storage_err)?; Ok(affected > 0) }) .await .map_err(storage_err)? } /// 恢复:清 deleted_at(从回收站还原) pub async fn restore(&self, id: &str) -> Result { 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 projects SET deleted_at = NULL, updated_at = ?1 WHERE id = ?2 AND deleted_at IS NOT NULL", params![now, id], ) .map_err(storage_err)?; Ok(affected > 0) }) .await .map_err(storage_err)? } /// 查找已绑定该规范化路径的项目(排除 exclude_id 自身)。无冲突返回 None。 /// /// DRY(R-PD-11):统一 project.rs::find_binding_conflict 与 tool_registry.rs::bind_dir_to_project /// 的「列项目→排除自身→按规范化路径比较」逻辑。本 crate 不依赖 df-project,故 norm_path 须由 /// 调用方先经 `df_project::scan::normalize_path`(内含 canonicalize,防 `C:\a\b` vs `C:/a/b/` 绕过)。 pub async fn find_path_conflict( &self, norm_path: &str, exclude_id: Option<&str>, ) -> Result> { let projects = self.list_active().await?; Ok(projects.into_iter().find(|p| { let excluded = exclude_id.is_some_and(|eid| p.id == eid); !excluded && p.path.as_ref().is_some_and(|pp| { // 路径规范化由调用方完成目标侧;这里为已存项目路径补规范化 // (历史路径写法可能不规范,统一规范化比较避免误判/漏判) normalize_stored_path(pp) == norm_path }) })) } /// 列出回收站(deleted_at IS NOT NULL),按更新时间(≈删除时间)降序 pub async fn list_deleted(&self) -> Result> { let conn = self.conn.clone(); tokio::task::spawn_blocking(move || { let guard = conn.blocking_lock(); let mut stmt = guard .prepare("SELECT id, name, description, status, idea_id, path, stack, created_at, updated_at FROM projects WHERE deleted_at IS NOT NULL ORDER BY updated_at DESC") .map_err(storage_err)?; let rows = stmt .query_map([], |row| project_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)? } /// 彻底删除:事务级联删 branches→releases→tasks→projects(不可恢复) /// /// SQLite 已开 PRAGMA foreign_keys=ON 但表无 ON DELETE CASCADE,ALTER 改不了 FK 约束, /// 故应用层级联:子表先于父表删,单事务保证一致性。 pub async fn purge_with_descendants(&self, id: &str) -> Result { let conn = self.conn.clone(); let id = id.to_owned(); tokio::task::spawn_blocking(move || { let mut guard = conn.blocking_lock(); let tx = guard.transaction().map_err(storage_err)?; tx.execute("DELETE FROM branches WHERE project_id = ?1", params![id]) .map_err(storage_err)?; tx.execute("DELETE FROM releases WHERE project_id = ?1", params![id]) .map_err(storage_err)?; tx.execute("DELETE FROM tasks WHERE project_id = ?1", params![id]) .map_err(storage_err)?; let affected = tx .execute("DELETE FROM projects WHERE id = ?1", params![id]) .map_err(storage_err)?; tx.commit().map_err(storage_err)?; Ok(affected > 0) }) .await .map_err(storage_err)? } } impl_repo!( /// 分支表 CRUD BranchRepo, BranchRecord, "branches", from_row => |row| branch_from_row(row), insert => |conn, rec| { conn.execute( "INSERT INTO branches (id, project_id, task_id, name, base, status, created_at, updated_at, merged_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)", params![ rec.id, rec.project_id, rec.task_id, rec.name, rec.base, rec.status, rec.created_at, rec.updated_at, rec.merged_at ], ) }, update => |conn, rec| { conn.execute( "UPDATE branches SET project_id = ?1, task_id = ?2, name = ?3, base = ?4, status = ?5, updated_at = ?6, merged_at = ?7 WHERE id = ?8", params![ rec.project_id, rec.task_id, rec.name, rec.base, rec.status, rec.updated_at, rec.merged_at, rec.id ], ) } ); impl_repo!( /// 发布表 CRUD ReleaseRepo, ReleaseRecord, "releases", from_row => |row| release_from_row(row), insert => |conn, rec| { conn.execute( "INSERT INTO releases (id, project_id, version, status, task_ids, changelog, created_at, released_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", params![ rec.id, rec.project_id, rec.version, rec.status, rec.task_ids, rec.changelog, rec.created_at, rec.released_at ], ) }, update => |conn, rec| { conn.execute( "UPDATE releases SET project_id = ?1, version = ?2, status = ?3, task_ids = ?4, changelog = ?5, released_at = ?6 WHERE id = ?7", params![ rec.project_id, rec.version, rec.status, rec.task_ids, rec.changelog, rec.released_at, rec.id ], ) } ); impl_repo!( /// 工作流执行表 CRUD WorkflowRepo, WorkflowRecord, "workflow_executions", from_row => |row| workflow_from_row(row), insert => |conn, rec| { conn.execute( "INSERT INTO workflow_executions (id, name, dag_json, status, triggered_by, project_id, task_id, created_at, completed_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)", params![ rec.id, rec.name, rec.dag_json, rec.status, rec.triggered_by, rec.project_id, rec.task_id, rec.created_at, rec.completed_at ], ) }, update => |conn, rec| { conn.execute( "UPDATE workflow_executions SET name = ?1, dag_json = ?2, status = ?3, triggered_by = ?4, project_id = ?5, task_id = ?6, completed_at = ?7 WHERE id = ?8", params![ rec.name, rec.dag_json, rec.status, rec.triggered_by, rec.project_id, rec.task_id, rec.completed_at, rec.id ], ) } ); impl_repo!( /// 节点执行表 CRUD NodeExecutionRepo, NodeExecutionRecord, "node_executions", from_row => |row| node_execution_from_row(row), insert => |conn, rec| { conn.execute( "INSERT INTO node_executions (id, workflow_id, node_id, node_type, status, input_json, output_json, error_message, started_at, completed_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)", params![ rec.id, rec.workflow_id, rec.node_id, rec.node_type, rec.status, rec.input_json, rec.output_json, rec.error_message, rec.started_at, rec.completed_at ], ) }, update => |conn, rec| { conn.execute( "UPDATE node_executions SET workflow_id = ?1, node_id = ?2, node_type = ?3, status = ?4, input_json = ?5, output_json = ?6, error_message = ?7, started_at = ?8, completed_at = ?9 WHERE id = ?10", params![ rec.workflow_id, rec.node_id, rec.node_type, rec.status, rec.input_json, rec.output_json, rec.error_message, rec.started_at, rec.completed_at, rec.id ], ) } );