use std::path::{Path, PathBuf};
use std::sync::RwLock;
use chrono::{DateTime, Utc};
use dashmap::DashMap;
use serde::{Deserialize, Serialize};
use tracing::{debug, info, warn};
use crate::hnsw_service::HnswServiceConfig;
use crate::vector_backend::{SearchResult, VectorBackend, VectorError, VectorResult};
use crate::vector_hnsw::HnswBackend;
#[cfg(feature = "exochain")]
use crate::chain::ChainManager;
#[cfg(feature = "exochain")]
use crate::governance::{EffectVector, GovernanceDecision, GovernanceEngine, GovernanceRequest};
#[cfg(feature = "exochain")]
use std::sync::Arc;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProfileMeta {
pub id: String,
pub name: String,
pub created_at: DateTime<Utc>,
pub vector_count: usize,
}
pub struct ProfileEntry {
pub meta: ProfileMeta,
pub backend: Box<dyn VectorBackend>,
}
impl std::fmt::Debug for ProfileEntry {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ProfileEntry")
.field("meta", &self.meta)
.field("backend", &self.meta.id)
.finish()
}
}
#[derive(Debug, thiserror::Error)]
pub enum ProfileError {
#[error("profile already exists: '{0}'")]
AlreadyExists(String),
#[error("profile not found: '{0}'")]
NotFound(String),
#[error("invalid profile id: '{0}' -- must be [a-zA-Z0-9_-]+")]
InvalidId(String),
#[error("profile io error: {0}")]
Io(#[from] std::io::Error),
#[error("profile json error: {0}")]
Json(#[from] serde_json::Error),
#[error("governance denied: {0}")]
GovernanceDenied(String),
}
pub type ProfileResult<T> = Result<T, ProfileError>;
pub struct ProfileStore {
storage_path: PathBuf,
profiles: DashMap<String, ProfileEntry>,
active_profile: RwLock<String>,
hnsw_config: HnswServiceConfig,
#[cfg(feature = "exochain")]
chain_manager: Option<Arc<ChainManager>>,
#[cfg(feature = "exochain")]
governance_engine: Option<Arc<GovernanceEngine>>,
}
impl ProfileStore {
pub fn new(storage_path: PathBuf, hnsw_config: HnswServiceConfig) -> Self {
Self {
storage_path,
profiles: DashMap::new(),
active_profile: RwLock::new("default".to_owned()),
hnsw_config,
#[cfg(feature = "exochain")]
chain_manager: None,
#[cfg(feature = "exochain")]
governance_engine: None,
}
}
#[cfg(feature = "exochain")]
pub fn set_chain_manager(&mut self, cm: Arc<ChainManager>) {
self.chain_manager = Some(cm);
}
#[cfg(feature = "exochain")]
pub fn set_governance_engine(&mut self, engine: Arc<GovernanceEngine>) {
self.governance_engine = Some(engine);
}
pub fn load_existing(&self) -> ProfileResult<usize> {
let dir = &self.storage_path;
if !dir.exists() {
std::fs::create_dir_all(dir)?;
}
let mut count = 0;
for entry in std::fs::read_dir(dir)? {
let entry = entry?;
if !entry.file_type()?.is_dir() {
continue;
}
let profile_dir = entry.path();
let meta_path = profile_dir.join("profile.json");
if !meta_path.exists() {
continue;
}
match Self::load_profile_meta(&meta_path) {
Ok(meta) => {
let backend = Box::new(HnswBackend::new(self.hnsw_config.clone()));
debug!(profile_id = %meta.id, "loaded profile from disk");
self.profiles
.insert(meta.id.clone(), ProfileEntry { meta, backend });
count += 1;
}
Err(e) => {
warn!(path = %meta_path.display(), error = %e, "failed to load profile");
}
}
}
if self.profiles.is_empty() {
info!("no profiles found, creating default profile");
self.create_profile_inner("default", "Default Profile")?;
count = 1;
}
Ok(count)
}
pub fn create_profile(&self, id: &str, name: &str) -> ProfileResult<ProfileMeta> {
Self::validate_id(id)?;
if self.profiles.contains_key(id) {
return Err(ProfileError::AlreadyExists(id.to_owned()));
}
#[cfg(feature = "exochain")]
if let Some(ref engine) = self.governance_engine {
let req =
GovernanceRequest::new("system", "profile.create").with_effect(EffectVector {
risk: 0.3,
privacy: 0.2,
..Default::default()
});
let result = engine.evaluate(&req);
match &result.decision {
GovernanceDecision::Deny(reason) | GovernanceDecision::EscalateToHuman(reason) => {
return Err(ProfileError::GovernanceDenied(reason.clone()));
}
_ => {}
}
}
let meta = self.create_profile_inner(id, name)?;
#[cfg(feature = "exochain")]
if let Some(ref cm) = self.chain_manager {
cm.append(
"profile_store",
crate::chain::EVENT_KIND_PROFILE_CREATE,
Some(serde_json::json!({
"profile_id": id,
"name": name,
})),
);
}
Ok(meta)
}
pub fn delete_profile(&self, id: &str) -> ProfileResult<()> {
if !self.profiles.contains_key(id) {
return Err(ProfileError::NotFound(id.to_owned()));
}
{
let active = self
.active_profile
.read()
.expect("active_profile lock poisoned");
if *active == id {
return Err(ProfileError::InvalidId(format!(
"cannot delete active profile '{id}'; switch first"
)));
}
}
#[cfg(feature = "exochain")]
if let Some(ref engine) = self.governance_engine {
let req =
GovernanceRequest::new("system", "profile.delete").with_effect(EffectVector {
risk: 0.7,
privacy: 0.4,
..Default::default()
});
let result = engine.evaluate(&req);
match &result.decision {
GovernanceDecision::Deny(reason) | GovernanceDecision::EscalateToHuman(reason) => {
return Err(ProfileError::GovernanceDenied(reason.clone()));
}
_ => {}
}
}
self.profiles.remove(id);
let dir = self.storage_path.join(id);
if dir.exists() {
std::fs::remove_dir_all(&dir)?;
}
#[cfg(feature = "exochain")]
if let Some(ref cm) = self.chain_manager {
cm.append(
"profile_store",
crate::chain::EVENT_KIND_PROFILE_DELETE,
Some(serde_json::json!({
"profile_id": id,
})),
);
}
info!(profile_id = %id, "deleted profile");
Ok(())
}
pub fn list_profiles(&self) -> Vec<ProfileMeta> {
self.profiles
.iter()
.map(|entry| {
let mut meta = entry.value().meta.clone();
meta.vector_count = entry.value().backend.len();
meta
})
.collect()
}
pub fn switch_profile(&self, id: &str) -> ProfileResult<()> {
if !self.profiles.contains_key(id) {
return Err(ProfileError::NotFound(id.to_owned()));
}
let previous = self.active_profile_id();
let mut active = self
.active_profile
.write()
.expect("active_profile lock poisoned");
*active = id.to_owned();
#[cfg(feature = "exochain")]
if let Some(ref cm) = self.chain_manager {
cm.append(
"profile_store",
crate::chain::EVENT_KIND_PROFILE_SWITCH,
Some(serde_json::json!({
"profile_id": id,
"previous": previous,
})),
);
}
info!(profile_id = %id, "switched active profile");
Ok(())
}
pub fn active_profile_id(&self) -> String {
self.active_profile
.read()
.expect("active_profile lock poisoned")
.clone()
}
pub fn get_profile_backend(
&self,
id: &str,
) -> Option<dashmap::mapref::one::Ref<'_, String, ProfileEntry>> {
self.profiles.get(id)
}
pub fn active_backend(&self) -> Option<dashmap::mapref::one::Ref<'_, String, ProfileEntry>> {
let id = self.active_profile_id();
self.profiles.get(&id)
}
pub fn insert(
&self,
id: u64,
key: &str,
vector: &[f32],
metadata: serde_json::Value,
) -> VectorResult<()> {
let profile_id = self.active_profile_id();
let entry = self.profiles.get(&profile_id).ok_or_else(|| {
VectorError::Other(format!("active profile '{profile_id}' not found"))
})?;
let result = entry.backend.insert(id, key, vector, metadata);
#[cfg(feature = "exochain")]
if result.is_ok()
&& let Some(ref cm) = self.chain_manager
{
cm.append(
"profile_store",
crate::chain::EVENT_KIND_PROFILE_VECTOR_INSERT,
Some(serde_json::json!({
"profile_id": profile_id,
"vector_id": id,
"key": key,
})),
);
}
result
}
pub fn search(&self, query: &[f32], k: usize) -> Vec<SearchResult> {
let profile_id = self.active_profile_id();
match self.profiles.get(&profile_id) {
Some(entry) => entry.backend.search(query, k),
None => Vec::new(),
}
}
pub fn len(&self) -> usize {
self.profiles.len()
}
pub fn is_empty(&self) -> bool {
self.profiles.is_empty()
}
pub fn persist_meta(&self, id: &str) -> ProfileResult<()> {
let entry = self
.profiles
.get(id)
.ok_or_else(|| ProfileError::NotFound(id.to_owned()))?;
let mut meta = entry.meta.clone();
meta.vector_count = entry.backend.len();
let dir = self.storage_path.join(id);
std::fs::create_dir_all(&dir)?;
let meta_path = dir.join("profile.json");
let json = serde_json::to_string_pretty(&meta)?;
std::fs::write(&meta_path, json)?;
Ok(())
}
fn create_profile_inner(&self, id: &str, name: &str) -> ProfileResult<ProfileMeta> {
let meta = ProfileMeta {
id: id.to_owned(),
name: name.to_owned(),
created_at: Utc::now(),
vector_count: 0,
};
let backend = Box::new(HnswBackend::new(self.hnsw_config.clone()));
let dir = self.storage_path.join(id);
std::fs::create_dir_all(dir.join("vectors"))?;
let meta_path = dir.join("profile.json");
let json = serde_json::to_string_pretty(&meta)?;
std::fs::write(&meta_path, json)?;
self.profiles.insert(
id.to_owned(),
ProfileEntry {
meta: meta.clone(),
backend,
},
);
info!(profile_id = %id, name = %name, "created profile");
Ok(meta)
}
fn load_profile_meta(path: &Path) -> ProfileResult<ProfileMeta> {
let data = std::fs::read_to_string(path)?;
let meta: ProfileMeta = serde_json::from_str(&data)?;
Ok(meta)
}
fn validate_id(id: &str) -> ProfileResult<()> {
if id.is_empty() {
return Err(ProfileError::InvalidId("empty id".to_owned()));
}
if !id
.chars()
.all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')
{
return Err(ProfileError::InvalidId(id.to_owned()));
}
Ok(())
}
}
impl std::fmt::Debug for ProfileStore {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ProfileStore")
.field("storage_path", &self.storage_path)
.field("profiles", &self.profiles.len())
.field("active", &self.active_profile_id())
.finish()
}
}
#[cfg(test)]
mod tests {
use super::*;
fn tmp_dir(name: &str) -> PathBuf {
std::env::temp_dir().join(format!(
"profile_store_test_{name}_{}",
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos()
))
}
fn make_store(name: &str) -> (ProfileStore, PathBuf) {
let dir = tmp_dir(name);
let store = ProfileStore::new(dir.clone(), HnswServiceConfig::default());
(store, dir)
}
fn cleanup(dir: &Path) {
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn create_and_list_profiles() {
let (store, dir) = make_store("create_list");
store.create_profile("alice", "Alice").unwrap();
store.create_profile("bob", "Bob").unwrap();
let profiles = store.list_profiles();
assert_eq!(profiles.len(), 2);
let names: Vec<&str> = profiles.iter().map(|p| p.name.as_str()).collect();
assert!(names.contains(&"Alice"));
assert!(names.contains(&"Bob"));
cleanup(&dir);
}
#[test]
fn duplicate_profile_rejected() {
let (store, dir) = make_store("duplicate");
store.create_profile("alice", "Alice").unwrap();
assert!(matches!(
store.create_profile("alice", "Alice Again"),
Err(ProfileError::AlreadyExists(_))
));
cleanup(&dir);
}
#[test]
fn invalid_id_rejected() {
let (store, dir) = make_store("invalid_id");
assert!(matches!(
store.create_profile("a b c", "Bad"),
Err(ProfileError::InvalidId(_))
));
assert!(matches!(
store.create_profile("", "Empty"),
Err(ProfileError::InvalidId(_))
));
assert!(matches!(
store.create_profile("a/b", "Slash"),
Err(ProfileError::InvalidId(_))
));
cleanup(&dir);
}
#[test]
fn switch_and_active_profile() {
let (store, dir) = make_store("switch");
store.create_profile("alice", "Alice").unwrap();
store.create_profile("bob", "Bob").unwrap();
assert_eq!(store.active_profile_id(), "default");
store.switch_profile("alice").unwrap();
assert_eq!(store.active_profile_id(), "alice");
store.switch_profile("bob").unwrap();
assert_eq!(store.active_profile_id(), "bob");
assert!(matches!(
store.switch_profile("nobody"),
Err(ProfileError::NotFound(_))
));
cleanup(&dir);
}
#[test]
fn delete_profile() {
let (store, dir) = make_store("delete");
store.create_profile("alice", "Alice").unwrap();
store.create_profile("bob", "Bob").unwrap();
assert!(matches!(
store.delete_profile("nobody"),
Err(ProfileError::NotFound(_))
));
store.switch_profile("alice").unwrap();
assert!(store.delete_profile("alice").is_err());
store.delete_profile("bob").unwrap();
assert_eq!(store.len(), 1);
cleanup(&dir);
}
#[test]
fn insert_and_search_active_profile() {
let (store, dir) = make_store("insert_search");
store.create_profile("alice", "Alice").unwrap();
store.switch_profile("alice").unwrap();
store
.insert(1, "v1", &[1.0, 0.0, 0.0], serde_json::json!({}))
.unwrap();
store
.insert(2, "v2", &[0.0, 1.0, 0.0], serde_json::json!({}))
.unwrap();
let results = store.search(&[1.0, 0.0, 0.0], 1);
assert_eq!(results.len(), 1);
assert_eq!(results[0].key, "v1");
cleanup(&dir);
}
#[test]
fn profiles_are_isolated() {
let (store, dir) = make_store("isolated");
store.create_profile("alice", "Alice").unwrap();
store.create_profile("bob", "Bob").unwrap();
store.switch_profile("alice").unwrap();
store
.insert(1, "alice-vec", &[1.0, 0.0], serde_json::json!({}))
.unwrap();
store.switch_profile("bob").unwrap();
store
.insert(1, "bob-vec", &[0.0, 1.0], serde_json::json!({}))
.unwrap();
store.switch_profile("alice").unwrap();
let results = store.search(&[1.0, 0.0], 1);
assert_eq!(results.len(), 1);
assert_eq!(results[0].key, "alice-vec");
store.switch_profile("bob").unwrap();
let results = store.search(&[0.0, 1.0], 1);
assert_eq!(results.len(), 1);
assert_eq!(results[0].key, "bob-vec");
cleanup(&dir);
}
#[test]
fn load_existing_creates_default() {
let (store, dir) = make_store("load_default");
let count = store.load_existing().unwrap();
assert_eq!(count, 1);
assert!(store.profiles.contains_key("default"));
cleanup(&dir);
}
#[test]
fn load_existing_reads_persisted() {
let dir = tmp_dir("load_persisted");
{
let store = ProfileStore::new(dir.clone(), HnswServiceConfig::default());
store.create_profile("alice", "Alice").unwrap();
store.create_profile("bob", "Bob").unwrap();
store.persist_meta("alice").unwrap();
store.persist_meta("bob").unwrap();
}
{
let store = ProfileStore::new(dir.clone(), HnswServiceConfig::default());
let count = store.load_existing().unwrap();
assert_eq!(count, 2);
assert!(store.profiles.contains_key("alice"));
assert!(store.profiles.contains_key("bob"));
}
cleanup(&dir);
}
#[test]
fn persist_meta_updates_vector_count() {
let (store, dir) = make_store("persist_count");
store.create_profile("alice", "Alice").unwrap();
store.switch_profile("alice").unwrap();
store
.insert(1, "v1", &[1.0, 0.0], serde_json::json!({}))
.unwrap();
store.persist_meta("alice").unwrap();
let meta_path = dir.join("alice").join("profile.json");
let data = std::fs::read_to_string(&meta_path).unwrap();
let meta: ProfileMeta = serde_json::from_str(&data).unwrap();
assert_eq!(meta.vector_count, 1);
cleanup(&dir);
}
}