/// Sync routes — peer delta exchange and peer registry. /// /// All routes under /sync require Authorization: Bearer {api_key}. /// /// GET /sync/delta?since={ms}&peer_id={uuid} — generate delta for caller /// POST /sync/push — receive delta from peer /// POST /sync/peers — register a new peer /// GET /sync/peers — list peers /// DELETE /sync/peers/{id} — remove peer use axum::{ extract::{Path, Query, State}, http::StatusCode, Json, }; use engram_core::types::MemoryTier; use engram_sync::{Peer, SyncDelta}; use serde::Deserialize; use std::sync::Arc; use uuid::Uuid; use crate::state::AppState; // ── GET /sync/delta ─────────────────────────────────────────────────────────── #[derive(Deserialize)] pub struct DeltaParams { pub since: Option, pub peer_id: Option, } pub async fn get_delta( State(state): State>, Query(params): Query, ) -> Result, StatusCode> { let since = params.since.unwrap_or(0); // Determine which tiers to expose based on caller's peer_id let tiers: Vec = { let engine = state.sync_engine.lock().await; if let Some(peer_id) = params.peer_id { if let Some(peer) = engine.get_peer(peer_id) { if peer.trusted { peer.sync_tiers.clone() } else { vec![MemoryTier::Semantic] } } else { vec![MemoryTier::Semantic] } } else { vec![MemoryTier::Semantic] } // engine lock released here }; let engine = state.sync_engine.lock().await; let delta = engine .generate_delta(since, &tiers) .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; Ok(Json(delta)) } // ── POST /sync/push ─────────────────────────────────────────────────────────── pub async fn push_delta( State(state): State>, Json(delta): Json, ) -> Result { // Accepted tiers for incoming pushes (Semantic and Procedural by default) let accepted_tiers = vec![MemoryTier::Semantic, MemoryTier::Procedural]; // Apply the delta using the DB handle directly (no async needed) let db = state.db.lock().map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; // Tombstones for id in &delta.tombstones { let _ = db.delete_node(*id); } // Nodes for node in delta.nodes { if !accepted_tiers.contains(&node.tier) { continue; } if db.get_node(node.id).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?.is_some() { continue; } db.put_node(node).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; } // Edges for edge in delta.edges { let from_ok = db.get_node(edge.from_id) .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? .is_some(); let to_ok = db.get_node(edge.to_id) .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? .is_some(); if from_ok && to_ok { let _ = db.put_edge(edge); } } Ok(StatusCode::OK) } // ── Peer registry ───────────────────────────────────────────────────────────── pub async fn list_peers( State(state): State>, ) -> Result>, StatusCode> { let engine = state.sync_engine.lock().await; Ok(Json(engine.list_peers().to_vec())) } pub async fn register_peer( State(state): State>, Json(peer): Json, ) -> Result { let mut engine = state.sync_engine.lock().await; engine.add_peer(peer); Ok(StatusCode::CREATED) } pub async fn delete_peer( State(state): State>, Path(id): Path, ) -> Result { let mut engine = state.sync_engine.lock().await; engine.remove_peer(id); Ok(StatusCode::NO_CONTENT) }