agentmux_srv/backend/storage/store.rs
1// Copyright 2025, Command Line Inc.
2// Copyright 2025-2026, AgentMux Corp.
3// SPDX-License-Identifier: Apache-2.0
4
5//! Store: generic OID-based CRUD for StoreObj types.
6//! Port of Go's pkg/mstore/mstore_dbops.go + mstore_dbsetup.go.
7//!
8//! Uses `Mutex<Connection>` matching Go's `MaxOpenConns(1)`.
9//! SQLite WAL mode + 5s busy timeout (same as Go).
10
11
12use std::path::{Path, PathBuf};
13use std::sync::{Arc, Mutex};
14
15use rusqlite::{params, Connection};
16
17use crate::backend::obj::{mux_obj_from_json, mux_obj_to_json, StoreObj};
18use crate::registry::{DefinitionStore, Registry};
19
20use super::error::StoreError;
21use super::migrations::{
22 check_schema_compat, run_identity_store_schema, run_object_schema, run_shared_store_schema,
23 stamp_version, IDENTITY_STORE_SCHEMA_VERSION, OBJECT_SCHEMA_VERSION,
24 SHARED_STORE_SCHEMA_VERSION,
25};
26
27/// SQLite-backed object store for StoreObj types.
28pub struct Store {
29 /// `pub(super)` so sibling subsystem modules (e.g. `bundles`)
30 /// can take the lock. Each per-subsystem file adds methods to `Store`
31 /// via `impl Store {}` and needs the connection.
32 pub(super) conn: Mutex<Connection>,
33 /// Cross-version named-agent registry. `None` for in-memory test
34 /// stores; `Some` for production srv. Mutations to
35 /// `db_agent_instances` parallel-write to this registry when set.
36 /// See `docs/specs/SPEC_SHARED_AGENT_REGISTRY_2026_05_12.md`.
37 registry: Mutex<Option<Arc<Registry>>>,
38 /// GLOBAL (cross-channel) agent-definition store. `None` for in-memory
39 /// test stores and when the shared dir can't be resolved; `Some` for
40 /// production srv. Definition mutations mirror to it so an agent created
41 /// in one channel is visible in every channel (cross-channel agent
42 /// persistence, `docs/specs/SPEC_CROSS_CHANNEL_AGENT_PERSISTENCE_2026-06-13.md`).
43 def_registry: Mutex<Option<Arc<DefinitionStore>>>,
44 /// token → UID for attributing requests (identity M4a). `None` except on
45 /// the channel's object store at runtime; see `attach_token_index`.
46 pub(super) token_index: Mutex<Option<Arc<super::agent_tokens::TokenIndex>>>,
47 /// The channel-wide WAN identity store (`wan-identity/wan.db`). `None`
48 /// except on the channel's object store at runtime, and when the file
49 /// couldn't be opened — WAN signing is then off. See `wan_identity.rs`.
50 wan_identity: Mutex<Option<Arc<super::wan_identity::WanIdentityStore>>>,
51 /// Base directory that named-instance `working_directory` values are
52 /// expressed **relative to** in the instance registry (write side:
53 /// `registry_mirror`; read side: the `listnamedagents` handler).
54 ///
55 /// This is the **current channel's** agents dir (`channels/<ch>/agents`,
56 /// i.e. `AGENTMUX_AGENTS_DIR`). It must be tracked separately from the
57 /// registry's own file location because P0.3 re-roots the registry to the
58 /// global `~/.agentmux/shared/agents/registry/` — once the registry no
59 /// longer sits under `channels/<ch>/agents/`, its parent (`agents_root()`)
60 /// stops coinciding with the channel agents dir, and using it to strip /
61 /// re-join `working_directory` would drop every live instance.
62 ///
63 /// In production it is wired from `AGENTMUX_AGENTS_DIR` in `main.rs`
64 /// (P0.3b), atomically with the re-root to the global shared registry —
65 /// which is why the wiring waited for the re-root: setting it earlier would
66 /// diverge in dev mode, where `AGENTMUX_AGENTS_DIR` ≠ the (then
67 /// channel-local) registry parent. `None` only for in-memory test stores
68 /// and odd envs where the var is unset; the accessor then falls back to the
69 /// registry's parent (which equals the channel agents dir in the pre-re-root
70 /// layout), so existing mirror tests are unchanged. When set, it is passed
71 /// in explicitly — never read from ambient env inside the Store — so tests
72 /// running inside an AgentMux pane don't pick up the host's
73 /// `AGENTMUX_AGENTS_DIR`. See
74 /// `docs/specs/SPEC_CROSS_CHANNEL_AGENT_PERSISTENCE_2026-06-13.md` (P0.3).
75 registry_agents_base: Mutex<Option<PathBuf>>,
76 /// Serializes `muxbus_save` end-to-end (keychain read, keychain write,
77 /// SQL write, and any rollback) — reagent P1 on #2260: two concurrent
78 /// `muxbus_save` calls (the `muxbus.login` RPC handler and the broker's
79 /// registered refresh closure can each call it independently) would
80 /// otherwise race on the keychain's "previous state" snapshot; if the
81 /// second call's SQL write then failed, its rollback could restore a
82 /// stale pre-first-call blob, silently reverting a credential the first
83 /// call already committed. `self.conn`'s own lock only covers the SQL
84 /// portion, not the keychain read/write either side of it, so a
85 /// dedicated lock is needed rather than reusing that one. `pub(super)`
86 /// like `conn` — `muxbus.rs` is a sibling module under `backend::storage`.
87 pub(super) muxbus_save_lock: Mutex<()>,
88 /// Told about every change to a Global Memory entry or its order, after
89 /// it commits — the Global Memory record's write-through
90 /// (`backend::global_memory_record`,
91 /// SPEC_MEMORY_FOLLOWS_THE_AGENT_2026_09_24.md §2.1.6). `None` except on
92 /// the channel's Global Memory store at runtime.
93 global_memory_observer: Mutex<Option<GlobalMemoryObserver>>,
94}
95
96/// A change to Global Memory, for [`Store::set_global_memory_observer`].
97#[derive(Debug, Clone, PartialEq, Eq, Hash)]
98pub enum GlobalMemoryChange {
99 /// The bundle with this id was written or deleted (it may or may not be
100 /// global, before or after).
101 Entry(String),
102 /// Global Memory was reordered.
103 Order,
104}
105
106/// Must not block or call back into the store: it runs where the change was
107/// made. The record's observer only queues the change for a worker.
108pub type GlobalMemoryObserver = Arc<dyn Fn(GlobalMemoryChange) + Send + Sync>;
109
110impl Store {
111 /// Open a Store backed by a file on disk.
112 /// Configures WAL mode and 5s busy timeout (matching Go).
113 pub fn open(path: &Path) -> Result<Self, StoreError> {
114 let conn = Connection::open(path)?;
115 Self::configure_and_migrate(conn)
116 }
117
118 /// Open an existing store **read-only**: `SQLITE_OPEN_READ_ONLY`, no
119 /// WAL/journal pragma, no schema setup, no version stamp — nothing that
120 /// could write. For doctor checks (`migrate --verify`, `muxspect
121 /// migrations`) that need the typed read API against a store they must
122 /// not touch, including one on a read-only backup (codex P2 on #3070).
123 /// Any read of a table the store predates simply errors. Fails if the
124 /// file does not exist — a doctor must never conjure a store.
125 pub fn open_read_only(path: &Path) -> Result<Self, StoreError> {
126 let conn = Connection::open_with_flags(path, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY)?;
127 conn.execute_batch("PRAGMA busy_timeout=5000;")?;
128 Ok(Self {
129 conn: Mutex::new(conn),
130 registry: Mutex::new(None),
131 def_registry: Mutex::new(None),
132 token_index: Mutex::new(None),
133 wan_identity: Mutex::new(None),
134 registry_agents_base: Mutex::new(None),
135 muxbus_save_lock: Mutex::new(()),
136 global_memory_observer: Mutex::new(None),
137 })
138 }
139
140 /// Open an in-memory Store for testing.
141 #[allow(dead_code)]
142 pub fn open_in_memory() -> Result<Self, StoreError> {
143 let conn = Connection::open_in_memory()?;
144 Self::configure_and_migrate(conn)
145 }
146
147 /// Which store this is, stable for its lifetime: the database file, or —
148 /// for an in-memory store — its address. Lets a record written into a
149 /// machine-wide zone say which channel's store wrote it.
150 pub fn origin_id(&self) -> String {
151 let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
152 match conn.path() {
153 Some(p) if !p.is_empty() => p.to_string(),
154 _ => format!("memory:{:p}", self),
155 }
156 }
157
158 /// TEST-ONLY: additionally install the **identity-store** schema on this
159 /// store, on top of whatever schema it already carries.
160 ///
161 /// `server::tests::test_state` deliberately aliases `mstore`, `id_store`
162 /// and `identity_store` to ONE in-memory store so a test can seed data
163 /// through any of them — a lot of existing setup code depends on that.
164 /// The cost is that the single store carried only the object schema, so a
165 /// handler touching a table that exists ONLY in the global identity store
166 /// (`db_work_queue`) failed with a bare "no such table" 500 in tests while
167 /// being correct in production — the harness silently answered a different
168 /// question than the one under test.
169 ///
170 /// Every statement in that schema is `CREATE TABLE/INDEX IF NOT EXISTS`,
171 /// so tables already present from the object schema are left untouched;
172 /// this only fills in the ones that are missing.
173 #[cfg(test)]
174 pub fn apply_identity_schema_for_tests(&self) -> Result<(), StoreError> {
175 let conn = self.conn.lock().unwrap();
176 run_identity_store_schema(&conn)?;
177 Ok(conn.execute_batch(super::work_claims::SCHEMA)?)
178 }
179
180 /// Open the GLOBAL shared store at `path` (`~/.agentmux/shared/store.db`).
181 ///
182 /// Uses the same WAL + busy-timeout config as `open()` but runs
183 /// `run_shared_store_schema` instead of `run_object_schema` so only
184 /// the durable-user-content tables are created (identity, memory, drone
185 /// definitions, muxbus creds). Per-channel session tables are intentionally
186 /// absent — calling per-channel CRUD methods on this store will error.
187 pub fn open_shared(path: &Path) -> Result<Self, StoreError> {
188 let conn = Connection::open(path)?;
189 conn.execute_batch(
190 "PRAGMA journal_mode=WAL;
191 PRAGMA busy_timeout=5000;
192 PRAGMA foreign_keys=ON;
193 PRAGMA synchronous=NORMAL;
194 PRAGMA cache_size=-8000;
195 PRAGMA mmap_size=268435456;
196 PRAGMA temp_store=MEMORY;",
197 )?;
198 check_schema_compat(&conn, SHARED_STORE_SCHEMA_VERSION, "store.db")?;
199 run_shared_store_schema(&conn)?;
200 stamp_version(&conn, SHARED_STORE_SCHEMA_VERSION)?;
201 Ok(Self {
202 conn: Mutex::new(conn),
203 registry: Mutex::new(None),
204 def_registry: Mutex::new(None),
205 token_index: Mutex::new(None),
206 wan_identity: Mutex::new(None),
207 registry_agents_base: Mutex::new(None),
208 muxbus_save_lock: Mutex::new(()),
209 global_memory_observer: Mutex::new(None),
210 })
211 }
212
213 /// Open the permanently-global identity store at `path`
214 /// (`~/.agentmux/shared/identity-store.db`) — see
215 /// `docs/specs/SPEC_IDENTITY_STORE_SPLIT_2026_08_17.md`.
216 ///
217 /// Same WAL + busy-timeout config as [`open_shared`](Self::open_shared),
218 /// but runs [`run_identity_store_schema`] — a strict subset of
219 /// [`open_shared`](Self::open_shared)'s tables (everything except
220 /// `db_accounts`) — instead of `run_shared_store_schema`. Crucially,
221 /// this store's PATH is never gated by `isolated_auth_enabled()` (see
222 /// its resolver, `registry::resolve_identity_store_path`): unlike
223 /// `open_shared`, there is no isolated/per-channel variant of this
224 /// store at all.
225 pub fn open_identity_store(path: &Path) -> Result<Self, StoreError> {
226 let conn = Connection::open(path)?;
227 conn.execute_batch(
228 "PRAGMA journal_mode=WAL;
229 PRAGMA busy_timeout=5000;
230 PRAGMA foreign_keys=ON;
231 PRAGMA synchronous=NORMAL;
232 PRAGMA cache_size=-8000;
233 PRAGMA mmap_size=268435456;
234 PRAGMA temp_store=MEMORY;",
235 )?;
236 check_schema_compat(&conn, IDENTITY_STORE_SCHEMA_VERSION, "identity-store.db")?;
237 run_identity_store_schema(&conn)?;
238 conn.execute_batch(super::work_claims::SCHEMA)?;
239 stamp_version(&conn, IDENTITY_STORE_SCHEMA_VERSION)?;
240 Ok(Self {
241 conn: Mutex::new(conn),
242 registry: Mutex::new(None),
243 def_registry: Mutex::new(None),
244 token_index: Mutex::new(None),
245 wan_identity: Mutex::new(None),
246 registry_agents_base: Mutex::new(None),
247 muxbus_save_lock: Mutex::new(()),
248 global_memory_observer: Mutex::new(None),
249 })
250 }
251
252 /// Open a sibling `objects.db` file for read-only backfill access.
253 ///
254 /// Does NOT run schema migrations or stamp a version — the source DB is
255 /// never modified (spec §3.1). WAL mode is not set either: the file is
256 /// opened with `SQLITE_OPEN_READ_ONLY | SQLITE_OPEN_NO_MUTEX` so concurrent
257 /// writers on the same DB are unaffected. Tables absent in older schemas
258 /// (e.g. `db_drone_definitions`) return empty results via the normal
259 /// `StoreError` path; callers use `.unwrap_or_default()`.
260 pub fn open_source_readonly(path: &Path) -> Result<Self, StoreError> {
261 use rusqlite::OpenFlags;
262 let conn = Connection::open_with_flags(
263 path,
264 OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
265 )?;
266 conn.busy_timeout(std::time::Duration::from_millis(500))?;
267 Ok(Self {
268 conn: Mutex::new(conn),
269 registry: Mutex::new(None),
270 def_registry: Mutex::new(None),
271 token_index: Mutex::new(None),
272 wan_identity: Mutex::new(None),
273 registry_agents_base: Mutex::new(None),
274 muxbus_save_lock: Mutex::new(()),
275 global_memory_observer: Mutex::new(None),
276 })
277 }
278
279 /// Crate-internal accessor for sibling modules that maintain their
280 /// own per-table CRUD via the `DroneStore` extension trait
281 /// pattern (see `crates/srv/src/drone/storage.rs`). Outside
282 /// callers must use the typed methods on this impl.
283 pub(crate) fn conn(&self) -> &Mutex<Connection> {
284 &self.conn
285 }
286
287 /// Run the `db_agents` consolidation backfill under the mstore's
288 /// exclusive connection lock. Idempotent — gated by a marker file in
289 /// `data_dir` (skip with `None` for tests).
290 pub fn run_agents_consolidate(
291 &self,
292 data_dir: Option<&Path>,
293 ) -> Result<super::agents_consolidate::ConsolidateStats, StoreError> {
294 let mut conn = self.conn.lock().unwrap();
295 super::agents_consolidate::run_consolidate_migration(&mut conn, data_dir)
296 }
297
298 /// Run `PRAGMA wal_checkpoint(TRUNCATE)` to fold the WAL back into the
299 /// main DB file. Called periodically and on clean shutdown. The 5s
300 /// `busy_timeout` set at open handles transient reader contention — if it
301 /// can't fully truncate it returns partial progress (safe; picked up on
302 /// the next pass).
303 pub fn checkpoint(&self) -> Result<(), StoreError> {
304 self.conn()
305 .lock()
306 .unwrap()
307 .execute_batch("PRAGMA wal_checkpoint(TRUNCATE);")?;
308 Ok(())
309 }
310
311 /// True when `db_agent_definitions`/`db_agent_instances` have rows but
312 /// `db_agents` doesn't — i.e. the consolidation backfill's marker/stamp
313 /// can't be trusted as proof it actually ran. See
314 /// `agents_consolidate::consolidate_looks_incomplete`.
315 pub fn agents_consolidate_looks_incomplete(&self) -> Result<bool, StoreError> {
316 let conn = self.conn.lock().unwrap();
317 super::agents_consolidate::consolidate_looks_incomplete(&conn)
318 }
319
320 fn configure_and_migrate(conn: Connection) -> Result<Self, StoreError> {
321 conn.execute_batch(
322 // `foreign_keys=ON` is per-connection and defaults to OFF in
323 // SQLite. The v6 schema (`db_agent_identity_links`,
324 // `db_agent_instances`) relies on `ON DELETE CASCADE` to clean
325 // up junction rows and instances when a parent agent or
326 // identity is removed. Without this pragma on the production
327 // connection, cascades silently no-op; migration tests set it
328 // explicitly, which would have masked the gap.
329 "PRAGMA journal_mode=WAL;
330 PRAGMA busy_timeout=5000;
331 PRAGMA foreign_keys=ON;
332 PRAGMA synchronous=NORMAL;
333 PRAGMA cache_size=-8000;
334 PRAGMA mmap_size=268435456;
335 PRAGMA temp_store=MEMORY;",
336 )?;
337 // Safety lock BEFORE any migration side effects — the legacy-
338 // table rename + seed-insert steps in `run_object_schema` are
339 // mutating, so we must refuse to open a newer-schema DB before
340 // we touch it. (codex P1 on #1029 — fixes the original PR's
341 // check-after-migrate order that let a downgraded binary
342 // partially mutate a newer DB before the error fired.)
343 check_schema_compat(&conn, OBJECT_SCHEMA_VERSION, "objects.db")?;
344 run_object_schema(&conn)?;
345 stamp_version(&conn, OBJECT_SCHEMA_VERSION)?;
346 Ok(Self {
347 conn: Mutex::new(conn),
348 registry: Mutex::new(None),
349 def_registry: Mutex::new(None),
350 token_index: Mutex::new(None),
351 wan_identity: Mutex::new(None),
352 registry_agents_base: Mutex::new(None),
353 muxbus_save_lock: Mutex::new(()),
354 global_memory_observer: Mutex::new(None),
355 })
356 }
357
358 /// Attach a shared cross-version agent registry. Called once on
359 /// srv startup after `Store::open` and before the store is
360 /// wrapped in `Arc`. Mutations to `db_agent_instances` will then
361 /// parallel-write to the registry; the SQLite table remains the
362 /// authoritative read path for PR A.
363 pub fn set_registry(&self, registry: Arc<Registry>) {
364 *self.registry.lock().unwrap_or_else(|e| e.into_inner()) = Some(registry);
365 }
366
367 /// Attach the Global Memory record's observer (see the field).
368 pub fn set_global_memory_observer(&self, observer: GlobalMemoryObserver) {
369 *self.global_memory_observer.lock().unwrap_or_else(|e| e.into_inner()) = Some(observer);
370 }
371
372 /// Tell the observer, if any, about a committed change. Callers invoke it
373 /// once their connection lock is released.
374 pub(super) fn notify_global_memory(&self, change: GlobalMemoryChange) {
375 let observer = self.global_memory_observer.lock().unwrap_or_else(|e| e.into_inner()).clone();
376 if let Some(o) = observer {
377 o(change);
378 }
379 }
380
381 pub(super) fn registry(&self) -> Option<Arc<Registry>> {
382 self.registry
383 .lock()
384 .unwrap_or_else(|e| e.into_inner())
385 .clone()
386 }
387
388 /// Set the channel agents dir that instance `working_directory` values
389 /// are stored relative to (see the `registry_agents_base` field). Wired
390 /// from `AGENTMUX_AGENTS_DIR` in `main.rs` in P0.3b (atomically with the
391 /// registry re-root); until then it is exercised only by tests.
392 pub fn set_registry_agents_base(&self, base: PathBuf) {
393 *self
394 .registry_agents_base
395 .lock()
396 .unwrap_or_else(|e| e.into_inner()) = Some(base);
397 }
398
399 /// The base dir for instance `working_directory` relative paths.
400 ///
401 /// Returns the explicitly-set channel agents dir
402 /// ([`set_registry_agents_base`]) when present; otherwise falls back to
403 /// the registry's parent (`agents_root()`), which equals the channel
404 /// agents dir in the pre-re-root layout. Used symmetrically by the write
405 /// mirror and the read handler so the two never disagree on the anchor.
406 pub fn registry_agents_base(&self) -> Option<PathBuf> {
407 if let Some(base) = self
408 .registry_agents_base
409 .lock()
410 .unwrap_or_else(|e| e.into_inner())
411 .clone()
412 {
413 return Some(base);
414 }
415 self.registry()
416 .and_then(|r| r.agents_root().map(|p| p.to_path_buf()))
417 }
418
419 /// Public accessor for the cross-version named-agent registry.
420 /// Returns `None` when the registry couldn't be resolved at
421 /// startup (CI / unusual envs); callers must handle the absent
422 /// case by falling back to SQLite.
423 pub fn shared_agent_registry(&self) -> Option<Arc<Registry>> {
424 self.registry()
425 }
426
427 /// Attach the GLOBAL (cross-channel) agent-definition store. Called
428 /// once on srv startup after `Store::open`, before the store is
429 /// wrapped in `Arc`. Definition mutations then mirror to it.
430 pub fn set_wan_identity(&self, wan_identity: Arc<super::wan_identity::WanIdentityStore>) {
431 *self.wan_identity.lock().unwrap_or_else(|e| e.into_inner()) = Some(wan_identity);
432 }
433
434 /// The channel-wide WAN identity store, if attached. `None` means WAN
435 /// signing is off for this process — never fall back to another store.
436 pub fn wan_identity(&self) -> Option<Arc<super::wan_identity::WanIdentityStore>> {
437 self.wan_identity.lock().unwrap_or_else(|e| e.into_inner()).clone()
438 }
439
440 pub fn set_def_registry(&self, def_registry: Arc<DefinitionStore>) {
441 *self.def_registry.lock().unwrap_or_else(|e| e.into_inner()) = Some(def_registry);
442 }
443
444 /// Public accessor for the global agent-definition store. `None` when
445 /// it couldn't be resolved at startup (CI / unusual envs / in-memory
446 /// tests); callers fall back to SQLite.
447 pub fn shared_def_registry(&self) -> Option<Arc<DefinitionStore>> {
448 self.def_registry
449 .lock()
450 .unwrap_or_else(|e| e.into_inner())
451 .clone()
452 }
453
454 /// Table name for a StoreObj type: `db_<otype>`.
455 fn table_name<T: StoreObj>() -> String {
456 format!("db_{}", T::get_otype())
457 }
458
459 /// Get a single object by OID. Returns `None` if not found.
460 pub fn get<T: StoreObj>(&self, oid: &str) -> Result<Option<T>, StoreError> {
461 let conn = self.conn.lock().unwrap();
462 let table = Self::table_name::<T>();
463 let mut stmt =
464 conn.prepare(&format!("SELECT version, data FROM {table} WHERE oid = ?1"))?;
465
466 let result = stmt.query_row(params![oid], |row| {
467 let version: i64 = row.get(0)?;
468 let data: Vec<u8> = row.get(1)?;
469 Ok((version, data))
470 });
471
472 match result {
473 Ok((version, data)) => {
474 let mut obj: T = mux_obj_from_json(&data)?;
475 obj.set_version(version);
476 Ok(Some(obj))
477 }
478 Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
479 Err(e) => Err(StoreError::Sqlite(e)),
480 }
481 }
482
483 /// Get a single object, returning `StoreError::NotFound` if missing.
484 pub fn must_get<T: StoreObj>(&self, oid: &str) -> Result<T, StoreError> {
485 self.get::<T>(oid)?.ok_or(StoreError::NotFound)
486 }
487
488 /// Get a single object as raw JSON Value by otype and OID.
489 /// Used by GetObject/GetObjects to return data without strict struct deserialization.
490 pub fn get_raw(&self, otype: &str, oid: &str) -> Result<Option<serde_json::Value>, StoreError> {
491 let conn = self.conn.lock().unwrap();
492 let table = format!("db_{}", otype);
493 let mut stmt =
494 conn.prepare(&format!("SELECT version, data FROM {table} WHERE oid = ?1"))?;
495
496 let result = stmt.query_row(params![oid], |row| {
497 let version: i64 = row.get(0)?;
498 let data: Vec<u8> = row.get(1)?;
499 Ok((version, data))
500 });
501
502 match result {
503 Ok((version, data)) => {
504 let mut val: serde_json::Value = serde_json::from_slice(&data)
505 .map_err(|e| StoreError::Json(e))?;
506 if let Some(obj) = val.as_object_mut() {
507 obj.insert("version".to_string(), serde_json::json!(version));
508 obj.insert("otype".to_string(), serde_json::json!(otype));
509 }
510 Ok(Some(val))
511 }
512 Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
513 Err(e) => Err(StoreError::Sqlite(e)),
514 }
515 }
516
517 /// Check if an object exists (by otype and OID).
518 #[allow(dead_code)]
519 pub fn exists_raw(&self, otype: &str, oid: &str) -> Result<bool, StoreError> {
520 let conn = self.conn.lock().unwrap();
521 let table = format!("db_{}", otype);
522 let count: i64 = conn.query_row(
523 &format!("SELECT COUNT(*) FROM {table} WHERE oid = ?1"),
524 params![oid],
525 |row| row.get(0),
526 )?;
527 Ok(count > 0)
528 }
529
530 /// Insert a new object. Sets version to 1.
531 pub fn insert<T: StoreObj>(&self, obj: &mut T) -> Result<(), StoreError> {
532 let oid = obj.get_oid().to_string();
533 if oid.is_empty() {
534 return Err(StoreError::EmptyOID);
535 }
536
537 obj.set_version(1);
538 let data = mux_obj_to_json(obj)?;
539
540 let conn = self.conn.lock().unwrap();
541 let table = Self::table_name::<T>();
542 conn.execute(
543 &format!("INSERT INTO {table} (oid, version, data) VALUES (?1, 1, ?2)"),
544 params![oid, data],
545 )?;
546
547 Ok(())
548 }
549
550 /// Update an existing object. Increments version atomically.
551 /// Returns the new version number.
552 pub fn update<T: StoreObj>(&self, obj: &mut T) -> Result<i64, StoreError> {
553 let oid = obj.get_oid().to_string();
554 if oid.is_empty() {
555 return Err(StoreError::EmptyOID);
556 }
557
558 let data = mux_obj_to_json(obj)?;
559
560 let conn = self.conn.lock().unwrap();
561 let table = Self::table_name::<T>();
562
563 // Optimistic locking: increment version and return new value.
564 // Matches Go: `UPDATE ... SET version = version+1 ... RETURNING version`
565 let new_version: i64 = conn.query_row(
566 &format!(
567 "UPDATE {table} SET data = ?1, version = version + 1 WHERE oid = ?2 RETURNING version"
568 ),
569 params![data, oid],
570 |row| row.get(0),
571 )?;
572
573 obj.set_version(new_version);
574 Ok(new_version)
575 }
576
577 /// Update an object using raw JSON (bypasses struct deserialization).
578 /// Used by UpdateObject where the frontend sends the full replacement object.
579 /// This matches Go's generic map-based UpdateObject behavior.
580 pub fn update_raw(&self, otype: &str, oid: &str, value: &serde_json::Value) -> Result<i64, StoreError> {
581 if oid.is_empty() {
582 return Err(StoreError::EmptyOID);
583 }
584 let data = serde_json::to_vec(value)?;
585 let conn = self.conn.lock().unwrap();
586 let table = format!("db_{}", otype);
587 let new_version: i64 = conn.query_row(
588 &format!(
589 "UPDATE {table} SET data = ?1, version = version + 1 WHERE oid = ?2 RETURNING version"
590 ),
591 params![data, oid],
592 |row| row.get(0),
593 )?;
594 Ok(new_version)
595 }
596
597 /// Delete an object by OID.
598 #[allow(dead_code)]
599 pub fn delete<T: StoreObj>(&self, oid: &str) -> Result<(), StoreError> {
600 let conn = self.conn.lock().unwrap();
601 let table = Self::table_name::<T>();
602 conn.execute(
603 &format!("DELETE FROM {table} WHERE oid = ?1"),
604 params![oid],
605 )?;
606 Ok(())
607 }
608
609 /// Delete by otype string and OID (for dynamic dispatch).
610 /// Validates `otype` against `VALID_OTYPES` to prevent SQL injection.
611 #[allow(dead_code)]
612 pub fn delete_by_otype(&self, otype: &str, oid: &str) -> Result<(), StoreError> {
613 if !crate::backend::obj::VALID_OTYPES.contains(&otype) {
614 return Err(StoreError::Other(format!("unknown otype: {otype:?}")));
615 }
616 let conn = self.conn.lock().unwrap();
617 let table = format!("db_{otype}");
618 conn.execute(
619 &format!("DELETE FROM {table} WHERE oid = ?1"),
620 params![oid],
621 )?;
622 Ok(())
623 }
624
625 /// Get all objects of a given type.
626 pub fn get_all<T: StoreObj>(&self) -> Result<Vec<T>, StoreError> {
627 let conn = self.conn.lock().unwrap();
628 let table = Self::table_name::<T>();
629 let mut stmt = conn.prepare(&format!("SELECT oid, version, data FROM {table}"))?;
630 let rows = stmt.query_map([], |row| {
631 let version: i64 = row.get(1)?;
632 let data: Vec<u8> = row.get(2)?;
633 Ok((version, data))
634 })?;
635
636 let mut result = Vec::new();
637 for row in rows {
638 let (version, data) = row?;
639 let mut obj: T = mux_obj_from_json(&data)?;
640 obj.set_version(version);
641 result.push(obj);
642 }
643 Ok(result)
644 }
645
646 /// Count objects of a given type.
647 #[allow(dead_code)]
648 pub fn count<T: StoreObj>(&self) -> Result<i64, StoreError> {
649 let conn = self.conn.lock().unwrap();
650 let table = Self::table_name::<T>();
651 let count: i64 =
652 conn.query_row(&format!("SELECT count(*) FROM {table}"), [], |row| {
653 row.get(0)
654 })?;
655 Ok(count)
656 }
657
658 /// Execute multiple operations in a single SQLite transaction.
659 /// Acquires the Mutex once, wraps all operations in BEGIN/COMMIT.
660 /// On error, rolls back and returns the error.
661 ///
662 /// This is the key performance primitive — reduces N lock acquisitions
663 /// and N fsyncs to 1 each.
664 pub fn with_tx<F, R>(&self, f: F) -> Result<R, StoreError>
665 where
666 F: FnOnce(&StoreTx) -> Result<R, StoreError>,
667 {
668 let conn = self.conn.lock().unwrap();
669 conn.execute_batch("BEGIN")?;
670 let tx = StoreTx { conn: &conn };
671 match f(&tx) {
672 Ok(result) => {
673 conn.execute_batch("COMMIT")?;
674 Ok(result)
675 }
676 Err(e) => {
677 let _ = conn.execute_batch("ROLLBACK");
678 Err(e)
679 }
680 }
681 }
682
683 // ── Migration state (db_migrations — shared store only) ──────────────
684
685 pub fn migration_is_applied(&self, id: &str) -> bool {
686 let conn = self.conn.lock().unwrap();
687 conn.query_row(
688 "SELECT 1 FROM db_migrations WHERE id = ?1",
689 [id],
690 |_| Ok(true),
691 )
692 .unwrap_or(false)
693 }
694
695 pub fn migration_mark_applied(
696 &self,
697 id: &str,
698 scope: &str,
699 duration_ms: u64,
700 ) -> Result<(), StoreError> {
701 let now = chrono::Utc::now().to_rfc3339();
702 let conn = self.conn.lock().unwrap();
703 conn.execute(
704 "INSERT OR REPLACE INTO db_migrations (id, applied_at, duration_ms, scope)
705 VALUES (?1, ?2, ?3, ?4)",
706 params![id, now, duration_ms as i64, scope],
707 )?;
708 Ok(())
709 }
710
711 pub fn migrations_list_applied(&self) -> Result<Vec<String>, StoreError> {
712 let conn = self.conn.lock().unwrap();
713 let mut stmt =
714 conn.prepare("SELECT id FROM db_migrations ORDER BY id")?;
715 let ids = stmt
716 .query_map([], |row| row.get(0))?
717 .collect::<Result<Vec<String>, _>>()?;
718 Ok(ids)
719 }
720}
721
722/// A borrowed connection handle for use inside [`Store::with_tx`].
723/// Provides the same CRUD methods as `Store` but operates on the
724/// already-locked connection without additional Mutex acquisition.
725pub struct StoreTx<'a> {
726 conn: &'a Connection,
727}
728
729impl<'a> StoreTx<'a> {
730 fn table_name<T: StoreObj>() -> String {
731 format!("db_{}", T::get_otype())
732 }
733
734 pub fn get<T: StoreObj>(&self, oid: &str) -> Result<Option<T>, StoreError> {
735 let table = Self::table_name::<T>();
736 let mut stmt =
737 self.conn.prepare(&format!("SELECT version, data FROM {table} WHERE oid = ?1"))?;
738
739 let result = stmt.query_row(params![oid], |row| {
740 let version: i64 = row.get(0)?;
741 let data: Vec<u8> = row.get(1)?;
742 Ok((version, data))
743 });
744
745 match result {
746 Ok((version, data)) => {
747 let mut obj: T = mux_obj_from_json(&data)?;
748 obj.set_version(version);
749 Ok(Some(obj))
750 }
751 Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
752 Err(e) => Err(StoreError::Sqlite(e)),
753 }
754 }
755
756 pub fn must_get<T: StoreObj>(&self, oid: &str) -> Result<T, StoreError> {
757 self.get::<T>(oid)?.ok_or(StoreError::NotFound)
758 }
759
760 pub fn insert<T: StoreObj>(&self, obj: &mut T) -> Result<(), StoreError> {
761 let oid = obj.get_oid().to_string();
762 if oid.is_empty() {
763 return Err(StoreError::EmptyOID);
764 }
765
766 obj.set_version(1);
767 let data = mux_obj_to_json(obj)?;
768
769 let table = Self::table_name::<T>();
770 self.conn.execute(
771 &format!("INSERT INTO {table} (oid, version, data) VALUES (?1, 1, ?2)"),
772 params![oid, data],
773 )?;
774
775 Ok(())
776 }
777
778 pub fn update<T: StoreObj>(&self, obj: &mut T) -> Result<i64, StoreError> {
779 let oid = obj.get_oid().to_string();
780 if oid.is_empty() {
781 return Err(StoreError::EmptyOID);
782 }
783
784 let data = mux_obj_to_json(obj)?;
785
786 let table = Self::table_name::<T>();
787 let new_version: i64 = self.conn.query_row(
788 &format!(
789 "UPDATE {table} SET data = ?1, version = version + 1 WHERE oid = ?2 RETURNING version"
790 ),
791 params![data, oid],
792 |row| row.get(0),
793 )?;
794
795 obj.set_version(new_version);
796 Ok(new_version)
797 }
798
799 pub fn get_all<T: StoreObj>(&self) -> Result<Vec<T>, StoreError> {
800 let table = Self::table_name::<T>();
801 let mut stmt = self.conn.prepare(&format!("SELECT oid, version, data FROM {table}"))?;
802 let rows = stmt.query_map([], |row| {
803 let version: i64 = row.get(1)?;
804 let data: Vec<u8> = row.get(2)?;
805 Ok((version, data))
806 })?;
807
808 let mut result = Vec::new();
809 for row in rows {
810 let (version, data) = row?;
811 let mut obj: T = mux_obj_from_json(&data)?;
812 obj.set_version(version);
813 result.push(obj);
814 }
815 Ok(result)
816 }
817
818 #[allow(dead_code)]
819 pub fn delete<T: StoreObj>(&self, oid: &str) -> Result<(), StoreError> {
820 let table = Self::table_name::<T>();
821 self.conn.execute(
822 &format!("DELETE FROM {table} WHERE oid = ?1"),
823 params![oid],
824 )?;
825 Ok(())
826 }
827}
828
829// Re-exports so existing `storage::store::*` imports keep working.
830pub use super::agents::{derive_slug, AgentDefinition, AgentInstance, InstanceStatus};
831pub use super::bundles::Bundle;
832pub use super::content::AgentContent;
833pub use super::skills::AgentSkill;
834
835// Identity system types.
836pub use super::identities::{AgentIdentityLink, IdentityAccount, SecretRef};
837
838
839
840
841// ====================================================================
842// Tests
843// ====================================================================
844
845#[cfg(test)]
846mod tests;