lix/mnote
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(control-plane): add libSQL Turso backend

Browse Source
This commit is contained in:
Agent Board
2026-07-03 13:24:20 +08:00
parent ee6612028d
commit a75b3d11f9
44 changed files with 11070 additions and 666 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rust/crates/control-plane/src/bin/control-plane-admin.rs
+1502
View File
File diff suppressed because it is too large Load Diff
rust/crates/control-plane/src/error.rs
+39
View File
@@ -8,6 +8,8 @@ pub enum ControlPlaneError {
Storage(String),
Unauthorized(String),
SessionExpired(String),
Timeout(String),
RateLimit(String),
}
impl fmt::Display for ControlPlaneError {
@@ -19,6 +21,8 @@ impl fmt::Display for ControlPlaneError {
ControlPlaneError::Storage(msg) => write!(f, "Storage: {msg}"),
ControlPlaneError::Unauthorized(msg) => write!(f, "Unauthorized: {msg}"),
ControlPlaneError::SessionExpired(msg) => write!(f, "SessionExpired: {msg}"),
ControlPlaneError::Timeout(msg) => write!(f, "Timeout: {msg}"),
ControlPlaneError::RateLimit(msg) => write!(f, "RateLimit: {msg}"),
}
}
}
@@ -31,6 +35,41 @@ impl From<rusqlite::Error> for ControlPlaneError {
}
}
impl From<libsql::Error> for ControlPlaneError {
fn from(e: libsql::Error) -> Self {
let message = e.to_string();
let lower = message.to_lowercase();
if message.contains("UNIQUE constraint failed")
|| message.contains("constraint failed")
|| message.contains("constraint violation")
{
ControlPlaneError::Conflict(message)
} else if message.contains("Unauthorized")
|| message.contains("authorization")
|| message.contains("401")
{
ControlPlaneError::Unauthorized(message)
} else if lower.contains("timeout")
|| lower.contains("timed out")
|| lower.contains("deadline")
|| lower.contains("connection refused")
|| lower.contains("connection reset")
|| lower.contains("eof")
{
ControlPlaneError::Timeout(message)
} else if lower.contains("rate limit")
|| lower.contains("rate_limit")
|| lower.contains("too many requests")
|| lower.contains("429")
{
ControlPlaneError::RateLimit(message)
} else {
ControlPlaneError::Storage(message)
}
}
}
impl From<serde_json::Error> for ControlPlaneError {
fn from(e: serde_json::Error) -> Self {
ControlPlaneError::Storage(format!("JSON error: {e}"))
rust/crates/control-plane/src/lib.rs
+2
View File
@@ -3,9 +3,11 @@ pub mod migrations;
pub mod model;
pub mod sqlite;
pub mod store;
pub mod turso;
pub use error::ControlPlaneError;
pub use migrations::run_migrations;
pub use model::*;
pub use sqlite::SqliteControlPlaneStore;
pub use store::*;
pub use turso::{TursoControlPlaneConfig, TursoControlPlaneMode, TursoControlPlaneStore};
rust/crates/control-plane/src/migrations.rs
+94
View File
@@ -3,6 +3,7 @@
//! Each migration is a named step that runs inside a transaction.
//! The `migrations` table tracks which steps have been applied.
use crate::error::ControlPlaneError;
use rusqlite::{Connection, Result as SqlResult};
const MIGRATIONS: &[(&str, &str)] = &[
@@ -72,6 +73,34 @@ pub fn run_migrations(conn: &Connection) -> SqlResult<()> {
Ok(())
}
pub async fn run_libsql_migrations(conn: &libsql::Connection) -> Result<(), ControlPlaneError> {
conn.execute_batch(
"CREATE TABLE IF NOT EXISTS _migrations (
name TEXT PRIMARY KEY,
applied_at TEXT NOT NULL DEFAULT (datetime('now'))
);",
)
.await?;
let mut rows = conn.query("SELECT name FROM _migrations", ()).await?;
let mut already_applied = Vec::new();
while let Some(row) = rows.next().await? {
already_applied.push(row.get::<String>(0)?);
}
for (name, sql) in MIGRATIONS {
if already_applied.iter().any(|value| value == name) {
continue;
}
tracing::info!("Running libSQL migration: {name}");
conn.execute_batch(sql).await?;
conn.execute("INSERT INTO _migrations (name) VALUES (?1)", [*name])
.await?;
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
@@ -96,6 +125,22 @@ mod tests {
> 0
}
#[cfg(feature = "turso-unit-tests")]
async fn libsql_table_exists(conn: &libsql::Connection, name: &str) -> bool {
let mut rows = conn
.query(
"SELECT count(*) FROM sqlite_master WHERE type='table' AND name=?1",
[name],
)
.await
.unwrap_or_else(|_| panic!("query libSQL table {name}"));
if let Some(row) = rows.next().await.unwrap_or(None) {
row.get::<i64>(0).unwrap_or(0) > 0
} else {
false
}
}
#[test]
fn migration_is_idempotent() {
let conn = fresh_db();
@@ -136,4 +181,53 @@ mod tests {
assert!(table_exists(&conn, "users"));
assert!(table_exists(&conn, "_migrations"));
}
#[cfg(feature = "turso-unit-tests")]
#[test]
fn libsql_migration_is_idempotent() {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("build tokio runtime");
let database = rt
.block_on(async { libsql::Builder::new_local(":memory:").build().await })
.expect("build libSQL in-memory database");
let conn = database.connect().expect("connect libSQL");
rt.block_on(async {
run_libsql_migrations(&conn)
.await
.expect("first libSQL migration run");
run_libsql_migrations(&conn)
.await
.expect("second libSQL migration run (idempotent)");
});
let tables: &[&str] = &[
"users",
"auth_identities",
"auth_sessions",
"workspaces",
"workspace_members",
"directory_grants",
"share_links",
"sync_state",
"ai_policies",
"audit_log",
"outbox_events",
"legacy_id_map",
"ai_runtime_runs",
"ai_runtime_events",
"sidebar_shortcuts",
"user_ui_preferences",
"user_navigation_recent",
"ai_agent_profiles",
"ai_agent_profile_grants",
"ai_external_conversation_bindings",
];
for table in tables {
let exists = rt.block_on(async { libsql_table_exists(&conn, table).await });
assert!(exists, "libSQL table {table} should exist");
}
}
}
rust/crates/control-plane/src/model.rs
+11
View File
@@ -112,6 +112,17 @@ pub struct WorkspaceRecord {
pub revision: i64,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct UpsertWorkspaceInput {
pub id: Option<EntityId>,
pub owner_user_id: EntityId,
pub name: String,
pub kind: Option<String>,
pub root_uri: String,
pub root_path: String,
pub source_kind: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DirectoryGrantRecord {
pub id: EntityId,
rust/crates/control-plane/src/sqlite.rs
+246 -49
View File
@@ -18,7 +18,8 @@ use crate::model::{
ResolvedAccess, ResolvedAuthSession, ShareLinkRecord, SidebarShortcutRecord, SyncStateRecord,
UpsertAiExternalConversationBindingInput, UpsertAiPolicyInput, UpsertAiRuntimeRunInput,
UpsertNavigationRecentInput, UpsertSidebarShortcutInput, UpsertSyncStateInput, UpsertUserInput,
UpsertUserUiPreferenceInput, UserRecord, UserUiPreferenceRecord, WorkspaceRecord,
UpsertUserUiPreferenceInput, UpsertWorkspaceInput, UserRecord, UserUiPreferenceRecord,
WorkspaceRecord,
};
use crate::store::ControlPlaneStore;
@@ -555,6 +556,14 @@ fn list_ai_agent_profile_access_rows(
Ok(by_profile.into_values().collect())
}
impl SqliteControlPlaneStore {
fn lock_conn(&self) -> Result<std::sync::MutexGuard<'_, rusqlite::Connection>, ControlPlaneError> {
self.conn
.lock()
.map_err(|e| ControlPlaneError::Storage(format!("sqlite lock poisoned: {e}")))
}
}
impl ControlPlaneStore for SqliteControlPlaneStore {
fn upsert_user(&self, input: UpsertUserInput) -> Result<UserRecord, ControlPlaneError> {
if input.username.trim().is_empty() {
@@ -568,7 +577,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
));
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let now = now_text();
let id = input.id.unwrap_or_else(|| new_id("usr"));
let role = input.role.unwrap_or_else(|| "user".to_string());
@@ -646,7 +655,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
));
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let now = now_text();
let user_id = input.user_id;
let password_hash = password_hash_v1(&input.password);
@@ -698,7 +707,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
));
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let provider = if account.contains('@') {
"password_email"
} else {
@@ -744,7 +753,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
));
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let now = now_text();
let expires_at = input
.expires_at
@@ -786,7 +795,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
return Ok(None);
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let now = now_text();
let resolved = conn
.query_row(
@@ -837,7 +846,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
"session_id 不能为空".to_string(),
));
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
conn.execute(
"UPDATE auth_sessions SET revoked_at = ?1 WHERE id = ?2 AND revoked_at IS NULL",
params![now_text(), session_id],
@@ -849,7 +858,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
&self,
actor_id: &str,
) -> Result<WorkspaceRecord, ControlPlaneError> {
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let existing = conn
.query_row(
"SELECT id, owner_user_id, name, kind, root_uri, root_path, source_kind, status, created_at, updated_at, revision
@@ -933,11 +942,89 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
Ok(workspace)
}
fn upsert_workspace(
&self,
input: UpsertWorkspaceInput,
) -> Result<WorkspaceRecord, ControlPlaneError> {
let owner_user_id = input.owner_user_id.trim().to_string();
let name = input.name.trim().to_string();
let root_uri = input.root_uri.trim().to_string();
let root_path = input.root_path.trim().to_string();
if owner_user_id.is_empty() || name.is_empty() || root_uri.is_empty() || root_path.is_empty()
{
return Err(ControlPlaneError::InvalidInput(
"workspace owner/name/root 不能为空".to_string(),
));
}
let kind = input
.kind
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or("personal")
.to_string();
let source_kind = input
.source_kind
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or("local_folder")
.to_string();
let now = now_text();
let id = input.id.unwrap_or_else(|| new_id("ws"));
let mut conn = self.lock_conn()?;
let tx = conn.transaction()?;
tx.query_row(
"SELECT id FROM users WHERE id = ?1",
params![&owner_user_id],
|row| row.get::<_, String>(0),
)
.optional()?
.ok_or_else(|| ControlPlaneError::NotFound(format!("user not found: {owner_user_id}")))?;
tx.execute(
"INSERT INTO workspaces (id, owner_user_id, name, kind, root_uri, root_path, source_kind, status, created_at, updated_at, revision)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 'active', ?8, ?9, 1)
ON CONFLICT(id) DO UPDATE SET
owner_user_id = excluded.owner_user_id,
name = excluded.name,
kind = excluded.kind,
root_uri = excluded.root_uri,
root_path = excluded.root_path,
source_kind = excluded.source_kind,
updated_at = excluded.updated_at,
revision = workspaces.revision + 1",
params![
&id,
&owner_user_id,
&name,
&kind,
&root_uri,
&root_path,
&source_kind,
&now,
&now
],
)?;
tx.execute(
"INSERT OR IGNORE INTO workspace_members (id, workspace_id, user_id, role, status, created_at, updated_at, revision)
VALUES (?1, ?2, ?3, 'owner', 'active', ?4, ?5, 1)",
params![new_id("wsm"), &id, &owner_user_id, &now, &now],
)?;
let workspace = tx.query_row(
"SELECT id, owner_user_id, name, kind, root_uri, root_path, source_kind, status, created_at, updated_at, revision
FROM workspaces WHERE id = ?1",
params![&id],
row_to_workspace,
)?;
tx.commit()?;
Ok(workspace)
}
fn grant_directory_access(
&self,
input: DirectoryGrantInput,
) -> Result<DirectoryGrantRecord, ControlPlaneError> {
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let now = now_text();
let record = DirectoryGrantRecord {
id: new_id("grant"),
@@ -979,7 +1066,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
}
fn list_directory_grants(&self) -> Result<Vec<DirectoryGrantRecord>, ControlPlaneError> {
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let mut stmt = conn.prepare(
"SELECT id, user_id, workspace_id, root_uri, root_path, permission, recursive, capabilities_json, source, status, created_by, created_at, updated_at, revision
FROM directory_grants
@@ -1001,7 +1088,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
if actor_id.is_empty() {
return Ok(Vec::new());
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let mut stmt = conn.prepare(
"SELECT id, user_id, workspace_id, root_uri, root_path, permission, recursive, capabilities_json, source, status, created_by, created_at, updated_at, revision
FROM directory_grants
@@ -1019,7 +1106,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
&self,
lookup: DirectoryGrantLookup,
) -> Result<Vec<DirectoryGrantRecord>, ControlPlaneError> {
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let grant_id = lookup
.grant_id
.as_deref()
@@ -1067,7 +1154,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
"grant_id 不能为空".to_string(),
));
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let existing_revision = conn
.query_row(
"SELECT revision FROM directory_grants WHERE id = ?1 AND status = 'active'",
@@ -1097,7 +1184,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
actor_id: &str,
root_uri: &str,
) -> Result<ResolvedAccess, ControlPlaneError> {
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let root_path = file_uri_to_legacy_path(root_uri);
let mut stmt = conn.prepare(
"SELECT id, user_id, workspace_id, root_uri, root_path, permission, recursive, capabilities_json, source, status, created_by, created_at, updated_at, revision
@@ -1150,7 +1237,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
.filter(|value| !value.trim().is_empty())
.unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
let token_hash = share_token_hash_v1(&token);
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let now = now_text();
let link = ShareLinkRecord {
id: input.id.unwrap_or_else(|| new_id("share")),
@@ -1193,7 +1280,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
if workspace_id.is_empty() {
return Ok(Vec::new());
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let mut stmt = conn.prepare(
"SELECT id, workspace_id, resource_kind, resource_id, token_hash, permission, created_by, expires_at, revoked_at, created_at, updated_at, revision
FROM share_links
@@ -1215,7 +1302,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
if token_hash.is_empty() {
return Ok(None);
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let link = conn
.query_row(
"SELECT id, workspace_id, resource_kind, resource_id, token_hash, permission, created_by, expires_at, revoked_at, created_at, updated_at, revision
@@ -1236,7 +1323,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
"link_id 不能为空".to_string(),
));
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let existed = conn.execute(
"UPDATE share_links
SET revoked_at = ?1, updated_at = ?2, revision = revision + 1
@@ -1259,7 +1346,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
"audit action/target_kind 不能为空".to_string(),
));
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
conn.execute(
"INSERT INTO audit_log (id, actor_user_id, action, target_kind, target_id, metadata_json, created_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
@@ -1277,7 +1364,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
}
fn list_audit_log(&self, limit: usize) -> Result<Vec<AuditLogRecord>, ControlPlaneError> {
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let mut stmt = conn.prepare(
"SELECT id, actor_user_id, action, target_kind, target_id, metadata_json, created_at
FROM audit_log
@@ -1295,7 +1382,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
&self,
input: OutboxEventInput,
) -> Result<OutboxEventRecord, ControlPlaneError> {
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let record = OutboxEventRecord {
id: new_id("evt"),
topic: input.topic,
@@ -1320,7 +1407,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
}
fn drain_outbox(&self, limit: usize) -> Result<Vec<OutboxEventRecord>, ControlPlaneError> {
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let mut stmt = conn.prepare(
"SELECT id, topic, event_type, payload_json, created_at, delivered_at, attempts
FROM outbox_events
@@ -1348,7 +1435,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
"event_id 不能为空".to_string(),
));
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let updated = conn.execute(
"UPDATE outbox_events
SET delivered_at = COALESCE(delivered_at, ?1), attempts = attempts + 1
@@ -1393,7 +1480,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
"sidebar shortcut kind 只能是 page 或 folder".to_string(),
));
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let now = now_text();
conn.execute(
"INSERT INTO sidebar_shortcuts (
@@ -1454,7 +1541,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
if user_id.is_empty() || workspace_id.is_empty() {
return Ok(Vec::new());
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let mut stmt = conn.prepare(
"SELECT id, user_id, workspace_id, root_uri, kind, source_kind, target_id, relative_path,
document_id, title, icon, sort_order, status, metadata_json, created_at,
@@ -1480,7 +1567,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
if user_id.is_empty() || workspace_id.is_empty() {
return Ok(Vec::new());
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let mut stmt = conn.prepare(
"SELECT id, user_id, workspace_id, root_uri, kind, source_kind, target_id, relative_path,
document_id, title, icon, sort_order, status, metadata_json, created_at,
@@ -1510,7 +1597,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
"sidebar shortcut user_id/shortcut_id 不能为空".to_string(),
));
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let changed = conn.execute(
"UPDATE sidebar_shortcuts
SET status = 'removed', updated_at = ?1, revision = revision + 1
@@ -1541,7 +1628,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
));
}
serde_json::from_str::<serde_json::Value>(&input.value_json)?;
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let now = now_text();
conn.execute(
"INSERT INTO user_ui_preferences (
@@ -1604,7 +1691,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
}
let workspace_id = workspace_id.map(str::trim).unwrap_or_default();
let source_kind = source_kind.map(str::trim).unwrap_or_default();
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let mut stmt = conn.prepare(
"SELECT id, user_id, workspace_id, source_kind, scope_kind, scope_id, key,
value_json, status, created_at, updated_at, revision
@@ -1641,7 +1728,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
if scope_kind.is_empty() || scope_id.is_empty() || key.is_empty() {
return Ok(Vec::new());
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let mut stmt = conn.prepare(
"SELECT id, user_id, workspace_id, source_kind, scope_kind, scope_id, key,
value_json, status, created_at, updated_at, revision
@@ -1675,7 +1762,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
"ai agent profile user_id 不能为空".to_string(),
));
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let now = now_text();
conn.execute(
"INSERT INTO ai_agent_profiles (
@@ -1897,7 +1984,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
relative_path.as_deref(),
document_id.as_deref(),
)?;
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let now = now_text();
conn.execute(
"INSERT INTO user_navigation_recent (
@@ -1961,7 +2048,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
return Ok(Vec::new());
}
let limit = limit.clamp(1, 100) as i64;
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let rows = if let Some(kind) = kind.map(str::trim).filter(|value| !value.is_empty()) {
let mut stmt = conn.prepare(
"SELECT id, user_id, workspace_id, kind, source_kind, root_uri, target_key,
@@ -2007,7 +2094,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
serde_json::from_str::<serde_json::Value>(&input.model_policy_json)?;
serde_json::from_str::<serde_json::Value>(&input.quota_json)?;
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let now = now_text();
let existing = conn
.query_row(
@@ -2090,7 +2177,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
if actor_id.is_empty() && workspace_id.is_none() {
return Ok(None);
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
if let Some(workspace_id) = workspace_id {
let workspace_policy = conn
.query_row(
@@ -2139,7 +2226,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
serde_json::from_str::<serde_json::Value>(error_json)?;
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let now = now_text();
let existing = conn
.query_row(
@@ -2222,7 +2309,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
if workspace_id.is_empty() || remote_kind.is_empty() {
return Ok(None);
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
conn.query_row(
"SELECT id, workspace_id, remote_kind, remote_id, cursor, last_synced_at, status, error_json, updated_at, revision
FROM sync_state
@@ -2250,7 +2337,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
serde_json::from_str::<serde_json::Value>(&input.runtime_json)?;
serde_json::from_str::<serde_json::Value>(&input.payload_json)?;
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let now = now_text();
let existing = conn
.query_row(
@@ -2354,7 +2441,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
session_id: Option<&str>,
limit: usize,
) -> Result<Vec<AiRuntimeRunRecord>, ControlPlaneError> {
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let mut stmt = conn.prepare(
"SELECT id, user_id, workspace_id, document_id, session_id, run_id, title, profile, acp_runtime, trace_id, status, runtime_json, payload_json, deleted_at, created_at, updated_at, revision
FROM ai_runtime_runs
@@ -2381,7 +2468,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
user_id: &str,
run_id: &str,
) -> Result<Option<AiRuntimeRunRecord>, ControlPlaneError> {
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
conn.query_row(
"SELECT id, user_id, workspace_id, document_id, session_id, run_id, title, profile, acp_runtime, trace_id, status, runtime_json, payload_json, deleted_at, created_at, updated_at, revision
FROM ai_runtime_runs
@@ -2408,7 +2495,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
}
serde_json::from_str::<serde_json::Value>(&input.payload_json)?;
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let record = AiRuntimeEventRecord {
id: input.id.unwrap_or_else(|| new_id("are")),
user_id: input.user_id,
@@ -2442,13 +2529,33 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
Ok(record)
}
fn delete_ai_runtime_events_for_run(
&self,
user_id: &str,
run_id: &str,
) -> Result<usize, ControlPlaneError> {
let user_id = user_id.trim();
let run_id = run_id.trim();
if user_id.is_empty() || run_id.is_empty() {
return Err(ControlPlaneError::InvalidInput(
"user_id/run_id 不能为空".to_string(),
));
}
let conn = self.lock_conn()?;
let rows = conn.execute(
"DELETE FROM ai_runtime_events WHERE user_id = ?1 AND run_id = ?2",
params![user_id, run_id],
)?;
Ok(rows)
}
fn list_ai_runtime_events(
&self,
user_id: &str,
run_id: &str,
limit: usize,
) -> Result<Vec<AiRuntimeEventRecord>, ControlPlaneError> {
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let mut stmt = conn.prepare(
"SELECT id, user_id, workspace_id, document_id, session_id, run_id, profile, acp_runtime, event_type, payload_json, created_at
FROM ai_runtime_events
@@ -2473,7 +2580,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
after_seq: i64,
limit: usize,
) -> Result<Vec<AiRuntimeJournalEventRecord>, ControlPlaneError> {
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let mut stmt = conn.prepare(
"WITH ordered AS (
SELECT
@@ -2521,7 +2628,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
serde_json::from_str::<serde_json::Value>(&input.metadata_json)?;
validate_ai_external_conversation_status(&input.status)?;
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let now = now_text();
let existing = conn
.query_row(
@@ -2639,7 +2746,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
if user_id.is_empty() || mnote_session_id.is_empty() || provider.is_empty() {
return Ok(None);
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
conn.query_row(
"SELECT id, user_id, workspace_id, mnote_session_id, acp_session_id, agent_id, profile, provider, remote_conversation_id, remote_url, status, metadata_json, created_at, updated_at, deleted_at, revision
FROM ai_external_conversation_bindings
@@ -2655,6 +2762,38 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
.map_err(ControlPlaneError::from)
}
fn list_ai_external_conversation_bindings(
&self,
user_id: &str,
workspace_id: Option<&str>,
mnote_session_id: &str,
limit: usize,
) -> Result<Vec<AiExternalConversationBindingRecord>, ControlPlaneError> {
let user_id = user_id.trim();
let mnote_session_id = mnote_session_id.trim();
if user_id.is_empty() || mnote_session_id.is_empty() {
return Ok(Vec::new());
}
let conn = self.lock_conn()?;
let mut stmt = conn.prepare(
"SELECT id, user_id, workspace_id, mnote_session_id, acp_session_id, agent_id, profile, provider, remote_conversation_id, remote_url, status, metadata_json, created_at, updated_at, deleted_at, revision
FROM ai_external_conversation_bindings
WHERE user_id = ?1
AND (?2 IS NULL OR workspace_id = ?2)
AND mnote_session_id = ?3
ORDER BY updated_at DESC
LIMIT ?4",
)?;
let rows = stmt
.query_map(
params![user_id, workspace_id, mnote_session_id, limit.max(1) as i64],
row_to_ai_external_conversation_binding,
)?
.collect::<Result<Vec<_>, _>>()
.map_err(ControlPlaneError::from)?;
Ok(rows)
}
fn mark_ai_external_conversation_binding_status(
&self,
user_id: &str,
@@ -2673,7 +2812,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
if user_id.is_empty() || mnote_session_id.is_empty() || provider.is_empty() {
return Ok(0);
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let now = now_text();
let deleted_at = if status == "active" {
None
@@ -2716,7 +2855,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
"ai runtime session user_id/session_id/title 不能为空".to_string(),
));
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let now = now_text();
conn.execute(
"UPDATE ai_runtime_runs
@@ -2759,7 +2898,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
"ai runtime session user_id/session_id 不能为空".to_string(),
));
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let existing = conn
.query_row(
"SELECT id, user_id, workspace_id, document_id, session_id, run_id, title, profile, acp_runtime, trace_id, status, runtime_json, payload_json, deleted_at, created_at, updated_at, revision
@@ -2813,7 +2952,7 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
"ai runtime session user_id/session_id 不能为空".to_string(),
));
}
let conn = self.conn.lock().unwrap();
let conn = self.lock_conn()?;
let changed = conn.execute(
"UPDATE ai_runtime_runs
SET deleted_at = ?1, updated_at = ?1, revision = revision + 1
@@ -2928,6 +3067,55 @@ mod tests {
assert_eq!(access.grant_ids.len(), 1);
}
#[test]
fn upsert_workspace_supports_explicit_local_root_for_dev_seed() {
let store = store();
create_user(&store, "seed_owner");
let workspace = store
.upsert_workspace(UpsertWorkspaceInput {
id: Some("local-ws:seed-owner:custom".to_string()),
owner_user_id: "seed_owner".to_string(),
name: "Seed Workspace".to_string(),
kind: Some("personal".to_string()),
root_uri: "file:///tmp/mnote-seed-workspace".to_string(),
root_path: "/tmp/mnote-seed-workspace".to_string(),
source_kind: Some("local_folder".to_string()),
})
.expect("upsert workspace");
assert_eq!(workspace.id, "local-ws:seed-owner:custom");
assert_eq!(workspace.status, "active");
assert_eq!(workspace.root_uri, "file:///tmp/mnote-seed-workspace");
let updated = store
.upsert_workspace(UpsertWorkspaceInput {
id: Some(workspace.id.clone()),
owner_user_id: "seed_owner".to_string(),
name: "Seed Workspace Renamed".to_string(),
kind: Some("personal".to_string()),
root_uri: "file:///tmp/mnote-seed-workspace".to_string(),
root_path: "/tmp/mnote-seed-workspace".to_string(),
source_kind: Some("local_folder".to_string()),
})
.expect("update workspace");
assert_eq!(updated.name, "Seed Workspace Renamed");
assert_eq!(updated.revision, workspace.revision + 1);
assert_eq!(updated.status, "active");
let missing_user = store
.upsert_workspace(UpsertWorkspaceInput {
id: Some("local-ws:missing:custom".to_string()),
owner_user_id: "missing".to_string(),
name: "Missing".to_string(),
kind: Some("personal".to_string()),
root_uri: "file:///tmp/missing".to_string(),
root_path: "/tmp/missing".to_string(),
source_kind: Some("local_folder".to_string()),
})
.expect_err("missing owner should fail");
assert!(matches!(missing_user, ControlPlaneError::NotFound(_)));
}
#[test]
fn explicit_write_grant_out_ranks_read_grant() {
let store = store();
@@ -3772,6 +3960,15 @@ mod tests {
assert_eq!(journal_events.len(), 1);
assert_eq!(journal_events[0].seq, 2);
assert_eq!(journal_events[0].event.event_type, "run.completed");
let deleted = store
.delete_ai_runtime_events_for_run("ai_runtime_user", "run_1")
.expect("delete runtime events");
assert_eq!(deleted, 2);
assert!(store
.list_ai_runtime_events("ai_runtime_user", "run_1", 10)
.expect("list after delete")
.is_empty());
}
#[test]
rust/crates/control-plane/src/store.rs
+31 -1
View File
@@ -11,7 +11,8 @@ use crate::model::{
ResolvedAccess, ResolvedAuthSession, ShareLinkRecord, SidebarShortcutRecord, SyncStateRecord,
UpsertAiExternalConversationBindingInput, UpsertAiPolicyInput, UpsertAiRuntimeRunInput,
UpsertNavigationRecentInput, UpsertSidebarShortcutInput, UpsertSyncStateInput, UpsertUserInput,
UpsertUserUiPreferenceInput, UserRecord, UserUiPreferenceRecord, WorkspaceRecord,
UpsertUserUiPreferenceInput, UpsertWorkspaceInput, UserRecord, UserUiPreferenceRecord,
WorkspaceRecord,
};
pub trait ControlPlaneStore: Send + Sync {
@@ -44,6 +45,16 @@ pub trait ControlPlaneStore: Send + Sync {
actor_id: &str,
) -> Result<WorkspaceRecord, ControlPlaneError>;
/// Upsert a workspace record after the caller has authorized the operation.
///
/// The store only enforces schema-level invariants and user existence. It does
/// not validate filesystem paths or decide whether the actor may manage the
/// workspace.
fn upsert_workspace(
&self,
input: UpsertWorkspaceInput,
) -> Result<WorkspaceRecord, ControlPlaneError>;
fn grant_directory_access(
&self,
input: DirectoryGrantInput,
@@ -219,6 +230,17 @@ pub trait ControlPlaneStore: Send + Sync {
input: AppendAiRuntimeEventInput,
) -> Result<AiRuntimeEventRecord, ControlPlaneError>;
/// Hard-delete AI runtime events for a run after the caller has authorized it.
///
/// This is intended for controlled test/dev seeding and recovery workflows.
/// Product flows should prefer append-only events or higher-level session
/// deletion APIs.
fn delete_ai_runtime_events_for_run(
&self,
user_id: &str,
run_id: &str,
) -> Result<usize, ControlPlaneError>;
fn list_ai_runtime_events(
&self,
user_id: &str,
@@ -247,6 +269,14 @@ pub trait ControlPlaneStore: Send + Sync {
provider: &str,
) -> Result<Option<AiExternalConversationBindingRecord>, ControlPlaneError>;
fn list_ai_external_conversation_bindings(
&self,
user_id: &str,
workspace_id: Option<&str>,
mnote_session_id: &str,
limit: usize,
) -> Result<Vec<AiExternalConversationBindingRecord>, ControlPlaneError>;
fn mark_ai_external_conversation_binding_status(
&self,
user_id: &str,
rust/crates/control-plane/src/turso.rs
+4903
View File
File diff suppressed because it is too large Load Diff