重构: crud.rs按表拆分(SMELL-P1-9) + F-09批1 PerConvState数据结构
- SMELL-P1-9: crud.rs 2212行→crud/6文件(mod/settings/project_repo/task_repo/conversation_repo/idea_repo) re-export pub use *_repo::* 零调用方改动,宏 pub(crate) use + 子模块 use super::impl_repo 基线测试锁12表白名单+13Repo构造 - F-09 批1: PerConvState struct(9字段对齐AiSession::new)+AiSession.per_conv HashMap+conv()/conv_read()访问器+3单测 纯新增无行为变化,b-1主代自主裁决采纳,批2迁移承接 主代统一兜底: cargo check --workspace 0 + df-storage 35+11 + devflow 96 passed
This commit is contained in:
File diff suppressed because it is too large
Load Diff
453
crates/df-storage/src/crud/conversation_repo.rs
Normal file
453
crates/df-storage/src/crud/conversation_repo.rs
Normal file
@@ -0,0 +1,453 @@
|
||||
//! 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")?,
|
||||
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")?,
|
||||
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, created_at, updated_at)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
|
||||
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.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, updated_at = ?10 WHERE id = ?11",
|
||||
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.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, 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)",
|
||||
params![
|
||||
rec.id, rec.conversation_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, tool_call_id = ?2, tool_name = ?3, arguments = ?4, result = ?5, status = ?6, risk_level = ?7, requested_at = ?8, executed_at = ?9, decided_by = ?10 WHERE id = ?11",
|
||||
params![
|
||||
rec.conversation_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)?
|
||||
}
|
||||
|
||||
/// 设置置顶标记(仅改 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");
|
||||
}
|
||||
}
|
||||
871
crates/df-storage/src/crud/idea_repo.rs
Normal file
871
crates/df-storage/src/crud/idea_repo.rs
Normal file
@@ -0,0 +1,871 @@
|
||||
//! 想法/知识域 Repo:IdeaRepo / KnowledgeRepo / KnowledgeEventsRepo + 向量工具(embedding BLOB 序列化 + 余弦相似度)
|
||||
|
||||
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::{IdeaRecord, KnowledgeEventRecord, KnowledgeRecord};
|
||||
|
||||
use super::impl_repo;
|
||||
use super::{now_millis_str, storage_err, validate_column_name};
|
||||
|
||||
// ============================================================
|
||||
// 知识库 SELECT 列清单(防 COLS 漂移)
|
||||
// ============================================================
|
||||
|
||||
/// `knowledges` 表对应 `KnowledgeRecord` 14 个字段的列名(顺序与结构体一致)。
|
||||
///
|
||||
/// 多处 `search`/`search_vector` 内联 COLS 串的 DRY 收口(CR-260615-03):集中一处定义,
|
||||
/// 配合下方 `KNOWLEDGE_COL_COUNT` 断言,任一处加列漏改会被测试 `test_knowledge_cols_matches_record`
|
||||
/// 立即捕获(`knowledge_from_row` 按 name 取列,SELECT 漏列会运行时 rusqlite 报错,故提前断言)。
|
||||
const KNOWLEDGE_COLS: &str = "id,kind,title,content,tags,status,confidence,reuse_count,verified,source_project,source_ref,reasoning,created_at,updated_at";
|
||||
/// `KnowledgeRecord` 字段数(与上面列清单的逗号分隔项数一致,被测试断言)。
|
||||
/// 仅测试期消费(列漂移断言);保留为非 `cfg(test)` 以便测试外的阅读者一眼看到字段数。
|
||||
#[cfg_attr(not(test), allow(dead_code))]
|
||||
const KNOWLEDGE_COL_COUNT: usize = 14;
|
||||
/// `search_vector` 用的列清单:KNOWLEDGE_COLS + embedding(余弦计算用,不入 KnowledgeRecord)。
|
||||
const KNOWLEDGE_COLS_WITH_EMBEDDING: &str = concat!(
|
||||
"id,kind,title,content,tags,status,confidence,reuse_count,verified,",
|
||||
"source_project,source_ref,reasoning,created_at,updated_at,embedding"
|
||||
);
|
||||
|
||||
// ============================================================
|
||||
// 向量工具 — embedding BLOB 序列化 + 余弦相似度
|
||||
// ============================================================
|
||||
|
||||
/// Vec<f32> → 小端字节 BLOB
|
||||
fn f32s_to_blob(v: &[f32]) -> Vec<u8> {
|
||||
v.iter().flat_map(|f| f.to_le_bytes()).collect()
|
||||
}
|
||||
|
||||
/// BLOB → Vec<f32>(长度非 4 倍数时截断尾部残字节)
|
||||
fn blob_to_f32s(blob: &[u8]) -> Vec<f32> {
|
||||
blob.chunks_exact(4)
|
||||
.map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]]))
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// 余弦相似度(确定性数学,与任何实现结果一致)
|
||||
fn cosine_similarity(a: &[f32], b: &[f32]) -> f32 {
|
||||
let dot: f32 = a.iter().zip(b).map(|(x, y)| x * y).sum();
|
||||
let norm_a: f32 = a.iter().map(|x| x * x).sum::<f32>().sqrt();
|
||||
let norm_b: f32 = b.iter().map(|x| x * x).sum::<f32>().sqrt();
|
||||
dot / (norm_a * norm_b + 1e-8)
|
||||
}
|
||||
|
||||
// ============================================================
|
||||
// from_row 辅助函数
|
||||
// ============================================================
|
||||
|
||||
fn idea_from_row(row: &Row<'_>) -> std::result::Result<IdeaRecord, rusqlite::Error> {
|
||||
Ok(IdeaRecord {
|
||||
id: row.get("id")?,
|
||||
title: row.get("title")?,
|
||||
description: row.get("description")?,
|
||||
status: row.get("status")?,
|
||||
priority: row.get("priority")?,
|
||||
score: row.get("score")?,
|
||||
tags: row.get("tags")?,
|
||||
source: row.get("source")?,
|
||||
promoted_to: row.get("promoted_to")?,
|
||||
ai_analysis: row.get("ai_analysis")?,
|
||||
scores: row.get("scores")?,
|
||||
created_at: row.get("created_at")?,
|
||||
updated_at: row.get("updated_at")?,
|
||||
})
|
||||
}
|
||||
|
||||
fn knowledge_from_row(row: &Row<'_>) -> std::result::Result<KnowledgeRecord, rusqlite::Error> {
|
||||
Ok(KnowledgeRecord {
|
||||
id: row.get("id")?,
|
||||
kind: row.get("kind")?,
|
||||
title: row.get("title")?,
|
||||
content: row.get("content")?,
|
||||
tags: row.get("tags")?,
|
||||
status: row.get("status")?,
|
||||
confidence: row.get("confidence")?,
|
||||
reuse_count: row.get("reuse_count")?,
|
||||
verified: row.get::<_, i32>("verified")? != 0,
|
||||
source_project: row.get("source_project")?,
|
||||
source_ref: row.get("source_ref")?,
|
||||
reasoning: row.get("reasoning")?,
|
||||
created_at: row.get("created_at")?,
|
||||
updated_at: row.get("updated_at")?,
|
||||
})
|
||||
}
|
||||
|
||||
fn knowledge_event_from_row(row: &Row<'_>) -> std::result::Result<KnowledgeEventRecord, rusqlite::Error> {
|
||||
Ok(KnowledgeEventRecord {
|
||||
id: row.get("id")?,
|
||||
knowledge_id: row.get("knowledge_id")?,
|
||||
event_type: row.get("event_type")?,
|
||||
source_ref: row.get("source_ref")?,
|
||||
context_json: row.get("context_json")?,
|
||||
timestamp: row.get("timestamp")?,
|
||||
})
|
||||
}
|
||||
|
||||
// ============================================================
|
||||
// Repo 实现
|
||||
// ============================================================
|
||||
|
||||
impl_repo!(
|
||||
/// 想法表 CRUD
|
||||
IdeaRepo,
|
||||
IdeaRecord,
|
||||
"ideas",
|
||||
from_row => |row| idea_from_row(row),
|
||||
insert => |conn, rec| {
|
||||
conn.execute(
|
||||
"INSERT INTO ideas (id, title, description, status, priority, score, tags, source, promoted_to, ai_analysis, scores, created_at, updated_at)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
|
||||
params![
|
||||
rec.id, rec.title, rec.description, rec.status, rec.priority,
|
||||
rec.score, rec.tags, rec.source, rec.promoted_to, rec.ai_analysis,
|
||||
rec.scores, rec.created_at, rec.updated_at
|
||||
],
|
||||
)
|
||||
},
|
||||
update => |conn, rec| {
|
||||
conn.execute(
|
||||
"UPDATE ideas SET title = ?1, description = ?2, status = ?3, priority = ?4, score = ?5, tags = ?6, source = ?7, promoted_to = ?8, ai_analysis = ?9, scores = ?10, updated_at = ?11 WHERE id = ?12",
|
||||
params![
|
||||
rec.title, rec.description, rec.status, rec.priority,
|
||||
rec.score, rec.tags, rec.source, rec.promoted_to, rec.ai_analysis,
|
||||
rec.scores, rec.updated_at, rec.id
|
||||
],
|
||||
)
|
||||
}
|
||||
);
|
||||
|
||||
impl_repo!(
|
||||
/// 知识库表 CRUD
|
||||
KnowledgeRepo,
|
||||
KnowledgeRecord,
|
||||
"knowledges",
|
||||
from_row => |row| knowledge_from_row(row),
|
||||
insert => |conn, rec| {
|
||||
let verified = if rec.verified { 1i32 } else { 0i32 };
|
||||
conn.execute(
|
||||
"INSERT INTO knowledges (id, kind, title, content, tags, status, confidence, reuse_count, verified, source_project, source_ref, reasoning, created_at, updated_at)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)",
|
||||
params![
|
||||
rec.id, rec.kind, rec.title, rec.content, rec.tags, rec.status, rec.confidence,
|
||||
rec.reuse_count, verified, rec.source_project, rec.source_ref, rec.reasoning,
|
||||
rec.created_at, rec.updated_at
|
||||
],
|
||||
)
|
||||
},
|
||||
update => |conn, rec| {
|
||||
let verified = if rec.verified { 1i32 } else { 0i32 };
|
||||
conn.execute(
|
||||
"UPDATE knowledges SET kind = ?1, title = ?2, content = ?3, tags = ?4, status = ?5, confidence = ?6, reuse_count = ?7, verified = ?8, source_project = ?9, source_ref = ?10, reasoning = ?11, updated_at = ?12 WHERE id = ?13",
|
||||
params![
|
||||
rec.kind, rec.title, rec.content, rec.tags, rec.status, rec.confidence,
|
||||
rec.reuse_count, verified, rec.source_project, rec.source_ref, rec.reasoning,
|
||||
rec.updated_at, rec.id
|
||||
],
|
||||
)
|
||||
}
|
||||
);
|
||||
|
||||
// KnowledgeRepo 的整体更新已由 impl_repo! 宏统一生成的 update_full 提供。
|
||||
|
||||
impl KnowledgeRepo {
|
||||
/// 检索知识: title/content LIKE 匹配,可选 kind 过滤,按 reuse_count 降序,top-N
|
||||
///
|
||||
/// 克制检索: top-N≤3(由调用方 limit 控制),精确匹配优先(语义模糊后做)。
|
||||
pub async fn search(&self, query: &str, kind: Option<&str>, limit: usize) -> Result<Vec<KnowledgeRecord>> {
|
||||
let conn = self.conn.clone();
|
||||
let pattern = format!("%{}%", query);
|
||||
let kind = kind.map(|s| s.to_owned());
|
||||
let limit_i = limit as i64;
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let guard = conn.blocking_lock();
|
||||
let mut results = Vec::new();
|
||||
if let Some(k) = &kind {
|
||||
let mut stmt = guard
|
||||
.prepare(&format!("SELECT {KNOWLEDGE_COLS} FROM knowledges WHERE status = 'published' AND (title LIKE ?1 OR content LIKE ?2) AND kind = ?3 ORDER BY reuse_count DESC LIMIT ?4"))
|
||||
.map_err(storage_err)?;
|
||||
let rows = stmt
|
||||
.query_map(params![pattern, pattern, k, limit_i], |row| knowledge_from_row(row))
|
||||
.map_err(storage_err)?;
|
||||
for r in rows {
|
||||
results.push(r.map_err(storage_err)?);
|
||||
}
|
||||
} else {
|
||||
let mut stmt = guard
|
||||
.prepare(&format!("SELECT {KNOWLEDGE_COLS} FROM knowledges WHERE status = 'published' AND (title LIKE ?1 OR content LIKE ?2) ORDER BY reuse_count DESC LIMIT ?3"))
|
||||
.map_err(storage_err)?;
|
||||
let rows = stmt
|
||||
.query_map(params![pattern, pattern, limit_i], |row| knowledge_from_row(row))
|
||||
.map_err(storage_err)?;
|
||||
for r in rows {
|
||||
results.push(r.map_err(storage_err)?);
|
||||
}
|
||||
}
|
||||
Ok(results)
|
||||
})
|
||||
.await
|
||||
.map_err(storage_err)?
|
||||
}
|
||||
|
||||
/// 按状态列出(审核收件箱用): 按 confidence 语义排序(high>medium>low),次按 created_at
|
||||
///
|
||||
/// 用 CASE WHEN 替代纯 TEXT 排序(字典序 high>low>medium 非预期语义)。
|
||||
pub async fn list_by_status(&self, status: &str) -> Result<Vec<KnowledgeRecord>> {
|
||||
let conn = self.conn.clone();
|
||||
let status = status.to_owned();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let guard = conn.blocking_lock();
|
||||
let mut stmt = guard
|
||||
.prepare(
|
||||
"SELECT id,kind,title,content,tags,status,confidence,reuse_count,verified,\
|
||||
source_project,source_ref,reasoning,created_at,updated_at \
|
||||
FROM knowledges WHERE status = ?1
|
||||
ORDER BY CASE confidence
|
||||
WHEN 'high' THEN 3
|
||||
WHEN 'medium' THEN 2
|
||||
WHEN 'low' THEN 1
|
||||
ELSE 0
|
||||
END DESC, created_at DESC",
|
||||
)
|
||||
.map_err(storage_err)?;
|
||||
let rows = stmt
|
||||
.query_map(params![status], |row| knowledge_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)?
|
||||
}
|
||||
|
||||
/// 复用计数 +1(SQL 行级原子操作,并发安全)
|
||||
pub async fn increment_reuse_count(&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 knowledges SET reuse_count = reuse_count + 1, updated_at = ?1 WHERE id = ?2",
|
||||
params![now, id],
|
||||
)
|
||||
.map_err(storage_err)?;
|
||||
Ok(affected > 0)
|
||||
})
|
||||
.await
|
||||
.map_err(storage_err)?
|
||||
}
|
||||
|
||||
/// 写入向量嵌入(BLOB = Vec<f32> 小端字节序列化)
|
||||
///
|
||||
/// embedding 列不进 KnowledgeRecord(IPC 不需要传向量给前端),专用方法读写。
|
||||
pub async fn set_embedding(&self, id: &str, embedding: &[f32]) -> Result<bool> {
|
||||
let conn = self.conn.clone();
|
||||
let id = id.to_owned();
|
||||
let blob = f32s_to_blob(embedding);
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let guard = conn.blocking_lock();
|
||||
let affected = guard
|
||||
.execute(
|
||||
"UPDATE knowledges SET embedding = ?1 WHERE id = ?2",
|
||||
params![blob, id],
|
||||
)
|
||||
.map_err(storage_err)?;
|
||||
Ok(affected > 0)
|
||||
})
|
||||
.await
|
||||
.map_err(storage_err)?
|
||||
}
|
||||
|
||||
/// 向量检索: 加载全部 published 且有 embedding 的记录,纯 Rust 余弦相似度取 top-N
|
||||
///
|
||||
/// 返回 (记录, 相似度分数)。数据量 <50k 时暴力遍历 <50ms,够用;
|
||||
/// 更大规模再升 sqlite-vec HNSW(结果不变,只提速)。
|
||||
///
|
||||
/// 列限定: 显式列出所需列(与 search/list_by_status/top_used 一致),避免 SELECT *
|
||||
/// 拉到未知新增列;embedding 单独取(不入 KnowledgeRecord)。
|
||||
///
|
||||
/// TODO(性能,低优先): 调用方(hybrid_search→merge_hybrid_results→build_knowledge_context)
|
||||
/// 实际只消费 id/kind/title/content/reuse_count;reasoning(AI 生成大文本)、tags、
|
||||
/// source_project/source_ref 等元字段未被使用却仍随每行读出。真正省 IO 需返回精简结构
|
||||
/// (如 KnowledgeVectorHit { id, kind, title, content, reuse_count })替换返回类型,
|
||||
/// 但这会改变 search_vector 签名与 merge_hybrid_results 调用契约——当前保守不动,
|
||||
/// 待向量检索量级或 reasoning 文本体积成为瓶颈再单独立项。SELECT 列化本身不省字段,
|
||||
/// 仅消除 SELECT * 的隐式依赖与未知列风险。
|
||||
pub async fn search_vector(&self, query_vec: &[f32], limit: usize) -> Result<Vec<(KnowledgeRecord, f32)>> {
|
||||
let conn = self.conn.clone();
|
||||
let query_vec = query_vec.to_vec();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let guard = conn.blocking_lock();
|
||||
// 显式列: 14 个 KnowledgeRecord 字段 + embedding(余弦计算用,不入 KnowledgeRecord)
|
||||
let mut stmt = guard
|
||||
.prepare(&format!(
|
||||
"SELECT {KNOWLEDGE_COLS_WITH_EMBEDDING} FROM knowledges WHERE status = 'published' AND embedding IS NOT NULL"
|
||||
))
|
||||
.map_err(storage_err)?;
|
||||
let rows = stmt
|
||||
.query_map([], |row| {
|
||||
let rec = knowledge_from_row(row)?;
|
||||
let blob: Vec<u8> = row.get("embedding")?;
|
||||
Ok((rec, blob))
|
||||
})
|
||||
.map_err(storage_err)?;
|
||||
|
||||
let mut scored: Vec<(KnowledgeRecord, f32)> = Vec::new();
|
||||
for r in rows {
|
||||
let (rec, blob) = r.map_err(storage_err)?;
|
||||
let emb = blob_to_f32s(&blob);
|
||||
// 维度不匹配(换过 embedding 模型的旧向量)跳过
|
||||
if emb.len() != query_vec.len() {
|
||||
continue;
|
||||
}
|
||||
let score = cosine_similarity(&query_vec, &emb);
|
||||
scored.push((rec, score));
|
||||
}
|
||||
scored.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
|
||||
scored.truncate(limit);
|
||||
Ok(scored)
|
||||
})
|
||||
.await
|
||||
.map_err(storage_err)?
|
||||
}
|
||||
|
||||
/// 列出非归档知识(全部 status != 'archived'),按 confidence 语义排序
|
||||
pub async fn list_non_archived(&self) -> Result<Vec<KnowledgeRecord>> {
|
||||
let conn = self.conn.clone();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let guard = conn.blocking_lock();
|
||||
let mut stmt = guard
|
||||
.prepare(
|
||||
"SELECT id,kind,title,content,tags,status,confidence,reuse_count,verified,\
|
||||
source_project,source_ref,reasoning,created_at,updated_at \
|
||||
FROM knowledges WHERE status != 'archived'
|
||||
ORDER BY CASE confidence
|
||||
WHEN 'high' THEN 3
|
||||
WHEN 'medium' THEN 2
|
||||
WHEN 'low' THEN 1
|
||||
ELSE 0
|
||||
END DESC, created_at DESC",
|
||||
)
|
||||
.map_err(storage_err)?;
|
||||
let rows = stmt
|
||||
.query_map([], |row| knowledge_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)?
|
||||
}
|
||||
|
||||
/// 热门知识(已发布,按复用次数降序)
|
||||
pub async fn top_used(&self, limit: usize) -> Result<Vec<KnowledgeRecord>> {
|
||||
let conn = self.conn.clone();
|
||||
let limit_i = limit as i64;
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let guard = conn.blocking_lock();
|
||||
let mut stmt = guard
|
||||
.prepare("SELECT id,kind,title,content,tags,status,confidence,reuse_count,verified,source_project,source_ref,reasoning,created_at,updated_at FROM knowledges WHERE status = 'published' ORDER BY reuse_count DESC LIMIT ?1")
|
||||
.map_err(storage_err)?;
|
||||
let rows = stmt
|
||||
.query_map(params![limit_i], |row| knowledge_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)?
|
||||
}
|
||||
}
|
||||
|
||||
impl_repo!(
|
||||
/// 知识生命线事件表 CRUD(追加型审计表)
|
||||
KnowledgeEventsRepo,
|
||||
KnowledgeEventRecord,
|
||||
"knowledge_events",
|
||||
from_row => |row| knowledge_event_from_row(row),
|
||||
insert => |conn, rec| {
|
||||
conn.execute(
|
||||
"INSERT INTO knowledge_events (id, knowledge_id, event_type, source_ref, context_json, timestamp)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
|
||||
params![rec.id, rec.knowledge_id, rec.event_type, rec.source_ref, rec.context_json, rec.timestamp],
|
||||
)
|
||||
},
|
||||
update => |conn, rec| {
|
||||
conn.execute(
|
||||
"UPDATE knowledge_events SET knowledge_id = ?1, event_type = ?2, source_ref = ?3, context_json = ?4, timestamp = ?5 WHERE id = ?6",
|
||||
params![rec.knowledge_id, rec.event_type, rec.source_ref, rec.context_json, rec.timestamp, rec.id],
|
||||
)
|
||||
}
|
||||
);
|
||||
|
||||
impl KnowledgeEventsRepo {
|
||||
/// 按知识 ID 查询全部事件(时间正序,构建生命线视图用)
|
||||
pub async fn list_by_knowledge(&self, knowledge_id: &str) -> Result<Vec<KnowledgeEventRecord>> {
|
||||
let conn = self.conn.clone();
|
||||
let knowledge_id = knowledge_id.to_owned();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let guard = conn.blocking_lock();
|
||||
let mut stmt = guard
|
||||
.prepare("SELECT id,knowledge_id,event_type,source_ref,context_json,timestamp FROM knowledge_events WHERE knowledge_id = ?1 ORDER BY timestamp ASC")
|
||||
.map_err(storage_err)?;
|
||||
let rows = stmt
|
||||
.query_map(params![knowledge_id], |row| knowledge_event_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)?
|
||||
}
|
||||
|
||||
/// 按知识 ID + event_type 查询最近 N 条(如引用记录翻页: type=referenced)
|
||||
pub async fn list_by_knowledge_type(
|
||||
&self,
|
||||
knowledge_id: &str,
|
||||
event_type: &str,
|
||||
limit: usize,
|
||||
) -> Result<Vec<KnowledgeEventRecord>> {
|
||||
let conn = self.conn.clone();
|
||||
let knowledge_id = knowledge_id.to_owned();
|
||||
let event_type = event_type.to_owned();
|
||||
let limit_i = limit as i64;
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let guard = conn.blocking_lock();
|
||||
let mut stmt = guard
|
||||
.prepare("SELECT id,knowledge_id,event_type,source_ref,context_json,timestamp FROM knowledge_events WHERE knowledge_id = ?1 AND event_type = ?2 ORDER BY timestamp DESC LIMIT ?3")
|
||||
.map_err(storage_err)?;
|
||||
let rows = stmt
|
||||
.query_map(params![knowledge_id, event_type, limit_i], |row| knowledge_event_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)?
|
||||
}
|
||||
|
||||
/// 跨知识列最近 N 条事件(全表 timestamp DESC,top-N)。
|
||||
///
|
||||
/// **专用兜底方法**:本表时间列名是 `timestamp` 而非 `created_at`,但
|
||||
/// `impl_repo!` 宏生成的 `query()` / `list_all()` 硬编码 `ORDER BY created_at`
|
||||
/// (见宏内 `ORDER BY created_at DESC` 字面量),误调 `state.knowledge_events.query(...)`
|
||||
/// 会触发 SQLite "no such column: created_at"。本方法走专用 SELECT 绕过宏硬编码,
|
||||
/// 供需要跨知识按时间倒序浏览事件的调用方使用(对标 AiToolExecutionRepo::list_recent
|
||||
/// 对 ai_tool_executions 表的同款兜底处理——那张表同样无 created_at 列)。
|
||||
/// limit 上限钳制 200,防前端恶意/失误传超大值。
|
||||
pub async fn list_recent(&self, limit: u32) -> Result<Vec<KnowledgeEventRecord>> {
|
||||
let conn = self.conn.clone();
|
||||
// 钳制 limit 防滥用(最大 200)
|
||||
let safe_limit = limit.min(200) as i64;
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let guard = conn.blocking_lock();
|
||||
let mut stmt = guard
|
||||
.prepare(
|
||||
"SELECT id,knowledge_id,event_type,source_ref,context_json,timestamp \
|
||||
FROM knowledge_events ORDER BY timestamp DESC LIMIT ?1",
|
||||
)
|
||||
.map_err(storage_err)?;
|
||||
let rows = stmt
|
||||
.query_map(params![safe_limit], |row| knowledge_event_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)?
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================
|
||||
// 单元测试 — 向量纯函数 + KnowledgeRepo 内存 DB + COLS 漂移防护
|
||||
// ============================================================
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::db::Database;
|
||||
|
||||
// ---------- COLS 漂移防护(CR-260615-03) ----------
|
||||
|
||||
/// KNOWLEDGE_COLS 列数须等于 KNOWLEDGE_COL_COUNT(任一处漂移:加列漏改 / 串错位 → 立即失败)。
|
||||
/// `knowledge_from_row` 按 name 取列,SELECT 漏列会在运行时被 rusqlite 报错;此断言提前到测试期捕获。
|
||||
#[test]
|
||||
fn test_knowledge_cols_matches_record() {
|
||||
let count = KNOWLEDGE_COLS.split(',').count();
|
||||
assert_eq!(
|
||||
count, KNOWLEDGE_COL_COUNT,
|
||||
"KNOWLEDGE_COLS({count}列) ≠ KNOWLEDGE_COL_COUNT({KNOWLEDGE_COL_COUNT}); \
|
||||
修改一处须同步另一处"
|
||||
);
|
||||
// search_vector 多一列 embedding
|
||||
let count_with_emb = KNOWLEDGE_COLS_WITH_EMBEDDING.split(',').count();
|
||||
assert_eq!(
|
||||
count_with_emb,
|
||||
KNOWLEDGE_COL_COUNT + 1,
|
||||
"KNOWLEDGE_COLS_WITH_EMBEDDING({count_with_emb}列) ≠ KNOWLEDGE_COL_COUNT+1({}); \
|
||||
embedding 列应单独追加",
|
||||
KNOWLEDGE_COL_COUNT + 1
|
||||
);
|
||||
// 每个列名须能被 split 出来(防末尾多逗号 / 空段)
|
||||
for col in KNOWLEDGE_COLS.split(',') {
|
||||
assert!(!col.is_empty(), "KNOWLEDGE_COLS 含空列名段");
|
||||
}
|
||||
}
|
||||
|
||||
// ---------- 向量纯函数 ----------
|
||||
|
||||
#[test]
|
||||
fn f32s_blob_roundtrip_nonempty() {
|
||||
let v = vec![0.0, 1.5, -2.25, 3.14159, -0.0001];
|
||||
let blob = f32s_to_blob(&v);
|
||||
assert_eq!(blob.len(), v.len() * 4);
|
||||
let back = blob_to_f32s(&blob);
|
||||
assert_eq!(back, v);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn f32s_blob_roundtrip_empty() {
|
||||
let v: Vec<f32> = vec![];
|
||||
let blob = f32s_to_blob(&v);
|
||||
assert!(blob.is_empty());
|
||||
assert!(blob_to_f32s(&blob).is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn blob_to_f32s_drops_trailing_partial_bytes() {
|
||||
// 1 个完整 f32 + 3 残字节 → chunks_exact 截断尾部
|
||||
let blob = f32s_to_blob(&[42.0]);
|
||||
let with_garbage: Vec<u8> = blob.into_iter().chain([0xff, 0xff, 0xff]).collect();
|
||||
assert_eq!(blob_to_f32s(&with_garbage), vec![42.0]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn cosine_identical_vectors_near_one() {
|
||||
let a = vec![1.0, 2.0, 3.0];
|
||||
let sim = cosine_similarity(&a, &a);
|
||||
assert!((sim - 1.0).abs() < 1e-5, "identical ≈ 1.0, got {}", sim);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn cosine_orthogonal_vectors_near_zero() {
|
||||
let a = vec![1.0, 0.0];
|
||||
let b = vec![0.0, 1.0];
|
||||
let sim = cosine_similarity(&a, &b);
|
||||
assert!(sim.abs() < 1e-5, "orthogonal ≈ 0.0, got {}", sim);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn cosine_opposite_vectors_near_neg_one() {
|
||||
let a = vec![1.0, 2.0, 3.0];
|
||||
let b = vec![-1.0, -2.0, -3.0];
|
||||
let sim = cosine_similarity(&a, &b);
|
||||
assert!((sim + 1.0).abs() < 1e-5, "opposite ≈ -1.0, got {}", sim);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn cosine_dimension_mismatch_zips_to_shortest() {
|
||||
// 实现用 zip(a,b),维度不匹配按较短的那个对齐,不 panic
|
||||
let a = vec![1.0, 0.0, 0.0]; // 3 维
|
||||
let b = vec![1.0, 0.0]; // 2 维 → zip 取前 2 个
|
||||
let sim = cosine_similarity(&a, &b);
|
||||
assert!((sim - 1.0).abs() < 1e-5, "zip 截断后应 ≈ 1.0, got {}", sim);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn cosine_zero_vector_does_not_panic() {
|
||||
// 分母有 +1e-8 保护,零向量返回有限值而非 NaN/inf
|
||||
let z = vec![0.0, 0.0, 0.0];
|
||||
let sim = cosine_similarity(&z, &z);
|
||||
assert!(sim.is_finite(), "零向量相似度应有限, got {}", sim);
|
||||
}
|
||||
|
||||
// ---------- KnowledgeRepo 内存 DB ----------
|
||||
|
||||
/// 构造一条 KnowledgeRecord fixture
|
||||
fn krec(
|
||||
id: &str,
|
||||
title: &str,
|
||||
content: &str,
|
||||
kind: &str,
|
||||
status: &str,
|
||||
confidence: Option<&str>,
|
||||
reuse: i32,
|
||||
) -> KnowledgeRecord {
|
||||
KnowledgeRecord {
|
||||
id: id.to_string(),
|
||||
kind: kind.to_string(),
|
||||
title: title.to_string(),
|
||||
content: content.to_string(),
|
||||
tags: Some("[]".to_string()),
|
||||
status: status.to_string(),
|
||||
confidence: confidence.map(|s| s.to_string()),
|
||||
reuse_count: reuse,
|
||||
verified: false,
|
||||
source_project: Some("proj-1".to_string()),
|
||||
source_ref: Some("conv:c1".to_string()),
|
||||
reasoning: None,
|
||||
created_at: "1700000000000".to_string(),
|
||||
updated_at: "1700000000000".to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
async fn setup_repo() -> KnowledgeRepo {
|
||||
let db = Database::open_in_memory().await.expect("open_in_memory");
|
||||
KnowledgeRepo::new(&db)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_matches_title_and_filters_non_published() {
|
||||
let repo = setup_repo().await;
|
||||
// 一条命中(title 含关键词)、一条 published 不命中、一条 status 非 published 但命中
|
||||
repo.insert(krec("k1", "Rust 异步并发模型", "tokio 运行时", "lesson", "published", Some("high"), 5))
|
||||
.await
|
||||
.unwrap();
|
||||
repo.insert(krec("k2", "无关标题", "无关内容", "lesson", "published", None, 0))
|
||||
.await
|
||||
.unwrap();
|
||||
repo.insert(krec("k3", "Rust 异步进阶", "...", "lesson", "candidate", Some("low"), 9))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let hits = repo.search("异步", None, 10).await.unwrap();
|
||||
// 只有 k1 是 published 且命中;k3 命中但非 published 被过滤
|
||||
assert_eq!(hits.len(), 1);
|
||||
assert_eq!(hits[0].id, "k1");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_matches_content_branch() {
|
||||
let repo = setup_repo().await;
|
||||
// title 不含关键词,content 含 → 命中 content LIKE 分支
|
||||
repo.insert(krec("k1", "标题", "深入理解 Rust 所有权与借用", "lesson", "published", None, 1))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let hits = repo.search("所有权", None, 10).await.unwrap();
|
||||
assert_eq!(hits.len(), 1);
|
||||
assert_eq!(hits[0].id, "k1");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_no_keyword_match_returns_empty() {
|
||||
let repo = setup_repo().await;
|
||||
repo.insert(krec("k1", "Rust", "tokio", "lesson", "published", None, 1))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let hits = repo.search("不存在的关键词xyz", None, 10).await.unwrap();
|
||||
assert!(hits.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_kind_filter_narrows_results() {
|
||||
let repo = setup_repo().await;
|
||||
repo.insert(krec("k1", "Rust 规范", "...", "lesson", "published", None, 1))
|
||||
.await
|
||||
.unwrap();
|
||||
repo.insert(krec("k2", "Rust 决策", "...", "decision", "published", None, 1))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// 不过滤 kind → 两条都命中
|
||||
let all = repo.search("Rust", None, 10).await.unwrap();
|
||||
assert_eq!(all.len(), 2);
|
||||
// 只取 lesson → 仅 k1
|
||||
let only_lesson = repo.search("Rust", Some("lesson"), 10).await.unwrap();
|
||||
assert_eq!(only_lesson.len(), 1);
|
||||
assert_eq!(only_lesson[0].id, "k1");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_orders_by_reuse_count_desc() {
|
||||
let repo = setup_repo().await;
|
||||
repo.insert(krec("low", "Rust A", "...", "lesson", "published", None, 1))
|
||||
.await
|
||||
.unwrap();
|
||||
repo.insert(krec("high", "Rust B", "...", "lesson", "published", None, 50))
|
||||
.await
|
||||
.unwrap();
|
||||
repo.insert(krec("mid", "Rust C", "...", "lesson", "published", None, 10))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let hits = repo.search("Rust", None, 10).await.unwrap();
|
||||
let ids: Vec<_> = hits.iter().map(|h| h.id.as_str()).collect();
|
||||
assert_eq!(ids, vec!["high", "mid", "low"]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_truncates_to_limit() {
|
||||
let repo = setup_repo().await;
|
||||
for i in 0..5 {
|
||||
repo.insert(krec(&format!("k{i}"), &format!("Rust-{i}"), "...", "lesson", "published", None, i))
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
let hits = repo.search("Rust", None, 2).await.unwrap();
|
||||
assert_eq!(hits.len(), 2);
|
||||
// reuse_count 最高的两条(k4=4, k3=3)
|
||||
assert_eq!(hits[0].id, "k4");
|
||||
assert_eq!(hits[1].id, "k3");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_by_status_orders_by_confidence_semantics() {
|
||||
// confidence 字典序 high>low>medium 非预期,实现用 CASE 强制 high>medium>low
|
||||
let repo = setup_repo().await;
|
||||
repo.insert(krec("low", "t", "c", "lesson", "pending_review", Some("low"), 0))
|
||||
.await
|
||||
.unwrap();
|
||||
repo.insert(krec("high", "t", "c", "lesson", "pending_review", Some("high"), 0))
|
||||
.await
|
||||
.unwrap();
|
||||
repo.insert(krec("medium", "t", "c", "lesson", "pending_review", Some("medium"), 0))
|
||||
.await
|
||||
.unwrap();
|
||||
// created_at 相同,纯靠 confidence 排序
|
||||
|
||||
let list = repo.list_by_status("pending_review").await.unwrap();
|
||||
let confidences: Vec<_> = list.iter().map(|r| r.confidence.as_deref().unwrap_or("")).collect();
|
||||
assert_eq!(confidences, vec!["high", "medium", "low"]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_by_status_filters_other_status() {
|
||||
let repo = setup_repo().await;
|
||||
repo.insert(krec("a", "t", "c", "lesson", "pending_review", Some("high"), 0))
|
||||
.await
|
||||
.unwrap();
|
||||
repo.insert(krec("b", "t", "c", "lesson", "published", Some("high"), 0))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let list = repo.list_by_status("pending_review").await.unwrap();
|
||||
assert_eq!(list.len(), 1);
|
||||
assert_eq!(list[0].id, "a");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn increment_reuse_count_is_atomic_plus_one() {
|
||||
let repo = setup_repo().await;
|
||||
repo.insert(krec("k1", "Rust", "...", "lesson", "published", None, 0))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// 连续 +1 两次
|
||||
assert!(repo.increment_reuse_count("k1").await.unwrap());
|
||||
assert!(repo.increment_reuse_count("k1").await.unwrap());
|
||||
|
||||
let rec = repo.get_by_id("k1").await.unwrap().expect("记录存在");
|
||||
assert_eq!(rec.reuse_count, 2);
|
||||
|
||||
// 不存在的 id → false
|
||||
assert!(!repo.increment_reuse_count("nope").await.unwrap());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_vector_orders_by_cosine_and_truncates() {
|
||||
let repo = setup_repo().await;
|
||||
// 三条 published 记录,embedding 维度均为 2
|
||||
repo.insert(krec("exact", "e", "c", "lesson", "published", None, 0))
|
||||
.await
|
||||
.unwrap();
|
||||
repo.insert(krec("orth", "e", "c", "lesson", "published", None, 0))
|
||||
.await
|
||||
.unwrap();
|
||||
repo.insert(krec("neg", "e", "c", "lesson", "published", None, 0))
|
||||
.await
|
||||
.unwrap();
|
||||
// 一条非 published(应被过滤)
|
||||
repo.insert(krec("hidden", "e", "c", "lesson", "candidate", None, 0))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
repo.set_embedding("exact", &[1.0, 0.0]).await.unwrap(); // 与 query 完全同向
|
||||
repo.set_embedding("orth", &[0.0, 1.0]).await.unwrap(); // 正交 ≈ 0
|
||||
repo.set_embedding("neg", &[-1.0, 0.0]).await.unwrap(); // 反向 ≈ -1
|
||||
repo.set_embedding("hidden", &[1.0, 0.0]).await.unwrap(); // 非 published 过滤掉
|
||||
|
||||
let results = repo.search_vector(&[1.0, 0.0], 10).await.unwrap();
|
||||
// hidden 被 status 过滤,剩 3 条
|
||||
assert_eq!(results.len(), 3);
|
||||
// 按相似度降序: exact(≈1) > orth(≈0) > neg(≈-1)
|
||||
assert_eq!(results[0].0.id, "exact");
|
||||
assert_eq!(results[1].0.id, "orth");
|
||||
assert_eq!(results[2].0.id, "neg");
|
||||
assert!((results[0].1 - 1.0).abs() < 1e-5);
|
||||
assert!(results[1].1.abs() < 1e-5);
|
||||
assert!((results[2].1 + 1.0).abs() < 1e-5);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_vector_skips_dimension_mismatch() {
|
||||
let repo = setup_repo().await;
|
||||
repo.insert(krec("dim2", "e", "c", "lesson", "published", None, 0))
|
||||
.await
|
||||
.unwrap();
|
||||
repo.insert(krec("dim3", "e", "c", "lesson", "published", None, 0))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
repo.set_embedding("dim2", &[1.0, 0.0]).await.unwrap();
|
||||
repo.set_embedding("dim3", &[1.0, 0.0, 0.0]).await.unwrap(); // 维度不匹配
|
||||
|
||||
// query 是 2 维,dim3 被跳过
|
||||
let results = repo.search_vector(&[1.0, 0.0], 10).await.unwrap();
|
||||
assert_eq!(results.len(), 1);
|
||||
assert_eq!(results[0].0.id, "dim2");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_vector_truncates_to_limit() {
|
||||
let repo = setup_repo().await;
|
||||
for i in 0..4 {
|
||||
repo.insert(krec(&format!("k{i}"), "e", "c", "lesson", "published", None, 0))
|
||||
.await
|
||||
.unwrap();
|
||||
// 全部与 query 同向,相似度相同 → 仅验证 truncate 生效
|
||||
repo.set_embedding(&format!("k{i}"), &[1.0, 0.0]).await.unwrap();
|
||||
}
|
||||
let results = repo.search_vector(&[1.0, 0.0], 2).await.unwrap();
|
||||
assert_eq!(results.len(), 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_vector_ignores_records_without_embedding() {
|
||||
let repo = setup_repo().await;
|
||||
// published 但未写 embedding
|
||||
repo.insert(krec("noemb", "e", "c", "lesson", "published", None, 0))
|
||||
.await
|
||||
.unwrap();
|
||||
let results = repo.search_vector(&[1.0, 0.0], 10).await.unwrap();
|
||||
assert!(results.is_empty());
|
||||
}
|
||||
}
|
||||
277
crates/df-storage/src/crud/mod.rs
Normal file
277
crates/df-storage/src/crud/mod.rs
Normal file
@@ -0,0 +1,277 @@
|
||||
//! Repository 模式的 CRUD 操作
|
||||
//!
|
||||
//! 为 7 张表各提供一个 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@conversation_repo`]:AiProviderRepo/AiConversationRepo/AiToolExecutionRepo
|
||||
//! - [`mod@idea_repo`]:IdeaRepo/KnowledgeRepo/KnowledgeEventsRepo + 向量工具
|
||||
//!
|
||||
//! re-export(`pub use ...::*`)保持 `df_storage::crud::XxxRepo` /
|
||||
//! `df_storage::crud::is_allowed_column` 路径不变,**调用方零改动**。
|
||||
|
||||
mod conversation_repo;
|
||||
mod idea_repo;
|
||||
mod project_repo;
|
||||
mod settings;
|
||||
mod task_repo;
|
||||
|
||||
pub use conversation_repo::*;
|
||||
pub use idea_repo::*;
|
||||
pub use project_repo::*;
|
||||
pub use settings::*;
|
||||
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)。
|
||||
///
|
||||
/// 宏通过 `#[macro_use]`(见本文件末 `mod` 声明上方的文本注入)对各子模块可见,
|
||||
/// 子模块 `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())
|
||||
}
|
||||
|
||||
/// 规范化路径用于比较: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",
|
||||
];
|
||||
for t in tables {
|
||||
assert!(
|
||||
allowed_columns_for(t).is_some(),
|
||||
"表 {t} 应登记列白名单(防 SQL 注入 + 按表隔离);拆分后白名单单一来源在 settings.rs"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// 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 _ = 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);
|
||||
}
|
||||
}
|
||||
360
crates/df-storage/src/crud/project_repo.rs
Normal file
360
crates/df-storage/src/crud/project_repo.rs
Normal file
@@ -0,0 +1,360 @@
|
||||
//! 项目域 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 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};
|
||||
|
||||
// ============================================================
|
||||
// from_row 辅助函数
|
||||
// ============================================================
|
||||
|
||||
fn project_from_row(row: &Row<'_>) -> std::result::Result<ProjectRecord, rusqlite::Error> {
|
||||
Ok(ProjectRecord {
|
||||
id: row.get("id")?,
|
||||
name: row.get("name")?,
|
||||
description: row.get("description")?,
|
||||
status: row.get("status")?,
|
||||
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<BranchRecord, rusqlite::Error> {
|
||||
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<ReleaseRecord, rusqlite::Error> {
|
||||
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<WorkflowRecord, rusqlite::Error> {
|
||||
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<NodeExecutionRecord, rusqlite::Error> {
|
||||
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, 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, 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<Vec<ProjectRecord>> {
|
||||
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)?
|
||||
}
|
||||
|
||||
/// 软删:标记 deleted_at(进回收站,可恢复)。仅作用于未删项目,返回是否命中。
|
||||
pub async fn soft_delete(&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 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<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 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<Option<ProjectRecord>> {
|
||||
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<Vec<ProjectRecord>> {
|
||||
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<bool> {
|
||||
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
|
||||
],
|
||||
)
|
||||
}
|
||||
);
|
||||
206
crates/df-storage/src/crud/settings.rs
Normal file
206
crates/df-storage/src/crud/settings.rs
Normal file
@@ -0,0 +1,206 @@
|
||||
//! 通用键值设置 — app_settings 表(KV,不适合 impl_repo! 宏)+ 列白名单(防 SQL 注入)
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use rusqlite::{params, Connection, OptionalExtension};
|
||||
use tokio::sync::Mutex;
|
||||
|
||||
use df_types::error::{Error, Result};
|
||||
|
||||
use crate::db::Database;
|
||||
|
||||
use super::{now_millis_str, storage_err};
|
||||
|
||||
// ============================================================
|
||||
// 通用键值设置 — app_settings 表
|
||||
// ============================================================
|
||||
|
||||
/// 通用应用设置 KV Repo(表 `app_settings`)
|
||||
///
|
||||
/// 前端 localStorage 迁移目标:存主题/语言/AI 偏好/连接配置等,`value` 为 JSON 字符串。
|
||||
/// KV 无固定 schema 记录结构,故不走 `impl_repo!` 宏,手写更直白。
|
||||
pub struct SettingsRepo {
|
||||
conn: Arc<Mutex<Connection>>,
|
||||
}
|
||||
|
||||
impl SettingsRepo {
|
||||
pub fn new(db: &Database) -> Self {
|
||||
Self { conn: db.conn() }
|
||||
}
|
||||
|
||||
/// 取单个 key 的值(JSON 字符串),不存在返回 None
|
||||
pub async fn get(&self, key: &str) -> Result<Option<String>> {
|
||||
let conn = self.conn.clone();
|
||||
let key = key.to_owned();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let guard = conn.blocking_lock();
|
||||
let v = guard
|
||||
.query_row(
|
||||
"SELECT value FROM app_settings WHERE key = ?1",
|
||||
params![key],
|
||||
|row| row.get::<_, String>(0),
|
||||
)
|
||||
.optional()
|
||||
.map_err(storage_err)?;
|
||||
Ok(v)
|
||||
})
|
||||
.await
|
||||
.map_err(storage_err)?
|
||||
}
|
||||
|
||||
/// 写 key/value(`INSERT OR REPLACE`),刷新 updated_at
|
||||
pub async fn set(&self, key: &str, value: &str) -> Result<bool> {
|
||||
let conn = self.conn.clone();
|
||||
let key = key.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(
|
||||
"INSERT OR REPLACE INTO app_settings (key, value, updated_at) VALUES (?1, ?2, ?3)",
|
||||
params![key, value, now],
|
||||
)
|
||||
.map_err(storage_err)?;
|
||||
Ok(affected > 0)
|
||||
})
|
||||
.await
|
||||
.map_err(storage_err)?
|
||||
}
|
||||
|
||||
/// 取全部 key/value
|
||||
pub async fn get_all(&self) -> Result<Vec<(String, String)>> {
|
||||
let conn = self.conn.clone();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let guard = conn.blocking_lock();
|
||||
let mut stmt = guard
|
||||
.prepare("SELECT key, value FROM app_settings")
|
||||
.map_err(storage_err)?;
|
||||
let rows = stmt
|
||||
.query_map([], |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)))
|
||||
.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)?
|
||||
}
|
||||
|
||||
/// 删除 key,返回是否实际删除
|
||||
pub async fn delete(&self, key: &str) -> Result<bool> {
|
||||
let conn = self.conn.clone();
|
||||
let key = key.to_owned();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let guard = conn.blocking_lock();
|
||||
let affected = guard
|
||||
.execute("DELETE FROM app_settings WHERE key = ?1", params![key])
|
||||
.map_err(storage_err)?;
|
||||
Ok(affected > 0)
|
||||
})
|
||||
.await
|
||||
.map_err(storage_err)?
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================
|
||||
// 列名白名单 — 防止 SQL 注入
|
||||
// ============================================================
|
||||
|
||||
/// 按表返回允许用于 query / update_field 的列名。
|
||||
///
|
||||
/// 白名单两重作用:① 防注入(列名参数化前校验);② 按表隔离(update_task 误传
|
||||
/// projects 的 "name" 会在校验阶段拒绝,而非靠 SQLite "no such column" 兜底报错)。
|
||||
/// 专用更新路径的列不列入:knowledges.embedding(set_embedding)、projects.deleted_at
|
||||
/// (soft_delete/restore)、tasks.deleted_at(soft_delete/restore)。未登记的表返回 None
|
||||
/// → 放行(仅靠参数化防注入,向后兼容)。
|
||||
pub fn allowed_columns_for(table: &str) -> Option<&'static [&'static str]> {
|
||||
Some(match table {
|
||||
"ideas" => &[
|
||||
"id", "title", "description", "status", "priority", "score", "tags", "source",
|
||||
"promoted_to", "ai_analysis", "scores", "created_at", "updated_at",
|
||||
],
|
||||
"projects" => &[
|
||||
"id", "name", "description", "status", "idea_id", "path", "stack", "created_at",
|
||||
"updated_at",
|
||||
],
|
||||
"tasks" => &[
|
||||
// id/created_at 不列入:主键与创建时间不可通过通用 update_field 改写
|
||||
// (防篡改主键/伪造创建时间/跨项目移动)。
|
||||
// status 不列入(D-260616-04 status 收口):所有 status 改动走
|
||||
// advance_status_atomic(CAS SQL `WHERE id AND status=expected`,唯一 status 写入路径),
|
||||
// 不经通用 update_field。防 update_task IPC / AI 工具旁路改 status 绕过状态机
|
||||
// (can_transition 校验)与 review_rounds 累加。后人勿把 status 补回白名单。
|
||||
// review_rounds 不列入:它仅 advance_status_atomic 退回转换
|
||||
// (in_review→in_progress / testing→in_review)时原子 +1(收口:仅该专用路径可改,
|
||||
// update_task/update_field 白名单均不含)。后人勿把 review_rounds 补进白名单,
|
||||
// 否则破坏「review_rounds 唯一写入路径」收口、引入旁路写导致计数错乱。
|
||||
"project_id", "title", "description", "priority", "branch_name",
|
||||
"assignee", "workflow_def_id", "base_branch",
|
||||
// output_json:ai_execute 写产出 / ai_self_review 读产出自审 / human_review 展示对象
|
||||
// (决策 a:task 中心,产出跟 task 走)。非状态机收口字段,合法可写。
|
||||
"output_json",
|
||||
"updated_at",
|
||||
// TODO(B-260616-16): project_id 跨表存在性校验待 commands/task.rs 层补。
|
||||
// 通用 CRUD 层(db repo)只懂表/列语义,不持有跨表业务约束(查 projects 表存在性)。
|
||||
// project_id 当前可在白名单内改写,合法目标存在性由上层命令层校验。
|
||||
],
|
||||
"releases" => &[
|
||||
"id", "project_id", "version", "status", "task_ids", "changelog", "created_at",
|
||||
"released_at",
|
||||
],
|
||||
"branches" => &[
|
||||
"id", "project_id", "task_id", "name", "base", "status", "created_at", "updated_at",
|
||||
"merged_at",
|
||||
],
|
||||
"workflow_executions" => &[
|
||||
"id", "name", "dag_json", "status", "triggered_by", "project_id", "task_id",
|
||||
"created_at", "completed_at",
|
||||
],
|
||||
"node_executions" => &[
|
||||
"id", "workflow_id", "node_id", "node_type", "status", "input_json", "output_json",
|
||||
"error_message", "started_at", "completed_at",
|
||||
],
|
||||
"ai_providers" => &[
|
||||
"id", "name", "provider_type", "api_key", "base_url", "default_model", "models",
|
||||
"is_default", "config", "created_at", "updated_at",
|
||||
],
|
||||
"ai_conversations" => &[
|
||||
"id", "title", "messages", "provider_id", "model", "models", "archived",
|
||||
"prompt_tokens", "completion_tokens", "created_at", "updated_at",
|
||||
],
|
||||
"ai_tool_executions" => &[
|
||||
"id", "conversation_id", "tool_call_id", "tool_name", "arguments", "result",
|
||||
"status", "risk_level", "requested_at", "executed_at", "decided_by",
|
||||
],
|
||||
"knowledges" => &[
|
||||
"id", "kind", "title", "content", "tags", "status", "confidence", "reuse_count",
|
||||
"verified", "source_project", "source_ref", "reasoning", "created_at", "updated_at",
|
||||
],
|
||||
"knowledge_events" => &[
|
||||
"id", "knowledge_id", "event_type", "source_ref", "context_json", "timestamp",
|
||||
],
|
||||
_ => return None,
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn validate_column_name(field: &str, table: &str) -> Result<()> {
|
||||
match allowed_columns_for(table) {
|
||||
Some(cols) if cols.contains(&field) => Ok(()),
|
||||
Some(_) => Err(Error::Storage(format!("表 {} 不允许的字段名: {}", table, field))),
|
||||
// 未登记表保守拒绝(FR-S6: 原放行 Ok,若未来未登记表走通用查询路径,列名直进字符串拼接即 SQL 注入;与 is_allowed_column 的 None=>false 对齐)
|
||||
None => Err(Error::Storage(format!("表 {} 未登记列白名单,拒绝防注入", table))),
|
||||
}
|
||||
}
|
||||
|
||||
/// 是否允许更新某表的某列(按表隔离白名单)。未登记表返回 false(保守拒绝,防误传)。
|
||||
///
|
||||
/// 供 AI 工具层等外部调用方复用 CRUD 白名单,避免双份字段列表不同步。
|
||||
pub fn is_allowed_column(table: &str, field: &str) -> bool {
|
||||
match allowed_columns_for(table) {
|
||||
Some(cols) => cols.contains(&field),
|
||||
None => false,
|
||||
}
|
||||
}
|
||||
218
crates/df-storage/src/crud/task_repo.rs
Normal file
218
crates/df-storage/src/crud/task_repo.rs
Normal file
@@ -0,0 +1,218 @@
|
||||
//! 任务域 Repo:TaskRepo(含 advance_status_atomic 状态机收口)
|
||||
|
||||
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::TaskRecord;
|
||||
|
||||
use super::impl_repo;
|
||||
use super::{now_millis_str, storage_err, validate_column_name};
|
||||
|
||||
// ============================================================
|
||||
// from_row 辅助函数
|
||||
// ============================================================
|
||||
|
||||
fn task_from_row(row: &Row<'_>) -> std::result::Result<TaskRecord, rusqlite::Error> {
|
||||
Ok(TaskRecord {
|
||||
id: row.get("id")?,
|
||||
project_id: row.get("project_id")?,
|
||||
title: row.get("title")?,
|
||||
description: row.get("description")?,
|
||||
status: row.get("status")?,
|
||||
priority: row.get("priority")?,
|
||||
branch_name: row.get("branch_name")?,
|
||||
assignee: row.get("assignee")?,
|
||||
workflow_def_id: row.get("workflow_def_id")?,
|
||||
base_branch: row.get("base_branch")?,
|
||||
review_rounds: row.get("review_rounds")?,
|
||||
output_json: row.get("output_json")?,
|
||||
created_at: row.get("created_at")?,
|
||||
updated_at: row.get("updated_at")?,
|
||||
})
|
||||
}
|
||||
|
||||
// ============================================================
|
||||
// Repo 实现
|
||||
// ============================================================
|
||||
|
||||
impl_repo!(
|
||||
/// 任务表 CRUD
|
||||
TaskRepo,
|
||||
TaskRecord,
|
||||
"tasks",
|
||||
from_row => |row| task_from_row(row),
|
||||
insert => |conn, rec| {
|
||||
conn.execute(
|
||||
"INSERT INTO tasks (id, project_id, title, description, status, priority, branch_name, assignee, workflow_def_id, base_branch, review_rounds, output_json, created_at, updated_at)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)",
|
||||
params![
|
||||
rec.id, rec.project_id, rec.title, rec.description, rec.status, rec.priority,
|
||||
rec.branch_name, rec.assignee, rec.workflow_def_id, rec.base_branch,
|
||||
rec.review_rounds, rec.output_json, rec.created_at, rec.updated_at
|
||||
],
|
||||
)
|
||||
},
|
||||
update => |conn, rec| {
|
||||
conn.execute(
|
||||
"UPDATE tasks SET project_id = ?1, title = ?2, description = ?3, status = ?4, priority = ?5, branch_name = ?6, assignee = ?7, workflow_def_id = ?8, base_branch = ?9, review_rounds = ?10, output_json = ?11, updated_at = ?12 WHERE id = ?13",
|
||||
params![
|
||||
rec.project_id, rec.title, rec.description, rec.status, rec.priority,
|
||||
rec.branch_name, rec.assignee, rec.workflow_def_id, rec.base_branch,
|
||||
rec.review_rounds, rec.output_json, rec.updated_at, rec.id
|
||||
],
|
||||
)
|
||||
}
|
||||
);
|
||||
|
||||
impl TaskRepo {
|
||||
/// 列出未删除任务(deleted_at IS NULL)— 对标 ProjectRepo::list_active
|
||||
///
|
||||
/// 显式列出 14 个 TaskRecord 列名(同 ProjectRepo::list_active 写法),
|
||||
/// 不 SELECT deleted_at:TaskRecord 不带该字段,取了 from_row 会因未知列报错。
|
||||
pub async fn list_active(&self) -> Result<Vec<TaskRecord>> {
|
||||
let conn = self.conn.clone();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let guard = conn.blocking_lock();
|
||||
let mut stmt = guard
|
||||
.prepare("SELECT id, project_id, title, description, status, priority, branch_name, assignee, workflow_def_id, base_branch, review_rounds, output_json, created_at, updated_at FROM tasks WHERE deleted_at IS NULL ORDER BY created_at DESC")
|
||||
.map_err(storage_err)?;
|
||||
let rows = stmt
|
||||
.query_map([], |row| task_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(进回收站,可恢复)。仅作用于未删任务,返回是否命中。
|
||||
/// 对标 ProjectRepo::soft_delete。
|
||||
pub async fn soft_delete(&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 tasks 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(从回收站还原)。仅作用于已删任务,返回是否命中。
|
||||
/// 对标 ProjectRepo::restore。
|
||||
pub async fn restore(&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 tasks 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)?
|
||||
}
|
||||
|
||||
/// 原子推进任务状态(任务推进链 F-260616-02 唯一 status 写入路径)
|
||||
///
|
||||
/// 下沉 SQL `WHERE id=? AND status=?expected` 做 CAS(Compare-And-Swap)防 TOCTOU:
|
||||
/// 并发推进/旁路修改若已改 status,affected_rows==0,本方法返回 None,调用方
|
||||
/// (task_advance_node)据此报「状态已变,推进中止」。`review_rounds` 不进通用
|
||||
/// update_field 白名单(收口:仅本方法可改 status 与 review_rounds)。
|
||||
///
|
||||
/// - `expected`:调用方读取的当前 status(状态机校验时的 from),CAS 前置。
|
||||
/// - `new_status`:目标 status(状态机 can_transition 已校验合法)。
|
||||
/// - `bump_rounds`:退回转换(in_review→in_progress / testing→in_review)传 true,
|
||||
/// 一并 `review_rounds = review_rounds + 1`(同 UPDATE 原子,避免读改写竞争)。
|
||||
/// 前向推进 / 进出 blocked / 进 cancelled 传 false,不动 review_rounds。
|
||||
///
|
||||
/// 返回:成功推进返回更新后的 TaskRecord;affected==0(状态已变/任务不存在)返回 None。
|
||||
pub async fn advance_status_atomic(
|
||||
&self,
|
||||
id: &str,
|
||||
expected: &str,
|
||||
new_status: &str,
|
||||
bump_rounds: bool,
|
||||
) -> Result<Option<TaskRecord>> {
|
||||
let conn = self.conn.clone();
|
||||
let id = id.to_owned();
|
||||
let expected = expected.to_owned();
|
||||
let new_status = new_status.to_owned();
|
||||
let now = now_millis_str();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let guard = conn.blocking_lock();
|
||||
// CAS:WHERE id AND status=expected 锁定当前态;affected==0 即并发已改动。
|
||||
// deleted_at IS NULL:回收站任务(soft_delete 设了 deleted_at)CAS 必败→affected=0,
|
||||
// 返回 None,杜绝回收站任务被推进(D-02 软删语义收口,一处关闭)。
|
||||
let sql = if bump_rounds {
|
||||
"UPDATE tasks SET status = ?1, review_rounds = review_rounds + 1, updated_at = ?2 \
|
||||
WHERE id = ?3 AND status = ?4 AND deleted_at IS NULL"
|
||||
} else {
|
||||
"UPDATE tasks SET status = ?1, updated_at = ?2 \
|
||||
WHERE id = ?3 AND status = ?4 AND deleted_at IS NULL"
|
||||
};
|
||||
let affected = guard
|
||||
.execute(sql, params![new_status, now, id, expected])
|
||||
.map_err(storage_err)?;
|
||||
if affected == 0 {
|
||||
return Ok(None);
|
||||
}
|
||||
// 回读更新后的记录(含新 status / 累加后的 review_rounds / 新 updated_at)。
|
||||
let mut stmt = guard
|
||||
.prepare("SELECT id, project_id, title, description, status, priority, branch_name, assignee, workflow_def_id, base_branch, review_rounds, output_json, created_at, updated_at FROM tasks WHERE id = ?1")
|
||||
.map_err(storage_err)?;
|
||||
let row = stmt
|
||||
.query_row(params![id], |row| task_from_row(row))
|
||||
.optional()
|
||||
.map_err(storage_err)?;
|
||||
Ok(row)
|
||||
})
|
||||
.await
|
||||
.map_err(storage_err)?
|
||||
}
|
||||
|
||||
/// 列出回收站(deleted_at IS NOT NULL),按更新时间(≈删除时间)降序。对标 ProjectRepo::list_deleted。
|
||||
///
|
||||
/// 注:任务表无专用 list_active_by_project 方法,按项目列活跃任务由 commands/task.rs
|
||||
/// 的 list_tasks 用 list_active 后内存过滤 project_id 实现(任务量小,无需 SQL 下推)。
|
||||
pub async fn list_deleted(&self) -> Result<Vec<TaskRecord>> {
|
||||
let conn = self.conn.clone();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let guard = conn.blocking_lock();
|
||||
let mut stmt = guard
|
||||
.prepare("SELECT id, project_id, title, description, status, priority, branch_name, assignee, workflow_def_id, base_branch, review_rounds, output_json, created_at, updated_at FROM tasks WHERE deleted_at IS NOT NULL ORDER BY updated_at DESC")
|
||||
.map_err(storage_err)?;
|
||||
let rows = stmt
|
||||
.query_map([], |row| task_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)?
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user