309 lines
12 KiB
Rust
309 lines
12 KiB
Rust
use std::fs;
|
|
use std::sync::Arc;
|
|
use std::thread;
|
|
|
|
use control_plane::{
|
|
AppendAiRuntimeEventInput, AppendAuditInput, ControlPlaneError, ControlPlaneStore,
|
|
CreateSessionInput, DirectoryGrantInput, OutboxEventInput, TursoControlPlaneConfig,
|
|
TursoControlPlaneMode, TursoControlPlaneStore, UpsertAiExternalConversationBindingInput,
|
|
UpsertAiRuntimeRunInput, UpsertSidebarShortcutInput, UpsertUserInput, UpsertWorkspaceInput,
|
|
};
|
|
|
|
fn local_turso_store() -> (TursoControlPlaneStore, std::path::PathBuf) {
|
|
let path = std::env::temp_dir().join(format!(
|
|
"mnote-control-plane-libsql-{}.db",
|
|
uuid::Uuid::new_v4().simple()
|
|
));
|
|
let _ = fs::remove_file(&path);
|
|
let _ = fs::remove_file(path.with_extension("db-wal"));
|
|
let _ = fs::remove_file(path.with_extension("db-shm"));
|
|
(
|
|
TursoControlPlaneStore::open_local(&path).expect("open libSQL local control plane"),
|
|
path,
|
|
)
|
|
}
|
|
|
|
#[test]
|
|
fn libsql_local_store_covers_control_plane_core_flows() {
|
|
let (store, path) = local_turso_store();
|
|
|
|
let user = store
|
|
.upsert_user(UpsertUserInput {
|
|
id: Some("libsql_user".to_string()),
|
|
email: Some("libsql_user@example.com".to_string()),
|
|
username: "libsql_user".to_string(),
|
|
display_name: "libSQL User".to_string(),
|
|
role: Some("admin".to_string()),
|
|
password_hash: None,
|
|
})
|
|
.expect("upsert user");
|
|
assert_eq!(user.id, "libsql_user");
|
|
|
|
let workspace = store
|
|
.upsert_workspace(UpsertWorkspaceInput {
|
|
id: Some("local-ws:libsql-user:workspace".to_string()),
|
|
owner_user_id: user.id.clone(),
|
|
name: "libSQL Workspace".to_string(),
|
|
kind: Some("personal".to_string()),
|
|
root_uri: "file:///tmp/mnote-libsql-workspace".to_string(),
|
|
root_path: "/tmp/mnote-libsql-workspace".to_string(),
|
|
source_kind: Some("local_folder".to_string()),
|
|
})
|
|
.expect("upsert workspace");
|
|
assert_eq!(workspace.owner_user_id, user.id);
|
|
store
|
|
.grant_directory_access(DirectoryGrantInput {
|
|
user_id: "libsql_user".to_string(),
|
|
workspace_id: Some(workspace.id.clone()),
|
|
root_uri: workspace.root_uri.clone(),
|
|
root_path: workspace.root_path.clone(),
|
|
permission: "write".to_string(),
|
|
recursive: true,
|
|
capabilities: vec!["ai".to_string()],
|
|
source: "test".to_string(),
|
|
created_by: Some("libsql_user".to_string()),
|
|
})
|
|
.expect("grant workspace access");
|
|
|
|
let access = store
|
|
.resolve_access("libsql_user", "file:///tmp/mnote-libsql-workspace/page.md")
|
|
.expect("resolve workspace access");
|
|
assert_eq!(access.permission, "write");
|
|
|
|
let run = store
|
|
.upsert_ai_runtime_run(UpsertAiRuntimeRunInput {
|
|
id: None,
|
|
user_id: "libsql_user".to_string(),
|
|
workspace_id: Some(workspace.id.clone()),
|
|
document_id: Some("doc_1".to_string()),
|
|
session_id: "sess_1".to_string(),
|
|
run_id: "run_1".to_string(),
|
|
title: Some("libSQL run".to_string()),
|
|
profile: "reasonix".to_string(),
|
|
acp_runtime: "reasonix".to_string(),
|
|
trace_id: Some("trace_1".to_string()),
|
|
status: "running".to_string(),
|
|
runtime_json: "{\"status\":\"running\"}".to_string(),
|
|
payload_json: "{\"message\":\"hello\"}".to_string(),
|
|
})
|
|
.expect("upsert ai runtime run");
|
|
assert_eq!(run.revision, 1);
|
|
|
|
store
|
|
.append_ai_runtime_event(AppendAiRuntimeEventInput {
|
|
id: None,
|
|
user_id: "libsql_user".to_string(),
|
|
workspace_id: Some(workspace.id.clone()),
|
|
document_id: Some("doc_1".to_string()),
|
|
session_id: "sess_1".to_string(),
|
|
run_id: "run_1".to_string(),
|
|
profile: "reasonix".to_string(),
|
|
acp_runtime: "reasonix".to_string(),
|
|
event_type: "message.delta".to_string(),
|
|
payload_json: "{\"text\":\"hello\"}".to_string(),
|
|
})
|
|
.expect("append runtime event");
|
|
assert_eq!(
|
|
store
|
|
.list_ai_runtime_events("libsql_user", "run_1", 10)
|
|
.expect("list runtime events")
|
|
.len(),
|
|
1
|
|
);
|
|
|
|
let binding = store
|
|
.upsert_ai_external_conversation_binding(UpsertAiExternalConversationBindingInput {
|
|
id: None,
|
|
user_id: "libsql_user".to_string(),
|
|
workspace_id: Some(workspace.id.clone()),
|
|
mnote_session_id: "sess_1".to_string(),
|
|
acp_session_id: Some("acp_1".to_string()),
|
|
agent_id: "chat_only".to_string(),
|
|
profile: "openclaw-doubao-chat".to_string(),
|
|
provider: "doubao-web".to_string(),
|
|
remote_conversation_id: "remote_1".to_string(),
|
|
remote_url: Some("https://www.doubao.com/chat/remote_1".to_string()),
|
|
status: "active".to_string(),
|
|
metadata_json: "{\"source\":\"test\"}".to_string(),
|
|
})
|
|
.expect("upsert external conversation binding");
|
|
assert_eq!(binding.status, "active");
|
|
|
|
let found = store
|
|
.find_ai_external_conversation_binding(
|
|
"libsql_user",
|
|
Some(&workspace.id),
|
|
"sess_1",
|
|
"doubao-web",
|
|
)
|
|
.expect("find binding")
|
|
.expect("binding exists");
|
|
assert_eq!(found.id, binding.id);
|
|
|
|
drop(store);
|
|
let _ = fs::remove_file(&path);
|
|
let _ = fs::remove_file(path.with_extension("db-wal"));
|
|
let _ = fs::remove_file(path.with_extension("db-shm"));
|
|
}
|
|
|
|
#[test]
|
|
fn libsql_local_store_handles_parallel_control_plane_writes() {
|
|
let (store, path) = local_turso_store();
|
|
let store = Arc::new(store);
|
|
|
|
let user = store
|
|
.upsert_user(UpsertUserInput {
|
|
id: Some("parallel_user".to_string()),
|
|
email: Some("parallel_user@example.com".to_string()),
|
|
username: "parallel_user".to_string(),
|
|
display_name: "Parallel User".to_string(),
|
|
role: Some("admin".to_string()),
|
|
password_hash: None,
|
|
})
|
|
.expect("upsert parallel user");
|
|
let workspace = store
|
|
.upsert_workspace(UpsertWorkspaceInput {
|
|
id: Some("local-ws:parallel-user:workspace".to_string()),
|
|
owner_user_id: user.id.clone(),
|
|
name: "Parallel Workspace".to_string(),
|
|
kind: Some("personal".to_string()),
|
|
root_uri: "file:///tmp/mnote-libsql-parallel-workspace".to_string(),
|
|
root_path: "/tmp/mnote-libsql-parallel-workspace".to_string(),
|
|
source_kind: Some("local_folder".to_string()),
|
|
})
|
|
.expect("upsert parallel workspace");
|
|
store
|
|
.upsert_ai_runtime_run(UpsertAiRuntimeRunInput {
|
|
id: Some("parallel_run_record".to_string()),
|
|
user_id: user.id.clone(),
|
|
workspace_id: Some(workspace.id.clone()),
|
|
document_id: Some("parallel_doc".to_string()),
|
|
session_id: "parallel_session".to_string(),
|
|
run_id: "parallel_run".to_string(),
|
|
title: Some("Parallel run".to_string()),
|
|
profile: "reasonix".to_string(),
|
|
acp_runtime: "reasonix".to_string(),
|
|
trace_id: None,
|
|
status: "running".to_string(),
|
|
runtime_json: "{}".to_string(),
|
|
payload_json: "{}".to_string(),
|
|
})
|
|
.expect("seed parallel run");
|
|
|
|
let mut handles = Vec::new();
|
|
for index in 0..8 {
|
|
let store = store.clone();
|
|
let workspace_id = workspace.id.clone();
|
|
handles.push(thread::spawn(move || {
|
|
let suffix = index.to_string();
|
|
store
|
|
.create_session(CreateSessionInput {
|
|
id: Some(format!("parallel_session_{suffix}")),
|
|
user_id: "parallel_user".to_string(),
|
|
token_hash: format!("parallel_token_hash_{suffix}"),
|
|
user_agent: Some("parallel-test".to_string()),
|
|
ip_hash: None,
|
|
expires_at: None,
|
|
})
|
|
.expect("create session in parallel");
|
|
store
|
|
.upsert_sidebar_shortcut(UpsertSidebarShortcutInput {
|
|
id: Some(format!("parallel_shortcut_{suffix}")),
|
|
user_id: "parallel_user".to_string(),
|
|
workspace_id: workspace_id.clone(),
|
|
root_uri: Some("file:///tmp/mnote-libsql-parallel-workspace".to_string()),
|
|
kind: "page".to_string(),
|
|
source_kind: "local_folder".to_string(),
|
|
target_id: format!("doc_{suffix}"),
|
|
relative_path: Some(format!("doc_{suffix}.md")),
|
|
document_id: Some(format!("local-md:doc_{suffix}.md")),
|
|
title: format!("Doc {suffix}"),
|
|
icon: None,
|
|
sort_order: index,
|
|
metadata_json: "{}".to_string(),
|
|
})
|
|
.expect("upsert shortcut in parallel");
|
|
store
|
|
.append_ai_runtime_event(AppendAiRuntimeEventInput {
|
|
id: Some(format!("parallel_event_{suffix}")),
|
|
user_id: "parallel_user".to_string(),
|
|
workspace_id: Some(workspace_id.clone()),
|
|
document_id: Some("parallel_doc".to_string()),
|
|
session_id: "parallel_session".to_string(),
|
|
run_id: "parallel_run".to_string(),
|
|
profile: "reasonix".to_string(),
|
|
acp_runtime: "reasonix".to_string(),
|
|
event_type: "message.delta".to_string(),
|
|
payload_json: format!("{{\"index\":{index}}}"),
|
|
})
|
|
.expect("append runtime event in parallel");
|
|
store
|
|
.append_audit(AppendAuditInput {
|
|
actor_user_id: Some("parallel_user".to_string()),
|
|
action: "parallel.write".to_string(),
|
|
target_kind: "control_plane".to_string(),
|
|
target_id: Some(format!("target_{suffix}")),
|
|
metadata_json: "{}".to_string(),
|
|
})
|
|
.expect("append audit in parallel");
|
|
store
|
|
.append_outbox(OutboxEventInput {
|
|
topic: "parallel".to_string(),
|
|
event_type: "parallel.write".to_string(),
|
|
payload_json: format!("{{\"index\":{index}}}"),
|
|
})
|
|
.expect("append outbox in parallel");
|
|
}));
|
|
}
|
|
|
|
for handle in handles {
|
|
handle
|
|
.join()
|
|
.expect("parallel writer thread should not panic");
|
|
}
|
|
|
|
assert_eq!(
|
|
store
|
|
.list_ai_runtime_events("parallel_user", "parallel_run", 100)
|
|
.expect("list parallel events")
|
|
.len(),
|
|
8
|
|
);
|
|
assert_eq!(
|
|
store
|
|
.list_sidebar_shortcuts("parallel_user", &workspace.id)
|
|
.expect("list parallel shortcuts")
|
|
.len(),
|
|
8
|
|
);
|
|
assert_eq!(
|
|
store.list_audit_log(20).expect("list parallel audit").len(),
|
|
8
|
|
);
|
|
assert_eq!(
|
|
store.drain_outbox(20).expect("drain parallel outbox").len(),
|
|
8
|
|
);
|
|
|
|
drop(store);
|
|
let _ = fs::remove_file(&path);
|
|
let _ = fs::remove_file(path.with_extension("db-wal"));
|
|
let _ = fs::remove_file(path.with_extension("db-shm"));
|
|
}
|
|
|
|
#[test]
|
|
fn turso_config_rejects_missing_remote_credentials_without_panic() {
|
|
let result = TursoControlPlaneStore::open_with_config(TursoControlPlaneConfig {
|
|
mode: TursoControlPlaneMode::Remote,
|
|
path: None,
|
|
remote_url: Some("libsql://example.turso.io".to_string()),
|
|
auth_token: None,
|
|
sync_interval: None,
|
|
});
|
|
let Err(error) = result else {
|
|
panic!("missing token should return an error");
|
|
};
|
|
|
|
assert!(matches!(error, ControlPlaneError::InvalidInput(_)));
|
|
}
|