diff --git a/.env.tmp b/.env.tmp new file mode 100644 index 0000000..e69de29 diff --git a/.gitea/workflows/build.yaml b/.gitea/workflows/build.yaml index 3d45825..0f1e31c 100644 --- a/.gitea/workflows/build.yaml +++ b/.gitea/workflows/build.yaml @@ -11,6 +11,7 @@ env: REGISTRY: forgejo.riotpiao.com IMAGE: forgejo.riotpiao.com/rock/poimen-memory DOCKER_HOST: tcp://localhost:2375 + SQLX_OFFLINE: "true" jobs: ci: @@ -25,11 +26,14 @@ jobs: - name: Checkout code uses: actions/checkout@v4 - - name: Cargo test - run: cargo test -p mem-ingest --lib 2>&1 | tail -50 || true + - name: Cargo build all + run: cargo build --all --verbose - - name: Cargo check - run: cargo check -p mem-ingest 2>&1 | tail -20 || true + - name: Cargo test all + run: cargo test --all --lib --verbose 2>&1 | tail -150 || true + + - name: Cargo clippy + run: cargo clippy --all --all-targets -- -D warnings 2>&1 | tail -50 || true - name: Get short SHA if: github.event_name == 'push' || github.event_name == 'workflow_dispatch' @@ -48,7 +52,7 @@ jobs: - name: Build Docker image if: github.event_name == 'push' || github.event_name == 'workflow_dispatch' run: | - docker build --no-cache \ + docker build --no-cache --progress=plain \ -t "${IMAGE}:${{ steps.sha.outputs.short_sha }}" \ -t "${IMAGE}:latest" \ -f Dockerfile . diff --git a/.sqlx/query-1e81bb729531ca33e4cef21623bcfe4fafb0c1bd435353b205f582bfda8873bc.json b/.sqlx/query-1e81bb729531ca33e4cef21623bcfe4fafb0c1bd435353b205f582bfda8873bc.json new file mode 100644 index 0000000..8cbf0c7 --- /dev/null +++ b/.sqlx/query-1e81bb729531ca33e4cef21623bcfe4fafb0c1bd435353b205f582bfda8873bc.json @@ -0,0 +1,52 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_entity_version\n WHERE entity_id = $1\n ORDER BY version_num DESC\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "version_num", + "type_info": "Int4" + }, + { + "ordinal": 1, + "name": "operation", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "snapshot", + "type_info": "Jsonb" + }, + { + "ordinal": 3, + "name": "changed_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 4, + "name": "changed_by", + "type_info": "Varchar" + }, + { + "ordinal": 5, + "name": "fields_changed!", + "type_info": "TextArray" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + false, + false, + false, + false, + null + ] + }, + "hash": "1e81bb729531ca33e4cef21623bcfe4fafb0c1bd435353b205f582bfda8873bc" +} diff --git a/.sqlx/query-62d65d4afc4d292b37de8e5cb59fbd51c602bdc1b437988f54e6c7fe268b9816.json b/.sqlx/query-62d65d4afc4d292b37de8e5cb59fbd51c602bdc1b437988f54e6c7fe268b9816.json new file mode 100644 index 0000000..6a40d9c --- /dev/null +++ b/.sqlx/query-62d65d4afc4d292b37de8e5cb59fbd51c602bdc1b437988f54e6c7fe268b9816.json @@ -0,0 +1,52 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_edge_version\n WHERE edge_id = $1\n ORDER BY version_num DESC\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "version_num", + "type_info": "Int4" + }, + { + "ordinal": 1, + "name": "operation", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "snapshot", + "type_info": "Jsonb" + }, + { + "ordinal": 3, + "name": "changed_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 4, + "name": "changed_by", + "type_info": "Varchar" + }, + { + "ordinal": 5, + "name": "fields_changed!", + "type_info": "TextArray" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + false, + false, + false, + false, + false, + null + ] + }, + "hash": "62d65d4afc4d292b37de8e5cb59fbd51c602bdc1b437988f54e6c7fe268b9816" +} diff --git a/.sqlx/query-aee5900f5e3d7cbba23729bbf2dd033dcc4cb41f6c851bf447a9238810684d18.json b/.sqlx/query-aee5900f5e3d7cbba23729bbf2dd033dcc4cb41f6c851bf447a9238810684d18.json new file mode 100644 index 0000000..61e13f9 --- /dev/null +++ b/.sqlx/query-aee5900f5e3d7cbba23729bbf2dd033dcc4cb41f6c851bf447a9238810684d18.json @@ -0,0 +1,53 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_entity_version\n WHERE entity_id = $1 AND changed_at <= $2\n ORDER BY version_num DESC\n LIMIT 1\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "version_num", + "type_info": "Int4" + }, + { + "ordinal": 1, + "name": "operation", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "snapshot", + "type_info": "Jsonb" + }, + { + "ordinal": 3, + "name": "changed_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 4, + "name": "changed_by", + "type_info": "Varchar" + }, + { + "ordinal": 5, + "name": "fields_changed!", + "type_info": "TextArray" + } + ], + "parameters": { + "Left": [ + "Text", + "Timestamptz" + ] + }, + "nullable": [ + false, + false, + false, + false, + false, + null + ] + }, + "hash": "aee5900f5e3d7cbba23729bbf2dd033dcc4cb41f6c851bf447a9238810684d18" +} diff --git a/.sqlx/query-c045466e1fe037dbdafea1008f262f4e48f104ea77732aa1d32ecb797f70e71d.json b/.sqlx/query-c045466e1fe037dbdafea1008f262f4e48f104ea77732aa1d32ecb797f70e71d.json new file mode 100644 index 0000000..9fe298e --- /dev/null +++ b/.sqlx/query-c045466e1fe037dbdafea1008f262f4e48f104ea77732aa1d32ecb797f70e71d.json @@ -0,0 +1,53 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_entity_version\n WHERE entity_id = $1 AND version_num = $2\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "version_num", + "type_info": "Int4" + }, + { + "ordinal": 1, + "name": "operation", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "snapshot", + "type_info": "Jsonb" + }, + { + "ordinal": 3, + "name": "changed_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 4, + "name": "changed_by", + "type_info": "Varchar" + }, + { + "ordinal": 5, + "name": "fields_changed!", + "type_info": "TextArray" + } + ], + "parameters": { + "Left": [ + "Text", + "Int4" + ] + }, + "nullable": [ + false, + false, + false, + false, + false, + null + ] + }, + "hash": "c045466e1fe037dbdafea1008f262f4e48f104ea77732aa1d32ecb797f70e71d" +} diff --git a/.sqlx/query-ca6872495bc04c6a65531279af8c758637c902dda2cc10366662988c6973ca48.json b/.sqlx/query-ca6872495bc04c6a65531279af8c758637c902dda2cc10366662988c6973ca48.json new file mode 100644 index 0000000..c4cca32 --- /dev/null +++ b/.sqlx/query-ca6872495bc04c6a65531279af8c758637c902dda2cc10366662988c6973ca48.json @@ -0,0 +1,53 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_edge_version\n WHERE edge_id = $1 AND version_num = $2\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "version_num", + "type_info": "Int4" + }, + { + "ordinal": 1, + "name": "operation", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "snapshot", + "type_info": "Jsonb" + }, + { + "ordinal": 3, + "name": "changed_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 4, + "name": "changed_by", + "type_info": "Varchar" + }, + { + "ordinal": 5, + "name": "fields_changed!", + "type_info": "TextArray" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Int4" + ] + }, + "nullable": [ + false, + false, + false, + false, + false, + null + ] + }, + "hash": "ca6872495bc04c6a65531279af8c758637c902dda2cc10366662988c6973ca48" +} diff --git a/Cargo.lock b/Cargo.lock index 0e3a8d0..cbc63ac 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -330,6 +330,28 @@ dependencies = [ "serde_json", ] +[[package]] +name = "async-stream" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b5a71a6f37880a80d1d7f19efd781e4b5de42c88f0722cc13bcb6cc2cfe8476" +dependencies = [ + "async-stream-impl", + "futures-core", + "pin-project-lite", +] + +[[package]] +name = "async-stream-impl" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "async-trait" version = "0.1.92" @@ -2017,11 +2039,13 @@ dependencies = [ "actix-rt", "actix-web", "anyhow", + "async-stream", "async-trait", "base64 0.21.7", "chrono", "clap", "futures", + "futures-util", "jsonwebtoken", "lru", "mem-chunk", @@ -2030,6 +2054,7 @@ dependencies = [ "mem-llm", "mem-store", "pgvector", + "rand 0.8.7", "redis", "reqwest", "serde", diff --git a/Dockerfile b/Dockerfile index eb5ec97..5643798 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,15 +1,16 @@ # Multi-stage build for Poimen Memory Service (Rust) # Stage 1: Builder -FROM rust:1.81-bookworm as builder +FROM rust:1-bookworm as builder WORKDIR /build # Copy source COPY . . -# Build in release mode -RUN cargo build --release +# Build the mem binary (offline sqlx - uses .sqlx/ cache) +ENV SQLX_OFFLINE=true +RUN cargo build --release -p mem-cli # Stage 2: Runtime FROM debian:bookworm-slim diff --git a/crates/mem-cli/Cargo.toml b/crates/mem-cli/Cargo.toml index 53d1a59..b846049 100644 --- a/crates/mem-cli/Cargo.toml +++ b/crates/mem-cli/Cargo.toml @@ -42,4 +42,7 @@ reqwest = { workspace = true } async-trait = { workspace = true } urlencoding = { workspace = true } walkdir = "2.5" +futures-util = "0.3" +async-stream = "0.3" +rand = "0.8" lru = "0.12" diff --git a/crates/mem-cli/src/agent/client_sdk.rs b/crates/mem-cli/src/agent/client_sdk.rs index e4262c7..2b87bc9 100644 --- a/crates/mem-cli/src/agent/client_sdk.rs +++ b/crates/mem-cli/src/agent/client_sdk.rs @@ -227,7 +227,7 @@ impl SynthesisClient { ) -> Vec> { let mut results = Vec::new(); for req in requests { - results.push(self.execute(&req).await); + results.push(self.execute(req).await); } results } @@ -280,11 +280,12 @@ impl SynthesisClient { tracing::debug!("Workflow executed in {}ms", elapsed_ms); Ok(body) } else { + let status = response.status(); let error_text = response .text() .await .unwrap_or_else(|_| "unknown error".to_string()); - Err(format!("Workflow failed ({}): {}", response.status(), error_text)) + Err(format!("Workflow failed ({}): {}", status, error_text)) } } } diff --git a/crates/mem-cli/src/agent/mod.rs b/crates/mem-cli/src/agent/mod.rs index 1b9150a..36e1145 100644 --- a/crates/mem-cli/src/agent/mod.rs +++ b/crates/mem-cli/src/agent/mod.rs @@ -11,3 +11,4 @@ pub use agent_interface::{Agent, AgentConfig, AgentCapability}; pub use webhook_handler::{WebhookEvent, WebhookPayload}; pub use observability::{AgentMetrics, MetricsCollector}; pub use client_sdk::{SynthesisClient, ClientRequest, ClientResponse}; +pub use agent_interface::DefaultAgent; diff --git a/crates/mem-cli/src/auth/authentik_service_account.rs b/crates/mem-cli/src/auth/authentik_service_account.rs index 419e232..7c58a2c 100644 --- a/crates/mem-cli/src/auth/authentik_service_account.rs +++ b/crates/mem-cli/src/auth/authentik_service_account.rs @@ -5,7 +5,7 @@ use std::sync::{Arc, RwLock}; use std::time::{Duration, Instant}; use serde::{Deserialize, Serialize}; use reqwest::Client; -use log::{debug, warn, error}; +use tracing::{debug, warn, error}; #[derive(Clone, Debug)] pub struct AuthentikServiceAccountConfig { diff --git a/crates/mem-cli/src/auth/mod.rs b/crates/mem-cli/src/auth/mod.rs new file mode 100644 index 0000000..26f5870 --- /dev/null +++ b/crates/mem-cli/src/auth/mod.rs @@ -0,0 +1,12 @@ +//! Authentication and Authorization Module +//! +//! Provides JWT validation, OIDC integration with Authentik, and RBAC. + +pub mod provider; +pub mod authentik_provider; +pub mod authentik_service_account; +pub mod guard; + +pub use provider::{AuthProvider, AuthError, Claims}; +pub use authentik_provider::AuthentikProvider; +pub use guard::{AuthGuard, PermissionGuard, Role}; diff --git a/crates/mem-cli/src/auth_middleware.rs b/crates/mem-cli/src/auth_middleware.rs index 5c18d09..6628cb8 100644 --- a/crates/mem-cli/src/auth_middleware.rs +++ b/crates/mem-cli/src/auth_middleware.rs @@ -59,15 +59,17 @@ pub fn auth_error_response(error: &AuthError) -> HttpResponse { let (status, message) = match error { AuthError::MissingToken => ("Unauthorized", "Missing or invalid Authorization header"), AuthError::InvalidSignature => ("Unauthorized", "Invalid token signature"), - AuthError::ExpiredToken => ("Unauthorized", "Token has expired"), + AuthError::TokenExpired => ("Unauthorized", "Token has expired"), AuthError::InvalidIssuer => ("Unauthorized", "Invalid token issuer"), - AuthError::AccessDenied => ("Forbidden", "Access denied for this resource"), - AuthError::InvalidClaims => ("Unauthorized", "Invalid or missing required claims"), + AuthError::InvalidAudience => ("Unauthorized", "Invalid token audience"), + AuthError::ProviderUnavailable(_) => ("ServiceUnavailable", "Auth provider unavailable"), + AuthError::Other(_) => ("Unauthorized", "Authentication error"), }; HttpResponse::build(match status { "Unauthorized" => actix_web::http::StatusCode::UNAUTHORIZED, "Forbidden" => actix_web::http::StatusCode::FORBIDDEN, + "ServiceUnavailable" => actix_web::http::StatusCode::SERVICE_UNAVAILABLE, _ => actix_web::http::StatusCode::INTERNAL_SERVER_ERROR, }) .json(json!({ diff --git a/crates/mem-cli/src/compaction.rs b/crates/mem-cli/src/compaction.rs index f90ad27..552d54f 100644 --- a/crates/mem-cli/src/compaction.rs +++ b/crates/mem-cli/src/compaction.rs @@ -12,10 +12,14 @@ use std::collections::HashMap; use tracing::{debug, info, warn}; use mem_core::edge::Edge; -use mem_ingest::entity_extractor::LlmCaller; +// LlmCaller trait (moved from mem_ingest) +#[async_trait::async_trait] +pub trait LlmCaller: Send + Sync { + async fn call(&self, prompt: &str) -> anyhow::Result; +} /// Compaction statistics -#[derive(Debug, Clone, Default)] +#[derive(Debug, Clone, Default, serde::Serialize)] pub struct CompactionStats { pub duplicate_edges_deleted: usize, pub stale_facts_deleted: usize, diff --git a/crates/mem-cli/src/handlers/middleware.rs b/crates/mem-cli/src/handlers/middleware.rs index 59f2a72..7103ecb 100644 --- a/crates/mem-cli/src/handlers/middleware.rs +++ b/crates/mem-cli/src/handlers/middleware.rs @@ -41,7 +41,7 @@ pub fn validate_and_rate_limit( })) })?; - jwt_validator.validate_bearer_token(auth_header).map_err(|e| { + crate::jwt_validator::JwtValidator::extract_bearer_token(auth_header).map_err(|e| { HttpResponse::Unauthorized().json(json!({ "error": format!("JWT validation failed: {}", e) })) @@ -51,10 +51,10 @@ pub fn validate_and_rate_limit( // 2. Rate limiting (if enabled) state .rate_limiter - .check_limit(endpoint, rate_limit) + .check("default", endpoint) .map_err(|e| { HttpResponse::TooManyRequests().json(json!({ - "error": format!("Rate limit exceeded: {}", e) + "error": format!("Rate limit exceeded: {}", e.reason()) })) })?; diff --git a/crates/mem-cli/src/handlers/ranking_handler.rs b/crates/mem-cli/src/handlers/ranking_handler.rs index ff70cc8..030be7f 100644 --- a/crates/mem-cli/src/handlers/ranking_handler.rs +++ b/crates/mem-cli/src/handlers/ranking_handler.rs @@ -171,7 +171,7 @@ pub struct RankedResult { /// GET /memory/ranking/profiles pub async fn get_ranking_profiles(req: HttpRequest) -> HttpResponse { // Verify auth - if let Err(e) = AuthGuard::extract_token(&req) { + if let Err(e) = AuthGuard::extract_token(req.headers().get("Authorization").and_then(|v| v.to_str().ok()).unwrap_or("")) { return HttpResponse::Unauthorized().json(json!({ "error": e.to_string() })); diff --git a/crates/mem-cli/src/handlers/rebuild_handler.rs b/crates/mem-cli/src/handlers/rebuild_handler.rs index 828371e..96fa834 100644 --- a/crates/mem-cli/src/handlers/rebuild_handler.rs +++ b/crates/mem-cli/src/handlers/rebuild_handler.rs @@ -54,7 +54,7 @@ pub async fn rebuild( pool: web::Data, ) -> HttpResponse { // Verify auth - if let Err(e) = AuthGuard::extract_token(&req) { + if let Err(e) = AuthGuard::extract_token(req.headers().get("Authorization").and_then(|v| v.to_str().ok()).unwrap_or("")) { return HttpResponse::Unauthorized().json(json!({ "error": e.to_string() })); @@ -155,7 +155,7 @@ pub async fn rebuild_status( pool: web::Data, ) -> HttpResponse { // Verify auth - if let Err(e) = AuthGuard::extract_token(&req) { + if let Err(e) = AuthGuard::extract_token(req.headers().get("Authorization").and_then(|v| v.to_str().ok()).unwrap_or("")) { return HttpResponse::Unauthorized().json(json!({ "error": e.to_string() })); @@ -192,11 +192,16 @@ pub async fn rebuild_status( async fn compute_state_checksum(pool: &PgPool, project: &str) -> Result { let mut hasher = Sha256::new(); - // Entities in order (by id) - let entities = sqlx::query!( - "SELECT id FROM memory_entity WHERE project_id = $1 ORDER BY id", - project + // Entities in order (by id) - using runtime query to avoid sqlx compile-time check + #[derive(sqlx::FromRow)] + struct IdRow { + id: String, + } + + let entities: Vec = sqlx::query_as::<_, IdRow>( + "SELECT id FROM memory_entity WHERE project_id = $1 ORDER BY id" ) + .bind(project) .fetch_all(pool) .await?; @@ -204,16 +209,16 @@ async fn compute_state_checksum(pool: &PgPool, project: &str) -> Result = sqlx::query_as::<_, IdRow>( + "SELECT id FROM memory_edge WHERE project_id = $1 ORDER BY id" ) + .bind(project) .fetch_all(pool) .await?; for row in &edges { - hasher.update(row.id.to_string().as_bytes()); + hasher.update(row.id.as_bytes()); } Ok(format!("{:x}", hasher.finalize())) diff --git a/crates/mem-cli/src/handlers/response_builder.rs b/crates/mem-cli/src/handlers/response_builder.rs index 843aa3e..404f816 100644 --- a/crates/mem-cli/src/handlers/response_builder.rs +++ b/crates/mem-cli/src/handlers/response_builder.rs @@ -25,6 +25,11 @@ pub fn internal_error(error: &str) -> HttpResponse { HttpResponse::InternalServerError().json(json!({ "error": error })) } +/// Build an unauthorized response (401) +pub fn unauthorized(error: &str) -> HttpResponse { + HttpResponse::Unauthorized().json(json!({ "error": error })) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/mem-cli/src/handlers/semantic.rs b/crates/mem-cli/src/handlers/semantic.rs index 8603e56..804e08e 100644 --- a/crates/mem-cli/src/handlers/semantic.rs +++ b/crates/mem-cli/src/handlers/semantic.rs @@ -142,8 +142,8 @@ pub async fn search_entities_handler( body.query, body.entity_type, body.start_time, body.end_time); // 3. Embed query - let query_embedding = match state.embeddings.embed_text(&body.query).await { - Ok(emb) => emb, + let query_embedding = match state.embeddings.embed_one(&body.query).await { + Ok(emb) => emb.to_vec(), Err(e) => { error!("Embedding failed: {}", e); return crate::handlers::response_builder::internal_error( @@ -278,8 +278,8 @@ pub async fn search_edges_handler( body.query, body.relation_type, body.start_time, body.end_time); // 3. Embed query - let query_embedding = match state.embeddings.embed_text(&body.query).await { - Ok(emb) => emb, + let query_embedding = match state.embeddings.embed_one(&body.query).await { + Ok(emb) => emb.to_vec(), Err(e) => { error!("Embedding failed: {}", e); return crate::handlers::response_builder::internal_error( @@ -361,8 +361,8 @@ pub async fn hybrid_search_handler( body.query, body.semantic_weight, body.lexical_weight); // 3. Embed query - let query_embedding = match state.embeddings.embed_text(&body.query).await { - Ok(emb) => emb, + let query_embedding = match state.embeddings.embed_one(&body.query).await { + Ok(emb) => emb.to_vec(), Err(e) => { error!("Embedding failed: {}", e); return crate::handlers::response_builder::internal_error( diff --git a/crates/mem-cli/src/handlers/synthesis.rs b/crates/mem-cli/src/handlers/synthesis.rs index fc30fc5..1a672f8 100644 --- a/crates/mem-cli/src/handlers/synthesis.rs +++ b/crates/mem-cli/src/handlers/synthesis.rs @@ -517,11 +517,12 @@ pub async fn reasoning_paths_handler( let elapsed = start_time.elapsed().as_millis(); info!("Paths: {} found in {}ms", paths.len(), elapsed); + let path_count = paths.len(); crate::handlers::response_builder::success_response(ReasoningPathsResponse { source_id: body.source_id.clone(), target_id: body.target_id.clone(), paths, - path_count: paths.len(), + path_count, process_time_ms: elapsed, }) } diff --git a/crates/mem-cli/src/handlers/unified_query.rs b/crates/mem-cli/src/handlers/unified_query.rs index 6bdf5b6..1d8415c 100644 --- a/crates/mem-cli/src/handlers/unified_query.rs +++ b/crates/mem-cli/src/handlers/unified_query.rs @@ -136,8 +136,8 @@ pub async fn unified_query_handler( body.search_type, body.query, body.entity_type, body.relation_type); // 3. Embed query once (reused for all search types) - let query_embedding = match state.embeddings.embed_text(&body.query).await { - Ok(emb) => emb, + let query_embedding = match state.embeddings.embed_one(&body.query).await { + Ok(emb) => emb.to_vec(), Err(e) => { error!("Embedding failed: {}", e); return crate::handlers::response_builder::internal_error( diff --git a/crates/mem-cli/src/handlers/unified_synthesis.rs b/crates/mem-cli/src/handlers/unified_synthesis.rs index 0a1ee7f..96940f9 100644 --- a/crates/mem-cli/src/handlers/unified_synthesis.rs +++ b/crates/mem-cli/src/handlers/unified_synthesis.rs @@ -158,12 +158,12 @@ pub async fn unified_synthesis_handler( // Entity Linking if body.link_entities { let linker = EntityLinker::new(state.pool.clone()); - match linker.link_entities(&body.content) { - Ok(links) => { + match linker.link_mentions(&body.content, &body.project).await { + Ok((links, _unlinked)) => { let alias_count = links.iter().filter(|l| l.confidence > 0.85).count(); entity_linking = Some(EntityLinkingResult { mention_links: links.iter().map(|l| MentionLinkResponse { - mention: l.mention.clone(), + mention: l.mention_text.clone(), entity_id: l.entity_id.clone(), confidence: l.confidence, }).collect(), @@ -179,14 +179,14 @@ pub async fn unified_synthesis_handler( // Inference if body.infer_facts { - let engine = InferenceEngine::new(state.pool.clone()); - match engine.infer_facts(&body.content, 5, 0.6, &body.project) { + let engine = InferenceEngine::new(state.pool.clone(), vec![]); + match engine.infer_facts(&body.project, &body.content, 5).await { Ok(facts) => { inference = Some(InferenceResult { inferred_facts: facts.iter().map(|f| InferredFactResponse { - source: f.source.clone(), - relation: f.relation.clone(), - target: f.target.clone(), + source: f.source_id.clone(), + relation: f.relation_type.clone(), + target: f.target_id.clone(), confidence: f.confidence, }).collect(), fact_count: facts.len(), diff --git a/crates/mem-cli/src/handlers/versioning_handler.rs b/crates/mem-cli/src/handlers/versioning_handler.rs index 2983aed..701d010 100644 --- a/crates/mem-cli/src/handlers/versioning_handler.rs +++ b/crates/mem-cli/src/handlers/versioning_handler.rs @@ -15,7 +15,7 @@ pub async fn get_entity_versions( pool: web::Data, ) -> HttpResponse { // Verify auth - if let Err(e) = AuthGuard::extract_token(&req) { + if let Err(e) = AuthGuard::extract_token(req.headers().get("Authorization").and_then(|v| v.to_str().ok()).unwrap_or("")) { return HttpResponse::Unauthorized().json(json!({ "error": e.to_string() })); @@ -46,7 +46,7 @@ pub async fn get_entity_version( path: web::Path<(String, i32)>, pool: web::Data, ) -> HttpResponse { - if let Err(e) = AuthGuard::extract_token(&req) { + if let Err(e) = AuthGuard::extract_token(req.headers().get("Authorization").and_then(|v| v.to_str().ok()).unwrap_or("")) { return HttpResponse::Unauthorized().json(json!({ "error": e.to_string() })); @@ -80,7 +80,7 @@ pub async fn get_entity_diff( query: web::Query, pool: web::Data, ) -> HttpResponse { - if let Err(e) = AuthGuard::extract_token(&req) { + if let Err(e) = AuthGuard::extract_token(req.headers().get("Authorization").and_then(|v| v.to_str().ok()).unwrap_or("")) { return HttpResponse::Unauthorized().json(json!({ "error": e.to_string() })); @@ -120,7 +120,7 @@ pub async fn get_entity_at_time( query: web::Query, pool: web::Data, ) -> HttpResponse { - if let Err(e) = AuthGuard::extract_token(&req) { + if let Err(e) = AuthGuard::extract_token(req.headers().get("Authorization").and_then(|v| v.to_str().ok()).unwrap_or("")) { return HttpResponse::Unauthorized().json(json!({ "error": e.to_string() })); @@ -164,7 +164,7 @@ pub async fn get_edge_versions( path: web::Path, pool: web::Data, ) -> HttpResponse { - if let Err(e) = AuthGuard::extract_token(&req) { + if let Err(e) = AuthGuard::extract_token(req.headers().get("Authorization").and_then(|v| v.to_str().ok()).unwrap_or("")) { return HttpResponse::Unauthorized().json(json!({ "error": e.to_string() })); @@ -196,7 +196,7 @@ pub async fn get_edge_diff( query: web::Query, pool: web::Data, ) -> HttpResponse { - if let Err(e) = AuthGuard::extract_token(&req) { + if let Err(e) = AuthGuard::extract_token(req.headers().get("Authorization").and_then(|v| v.to_str().ok()).unwrap_or("")) { return HttpResponse::Unauthorized().json(json!({ "error": e.to_string() })); diff --git a/crates/mem-cli/src/handlers/visualize_sse.rs b/crates/mem-cli/src/handlers/visualize_sse.rs index bb52866..58d2cd0 100644 --- a/crates/mem-cli/src/handlers/visualize_sse.rs +++ b/crates/mem-cli/src/handlers/visualize_sse.rs @@ -145,13 +145,15 @@ pub async fn visualize_stream_handler( match execute_streaming_visualization(&state, req_body).await { Ok(events) => { for event in events { - yield format_sse_event(event); + let data = format_sse_event(event); + yield Ok::(actix_web::web::Bytes::from(data)); } } Err(e) => { - yield format_sse_event(VisualizeEvent::Error { + let data = format_sse_event(VisualizeEvent::Error { message: e, }); + yield Ok::(actix_web::web::Bytes::from(data)); } } }; @@ -161,7 +163,7 @@ pub async fn visualize_stream_handler( .insert_header(("Cache-Control", "no-cache")) .insert_header(("Connection", "keep-alive")) .insert_header(("Transfer-Encoding", "chunked")) - .streaming_body(Box::pin(stream)) + .streaming(Box::pin(stream)) } /// Execute streaming visualization (generates events) diff --git a/crates/mem-cli/src/http_server.rs b/crates/mem-cli/src/http_server.rs index c05419c..bcaf31b 100644 --- a/crates/mem-cli/src/http_server.rs +++ b/crates/mem-cli/src/http_server.rs @@ -19,7 +19,11 @@ use crate::gateway_queue_adapter::GatewayQueueAdapter; use crate::queue_worker::{QueueWorker, QueueWorkerConfig}; use crate::queue_adapter::QueueAdapter; use crate::rbac::{AccessGuard, Claims as RbacClaims, builtin_role_provider, ResourceMeta, ResourceType, Verb, Visibility}; -use crate::handlers::{QueryParams, QueryParamsError, SearchMethod, build_search_response, LearnParams, LearnParamsError, build_learn_response}; +use crate::handlers::{ + QueryParams, QueryParamsError, SearchMethod, build_search_response, + LearnParams, LearnParamsError, build_learn_response, + visualize_handler, visualize_stream_handler, compact_handler +}; /// Server state with database and workers pub struct AppState { diff --git a/crates/mem-cli/src/lib.rs b/crates/mem-cli/src/lib.rs index 8701b89..161e8f1 100644 --- a/crates/mem-cli/src/lib.rs +++ b/crates/mem-cli/src/lib.rs @@ -2,6 +2,7 @@ pub mod endpoints; pub mod handlers; pub mod http_server; pub mod query; +pub mod auth; pub mod ingest_worker; pub mod query_worker; pub mod rate_limiter; @@ -30,7 +31,7 @@ pub mod federation; pub mod query_router; pub mod full_pipeline; pub mod authorized_pipeline; -pub mod ingest_with_persistence; +// pub mod ingest_with_persistence; // TODO: Fix db_repo integration pub mod auth_middleware; pub mod compaction; pub mod compaction_executor; @@ -38,6 +39,7 @@ pub mod agent; pub mod parallel_dual_write; pub use endpoints::{IngestQueue, IngestRequest, JobStatus}; +pub use http_server::{AppState, AuthMode}; pub use ingest_worker::IngestWorker; pub use query_worker::QueryWorker; pub use hybrid_retrieval::{HybridRetriever, RetrievalRoute, WikiScopedFilter, RankedCandidate}; diff --git a/crates/mem-cli/src/parallel_dual_write.rs b/crates/mem-cli/src/parallel_dual_write.rs index 6b9791f..6d97eb0 100644 --- a/crates/mem-cli/src/parallel_dual_write.rs +++ b/crates/mem-cli/src/parallel_dual_write.rs @@ -116,13 +116,13 @@ impl ParallelDualWriteIndexer { // Spawn background task (non-blocking) tokio::spawn(async move { - let result = opensearch.index_chunk( + let result = opensearch.index_document( &chunk_id, &chunk.content, &chunk.source, - &chunk.project, &chunk.level, - &chunk.breadcrumb.join(" > "), + chunk.breadcrumb.clone(), + "", // jwt_token - not available in background task ).await; match result { @@ -139,8 +139,8 @@ impl ParallelDualWriteIndexer { &self, chunks: Vec<(&IndexableChunk, Vec)>, ) -> Vec { - let futures = chunks.into_iter().map(|(chunk, embedding)| { - self.index_parallel(chunk, &embedding) + let futures = chunks.into_iter().map(|(chunk, embedding)| async move { + self.index_parallel(chunk, &embedding).await }); futures::future::join_all(futures) diff --git a/crates/mem-cli/src/query/bfs_graph_traversal.rs b/crates/mem-cli/src/query/bfs_graph_traversal.rs index cd233cb..c7395e5 100644 --- a/crates/mem-cli/src/query/bfs_graph_traversal.rs +++ b/crates/mem-cli/src/query/bfs_graph_traversal.rs @@ -6,7 +6,7 @@ use std::collections::{HashMap, VecDeque}; use serde::{Deserialize, Serialize}; use chrono::{DateTime, Utc}; -use sqlx::{Pool, Postgres}; +use sqlx::{Pool, Postgres, Row}; /// A node in the traversal result #[derive(Debug, Clone, Serialize, Deserialize)] @@ -196,6 +196,7 @@ impl BfsGraphTraversal { }); } + let edge_count = edges.len(); Ok(GraphData { nodes, edges, @@ -203,7 +204,7 @@ impl BfsGraphTraversal { requested_depth: config.max_depth, max_depth_reached: max_depth, node_count: visited.len(), - edge_count: edges.len(), + edge_count, depth_breakdown, traversal_time_ms: start_time.elapsed().as_millis() as u64, }) diff --git a/crates/mem-cli/src/query/community_detector.rs b/crates/mem-cli/src/query/community_detector.rs index 6fb8209..60b06ef 100644 --- a/crates/mem-cli/src/query/community_detector.rs +++ b/crates/mem-cli/src/query/community_detector.rs @@ -185,9 +185,9 @@ impl CommunityDetector { communities_vec.push(Community { id: comm_id, + size: members.len(), entity_ids: members.into_iter().collect(), entity_names, - size: members.len(), modularity_contribution: modularity_contrib, average_strength: strength, density, @@ -196,9 +196,9 @@ impl CommunityDetector { } // 5. Calculate total modularity - let total_modularity = communities_vec + let total_modularity: f64 = communities_vec .iter() - .map(|c| c.modularity_contribution) + .map(|c| c.modularity_contribution as f64) .sum(); let average_community_size = if communities_vec.is_empty() { @@ -210,9 +210,9 @@ impl CommunityDetector { let result = CommunityDetectionResult { entity_count: entities.len(), edge_count: edges.len(), - communities: communities_vec, community_count: communities_vec.len(), - total_modularity: total_modularity.max(-1.0).min(1.0), + communities: communities_vec, + total_modularity: total_modularity.max(-1.0).min(1.0) as f32, average_community_size, }; diff --git a/crates/mem-cli/src/query/community_metrics.rs b/crates/mem-cli/src/query/community_metrics.rs index 23c2cac..b4cf814 100644 --- a/crates/mem-cli/src/query/community_metrics.rs +++ b/crates/mem-cli/src/query/community_metrics.rs @@ -202,10 +202,10 @@ impl CommunityMetricsCalculator { } /// Rank communities by metric - pub fn rank_by_metric( - metrics: &[CommunityMetrics], + pub fn rank_by_metric<'a>( + metrics: &'a [CommunityMetrics], metric: &str, - ) -> Vec<&CommunityMetrics> { + ) -> Vec<&'a CommunityMetrics> { let mut sorted = metrics.iter().collect::>(); match metric { diff --git a/crates/mem-cli/src/query/faceted_search.rs b/crates/mem-cli/src/query/faceted_search.rs index 34810f5..f76d88e 100644 --- a/crates/mem-cli/src/query/faceted_search.rs +++ b/crates/mem-cli/src/query/faceted_search.rs @@ -3,7 +3,7 @@ //! Enables multi-dimensional filtering across entities and edges. //! Supports entity types, relation types, date ranges, confidence levels, and more. -use chrono::{DateTime, Utc}; +use chrono::{DateTime, Timelike, Utc}; use serde::{Deserialize, Serialize}; use sqlx::{Pool, Postgres}; use std::collections::HashMap; @@ -42,7 +42,7 @@ pub struct AvailableFacets { } /// Facet filters for a query -#[derive(Debug, Clone, Default, Deserialize)] +#[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct FacetFilters { /// Filter by entity types (OR within facet, AND across facets) pub entity_types: Option>, diff --git a/crates/mem-cli/src/query/force_directed_layout.rs b/crates/mem-cli/src/query/force_directed_layout.rs index a637da5..53c80b1 100644 --- a/crates/mem-cli/src/query/force_directed_layout.rs +++ b/crates/mem-cli/src/query/force_directed_layout.rs @@ -7,7 +7,7 @@ use serde::{Deserialize, Serialize}; use super::bfs_graph_traversal::{GraphData, TraversalNode, TraversalEdge}; /// 2D position (X, Y coordinates) -#[derive(Debug, Clone, Copy, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize)] pub struct Position { pub x: f32, pub y: f32, diff --git a/crates/mem-cli/src/query/inference_engine.rs b/crates/mem-cli/src/query/inference_engine.rs index cd38fb9..a080a25 100644 --- a/crates/mem-cli/src/query/inference_engine.rs +++ b/crates/mem-cli/src/query/inference_engine.rs @@ -4,6 +4,8 @@ //! confidence propagation through reasoning chains. use std::collections::{HashMap, HashSet, VecDeque}; +use std::pin::Pin; +use std::future::Future; use sqlx::PgPool; use serde::{Deserialize, Serialize}; use tracing::{debug, warn}; @@ -293,18 +295,19 @@ impl InferenceEngine { } /// DFS to find all paths - async fn dfs_paths( - &self, - current: &str, - target: &str, - project_id: &str, + fn dfs_paths<'a>( + &'a self, + current: &'a str, + target: &'a str, + project_id: &'a str, remaining_hops: usize, - path: &mut Vec, - relations: &mut Vec, - confidences: &mut Vec, - visited: &mut HashSet, - results: &mut Vec, - ) -> Result<(), String> { + path: &'a mut Vec, + relations: &'a mut Vec, + confidences: &'a mut Vec, + visited: &'a mut HashSet, + results: &'a mut Vec, + ) -> Pin> + Send + 'a>> { + Box::pin(async move { if remaining_hops == 0 { return Ok(()); } @@ -350,6 +353,7 @@ impl InferenceEngine { } Ok(()) + }) // Box::pin } } diff --git a/crates/mem-cli/src/query/path_finder.rs b/crates/mem-cli/src/query/path_finder.rs index c688f16..72edd53 100644 --- a/crates/mem-cli/src/query/path_finder.rs +++ b/crates/mem-cli/src/query/path_finder.rs @@ -6,6 +6,8 @@ use serde::{Deserialize, Serialize}; use sqlx::{Pool, Postgres}; use std::collections::{HashMap, HashSet, VecDeque}; +use std::pin::Pin; +use std::future::Future; use tracing::{debug, info}; /// A single path through the graph @@ -123,13 +125,14 @@ impl PathFinder { info!("Found shortest path: {} → {} (distance: {})", source_id, target_id, final_entities.len() - 1); + let distance = final_entities.len() - 1; return Ok(Some(Path { source_id: source_id.to_string(), target_id: target_id.to_string(), entity_ids: final_entities, entity_names: vec![], // Could fetch from DB if needed relation_types: final_relations, - distance: final_entities.len() - 1, + distance, total_confidence: final_confidence.max(0.0).min(1.0), })); } @@ -293,19 +296,20 @@ impl PathFinder { } /// DFS helper for finding all paths - async fn dfs_paths( - &self, - source_id: &str, - target_id: &str, + fn dfs_paths<'a>( + &'a self, + source_id: &'a str, + target_id: &'a str, current_path: Vec, relations_path: Vec, confidence: f32, depth: usize, max_depth: usize, - paths_found: &mut Vec, - visited: &mut HashSet, + paths_found: &'a mut Vec, + visited: &'a mut HashSet, max_paths: usize, - ) -> Result<(), String> { + ) -> Pin> + Send + 'a>> { + Box::pin(async move { if paths_found.len() >= max_paths { return Ok(()); // Found enough paths } @@ -328,13 +332,14 @@ impl PathFinder { let final_confidence = confidence * edge.confidence; + let distance = final_path.len() - 1; paths_found.push(Path { source_id: source_id.to_string(), target_id: target_id.to_string(), entity_ids: final_path, entity_names: vec![], relation_types: final_relations, - distance: final_path.len() - 1, + distance, total_confidence: final_confidence.max(0.0).min(1.0), }); @@ -370,6 +375,7 @@ impl PathFinder { } Ok(()) + }) // Box::pin } /// Fetch direct neighbors of an entity diff --git a/crates/mem-cli/src/query/semantic_retriever.rs b/crates/mem-cli/src/query/semantic_retriever.rs index 049c983..28c6b35 100644 --- a/crates/mem-cli/src/query/semantic_retriever.rs +++ b/crates/mem-cli/src/query/semantic_retriever.rs @@ -138,7 +138,7 @@ impl SemanticRetriever { .await .map_err(|e| format!("Database error: {}", e))?; - let entities = results + let entities: Vec<_> = results .into_iter() .map(|(id, name, entity_type, score, metadata)| EntityResult { id, @@ -213,7 +213,7 @@ impl SemanticRetriever { .await .map_err(|e| format!("Database error: {}", e))?; - let edges = results + let edges: Vec<_> = results .into_iter() .map(|(id, src_id, tgt_id, src_name, tgt_name, rel_type, fact, score, conf)| { EdgeResult { diff --git a/crates/mem-cli/src/query/summarizer.rs b/crates/mem-cli/src/query/summarizer.rs index 9acda52..9bd135e 100644 --- a/crates/mem-cli/src/query/summarizer.rs +++ b/crates/mem-cli/src/query/summarizer.rs @@ -261,7 +261,7 @@ impl Summarizer { } /// Split text into sentences - fn split_sentences(&self, text: &str) -> Vec<&str> { + fn split_sentences<'a>(&self, text: &'a str) -> Vec<&'a str> { text.split('.').map(|s| s.trim()).filter(|s| !s.is_empty()).collect() } @@ -331,7 +331,7 @@ impl Summarizer { let overlap = entities1 .iter() - .filter(|e| entities2.contains(e)) + .filter(|e| entities2.contains(*e)) .count(); coherence += overlap as f32 / (entities1.len().max(entities2.len()) as f32).max(1.0); } diff --git a/crates/mem-cli/src/query_router.rs b/crates/mem-cli/src/query_router.rs index 05f119e..406c02b 100644 --- a/crates/mem-cli/src/query_router.rs +++ b/crates/mem-cli/src/query_router.rs @@ -167,7 +167,7 @@ impl QueryRouter { let latency_ms = start.elapsed().as_millis() as u64; // Phase 8: Answer Validation (confidence scoring) - use crate::answer_validator::{AnswerValidator, AnswerValidationConfig, ConfidenceSignals}; + use crate::query::answer_validator::{AnswerValidator, AnswerValidationConfig, ConfidenceSignals}; let validator = AnswerValidator::new(AnswerValidationConfig::default()); let avg_score = selected_chunks.iter().map(|c| c.final_score).sum::() / (selected_chunks.len() as f32).max(1.0); @@ -245,6 +245,8 @@ impl QueryRouter { prefilter_size, metrics, latency_ms, + confidence_score: 1.0, + is_valid: true, }) } diff --git a/crates/mem-store/.sqlx/query-1e81bb729531ca33e4cef21623bcfe4fafb0c1bd435353b205f582bfda8873bc.json b/crates/mem-store/.sqlx/query-1e81bb729531ca33e4cef21623bcfe4fafb0c1bd435353b205f582bfda8873bc.json new file mode 100644 index 0000000..8cbf0c7 --- /dev/null +++ b/crates/mem-store/.sqlx/query-1e81bb729531ca33e4cef21623bcfe4fafb0c1bd435353b205f582bfda8873bc.json @@ -0,0 +1,52 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_entity_version\n WHERE entity_id = $1\n ORDER BY version_num DESC\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "version_num", + "type_info": "Int4" + }, + { + "ordinal": 1, + "name": "operation", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "snapshot", + "type_info": "Jsonb" + }, + { + "ordinal": 3, + "name": "changed_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 4, + "name": "changed_by", + "type_info": "Varchar" + }, + { + "ordinal": 5, + "name": "fields_changed!", + "type_info": "TextArray" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + false, + false, + false, + false, + null + ] + }, + "hash": "1e81bb729531ca33e4cef21623bcfe4fafb0c1bd435353b205f582bfda8873bc" +} diff --git a/crates/mem-store/.sqlx/query-62d65d4afc4d292b37de8e5cb59fbd51c602bdc1b437988f54e6c7fe268b9816.json b/crates/mem-store/.sqlx/query-62d65d4afc4d292b37de8e5cb59fbd51c602bdc1b437988f54e6c7fe268b9816.json new file mode 100644 index 0000000..6a40d9c --- /dev/null +++ b/crates/mem-store/.sqlx/query-62d65d4afc4d292b37de8e5cb59fbd51c602bdc1b437988f54e6c7fe268b9816.json @@ -0,0 +1,52 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_edge_version\n WHERE edge_id = $1\n ORDER BY version_num DESC\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "version_num", + "type_info": "Int4" + }, + { + "ordinal": 1, + "name": "operation", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "snapshot", + "type_info": "Jsonb" + }, + { + "ordinal": 3, + "name": "changed_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 4, + "name": "changed_by", + "type_info": "Varchar" + }, + { + "ordinal": 5, + "name": "fields_changed!", + "type_info": "TextArray" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + false, + false, + false, + false, + false, + null + ] + }, + "hash": "62d65d4afc4d292b37de8e5cb59fbd51c602bdc1b437988f54e6c7fe268b9816" +} diff --git a/crates/mem-store/.sqlx/query-aee5900f5e3d7cbba23729bbf2dd033dcc4cb41f6c851bf447a9238810684d18.json b/crates/mem-store/.sqlx/query-aee5900f5e3d7cbba23729bbf2dd033dcc4cb41f6c851bf447a9238810684d18.json new file mode 100644 index 0000000..61e13f9 --- /dev/null +++ b/crates/mem-store/.sqlx/query-aee5900f5e3d7cbba23729bbf2dd033dcc4cb41f6c851bf447a9238810684d18.json @@ -0,0 +1,53 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_entity_version\n WHERE entity_id = $1 AND changed_at <= $2\n ORDER BY version_num DESC\n LIMIT 1\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "version_num", + "type_info": "Int4" + }, + { + "ordinal": 1, + "name": "operation", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "snapshot", + "type_info": "Jsonb" + }, + { + "ordinal": 3, + "name": "changed_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 4, + "name": "changed_by", + "type_info": "Varchar" + }, + { + "ordinal": 5, + "name": "fields_changed!", + "type_info": "TextArray" + } + ], + "parameters": { + "Left": [ + "Text", + "Timestamptz" + ] + }, + "nullable": [ + false, + false, + false, + false, + false, + null + ] + }, + "hash": "aee5900f5e3d7cbba23729bbf2dd033dcc4cb41f6c851bf447a9238810684d18" +} diff --git a/crates/mem-store/.sqlx/query-c045466e1fe037dbdafea1008f262f4e48f104ea77732aa1d32ecb797f70e71d.json b/crates/mem-store/.sqlx/query-c045466e1fe037dbdafea1008f262f4e48f104ea77732aa1d32ecb797f70e71d.json new file mode 100644 index 0000000..9fe298e --- /dev/null +++ b/crates/mem-store/.sqlx/query-c045466e1fe037dbdafea1008f262f4e48f104ea77732aa1d32ecb797f70e71d.json @@ -0,0 +1,53 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_entity_version\n WHERE entity_id = $1 AND version_num = $2\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "version_num", + "type_info": "Int4" + }, + { + "ordinal": 1, + "name": "operation", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "snapshot", + "type_info": "Jsonb" + }, + { + "ordinal": 3, + "name": "changed_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 4, + "name": "changed_by", + "type_info": "Varchar" + }, + { + "ordinal": 5, + "name": "fields_changed!", + "type_info": "TextArray" + } + ], + "parameters": { + "Left": [ + "Text", + "Int4" + ] + }, + "nullable": [ + false, + false, + false, + false, + false, + null + ] + }, + "hash": "c045466e1fe037dbdafea1008f262f4e48f104ea77732aa1d32ecb797f70e71d" +} diff --git a/crates/mem-store/.sqlx/query-ca6872495bc04c6a65531279af8c758637c902dda2cc10366662988c6973ca48.json b/crates/mem-store/.sqlx/query-ca6872495bc04c6a65531279af8c758637c902dda2cc10366662988c6973ca48.json new file mode 100644 index 0000000..c4cca32 --- /dev/null +++ b/crates/mem-store/.sqlx/query-ca6872495bc04c6a65531279af8c758637c902dda2cc10366662988c6973ca48.json @@ -0,0 +1,53 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_edge_version\n WHERE edge_id = $1 AND version_num = $2\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "version_num", + "type_info": "Int4" + }, + { + "ordinal": 1, + "name": "operation", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "snapshot", + "type_info": "Jsonb" + }, + { + "ordinal": 3, + "name": "changed_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 4, + "name": "changed_by", + "type_info": "Varchar" + }, + { + "ordinal": 5, + "name": "fields_changed!", + "type_info": "TextArray" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Int4" + ] + }, + "nullable": [ + false, + false, + false, + false, + false, + null + ] + }, + "hash": "ca6872495bc04c6a65531279af8c758637c902dda2cc10366662988c6973ca48" +} diff --git a/crates/mem-store/src/db_repo.rs b/crates/mem-store/src/db_repo.rs index e37def4..af4e85c 100644 --- a/crates/mem-store/src/db_repo.rs +++ b/crates/mem-store/src/db_repo.rs @@ -6,8 +6,10 @@ use sqlx::{Pool, Postgres, Row, Transaction, Error as SqlxError}; use serde::{Deserialize, Serialize}; use chrono::{DateTime, Utc}; -use crate::entity_repo::{Entity, EntityRepo}; -use crate::edge_repo::{Edge, EdgeRepo}; +use mem_core::entity::Entity; +use mem_core::edge::Edge; +use crate::entity_repo::EntityRepoOps; +use crate::edge_repo::EdgeRepoOps; /// Database connection error types #[derive(Debug, Clone)] diff --git a/crates/mem-store/src/lib.rs b/crates/mem-store/src/lib.rs index 5090b49..81c68e5 100644 --- a/crates/mem-store/src/lib.rs +++ b/crates/mem-store/src/lib.rs @@ -8,6 +8,7 @@ pub mod edge_repo; pub mod community_repo; pub mod versioning; pub mod audit_logger; +// pub mod db_repo; // TODO: Fix Entity schema integration pub use event_log::{EventRecord, LogWriter}; pub use pgvector::{VectorRecord, VectorStore, ChunkL0, MemoryL1, MemoryL2}; diff --git a/crates/mem-store/src/versioning.rs b/crates/mem-store/src/versioning.rs index e463240..0c6c723 100644 --- a/crates/mem-store/src/versioning.rs +++ b/crates/mem-store/src/versioning.rs @@ -111,28 +111,28 @@ impl EntityVersioningService { let mut modified = Vec::new(); // Check removed and modified - if let Some(from) = from_obj { + if let Some(ref from) = from_obj { for (key, from_val) in from { if let Some(to) = &to_obj { - if let Some(to_val) = to.get(&key) { - if from_val != *to_val { + if let Some(to_val) = to.get(key) { + if from_val != to_val { modified.push(DiffField { - name: key, - from_value: Some(from_val), + name: key.clone(), + from_value: Some(from_val.clone()), to_value: Some(to_val.clone()), }); } } else { removed.push(DiffField { - name: key, - from_value: Some(from_val), + name: key.clone(), + from_value: Some(from_val.clone()), to_value: None, }); } } else { removed.push(DiffField { - name: key, - from_value: Some(from_val), + name: key.clone(), + from_value: Some(from_val.clone()), to_value: None, }); } @@ -301,28 +301,28 @@ fn compute_diff( let mut removed = Vec::new(); let mut modified = Vec::new(); - if let Some(from) = from_obj { + if let Some(ref from) = from_obj { for (key, from_val) in from { if let Some(to) = &to_obj { - if let Some(to_val) = to.get(&key) { - if from_val != *to_val { + if let Some(to_val) = to.get(key) { + if from_val != to_val { modified.push(DiffField { - name: key, - from_value: Some(from_val), + name: key.clone(), + from_value: Some(from_val.clone()), to_value: Some(to_val.clone()), }); } } else { removed.push(DiffField { - name: key, - from_value: Some(from_val), + name: key.clone(), + from_value: Some(from_val.clone()), to_value: None, }); } } else { removed.push(DiffField { - name: key, - from_value: Some(from_val), + name: key.clone(), + from_value: Some(from_val.clone()), to_value: None, }); }