diff --git a/omem-server/src/api/handlers/mod.rs b/omem-server/src/api/handlers/mod.rs index 4a70f8c..798784d 100644 --- a/omem-server/src/api/handlers/mod.rs +++ b/omem-server/src/api/handlers/mod.rs @@ -19,9 +19,10 @@ pub use memory::{ }; pub use profile::get_profile; pub use sharing::{ - batch_share, create_auto_share_rule, delete_auto_share_rule, list_auto_share_rules, - org_publish, org_setup, pull_memory, reshare_memory, share_all, share_all_to_user, - share_memory, share_to_user, unshare_memory, + approve_pending_share, batch_share, create_auto_share_rule, delete_auto_share_rule, + list_auto_share_rules, list_pending_shares, org_publish, org_setup, pull_memory, + reject_pending_share, reshare_memory, share_all, share_all_to_user, share_memory, + share_to_user, unshare_memory, }; pub use spaces::{ add_member, create_space, delete_space, get_space, list_spaces, remove_member, diff --git a/omem-server/src/api/handlers/sharing.rs b/omem-server/src/api/handlers/sharing.rs index 11fea67..282e161 100644 --- a/omem-server/src/api/handlers/sharing.rs +++ b/omem-server/src/api/handlers/sharing.rs @@ -11,8 +11,8 @@ use crate::api::server::{normalize_space_id, personal_space_id, AppState}; use crate::domain::error::OmemError; use crate::domain::memory::Memory; use crate::domain::space::{ - AutoShareRule, MemberRole, Provenance, SharingAction, SharingEvent, Space, SpaceMember, - SpaceType, + AutoShareRule, MemberRole, PendingShare, Provenance, SharingAction, SharingEvent, Space, + SpaceMember, SpaceType, }; use crate::domain::tenant::AuthInfo; use crate::store::StoreManager; @@ -1357,7 +1357,19 @@ pub async fn check_auto_share( continue; } if rule.require_approval { - continue; + let pending = PendingShare { + id: Uuid::new_v4().to_string(), + source_space: memory.space_id.clone(), + source_memory: memory.id.clone(), + target_space: space.id.clone(), + rule_id: rule.id.clone(), + requested_by_user: user_id.to_string(), + requested_by_agent: agent_id.to_string(), + content_preview: content_preview(&memory.content), + created_at: chrono::Utc::now().to_rfc3339(), + }; + space_store.record_pending_share(&pending).await?; + break; } let target_store = store_manager.get_store(&space.id).await?; @@ -1385,6 +1397,117 @@ pub async fn check_auto_share( Ok(shared_to) } +/// Materialise an approved pending share: copy the source memory (content + +/// vector) into the target space, record a Share event, and drop the queue +/// entry. Returns the new copy. Kept free of `AppState` so it stays unit-testable. +pub(crate) async fn apply_pending_share( + store_manager: &StoreManager, + space_store: &crate::store::SpaceStore, + pending: &PendingShare, +) -> Result { + let source_store = store_manager.get_store(&pending.source_space).await?; + let source = source_store + .get_by_id(&pending.source_memory) + .await? + .ok_or_else(|| { + OmemError::NotFound(format!( + "source memory {} no longer exists", + pending.source_memory + )) + })?; + let source_vector = source_store + .get_vector_by_id(&pending.source_memory) + .await?; + + let copy = make_shared_copy( + &source, + &pending.target_space, + &pending.requested_by_user, + &pending.requested_by_agent, + ); + let target_store = store_manager.get_store(&pending.target_space).await?; + target_store.create(©, source_vector.as_deref()).await?; + + let event = make_sharing_event( + SharingAction::Share, + ©.id, + &pending.source_space, + &pending.target_space, + &pending.requested_by_user, + &pending.requested_by_agent, + &content_preview(&source.content), + ); + space_store.record_sharing_event(&event).await?; + space_store.delete_pending_share(&pending.id).await?; + + Ok(copy) +} + +/// GET /v1/shares/pending +/// +/// List shares awaiting approval in spaces the caller can write to. Poll this; +/// there is no notification system. +pub async fn list_pending_shares( + State(state): State>, + Extension(auth): Extension, +) -> Result>, OmemError> { + let spaces = state + .space_store + .list_spaces_for_user(&auth.tenant_id) + .await?; + let mut pending = Vec::new(); + for space in &spaces { + if verify_space_write_access(space, &auth.tenant_id).is_ok() { + pending.extend(state.space_store.list_pending_shares(&space.id).await?); + } + } + Ok(Json(pending)) +} + +/// POST /v1/shares/pending/{id}/approve +pub async fn approve_pending_share( + State(state): State>, + Extension(auth): Extension, + Path(id): Path, +) -> Result, OmemError> { + let pending = state + .space_store + .get_pending_share(&id) + .await? + .ok_or_else(|| OmemError::NotFound(format!("pending share {id}")))?; + let space = state + .space_store + .get_space(&pending.target_space) + .await? + .ok_or_else(|| OmemError::NotFound(format!("space {}", pending.target_space)))?; + verify_space_write_access(&space, &auth.tenant_id)?; + + let copy = apply_pending_share(&state.store_manager, &state.space_store, &pending).await?; + Ok(Json(copy)) +} + +/// POST /v1/shares/pending/{id}/reject +pub async fn reject_pending_share( + State(state): State>, + Extension(auth): Extension, + Path(id): Path, +) -> Result, OmemError> { + let pending = state + .space_store + .get_pending_share(&id) + .await? + .ok_or_else(|| OmemError::NotFound(format!("pending share {id}")))?; + let space = state + .space_store + .get_space(&pending.target_space) + .await? + .ok_or_else(|| OmemError::NotFound(format!("space {}", pending.target_space)))?; + verify_space_write_access(&space, &auth.tenant_id)?; + + state.space_store.delete_pending_share(&id).await?; + Ok(Json(serde_json::json!({ "rejected": true, "id": id }))) +} + // ── Tests ──────────────────────────────────────────────────────────── #[cfg(test)] @@ -1594,6 +1717,93 @@ mod tests { assert_eq!(team_list.len(), 3); } + #[tokio::test] + async fn test_pending_share_approval_flow() { + let env = setup().await; + let dim = env.store_manager.vector_dim() as usize; + + // Team space with an auto-share rule that REQUIRES approval. + let mut team_space = make_space("team:backend", "user-001"); + team_space.auto_share_rules.push(AutoShareRule { + id: "rule-1".to_string(), + source_space: "user-001".to_string(), + categories: vec!["preferences".to_string()], + tags: Vec::new(), + min_importance: 0.0, + require_approval: true, + created_at: "2025-01-01T00:00:00Z".to_string(), + }); + env.space_store + .create_space(&team_space) + .await + .expect("create space"); + + // Source memory (with a vector) that matches the rule. + let source_store = env + .store_manager + .get_store("user-001") + .await + .expect("source store"); + let mem = make_memory("prefers tabs over spaces", "user-001", "user-001"); + source_store + .create(&mem, Some(&vec![0.3f32; dim])) + .await + .expect("create source"); + + // require_approval => ENQUEUE, do not auto-share. + let shared_to = check_auto_share( + &mem, + &env.space_store, + &env.store_manager, + "user-001", + "agent-1", + ) + .await + .expect("auto share"); + assert!(shared_to.is_empty(), "require_approval must not auto-share"); + + let team_store = env + .store_manager + .get_store("team:backend") + .await + .expect("team store"); + assert_eq!( + team_store.list_all_active().await.expect("list").len(), + 0, + "nothing shared before approval" + ); + + let pending = env + .space_store + .list_pending_shares("team:backend") + .await + .expect("list pending"); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].source_memory, mem.id); + assert_eq!(pending[0].target_space, "team:backend"); + + // Approve => copy materialises (with vector) and the queue drains. + let copy = apply_pending_share(&env.store_manager, &env.space_store, &pending[0]) + .await + .expect("apply"); + assert_eq!(copy.content, "prefers tabs over spaces"); + + let active = team_store.list_all_active().await.expect("list"); + assert_eq!(active.len(), 1); + let v = team_store + .get_vector_by_id(&active[0].id) + .await + .expect("vec") + .expect("vector present after approval"); + assert!(v.iter().any(|x| *x != 0.0)); + assert!(env + .space_store + .list_pending_shares("team:backend") + .await + .expect("list") + .is_empty()); + } + #[tokio::test] async fn test_auto_share_rule() { let env = setup().await; diff --git a/omem-server/src/api/router.rs b/omem-server/src/api/router.rs index dd12030..4fdec48 100644 --- a/omem-server/src/api/router.rs +++ b/omem-server/src/api/router.rs @@ -70,6 +70,15 @@ pub fn build_router(state: Arc) -> Router { .route("/v1/memories/{id}/pull", post(handlers::pull_memory)) .route("/v1/memories/{id}/unshare", post(handlers::unshare_memory)) .route("/v1/memories/{id}/reshare", post(handlers::reshare_memory)) + .route("/v1/shares/pending", get(handlers::list_pending_shares)) + .route( + "/v1/shares/pending/{id}/approve", + post(handlers::approve_pending_share), + ) + .route( + "/v1/shares/pending/{id}/reject", + post(handlers::reject_pending_share), + ) .route("/v1/memories/batch-share", post(handlers::batch_share)) .route("/v1/memories/share-all", post(handlers::share_all)) .route( diff --git a/omem-server/src/domain/space.rs b/omem-server/src/domain/space.rs index f085c7f..c8a80f2 100644 --- a/omem-server/src/domain/space.rs +++ b/omem-server/src/domain/space.rs @@ -131,6 +131,22 @@ pub struct SharingEvent { pub timestamp: String, } +/// A share that matched an auto-share rule with `require_approval = true`, +/// awaiting a human decision. Held in a queue (no notifications) until an +/// approver with write access to `target_space` approves or rejects it. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PendingShare { + pub id: String, + pub source_space: String, + pub source_memory: String, + pub target_space: String, + pub rule_id: String, + pub requested_by_user: String, + pub requested_by_agent: String, + pub content_preview: String, + pub created_at: String, +} + #[cfg(test)] mod tests { use super::*; diff --git a/omem-server/src/store/spaces.rs b/omem-server/src/store/spaces.rs index eabef7a..b035067 100644 --- a/omem-server/src/store/spaces.rs +++ b/omem-server/src/store/spaces.rs @@ -8,11 +8,12 @@ use lancedb::Connection; use serde::{Deserialize, Serialize}; use crate::domain::error::OmemError; -use crate::domain::space::{SharingEvent, Space}; +use crate::domain::space::{PendingShare, SharingEvent, Space}; const SPACES_TABLE: &str = "spaces"; const SHARING_EVENTS_TABLE: &str = "sharing_events"; const IMPORT_TASKS_TABLE: &str = "import_tasks"; +const PENDING_SHARES_TABLE: &str = "pending_shares"; #[derive(Serialize, Deserialize, Debug, Clone)] pub struct ImportTaskRecord { @@ -102,6 +103,16 @@ impl SpaceStore { })?; } + if !existing.contains(&PENDING_SHARES_TABLE.to_string()) { + self.spaces_db + .create_empty_table(PENDING_SHARES_TABLE, Self::pending_shares_schema()) + .execute() + .await + .map_err(|e| { + OmemError::Storage(format!("failed to create pending_shares table: {e}")) + })?; + } + Ok(()) } @@ -428,6 +439,113 @@ impl SpaceStore { Ok(tasks) } + fn pending_shares_schema() -> Arc { + Arc::new(Schema::new(vec![ + Field::new("id", DataType::Utf8, false), + Field::new("target_space", DataType::Utf8, false), + Field::new("data", DataType::Utf8, false), + Field::new("created_at", DataType::Utf8, false), + ])) + } + + async fn open_pending_shares_table(&self) -> Result { + self.spaces_db + .open_table(PENDING_SHARES_TABLE) + .execute() + .await + .map_err(|e| OmemError::Storage(format!("failed to open pending_shares table: {e}"))) + } + + pub async fn record_pending_share(&self, pending: &PendingShare) -> Result<(), OmemError> { + let data_json = serde_json::to_string(pending) + .map_err(|e| OmemError::Storage(format!("failed to serialize pending share: {e}")))?; + + let batch = RecordBatch::try_new( + Self::pending_shares_schema(), + vec![ + Arc::new(StringArray::from(vec![pending.id.as_str()])), + Arc::new(StringArray::from(vec![pending.target_space.as_str()])), + Arc::new(StringArray::from(vec![data_json.as_str()])), + Arc::new(StringArray::from(vec![pending.created_at.as_str()])), + ], + ) + .map_err(|e| OmemError::Storage(format!("failed to build pending share batch: {e}")))?; + + let table = self.open_pending_shares_table().await?; + let reader = RecordBatchIterator::new(vec![Ok(batch)], Self::pending_shares_schema()); + table + .add(Box::new(reader) as Box) + .execute() + .await + .map_err(|e| OmemError::Storage(format!("failed to insert pending share: {e}")))?; + + Ok(()) + } + + pub async fn list_pending_shares( + &self, + target_space: &str, + ) -> Result, OmemError> { + let table = self.open_pending_shares_table().await?; + let batches: Vec = table + .query() + .only_if(format!("target_space = '{}'", escape_sql(target_space))) + .execute() + .await + .map_err(|e| OmemError::Storage(format!("list pending shares query failed: {e}")))? + .try_collect() + .await + .map_err(|e| OmemError::Storage(format!("collect failed: {e}")))?; + + let mut out = Vec::new(); + for batch in &batches { + for i in 0..batch.num_rows() { + out.push(Self::row_to_pending_share(batch, i)?); + } + } + Ok(out) + } + + pub async fn get_pending_share(&self, id: &str) -> Result, OmemError> { + let table = self.open_pending_shares_table().await?; + let batches: Vec = table + .query() + .only_if(format!("id = '{}'", escape_sql(id))) + .limit(1) + .execute() + .await + .map_err(|e| OmemError::Storage(format!("pending share query failed: {e}")))? + .try_collect() + .await + .map_err(|e| OmemError::Storage(format!("collect failed: {e}")))?; + + for batch in &batches { + if batch.num_rows() > 0 { + return Ok(Some(Self::row_to_pending_share(batch, 0)?)); + } + } + Ok(None) + } + + pub async fn delete_pending_share(&self, id: &str) -> Result<(), OmemError> { + let table = self.open_pending_shares_table().await?; + table + .delete(&format!("id = '{}'", escape_sql(id))) + .await + .map_err(|e| OmemError::Storage(format!("pending share delete failed: {e}")))?; + Ok(()) + } + + fn row_to_pending_share(batch: &RecordBatch, row: usize) -> Result { + let data = batch + .column_by_name("data") + .and_then(|c| c.as_any().downcast_ref::()) + .ok_or_else(|| OmemError::Storage("pending_shares missing data column".to_string()))? + .value(row); + serde_json::from_str(data) + .map_err(|e| OmemError::Storage(format!("failed to parse pending share: {e}"))) + } + fn row_to_space(batch: &RecordBatch, row: usize) -> Result { let col = batch .column_by_name("data")