use std::collections::HashMap;
use std::fmt::Write as _;
use std::path::Path;
use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params};
use crate::model::{ListEventsOptions, ListObjectsOptions};
use crate::{
AggregateFunction, AggregateGroupResult, AggregateOptions, AggregateResult, CollectionSchema,
EventLog, FieldSchema, MemoryEvent, MemoryObject, ObjectKey, ObjectStore, QueueClaimOptions,
QueueJob, QueueJobStatus, QueueNackOptions, QueueStore, SchemaOptions, ThingdError,
ThingdResult, TimeSeriesBucket, TimeSeriesOptions, TimeSeriesResult, u64_to_i64,
unix_timestamp_millis,
};
pub const SQLITE_SCHEMA_VERSION: u32 = 6;
pub struct SqliteThingStore {
connection: Connection,
db_path: Option<String>,
event_idempotency_keys: HashMap<(String, String), u64>,
}
impl SqliteThingStore {
pub fn open(path: impl AsRef<Path>) -> ThingdResult<Self> {
let path_str = path.as_ref().to_str().map(String::from);
let connection = Connection::open(path).map_err(ThingdError::from)?;
connection
.busy_timeout(std::time::Duration::from_secs(5))
.map_err(ThingdError::from)?;
let store = Self {
connection,
db_path: path_str,
event_idempotency_keys: HashMap::new(),
};
store.initialize()?;
Ok(store)
}
pub fn open_in_memory() -> ThingdResult<Self> {
let connection = Connection::open_in_memory().map_err(ThingdError::from)?;
connection
.busy_timeout(std::time::Duration::from_secs(5))
.map_err(ThingdError::from)?;
let store = Self {
connection,
db_path: None,
event_idempotency_keys: HashMap::new(),
};
store.initialize()?;
Ok(store)
}
fn initialize(&self) -> ThingdResult<()> {
let current_mode: String = self
.connection
.query_row("PRAGMA journal_mode;", [], |row| row.get(0))
.unwrap_or_else(|_| "delete".to_string());
if current_mode.to_lowercase() != "wal" {
self.connection
.query_row("PRAGMA journal_mode = WAL;", [], |_| Ok(()))
.map_err(|e| {
eprintln!("warning: failed to enable WAL journal mode: {e}");
})
.ok();
}
self.connection
.execute_batch(
r"
PRAGMA synchronous = NORMAL;
PRAGMA foreign_keys = ON;
PRAGMA busy_timeout = 5000;
",
)
.map_err(ThingdError::from)?;
self.connection
.execute_batch(
r"
CREATE TABLE IF NOT EXISTS thingd_schema_migrations (
version INTEGER PRIMARY KEY,
name TEXT NOT NULL,
applied_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);
",
)
.map_err(ThingdError::from)?;
let current_version = self.schema_version()?;
if current_version > 0 && current_version < SQLITE_SCHEMA_VERSION {
#[allow(clippy::collapsible_if)]
if let Some(ref path) = self.db_path {
let backup_path = format!("{path}.pre-v{current_version}");
let escaped = backup_path.replace('\'', "''");
let _ = self
.connection
.execute_batch(&format!("VACUUM INTO '{escaped}'"));
}
}
if current_version < 1 {
self.apply_schema_v1()?;
}
let current_version = self.schema_version()?;
if current_version < 2 {
self.apply_schema_v2()?;
}
let current_version = self.schema_version()?;
if current_version < 3 {
self.apply_schema_v3()?;
}
let current_version = self.schema_version()?;
if current_version < 4 {
self.apply_schema_v4()?;
}
let current_version = self.schema_version()?;
if current_version < 5 {
self.apply_schema_v5()?;
}
let current_version = self.schema_version()?;
if current_version < 6 {
self.apply_schema_v6()?;
}
if current_version > SQLITE_SCHEMA_VERSION {
eprintln!(
"warning: database schema version {current_version} is newer than supported version {SQLITE_SCHEMA_VERSION}. Proceeding in forward-compatibility mode."
);
}
let ok: String = self
.connection
.query_row("PRAGMA quick_check", [], |row| row.get(0))
.map_err(ThingdError::from)?;
if ok != "ok" {
return Err(ThingdError::Storage(format!(
"database integrity check failed: {ok}"
)));
}
Ok(())
}
pub fn schema_version(&self) -> ThingdResult<u32> {
let version = self
.connection
.query_row(
"SELECT COALESCE(MAX(version), 0) FROM thingd_schema_migrations",
[],
|row| row.get::<_, i64>(0),
)
.map_err(ThingdError::from)?;
u32::try_from(version).map_err(|error| ThingdError::Storage(error.to_string()))
}
pub fn wal_checkpoint(&self) -> ThingdResult<(i32, i32)> {
self.connection
.query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |row| {
Ok((row.get::<_, i32>(0)?, row.get::<_, i32>(1)?))
})
.map_err(ThingdError::from)
}
pub fn optimize_search_index(&self) -> ThingdResult<()> {
self.connection
.execute_batch("INSERT INTO search_index(search_index) VALUES('optimize')")
.map_err(ThingdError::from)
}
pub fn close(&self) -> ThingdResult<()> {
self.wal_checkpoint()?;
Ok(())
}
pub fn backup_to(&self, path: &str) -> ThingdResult<()> {
if Path::new(path)
.components()
.filter_map(|c| match c {
std::path::Component::Normal(s) => Some(s.to_str().unwrap_or("")),
std::path::Component::ParentDir => Some(".."),
_ => None,
})
.any(|x| x == "..")
{
return Err(ThingdError::InvalidInput(
"Backup path must not contain '..' (path traversal)".to_string(),
));
}
let escaped = path.replace('\'', "''");
self.connection
.execute_batch(&format!("VACUUM INTO '{escaped}'"))
.map_err(ThingdError::from)
}
fn apply_schema_v1(&self) -> ThingdResult<()> {
self.connection
.execute_batch(
r"
BEGIN;
CREATE TABLE IF NOT EXISTS objects (
collection TEXT NOT NULL,
id TEXT NOT NULL,
body TEXT NOT NULL,
version INTEGER NOT NULL,
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
PRIMARY KEY (collection, id)
);
CREATE TABLE IF NOT EXISTS events (
sequence INTEGER PRIMARY KEY AUTOINCREMENT,
stream TEXT NOT NULL,
event_type TEXT NOT NULL,
body TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);
CREATE INDEX IF NOT EXISTS idx_events_stream_sequence
ON events (stream, sequence);
CREATE TABLE IF NOT EXISTS queue_jobs (
queue TEXT NOT NULL,
id TEXT NOT NULL,
body TEXT NOT NULL,
attempts INTEGER NOT NULL,
max_attempts INTEGER NOT NULL,
status TEXT NOT NULL,
available_at_ms INTEGER NOT NULL,
leased_at_ms INTEGER,
lease_expires_at_ms INTEGER,
completed_at_ms INTEGER,
dead_at_ms INTEGER,
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
PRIMARY KEY (queue, id)
);
CREATE INDEX IF NOT EXISTS idx_queue_jobs_queue_status_created
ON queue_jobs (queue, status, created_at);
CREATE INDEX IF NOT EXISTS idx_queue_jobs_status
ON queue_jobs (status);
INSERT OR IGNORE INTO thingd_schema_migrations (version, name, applied_at)
VALUES (1, 'initial_objects_events_queues', strftime('%Y-%m-%dT%H:%M:%fZ', 'now'));
COMMIT;
",
)
.map_err(ThingdError::from)?;
Ok(())
}
fn apply_schema_v2(&self) -> ThingdResult<()> {
let tx = self
.connection
.unchecked_transaction()
.map_err(ThingdError::from)?;
tx.execute_batch(
"CREATE VIRTUAL TABLE IF NOT EXISTS search_index USING fts5(
collection UNINDEXED,
id UNINDEXED,
kind UNINDEXED,
text,
tokenize='porter unicode61'
);",
)
.map_err(ThingdError::from)?;
Self::reindex_all_into(&tx)?;
tx.execute(
"INSERT OR IGNORE INTO thingd_schema_migrations (version, name, applied_at)
VALUES (2, 'fts5_search_index', strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))",
[],
)
.map_err(ThingdError::from)?;
tx.commit().map_err(ThingdError::from)?;
Ok(())
}
fn apply_schema_v3(&self) -> ThingdResult<()> {
self.connection
.execute(
"ALTER TABLE queue_jobs ADD COLUMN last_error TEXT NOT NULL DEFAULT ''",
[],
)
.map_err(ThingdError::from)?;
self.connection
.execute(
"INSERT OR IGNORE INTO thingd_schema_migrations (version, name, applied_at)
VALUES (3, 'queue_jobs_last_error', strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))",
[],
)
.map_err(ThingdError::from)?;
Ok(())
}
fn apply_schema_v4(&self) -> ThingdResult<()> {
self.connection
.execute_batch(
r"
CREATE TABLE IF NOT EXISTS links (
id TEXT PRIMARY KEY,
from_ref TEXT NOT NULL,
type TEXT NOT NULL,
to_ref TEXT NOT NULL,
weight REAL,
metadata_json TEXT NOT NULL DEFAULT '{}',
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);
CREATE INDEX IF NOT EXISTS idx_links_from_ref ON links (from_ref);
CREATE INDEX IF NOT EXISTS idx_links_to_ref ON links (to_ref);
CREATE INDEX IF NOT EXISTS idx_links_type ON links (type);
",
)
.map_err(ThingdError::from)?;
self.connection
.execute(
"INSERT OR IGNORE INTO thingd_schema_migrations (version, name, applied_at)
VALUES (4, 'graph_links', strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))",
[],
)
.map_err(ThingdError::from)?;
Ok(())
}
fn apply_schema_v5(&self) -> ThingdResult<()> {
self.connection
.execute_batch(
r"
CREATE INDEX IF NOT EXISTS idx_objects_collection ON objects (collection);
CREATE INDEX IF NOT EXISTS idx_objects_created_at ON objects (created_at);
CREATE INDEX IF NOT EXISTS idx_events_stream ON events (stream);
",
)
.map_err(ThingdError::from)?;
self.connection
.execute(
"INSERT OR IGNORE INTO thingd_schema_migrations (version, name, applied_at)
VALUES (5, 'performance_indexes', strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))",
[],
)
.map_err(ThingdError::from)?;
Ok(())
}
fn apply_schema_v6(&self) -> ThingdResult<()> {
self.connection
.execute_batch(
r"
CREATE INDEX IF NOT EXISTS idx_events_stream_created_at
ON events (stream, created_at);
",
)
.map_err(ThingdError::from)?;
self.connection
.execute(
"INSERT OR IGNORE INTO thingd_schema_migrations (version, name, applied_at)
VALUES (6, 'events_stream_created_at_index', strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))",
[],
)
.map_err(ThingdError::from)?;
Ok(())
}
fn reindex_all_into(tx: &rusqlite::Transaction<'_>) -> ThingdResult<()> {
let mut stmt_objects = tx
.prepare("SELECT collection, id, body FROM objects")
.map_err(ThingdError::from)?;
let rows_objects = stmt_objects
.query_map([], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
))
})
.map_err(ThingdError::from)?;
let mut stmt_events = tx
.prepare("SELECT stream, sequence, body FROM events")
.map_err(ThingdError::from)?;
let rows_events = stmt_events
.query_map([], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, i64>(1)?.to_string(),
row.get::<_, String>(2)?,
))
})
.map_err(ThingdError::from)?;
tx.execute("DELETE FROM search_index", [])
.map_err(ThingdError::from)?;
for row in rows_objects {
let (collection, id, body) = row.map_err(ThingdError::from)?;
let text = extract_text_from_json(&body);
tx.execute(
"INSERT INTO search_index (collection, id, kind, text) VALUES (?1, ?2, 'object', ?3)",
params![collection, id, text],
)
.map_err(ThingdError::from)?;
}
for row in rows_events {
let (stream, sequence, body) = row.map_err(ThingdError::from)?;
let text = extract_text_from_json(&body);
tx.execute(
"INSERT INTO search_index (collection, id, kind, text) VALUES (?1, ?2, 'event', ?3)",
params![stream, sequence, text],
)
.map_err(ThingdError::from)?;
}
Ok(())
}
}
impl ObjectStore for SqliteThingStore {
fn put_object(&mut self, mut object: MemoryObject) -> ThingdResult<MemoryObject> {
let transaction = self.connection.transaction().map_err(ThingdError::from)?;
let row = transaction
.query_row(
r"
INSERT INTO objects (collection, id, body, version, created_at, updated_at)
VALUES (?1, ?2, ?3, 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'), strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
ON CONFLICT(collection, id) DO UPDATE SET
body = excluded.body,
version = version + 1,
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
RETURNING version, created_at, updated_at
",
params![&object.key.collection, &object.key.id, &object.body],
|row| {
Ok((
row.get::<_, i64>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
))
},
)
.map_err(ThingdError::from)?;
object.version = u64::try_from(row.0).map_err(|e| ThingdError::Storage(e.to_string()))?;
object.created_at = row.1;
object.updated_at = row.2;
let text = extract_text_from_json(&object.body);
transaction
.execute(
"DELETE FROM search_index WHERE collection = ?1 AND id = ?2 AND kind = 'object'",
params![&object.key.collection, &object.key.id],
)
.map_err(ThingdError::from)?;
transaction
.execute(
"INSERT INTO search_index (collection, id, kind, text) VALUES (?1, ?2, 'object', ?3)",
params![&object.key.collection, &object.key.id, text],
)
.map_err(ThingdError::from)?;
transaction.commit().map_err(ThingdError::from)?;
Ok(object)
}
fn put_objects_batch(&mut self, objects: Vec<MemoryObject>) -> ThingdResult<Vec<MemoryObject>> {
if objects.is_empty() {
return Ok(Vec::new());
}
let transaction = self.connection.transaction().map_err(ThingdError::from)?;
let mut fts_updates: Vec<(String, String, String)> = Vec::with_capacity(objects.len());
let mut results = Vec::with_capacity(objects.len());
let mut param_values: Vec<String> = Vec::new();
let mut value_sql = String::new();
for (i, object) in objects.into_iter().enumerate() {
if i > 0 {
value_sql.push_str(", ");
}
let ci = i * 3 + 1;
let ii = i * 3 + 2;
let bi = i * 3 + 3;
let _ = std::fmt::Write::write_fmt(
&mut value_sql,
format_args!(
"(?{ci}, ?{ii}, ?{bi}, 1, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'), strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))"
),
);
let text = extract_text_from_json(&object.body);
fts_updates.push((object.key.collection.clone(), object.key.id.clone(), text));
param_values.push(object.key.collection.clone());
param_values.push(object.key.id.clone());
param_values.push(object.body.clone());
results.push(object);
}
let sql = format!(
"INSERT INTO objects (collection, id, body, version, created_at, updated_at) VALUES {value_sql} \
ON CONFLICT(collection, id) DO UPDATE SET \
body = excluded.body, version = version + 1, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') \
RETURNING version, created_at, updated_at"
);
let param_slices: Vec<&dyn rusqlite::types::ToSql> = param_values
.iter()
.map(|s| s as &dyn rusqlite::types::ToSql)
.collect();
let mut statement = transaction.prepare(&sql).map_err(ThingdError::from)?;
let rows = statement
.query_map(param_slices.as_slice(), |row| {
Ok((
row.get::<_, i64>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
))
})
.map_err(ThingdError::from)?;
for (result, row) in results.iter_mut().zip(rows) {
let (version, created_at, updated_at) = row.map_err(ThingdError::from)?;
result.version =
u64::try_from(version).map_err(|e| ThingdError::Storage(e.to_string()))?;
result.created_at = created_at;
result.updated_at = updated_at;
}
drop(statement);
for (collection, id, text) in &fts_updates {
transaction
.execute(
"DELETE FROM search_index WHERE collection = ?1 AND id = ?2 AND kind = 'object'",
params![collection, id],
)
.map_err(ThingdError::from)?;
transaction
.execute(
"INSERT INTO search_index (collection, id, kind, text) VALUES (?1, ?2, 'object', ?3)",
params![collection, id, text],
)
.map_err(ThingdError::from)?;
}
transaction.commit().map_err(ThingdError::from)?;
Ok(results)
}
fn put_object_with_options(
&mut self,
mut object: MemoryObject,
options: crate::PutObjectOptions,
) -> ThingdResult<MemoryObject> {
let transaction = self.connection.transaction().map_err(ThingdError::from)?;
let current_version = transaction
.query_row(
"SELECT version FROM objects WHERE collection = ?1 AND id = ?2",
params![&object.key.collection, &object.key.id],
|row| row.get::<_, i64>(0),
)
.optional()
.map_err(ThingdError::from)?;
if let Some(expected) = options.expected_version {
match current_version {
Some(actual) if u64::try_from(actual).unwrap_or(0) != expected => {
return Err(ThingdError::Conflict(format!(
"Version mismatch for {}/{}: expected {expected}, got {actual}",
object.key.collection, object.key.id,
)));
},
None => {
return Err(ThingdError::Conflict(format!(
"Version mismatch for {}/{}: expected {expected}, object does not exist",
object.key.collection, object.key.id,
)));
},
_ => {},
}
}
let version = current_version.map_or(Ok::<u64, ThingdError>(1), |existing| {
u64::try_from(existing)
.map(|existing| existing + 1)
.map_err(|error| ThingdError::Storage(error.to_string()))
})?;
object.version = version;
let stored_version = i64::try_from(object.version)
.map_err(|error| ThingdError::Storage(error.to_string()))?;
let timestamps = transaction
.query_row(
r"
INSERT INTO objects (collection, id, body, version, created_at, updated_at)
VALUES (?1, ?2, ?3, ?4, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'), strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
ON CONFLICT(collection, id) DO UPDATE SET
body = excluded.body,
version = excluded.version,
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
RETURNING created_at, updated_at
",
params![
&object.key.collection,
&object.key.id,
&object.body,
stored_version
],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
))
},
)
.map_err(ThingdError::from)?;
object.created_at = timestamps.0;
object.updated_at = timestamps.1;
if options.index {
let text = extract_text_from_json(&object.body);
transaction
.execute(
"DELETE FROM search_index WHERE collection = ?1 AND id = ?2 AND kind = 'object'",
params![&object.key.collection, &object.key.id],
)
.map_err(ThingdError::from)?;
transaction
.execute(
"INSERT INTO search_index (collection, id, kind, text) VALUES (?1, ?2, 'object', ?3)",
params![&object.key.collection, &object.key.id, text],
)
.map_err(ThingdError::from)?;
}
transaction.commit().map_err(ThingdError::from)?;
Ok(object)
}
fn get_object(&self, collection: &str, id: &str) -> ThingdResult<Option<MemoryObject>> {
self.connection
.query_row(
"SELECT collection, id, body, version, created_at, updated_at FROM objects WHERE collection = ?1 AND id = ?2",
params![collection, id],
|row| {
let version = row.get::<_, i64>(3)?;
Ok(MemoryObject {
key: ObjectKey::new(row.get::<_, String>(0)?, row.get::<_, String>(1)?),
body: row.get(2)?,
version: u64::try_from(version).map_err(|error| {
rusqlite::Error::FromSqlConversionFailure(
3,
rusqlite::types::Type::Integer,
Box::new(error),
)
})?,
created_at: row.get::<_, String>(4).unwrap_or_default(),
updated_at: row.get::<_, String>(5).unwrap_or_default(),
})
},
)
.optional()
.map_err(ThingdError::from)
}
fn list_objects(
&self,
collections: Option<&[String]>,
options: &ListObjectsOptions,
) -> ThingdResult<Vec<MemoryObject>> {
let mut conditions: Vec<String> = Vec::new();
let mut bound_values: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
let has_collections = collections.is_some_and(|c| !c.is_empty());
if has_collections {
let cols = collections.unwrap_or_default();
let placeholders = cols.iter().map(|_| "?").collect::<Vec<_>>().join(", ");
conditions.push(format!("collection IN ({placeholders})"));
for col in cols {
bound_values.push(Box::new(col.clone()));
}
}
for (key, value) in &options.filter {
if !key
.chars()
.all(|c| c.is_alphanumeric() || c == '_' || c == '.')
{
return Err(ThingdError::InvalidInput(format!(
"Invalid filter key: '{key}'. Only alphanumeric, underscore, and dot characters are allowed."
)));
}
conditions.push(format!("json_extract(body, '$.{key}') = ?"));
let sql_val: Box<dyn rusqlite::types::ToSql> = match value {
serde_json::Value::String(s) => Box::new(s.clone()),
serde_json::Value::Number(n) => {
if let Some(i) = n.as_i64() {
Box::new(i)
} else {
Box::new(n.as_f64().unwrap_or(0.0))
}
},
serde_json::Value::Bool(b) => Box::new(i64::from(*b)),
serde_json::Value::Null => Box::new(rusqlite::types::Null),
other => Box::new(other.to_string()),
};
bound_values.push(sql_val);
}
let where_clause = if conditions.is_empty() {
String::new()
} else {
format!("WHERE {}", conditions.join(" AND "))
};
let limit_clause = match (options.limit, options.offset) {
(Some(l), Some(o)) => {
bound_values.push(Box::new(i64::try_from(l).unwrap_or(i64::MAX)));
bound_values.push(Box::new(i64::try_from(o).unwrap_or(i64::MAX)));
" LIMIT ? OFFSET ?"
},
(Some(l), None) => {
bound_values.push(Box::new(i64::try_from(l).unwrap_or(i64::MAX)));
" LIMIT ?"
},
(None, Some(o)) => {
bound_values.push(Box::new(i64::try_from(o).unwrap_or(i64::MAX)));
" LIMIT -1 OFFSET ?"
},
(None, None) => "",
};
let order_clause = options.sort_by.as_ref().map_or_else(
|| "ORDER BY collection, id".to_string(),
|sort_by| {
let col = match sort_by.field.as_str() {
"id" => "id",
"collection" => "collection",
"created_at" => "created_at",
"updated_at" => "updated_at",
"version" => "version",
_ => "collection, id",
};
let dir = match sort_by.direction {
crate::model::SortDirection::Asc => "ASC",
crate::model::SortDirection::Desc => "DESC",
};
format!("ORDER BY {col} {dir}")
},
);
let sql = format!(
"SELECT collection, id, body, version, created_at, updated_at FROM objects {where_clause} {order_clause} {limit_clause}"
);
let mut statement = self.connection.prepare(&sql).map_err(ThingdError::from)?;
let params: Vec<&dyn rusqlite::types::ToSql> =
bound_values.iter().map(AsRef::as_ref).collect();
let rows = statement
.query_map(params.as_slice(), row_to_object)
.map_err(ThingdError::from)?;
let mut objects = Vec::new();
for row in rows {
objects.push(row.map_err(ThingdError::from)?);
}
Ok(objects)
}
fn delete_object(&mut self, collection: &str, id: &str) -> ThingdResult<bool> {
let transaction = self.connection.transaction().map_err(ThingdError::from)?;
let changed = transaction
.execute(
"DELETE FROM objects WHERE collection = ?1 AND id = ?2",
params![collection, id],
)
.map_err(ThingdError::from)?;
if changed > 0 {
transaction
.execute(
"DELETE FROM search_index WHERE collection = ?1 AND id = ?2 AND kind = 'object'",
params![collection, id],
)
.map_err(ThingdError::from)?;
}
transaction.commit().map_err(ThingdError::from)?;
Ok(changed > 0)
}
fn delete_objects_batch(&mut self, keys: &[(String, String)]) -> ThingdResult<u64> {
use std::fmt::Write;
let transaction = self.connection.transaction().map_err(ThingdError::from)?;
if keys.is_empty() {
return Ok(0);
}
let mut total_deleted = 0u64;
for chunk in keys.chunks(500) {
let mut sql = String::from("DELETE FROM objects WHERE ");
let mut fts_sql = String::from("DELETE FROM search_index WHERE kind = 'object' AND (");
let mut param_values: Vec<String> = Vec::with_capacity(chunk.len() * 2);
for (i, (collection, id)) in chunk.iter().enumerate() {
if i > 0 {
sql.push_str(" OR ");
fts_sql.push_str(" OR ");
}
let ci = i * 2 + 1;
let ii = i * 2 + 2;
let _ = write!(sql, "(collection = ?{ci} AND id = ?{ii})");
let _ = write!(fts_sql, "(collection = ?{ci} AND id = ?{ii})");
param_values.push(collection.clone());
param_values.push(id.clone());
}
fts_sql.push(')');
let param_slices: Vec<&dyn rusqlite::types::ToSql> = param_values
.iter()
.map(|s| s as &dyn rusqlite::types::ToSql)
.collect();
let deleted = transaction
.execute(&sql, param_slices.as_slice())
.map_err(ThingdError::from)?;
transaction
.execute(&fts_sql, param_slices.as_slice())
.map_err(ThingdError::from)?;
total_deleted += deleted as u64;
}
transaction.commit().map_err(ThingdError::from)?;
Ok(total_deleted)
}
fn count_objects(&self) -> ThingdResult<u64> {
let count: i64 = self
.connection
.query_row("SELECT COUNT(*) FROM objects", [], |row| row.get(0))
.map_err(ThingdError::from)?;
Ok(u64::try_from(count).unwrap_or(0))
}
fn list_collections(&self) -> ThingdResult<Vec<String>> {
let mut statement = self
.connection
.prepare("SELECT DISTINCT collection FROM objects ORDER BY collection")
.map_err(ThingdError::from)?;
let rows = statement
.query_map([], |row| row.get::<_, String>(0))
.map_err(ThingdError::from)?;
let mut collections = Vec::new();
for row in rows {
collections.push(row.map_err(ThingdError::from)?);
}
Ok(collections)
}
fn schema(
&self,
collection: Option<&str>,
options: &SchemaOptions,
) -> ThingdResult<Vec<CollectionSchema>> {
let sample_size = options.sample_size.unwrap_or(50);
let collections: Vec<String> = if let Some(name) = collection {
vec![name.to_string()]
} else {
let mut stmt = self
.connection
.prepare("SELECT DISTINCT collection FROM objects ORDER BY collection")
.map_err(ThingdError::from)?;
let rows = stmt
.query_map([], |row| row.get::<_, String>(0))
.map_err(ThingdError::from)?;
let mut cols = Vec::new();
for row in rows {
cols.push(row.map_err(ThingdError::from)?);
}
cols
};
let mut schemas = Vec::new();
for col in &collections {
let object_count: u64 = self
.connection
.query_row(
"SELECT COUNT(*) FROM objects WHERE collection = ?1",
[col],
|row| row.get::<_, i64>(0),
)
.map_err(ThingdError::from)? as u64;
if object_count == 0 {
continue;
}
let mut stmt = self
.connection
.prepare("SELECT body FROM objects WHERE collection = ?1 LIMIT ?2")
.map_err(ThingdError::from)?;
let rows = stmt
.query_map(rusqlite::params![col, sample_size as i64], |row| {
row.get::<_, String>(0)
})
.map_err(ThingdError::from)?;
let mut field_map: std::collections::BTreeMap<
String,
(String, bool, Vec<serde_json::Value>),
> = std::collections::BTreeMap::new();
for row in rows {
let body_str = row.map_err(ThingdError::from)?;
let body: serde_json::Value =
serde_json::from_str(&body_str).unwrap_or(serde_json::Value::Null);
let map = match &body {
serde_json::Value::Object(m) => m,
_ => continue,
};
for (key, value) in map {
let entry = field_map
.entry(key.clone())
.or_insert_with(|| (infer_sqlite_json_type(value), false, Vec::new()));
if value.is_null() {
entry.1 = true;
}
if entry.2.len() < 3 && !value.is_null() {
entry.2.push(value.clone());
}
let t = infer_sqlite_json_type(value);
if entry.0 != t && !value.is_null() {
entry.0 = "unknown".to_string();
}
}
}
let fields: Vec<FieldSchema> = field_map
.into_iter()
.map(
|(name, (field_type, nullable, sample_values))| FieldSchema {
name,
field_type,
nullable,
sample_values,
},
)
.collect();
schemas.push(CollectionSchema {
name: col.clone(),
object_count,
fields,
});
}
Ok(schemas)
}
}
impl EventLog for SqliteThingStore {
fn is_protected_stream(&self, stream: &str) -> bool {
stream == "__thingd:mcp:audit"
}
fn append_event(&mut self, mut event: MemoryEvent) -> ThingdResult<MemoryEvent> {
if !event.idempotency_key.is_empty()
&& let Some(&existing_seq) = self
.event_idempotency_keys
.get(&(event.stream.clone(), event.idempotency_key.clone()))
{
let existing = self.connection.query_row(
"SELECT stream, event_type, body, sequence, created_at FROM events WHERE stream = ?1 AND sequence = ?2",
params![&event.stream, existing_seq.cast_signed()],
|row| {
Ok(MemoryEvent {
stream: row.get(0)?,
event_type: row.get(1)?,
body: row.get(2)?,
sequence: row.get::<_, i64>(3)?.cast_unsigned(),
created_at: row.get(4)?,
idempotency_key: event.idempotency_key.clone(),
})
},
).map_err(ThingdError::from)?;
return Ok(existing);
}
let transaction = self.connection.transaction().map_err(ThingdError::from)?;
let (sequence, created_at): (i64, String) = transaction
.query_row(
r"
INSERT INTO events (stream, event_type, body, created_at)
VALUES (?1, ?2, ?3, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
RETURNING sequence, created_at
",
params![&event.stream, &event.event_type, &event.body],
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)),
)
.map_err(ThingdError::from)?;
event.sequence =
u64::try_from(sequence).map_err(|error| ThingdError::Storage(error.to_string()))?;
event.created_at = created_at;
if !event.idempotency_key.is_empty() {
self.event_idempotency_keys.insert(
(event.stream.clone(), event.idempotency_key.clone()),
event.sequence,
);
}
let text = extract_text_from_json(&event.body);
let seq_str = sequence.to_string();
transaction
.execute(
"INSERT INTO search_index (collection, id, kind, text) VALUES (?1, ?2, 'event', ?3)",
params![&event.stream, seq_str, text],
)
.map_err(ThingdError::from)?;
transaction.commit().map_err(ThingdError::from)?;
Ok(event)
}
fn append_events_batch(&mut self, events: Vec<MemoryEvent>) -> ThingdResult<Vec<MemoryEvent>> {
if events.is_empty() {
return Ok(Vec::new());
}
let transaction = self.connection.transaction().map_err(ThingdError::from)?;
let mut param_values: Vec<String> = Vec::new();
let mut value_sql = String::new();
let mut results: Vec<MemoryEvent> = Vec::with_capacity(events.len());
for (i, event) in events.into_iter().enumerate() {
if i > 0 {
value_sql.push_str(", ");
}
let si = i * 3 + 1;
let ti = i * 3 + 2;
let bi = i * 3 + 3;
let _ = std::fmt::Write::write_fmt(
&mut value_sql,
format_args!("(?{si}, ?{ti}, ?{bi}, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))"),
);
param_values.push(event.stream.clone());
param_values.push(event.event_type.clone());
param_values.push(event.body.clone());
results.push(event);
}
let sql = format!(
"INSERT INTO events (stream, event_type, body, created_at) VALUES {value_sql} \
RETURNING sequence, created_at"
);
let param_slices: Vec<&dyn rusqlite::types::ToSql> = param_values
.iter()
.map(|s| s as &dyn rusqlite::types::ToSql)
.collect();
let mut statement = transaction.prepare(&sql).map_err(ThingdError::from)?;
let rows = statement
.query_map(param_slices.as_slice(), |row| {
Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?))
})
.map_err(ThingdError::from)?;
for (result, row) in results.iter_mut().zip(rows) {
let (sequence, created_at) = row.map_err(ThingdError::from)?;
result.sequence =
u64::try_from(sequence).map_err(|error| ThingdError::Storage(error.to_string()))?;
result.created_at = created_at;
let text = extract_text_from_json(&result.body);
let seq_str = sequence.to_string();
transaction
.execute(
"INSERT INTO search_index (collection, id, kind, text) VALUES (?1, ?2, 'event', ?3)",
params![&result.stream, seq_str, text],
)
.map_err(ThingdError::from)?;
}
drop(statement);
transaction.commit().map_err(ThingdError::from)?;
Ok(results)
}
fn list_events(
&self,
stream: Option<&str>,
options: ListEventsOptions,
) -> ThingdResult<Vec<MemoryEvent>> {
let mut events = Vec::new();
let mut sql = String::from(
"SELECT stream, event_type, body, sequence, created_at FROM events WHERE 1=1",
);
let mut param_values: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
if let Some(stream) = stream {
let idx = param_values.len() + 1;
write!(sql, " AND stream = ?{idx}").unwrap();
param_values.push(Box::new(stream.to_string()));
}
if let Some(from_sequence) = options.from_sequence {
let idx = param_values.len() + 1;
write!(sql, " AND sequence > ?{idx}").unwrap();
param_values.push(Box::new(from_sequence.cast_signed()));
}
if let Some(ref since) = options.since {
let idx = param_values.len() + 1;
write!(sql, " AND created_at >= ?{idx}").unwrap();
param_values.push(Box::new(since.clone()));
}
sql.push_str(" ORDER BY sequence");
if let Some(limit) = options.limit {
sql.push_str(" LIMIT ?");
param_values.push(Box::new(limit.cast_signed()));
}
let mut statement = self.connection.prepare(&sql).map_err(ThingdError::from)?;
let param_refs: Vec<&dyn rusqlite::types::ToSql> =
param_values.iter().map(AsRef::as_ref).collect();
let rows = statement
.query_map(param_refs.as_slice(), row_to_event)
.map_err(ThingdError::from)?;
for row in rows {
events.push(row.map_err(ThingdError::from)?);
}
Ok(events)
}
fn count_events(&self) -> ThingdResult<u64> {
let count: i64 = self
.connection
.query_row("SELECT COUNT(*) FROM events", [], |row| row.get(0))
.map_err(ThingdError::from)?;
Ok(u64::try_from(count).unwrap_or(0))
}
fn list_streams(&self) -> ThingdResult<Vec<String>> {
let mut statement = self
.connection
.prepare("SELECT DISTINCT stream FROM events ORDER BY stream")
.map_err(ThingdError::from)?;
let rows = statement
.query_map([], |row| row.get::<_, String>(0))
.map_err(ThingdError::from)?;
let mut streams = Vec::new();
for row in rows {
streams.push(row.map_err(ThingdError::from)?);
}
Ok(streams)
}
fn delete_last_event(&mut self, stream: &str) -> ThingdResult<Option<MemoryEvent>> {
if self.is_protected_stream(stream) {
return Err(ThingdError::Protected(format!(
"stream '{stream}' is protected and cannot be modified"
)));
}
let transaction = self.connection.transaction().map_err(ThingdError::from)?;
let result = transaction
.query_row(
r"
DELETE FROM events
WHERE sequence = (
SELECT MAX(sequence) FROM events WHERE stream = ?1
)
RETURNING stream, event_type, body, sequence, created_at
",
params![stream],
row_to_event,
)
.optional()
.map_err(ThingdError::from)?;
if let Some(ref event) = result {
let seq_str = event.sequence.to_string();
transaction
.execute(
"DELETE FROM search_index WHERE collection = ?1 AND kind = 'event' AND id = ?2",
params![stream, seq_str],
)
.map_err(ThingdError::from)?;
}
transaction.commit().map_err(ThingdError::from)?;
Ok(result)
}
fn delete_stream(&mut self, stream: &str) -> ThingdResult<u64> {
if self.is_protected_stream(stream) {
return Err(ThingdError::Protected(format!(
"stream '{stream}' is protected and cannot be modified"
)));
}
let transaction = self.connection.transaction().map_err(ThingdError::from)?;
let count = transaction
.execute("DELETE FROM events WHERE stream = ?1", params![stream])
.map_err(ThingdError::from)?;
transaction
.execute(
"DELETE FROM search_index WHERE collection = ?1 AND kind = 'event'",
params![stream],
)
.map_err(ThingdError::from)?;
transaction.commit().map_err(ThingdError::from)?;
Ok(count as u64)
}
}
impl QueueStore for SqliteThingStore {
fn push_job(&mut self, job: QueueJob) -> ThingdResult<QueueJob> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)
.map_err(ThingdError::from)?;
if let Some(existing) = transaction
.query_row(
&queue_job_select_sql("WHERE queue = ?1 AND id = ?2"),
params![&job.queue, &job.id],
row_to_queue_job,
)
.optional()
.map_err(ThingdError::from)?
{
transaction.commit().map_err(ThingdError::from)?;
return Ok(existing);
}
let created_at: String = transaction
.query_row(
r"
INSERT INTO queue_jobs (
queue,
id,
body,
attempts,
max_attempts,
status,
available_at_ms,
leased_at_ms,
lease_expires_at_ms,
completed_at_ms,
dead_at_ms,
created_at,
updated_at
)
VALUES (
?1,
?2,
?3,
?4,
?5,
?6,
?7,
?8,
?9,
?10,
?11,
strftime('%Y-%m-%dT%H:%M:%fZ', 'now'),
strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
)
RETURNING created_at
",
params![
&job.queue,
&job.id,
&job.body,
u32_to_i64(job.attempts),
u32_to_i64(job.max_attempts),
status_to_str(job.status),
job.available_at_ms,
job.leased_at_ms,
job.lease_expires_at_ms,
job.completed_at_ms,
job.dead_at_ms
],
|row| row.get(0),
)
.map_err(ThingdError::from)?;
transaction.commit().map_err(ThingdError::from)?;
Ok(QueueJob { created_at, ..job })
}
fn push_jobs_batch(&mut self, jobs: Vec<QueueJob>) -> ThingdResult<Vec<QueueJob>> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)
.map_err(ThingdError::from)?;
let mut results = Vec::with_capacity(jobs.len());
for job in jobs {
if let Some(existing) = transaction
.query_row(
&queue_job_select_sql("WHERE queue = ?1 AND id = ?2"),
params![&job.queue, &job.id],
row_to_queue_job,
)
.optional()
.map_err(ThingdError::from)?
{
results.push(existing);
continue;
}
let created_at: String = transaction
.query_row(
r"
INSERT INTO queue_jobs (
queue,
id,
body,
attempts,
max_attempts,
status,
available_at_ms,
leased_at_ms,
lease_expires_at_ms,
completed_at_ms,
dead_at_ms,
created_at,
updated_at
)
VALUES (
?1,
?2,
?3,
?4,
?5,
?6,
?7,
?8,
?9,
?10,
?11,
strftime('%Y-%m-%dT%H:%M:%fZ', 'now'),
strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
)
RETURNING created_at
",
params![
&job.queue,
&job.id,
&job.body,
u32_to_i64(job.attempts),
u32_to_i64(job.max_attempts),
status_to_str(job.status),
job.available_at_ms,
job.leased_at_ms,
job.lease_expires_at_ms,
job.completed_at_ms,
job.dead_at_ms
],
|row| row.get(0),
)
.map_err(ThingdError::from)?;
results.push(QueueJob { created_at, ..job });
}
transaction.commit().map_err(ThingdError::from)?;
Ok(results)
}
fn claim_job_with_options(
&mut self,
queue: &str,
options: QueueClaimOptions,
) -> ThingdResult<Option<QueueJob>> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)
.map_err(ThingdError::from)?;
release_expired_leases(&transaction, queue)?;
let now = unix_timestamp_millis();
let Some(mut job) = transaction
.query_row(
&queue_job_select_sql(
"WHERE queue = ?1 AND status = 'ready' AND available_at_ms <= ?2 ORDER BY created_at LIMIT 1",
),
params![queue, now],
row_to_queue_job,
)
.optional()
.map_err(ThingdError::from)?
else {
transaction.commit().map_err(ThingdError::from)?;
return Ok(None);
};
job.status = QueueJobStatus::Leased;
job.attempts += 1;
job.leased_at_ms = Some(now);
job.lease_expires_at_ms = Some(now.saturating_add(u64_to_i64(options.lease_ms)));
transaction
.execute(
r"
UPDATE queue_jobs
SET attempts = ?3,
status = ?4,
leased_at_ms = ?5,
lease_expires_at_ms = ?6,
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE queue = ?1 AND id = ?2
",
params![
&job.queue,
&job.id,
u32_to_i64(job.attempts),
status_to_str(job.status),
job.leased_at_ms,
job.lease_expires_at_ms
],
)
.map_err(ThingdError::from)?;
transaction.commit().map_err(ThingdError::from)?;
Ok(Some(job))
}
fn ack_job(&mut self, queue: &str, id: &str) -> ThingdResult<Option<QueueJob>> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)
.map_err(ThingdError::from)?;
let Some(mut job) = transaction
.query_row(
&queue_job_select_sql("WHERE queue = ?1 AND id = ?2"),
params![queue, id],
row_to_queue_job,
)
.optional()
.map_err(ThingdError::from)?
else {
transaction.commit().map_err(ThingdError::from)?;
return Ok(None);
};
if job.status != QueueJobStatus::Leased {
return Err(ThingdError::Conflict(format!(
"job {id} must be leased before ack"
)));
}
job.status = QueueJobStatus::Completed;
job.completed_at_ms = Some(unix_timestamp_millis());
transaction
.execute(
r"
UPDATE queue_jobs
SET status = ?3,
completed_at_ms = ?4,
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE queue = ?1 AND id = ?2
",
params![queue, id, status_to_str(job.status), job.completed_at_ms],
)
.map_err(ThingdError::from)?;
transaction.commit().map_err(ThingdError::from)?;
Ok(Some(job))
}
fn claim_and_ack(
&mut self,
queue: &str,
options: QueueClaimOptions,
) -> ThingdResult<Option<QueueJob>> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)
.map_err(ThingdError::from)?;
release_expired_leases(&transaction, queue)?;
let now = unix_timestamp_millis();
let Some(mut job) = transaction
.query_row(
&queue_job_select_sql(
"WHERE queue = ?1 AND status = 'ready' AND available_at_ms <= ?2 ORDER BY created_at LIMIT 1",
),
params![queue, now],
row_to_queue_job,
)
.optional()
.map_err(ThingdError::from)?
else {
transaction.commit().map_err(ThingdError::from)?;
return Ok(None);
};
job.status = QueueJobStatus::Leased;
job.attempts += 1;
job.leased_at_ms = Some(now);
job.lease_expires_at_ms = Some(now.saturating_add(u64_to_i64(options.lease_ms)));
transaction
.execute(
r"
UPDATE queue_jobs
SET attempts = ?3,
status = ?4,
leased_at_ms = ?5,
lease_expires_at_ms = ?6,
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE queue = ?1 AND id = ?2
",
params![
&job.queue,
&job.id,
u32_to_i64(job.attempts),
status_to_str(job.status),
job.leased_at_ms,
job.lease_expires_at_ms
],
)
.map_err(ThingdError::from)?;
job.status = QueueJobStatus::Completed;
job.completed_at_ms = Some(unix_timestamp_millis());
transaction
.execute(
r"
UPDATE queue_jobs
SET status = ?3,
completed_at_ms = ?4,
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE queue = ?1 AND id = ?2
",
params![
&job.queue,
&job.id,
status_to_str(job.status),
job.completed_at_ms
],
)
.map_err(ThingdError::from)?;
transaction.commit().map_err(ThingdError::from)?;
Ok(Some(job))
}
fn nack_job_with_options(
&mut self,
queue: &str,
id: &str,
options: QueueNackOptions,
) -> ThingdResult<Option<QueueJob>> {
let transaction = self
.connection
.transaction_with_behavior(TransactionBehavior::Immediate)
.map_err(ThingdError::from)?;
let Some(mut job) = transaction
.query_row(
&queue_job_select_sql("WHERE queue = ?1 AND id = ?2"),
params![queue, id],
row_to_queue_job,
)
.optional()
.map_err(ThingdError::from)?
else {
transaction.commit().map_err(ThingdError::from)?;
return Ok(None);
};
if job.status != QueueJobStatus::Leased {
return Err(ThingdError::Conflict(format!(
"job {id} must be leased before nack"
)));
}
let now = unix_timestamp_millis();
job.leased_at_ms = None;
job.lease_expires_at_ms = None;
job.status = if job.attempts >= job.max_attempts {
job.dead_at_ms = Some(now);
QueueJobStatus::Dead
} else {
job.available_at_ms = now.saturating_add(u64_to_i64(options.delay_ms));
QueueJobStatus::Ready
};
if !options.error.is_empty() {
job.last_error = options.error;
}
transaction
.execute(
r"
UPDATE queue_jobs
SET attempts = ?3,
status = ?4,
available_at_ms = ?5,
leased_at_ms = NULL,
lease_expires_at_ms = NULL,
dead_at_ms = ?6,
last_error = ?7,
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE queue = ?1 AND id = ?2
",
params![
queue,
id,
u32_to_i64(job.attempts),
status_to_str(job.status),
job.available_at_ms,
job.dead_at_ms,
job.last_error
],
)
.map_err(ThingdError::from)?;
transaction.commit().map_err(ThingdError::from)?;
Ok(Some(job))
}
fn list_jobs(&self, queue: &str) -> ThingdResult<Vec<QueueJob>> {
let mut statement = self
.connection
.prepare(&queue_job_select_sql(
"WHERE queue = ?1 ORDER BY created_at LIMIT 1000",
))
.map_err(ThingdError::from)?;
let rows = statement
.query_map(params![queue], row_to_queue_job)
.map_err(ThingdError::from)?;
let mut jobs = Vec::new();
for row in rows {
jobs.push(row.map_err(ThingdError::from)?);
}
Ok(jobs)
}
fn list_dead_jobs(&self, queue: &str) -> ThingdResult<Vec<QueueJob>> {
let mut statement = self
.connection
.prepare(&queue_job_select_sql(
"WHERE queue = ?1 AND status = 'dead' ORDER BY created_at LIMIT 1000",
))
.map_err(ThingdError::from)?;
let rows = statement
.query_map(params![queue], row_to_queue_job)
.map_err(ThingdError::from)?;
let mut jobs = Vec::new();
for row in rows {
jobs.push(row.map_err(ThingdError::from)?);
}
Ok(jobs)
}
fn list_queues(&self) -> ThingdResult<Vec<String>> {
let mut statement = self
.connection
.prepare("SELECT DISTINCT queue FROM queue_jobs ORDER BY queue")
.map_err(ThingdError::from)?;
let rows = statement
.query_map([], |row| row.get::<_, String>(0))
.map_err(ThingdError::from)?;
let mut queues = Vec::new();
for row in rows {
queues.push(row.map_err(ThingdError::from)?);
}
Ok(queues)
}
fn count_active_jobs(&self) -> ThingdResult<u64> {
let count = self
.connection
.query_row(
"SELECT COUNT(id) FROM queue_jobs WHERE status != 'dead'",
[],
|row| row.get::<_, i64>(0),
)
.map_err(ThingdError::from)?;
Ok(u64::try_from(count).unwrap_or(0))
}
fn count_dead_jobs(&self) -> ThingdResult<u64> {
let count = self
.connection
.query_row(
"SELECT COUNT(id) FROM queue_jobs WHERE status = 'dead'",
[],
|row| row.get::<_, i64>(0),
)
.map_err(ThingdError::from)?;
Ok(u64::try_from(count).unwrap_or(0))
}
}
impl crate::store::Searcher for SqliteThingStore {
#[allow(clippy::too_many_lines)]
fn search(
&self,
query: &str,
options: crate::SearchOptions,
) -> ThingdResult<Vec<crate::SearchHit>> {
let sanitized = sanitize_fts_query(query);
if sanitized.is_empty() {
return Ok(Vec::new());
}
if query.chars().filter(|c| c.is_alphanumeric()).count() < 2 {
return Ok(Vec::new());
}
let mut sql = String::from(
r"
SELECT
s.kind,
s.collection,
s.id,
s.text,
o.body AS object_body,
o.version AS object_version,
o.created_at AS object_created_at,
o.updated_at AS object_updated_at,
e.event_type AS event_type,
e.body AS event_body,
e.created_at AS event_created_at,
bm25(search_index) AS bm25_score,
(strftime('%s', 'now') - strftime('%s', coalesce(o.created_at, e.created_at))) AS age_seconds
FROM search_index s
LEFT JOIN objects o ON s.kind = 'object' AND s.collection = o.collection AND s.id = o.id
LEFT JOIN events e ON s.kind = 'event' AND s.collection = e.stream AND s.id = CAST(e.sequence AS TEXT)
WHERE search_index MATCH ?1
AND (s.kind != 'object' OR o.collection IS NOT NULL)
AND (s.kind != 'event' OR e.stream IS NOT NULL)
",
);
let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![Box::new(sanitized)];
if let Some(ref collections) = options.collections
&& !collections.is_empty()
{
let placeholders: Vec<String> = (0..collections.len())
.map(|i| format!("?{}", params.len() + i + 1))
.collect();
write!(sql, " AND s.collection IN ({})", placeholders.join(",")).unwrap();
for coll in collections {
params.push(Box::new(coll.clone()));
}
}
let effective_limit = options.limit.map_or(100, |l| l.min(1000));
sql.push_str(" ORDER BY bm25_score");
if options.filter.is_none() {
write!(sql, " LIMIT ?{}", params.len() + 1).unwrap();
params.push(Box::new(i64::try_from(effective_limit).unwrap_or(100)));
} else {
let fetch_limit = (effective_limit * 3).min(1000);
write!(sql, " LIMIT ?{}", params.len() + 1).unwrap();
params.push(Box::new(i64::try_from(fetch_limit).unwrap_or(1000)));
}
let mut statement = self.connection.prepare(&sql).map_err(ThingdError::from)?;
let param_refs: Vec<&dyn rusqlite::types::ToSql> =
params.iter().map(AsRef::as_ref).collect();
let rows = statement
.query_map(param_refs.as_slice(), |row| {
let kind: String = row.get(0)?;
let collection: String = row.get(1)?;
let id: String = row.get(2)?;
let text: String = row.get(3)?;
let bm25_score: f64 = row.get(11)?;
let age_seconds: Option<i64> = row.get(12)?;
let relevance_score = -bm25_score;
let age =
f64::from(i32::try_from(age_seconds.unwrap_or(0).max(0)).unwrap_or(i32::MAX));
let recency_factor = 1.0 / (1.0 + age / 86400.0);
let score = relevance_score * recency_factor;
let (body, version, created_at, updated_at, event_type) = if kind == "object" {
let object_body: String = row.get(4)?;
let object_version: i64 = row.get(5)?;
let object_created_at: String = row.get(6)?;
let object_updated_at: String = row.get(7)?;
(
object_body,
Some(object_version.cast_unsigned()),
object_created_at,
Some(object_updated_at),
None,
)
} else {
let event_type_val: String = row.get(8)?;
let event_body: String = row.get(9)?;
let event_created_at: String = row.get(10)?;
(
event_body,
None,
event_created_at,
None,
Some(event_type_val),
)
};
Ok(crate::SearchHit {
kind,
collection,
id,
text,
score,
body,
version,
created_at,
updated_at,
event_type,
})
})
.map_err(ThingdError::from)?;
let mut hits = Vec::new();
for row in rows {
let hit = row.map_err(ThingdError::from)?;
if let Some(ref filter) = options.filter
&& !matches_filter(&hit.body, filter)
{
continue;
}
hits.push(hit);
}
hits.sort_by(|a, b| {
b.score
.partial_cmp(&a.score)
.unwrap_or(std::cmp::Ordering::Equal)
});
if let Some(limit) = options.limit {
hits.truncate(limit);
}
Ok(hits)
}
}
impl crate::store::LinkStore for SqliteThingStore {
fn create_link(&mut self, link: crate::Link) -> ThingdResult<crate::Link> {
let id = uuid::Uuid::new_v4().to_string();
self.connection
.execute(
r"
INSERT INTO links (id, from_ref, type, to_ref, weight, metadata_json, created_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
",
params![
id,
link.from_ref,
link.link_type,
link.to_ref,
link.weight,
link.metadata_json
],
)
.map_err(ThingdError::from)?;
let created_at: String = self
.connection
.query_row(
"SELECT created_at FROM links WHERE id = ?1",
params![id],
|row| row.get(0),
)
.map_err(ThingdError::from)?;
Ok(crate::Link {
id,
from_ref: link.from_ref,
link_type: link.link_type,
to_ref: link.to_ref,
weight: link.weight,
metadata_json: link.metadata_json,
created_at,
})
}
fn delete_link(&mut self, id: &str) -> ThingdResult<bool> {
let changed = self
.connection
.execute("DELETE FROM links WHERE id = ?1", params![id])
.map_err(ThingdError::from)?;
Ok(changed > 0)
}
fn get_link(&self, id: &str) -> ThingdResult<Option<crate::Link>> {
self.connection
.query_row(
"SELECT id, from_ref, type, to_ref, weight, metadata_json, created_at FROM links WHERE id = ?1",
params![id],
row_to_link,
)
.optional()
.map_err(ThingdError::from)
}
fn get_neighbors(
&self,
reference: &str,
direction: crate::LinkDirection,
options: crate::LinkQueryOptions,
) -> ThingdResult<Vec<crate::Link>> {
let (where_clause, param_value): (&str, String) = match direction {
crate::LinkDirection::Outgoing => ("WHERE from_ref = ?1", reference.to_string()),
crate::LinkDirection::Incoming => ("WHERE to_ref = ?1", reference.to_string()),
crate::LinkDirection::Both => (
"WHERE (from_ref = ?1 OR to_ref = ?1)",
reference.to_string(),
),
};
let (type_filter_sql, type_param) = options.link_type.as_ref().map_or_else(
|| (String::new(), None),
|t| (" AND type = ?2".to_string(), Some(t.clone())),
);
let limit_clause = options
.limit
.map_or_else(String::new, |_| " LIMIT ?".to_string());
let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
params.push(Box::new(param_value));
if let Some(ref t) = type_param {
params.push(Box::new(t.clone()));
}
if let Some(l) = options.limit {
params.push(Box::new(i64::try_from(l).unwrap_or(1000)));
}
let sql = format!(
"SELECT id, from_ref, type, to_ref, weight, metadata_json, created_at FROM links {where_clause}{type_filter_sql}{limit_clause}"
);
let mut statement = self.connection.prepare(&sql).map_err(ThingdError::from)?;
let param_refs: Vec<&dyn rusqlite::types::ToSql> =
params.iter().map(AsRef::as_ref).collect();
let rows = statement
.query_map(param_refs.as_slice(), row_to_link)
.map_err(ThingdError::from)?;
let mut links = Vec::new();
for row in rows {
links.push(row.map_err(ThingdError::from)?);
}
Ok(links)
}
fn count_links(&self) -> ThingdResult<u64> {
let count: i64 = self
.connection
.query_row("SELECT COUNT(*) FROM links", [], |row| row.get(0))
.map_err(ThingdError::from)?;
Ok(u64::try_from(count).unwrap_or(0))
}
}
impl crate::store::AggregateStore for SqliteThingStore {
fn aggregate(
&self,
collection: &str,
options: &AggregateOptions,
) -> ThingdResult<AggregateResult> {
let mut conditions = vec!["collection = ?".to_string()];
let mut bound_values: Vec<Box<dyn rusqlite::types::ToSql>> =
vec![Box::new(collection.to_string())];
for (key, value) in &options.filter {
if !key
.chars()
.all(|c| c.is_alphanumeric() || c == '_' || c == '.')
{
return Err(ThingdError::InvalidInput(format!(
"Invalid filter key: '{key}'. Only alphanumeric, underscore, and dot characters are allowed."
)));
}
conditions.push(format!("json_extract(body, '$.{key}') = ?"));
let sql_val: Box<dyn rusqlite::types::ToSql> = match value {
serde_json::Value::String(s) => Box::new(s.clone()),
serde_json::Value::Number(n) => {
if let Some(i) = n.as_i64() {
Box::new(i)
} else {
Box::new(n.as_f64().unwrap_or(0.0))
}
},
serde_json::Value::Bool(b) => Box::new(i64::from(*b)),
serde_json::Value::Null => Box::new(rusqlite::types::Null),
other => Box::new(other.to_string()),
};
bound_values.push(sql_val);
}
let where_clause = format!("WHERE {}", conditions.join(" AND "));
if let Some(group_field) = &options.group_by {
if !group_field
.chars()
.all(|c| c.is_alphanumeric() || c == '_' || c == '.')
{
return Err(ThingdError::InvalidInput(format!(
"Invalid groupBy field: '{group_field}'. Only alphanumeric, underscore, and dot characters are allowed."
)));
}
let sql = if options.function == AggregateFunction::Count {
format!(
"SELECT json_extract(body, '$.{group_field}') AS grp, COUNT(*) AS val FROM objects {where_clause} GROUP BY grp ORDER BY grp"
)
} else {
let field = options.field.as_deref().unwrap_or_default();
if !field
.chars()
.all(|c| c.is_alphanumeric() || c == '_' || c == '.')
{
return Err(ThingdError::InvalidInput(format!(
"Invalid field: '{field}'. Only alphanumeric, underscore, and dot characters are allowed."
)));
}
let func = options.function.sql_func();
format!(
"SELECT json_extract(body, '$.{group_field}') AS grp, {func}(json_extract(body, '$.{field}')) AS val FROM objects {where_clause} GROUP BY grp ORDER BY grp"
)
};
let mut statement = self.connection.prepare(&sql).map_err(ThingdError::from)?;
let param_refs: Vec<&dyn rusqlite::types::ToSql> =
bound_values.iter().map(AsRef::as_ref).collect();
let rows = statement
.query_map(param_refs.as_slice(), |row| {
Ok(AggregateGroupResult {
key: row.get::<_, String>(0).unwrap_or_default(),
value: row.get::<_, f64>(1).unwrap_or(0.0),
})
})
.map_err(ThingdError::from)?;
let mut groups = Vec::new();
for row in rows {
groups.push(row.map_err(ThingdError::from)?);
}
let total: f64 = groups.iter().map(|g| g.value).sum();
Ok(AggregateResult { total, groups })
} else {
let sql = if options.function == AggregateFunction::Count {
format!("SELECT COUNT(*) FROM objects {where_clause}")
} else {
let field = options.field.as_deref().unwrap_or_default();
if !field
.chars()
.all(|c| c.is_alphanumeric() || c == '_' || c == '.')
{
return Err(ThingdError::InvalidInput(format!(
"Invalid field: '{field}'. Only alphanumeric, underscore, and dot characters are allowed."
)));
}
let func = options.function.sql_func();
format!(
"SELECT {func}(json_extract(body, '$.{field}')) FROM objects {where_clause}"
)
};
let mut statement = self.connection.prepare(&sql).map_err(ThingdError::from)?;
let param_refs: Vec<&dyn rusqlite::types::ToSql> =
bound_values.iter().map(AsRef::as_ref).collect();
let total = statement
.query_row(param_refs.as_slice(), |row| row.get::<_, f64>(0))
.map_err(ThingdError::from)?;
Ok(AggregateResult {
total,
groups: Vec::new(),
})
}
}
fn timeseries(
&self,
collection: &str,
options: &TimeSeriesOptions,
) -> ThingdResult<TimeSeriesResult> {
let mut conditions = vec!["collection = ?".to_string()];
let mut bound_values: Vec<Box<dyn rusqlite::types::ToSql>> =
vec![Box::new(collection.to_string())];
for (key, value) in &options.filter {
if !key
.chars()
.all(|c| c.is_alphanumeric() || c == '_' || c == '.')
{
return Err(ThingdError::InvalidInput(format!(
"Invalid filter key: '{key}'. Only alphanumeric, underscore, and dot characters are allowed."
)));
}
conditions.push(format!("json_extract(body, '$.{key}') = ?"));
let sql_val: Box<dyn rusqlite::types::ToSql> = match value {
serde_json::Value::String(s) => Box::new(s.clone()),
serde_json::Value::Number(n) => {
if let Some(i) = n.as_i64() {
Box::new(i)
} else {
Box::new(n.as_f64().unwrap_or(0.0))
}
},
serde_json::Value::Bool(b) => Box::new(i64::from(*b)),
serde_json::Value::Null => Box::new(rusqlite::types::Null),
other => Box::new(other.to_string()),
};
bound_values.push(sql_val);
}
if let Some(ref from) = options.from {
conditions.push("created_at >= ?".to_string());
bound_values.push(Box::new(from.clone()));
}
if let Some(ref to) = options.to {
conditions.push("created_at < ?".to_string());
bound_values.push(Box::new(to.clone()));
}
let where_clause = format!("WHERE {}", conditions.join(" AND "));
let strftime_format = options.bucket.strftime_format();
let sql = if options.function == AggregateFunction::Count {
format!(
"SELECT strftime('{strftime_format}', created_at) AS label, COUNT(*) AS val FROM objects {where_clause} GROUP BY label ORDER BY label"
)
} else {
let field = options.field.as_deref().unwrap_or_default();
if !field
.chars()
.all(|c| c.is_alphanumeric() || c == '_' || c == '.')
{
return Err(ThingdError::InvalidInput(format!(
"Invalid field: '{field}'. Only alphanumeric, underscore, and dot characters are allowed."
)));
}
let func = options.function.sql_func();
format!(
"SELECT strftime('{strftime_format}', created_at) AS label, {func}(json_extract(body, '$.{field}')) AS val FROM objects {where_clause} GROUP BY label ORDER BY label"
)
};
let mut statement = self.connection.prepare(&sql).map_err(ThingdError::from)?;
let param_refs: Vec<&dyn rusqlite::types::ToSql> =
bound_values.iter().map(AsRef::as_ref).collect();
let rows = statement
.query_map(param_refs.as_slice(), |row| {
Ok(TimeSeriesBucket {
label: row.get::<_, String>(0).unwrap_or_default(),
value: row.get::<_, f64>(1).unwrap_or(0.0),
})
})
.map_err(ThingdError::from)?;
let mut buckets = Vec::new();
for row in rows {
buckets.push(row.map_err(ThingdError::from)?);
}
Ok(TimeSeriesResult { buckets })
}
}
fn row_to_link(row: &rusqlite::Row<'_>) -> rusqlite::Result<crate::Link> {
Ok(crate::Link {
id: row.get(0)?,
from_ref: row.get(1)?,
link_type: row.get(2)?,
to_ref: row.get(3)?,
weight: row.get(4)?,
metadata_json: row.get(5)?,
created_at: row.get(6)?,
})
}
fn row_to_object(row: &rusqlite::Row<'_>) -> rusqlite::Result<MemoryObject> {
let version = row.get::<_, i64>(3)?;
Ok(MemoryObject {
key: ObjectKey::new(row.get::<_, String>(0)?, row.get::<_, String>(1)?),
body: row.get(2)?,
version: u64::try_from(version).map_err(|error| {
rusqlite::Error::FromSqlConversionFailure(
3,
rusqlite::types::Type::Integer,
Box::new(error),
)
})?,
created_at: row.get::<_, String>(4).unwrap_or_default(),
updated_at: row.get::<_, String>(5).unwrap_or_default(),
})
}
fn row_to_event(row: &rusqlite::Row<'_>) -> rusqlite::Result<MemoryEvent> {
let sequence = row.get::<_, i64>(3)?;
Ok(MemoryEvent {
stream: row.get(0)?,
event_type: row.get(1)?,
body: row.get(2)?,
sequence: u64::try_from(sequence).map_err(|error| {
rusqlite::Error::FromSqlConversionFailure(
3,
rusqlite::types::Type::Integer,
Box::new(error),
)
})?,
created_at: row.get::<_, String>(4).unwrap_or_default(),
idempotency_key: String::new(),
})
}
fn queue_job_select_sql(predicate: &str) -> String {
format!(
"SELECT queue, id, body, attempts, max_attempts, status, available_at_ms, leased_at_ms, lease_expires_at_ms, completed_at_ms, dead_at_ms, created_at, last_error FROM queue_jobs {predicate}"
)
}
fn row_to_queue_job(row: &rusqlite::Row<'_>) -> rusqlite::Result<QueueJob> {
let attempts = row.get::<_, i64>(3)?;
let max_attempts = row.get::<_, i64>(4)?;
let status = row.get::<_, String>(5)?;
Ok(QueueJob {
queue: row.get(0)?,
id: row.get(1)?,
body: row.get(2)?,
attempts: u32::try_from(attempts).map_err(|error| {
rusqlite::Error::FromSqlConversionFailure(
3,
rusqlite::types::Type::Integer,
Box::new(error),
)
})?,
max_attempts: u32::try_from(max_attempts).map_err(|error| {
rusqlite::Error::FromSqlConversionFailure(
4,
rusqlite::types::Type::Integer,
Box::new(error),
)
})?,
status: match status.as_str() {
"ready" => QueueJobStatus::Ready,
"leased" => QueueJobStatus::Leased,
"completed" => QueueJobStatus::Completed,
"dead" => QueueJobStatus::Dead,
_other => {
return Err(rusqlite::Error::FromSqlConversionFailure(
5,
rusqlite::types::Type::Text,
Box::new(std::fmt::Error),
));
},
},
available_at_ms: row.get(6)?,
leased_at_ms: row.get(7)?,
lease_expires_at_ms: row.get(8)?,
completed_at_ms: row.get(9)?,
dead_at_ms: row.get(10)?,
created_at: row.get::<_, String>(11).unwrap_or_default(),
last_error: row.get::<_, String>(12).unwrap_or_default(),
})
}
fn release_expired_leases(connection: &rusqlite::Connection, queue: &str) -> ThingdResult<()> {
connection
.execute(
r"
UPDATE queue_jobs
SET status = 'ready',
leased_at_ms = NULL,
lease_expires_at_ms = NULL,
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE queue = ?1
AND status = 'leased'
AND lease_expires_at_ms IS NOT NULL
AND lease_expires_at_ms <= ?2
",
params![queue, unix_timestamp_millis()],
)
.map_err(ThingdError::from)?;
Ok(())
}
const fn status_to_str(status: QueueJobStatus) -> &'static str {
match status {
QueueJobStatus::Ready => "ready",
QueueJobStatus::Leased => "leased",
QueueJobStatus::Completed => "completed",
QueueJobStatus::Dead => "dead",
}
}
fn u32_to_i64(value: u32) -> i64 {
i64::from(value)
}
fn sanitize_fts_query(query: &str) -> String {
let mut cleaned = String::new();
let normalized: String = query
.chars()
.map(|c| {
if c.is_alphanumeric() || c.is_whitespace() {
c
} else {
' '
}
})
.collect();
for word in normalized.split_whitespace() {
if !word.is_empty() {
if !cleaned.is_empty() {
cleaned.push(' ');
}
cleaned.push_str(word);
cleaned.push('*');
}
}
cleaned
}
fn extract_text_from_json(json_str: &str) -> String {
serde_json::from_str::<serde_json::Value>(json_str).map_or_else(
|_| json_str.to_string(),
|value| {
let mut out = String::new();
collect_strings(&value, &mut out);
out.trim().to_string()
},
)
}
fn collect_strings(value: &serde_json::Value, out: &mut String) {
match value {
serde_json::Value::String(s) => {
if !out.is_empty() {
out.push(' ');
}
out.push_str(s);
},
serde_json::Value::Array(arr) => {
for val in arr {
collect_strings(val, out);
}
},
serde_json::Value::Object(obj) => {
for (key, val) in obj {
if !out.is_empty() {
out.push(' ');
}
out.push_str(key);
collect_strings(val, out);
}
},
serde_json::Value::Number(num) => {
if !out.is_empty() {
out.push(' ');
}
out.push_str(&num.to_string());
},
serde_json::Value::Bool(b) => {
if !out.is_empty() {
out.push(' ');
}
out.push_str(&b.to_string());
},
serde_json::Value::Null => {},
}
}
fn matches_filter(body_str: &str, filter: &serde_json::Value) -> bool {
let Ok(body) = serde_json::from_str::<serde_json::Value>(body_str) else {
return false;
};
let Some(filter_obj) = filter.as_object() else {
return true;
};
for (k, v) in filter_obj {
if body.get(k) != Some(v) {
return false;
}
}
true
}
fn infer_sqlite_json_type(value: &serde_json::Value) -> String {
match value {
serde_json::Value::Null => "null".to_string(),
serde_json::Value::Bool(_) => "boolean".to_string(),
serde_json::Value::Number(_) => "number".to_string(),
serde_json::Value::String(s) => {
if s.len() > 10
&& (s.contains('T') || s.contains('-'))
&& chrono::DateTime::parse_from_rfc3339(s).is_ok()
{
"date".to_string()
} else {
"string".to_string()
}
},
serde_json::Value::Array(_) => "array".to_string(),
serde_json::Value::Object(_) => "object".to_string(),
}
}
#[cfg(test)]
mod tests {
use rusqlite::Connection;
use tempfile::NamedTempFile;
use super::*;
use crate::store::Searcher;
use crate::{ListObjectsOptions, SearchOptions};
#[test]
fn records_schema_version_on_initialize() {
let store = SqliteThingStore::open_in_memory().unwrap();
assert_eq!(store.schema_version().unwrap(), SQLITE_SCHEMA_VERSION);
}
#[test]
fn integrity_check_passes_on_fresh_store() {
let store = SqliteThingStore::open_in_memory().unwrap();
let ok: String = store
.connection
.query_row("PRAGMA quick_check", [], |row| row.get(0))
.unwrap();
assert_eq!(ok, "ok");
}
#[test]
fn wal_checkpoint_returns_zero_frames_on_in_memory() {
let store = SqliteThingStore::open_in_memory().unwrap();
let (_busy, _frames) = store.wal_checkpoint().unwrap();
}
#[test]
fn backup_to_creates_valid_database() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
store
.put_object(MemoryObject::new("test", "1", r#"{"v":1}"#))
.unwrap();
let dir = tempfile::tempdir().unwrap();
let backup_path = dir.path().join("backup.db");
let path_str = backup_path.to_str().unwrap().to_string();
store.backup_to(&path_str).unwrap();
assert!(backup_path.exists());
let backup = SqliteThingStore::open(&backup_path).unwrap();
let obj = backup.get_object("test", "1").unwrap();
assert!(obj.is_some());
assert_eq!(obj.unwrap().key.id, "1");
}
#[test]
fn allows_newer_schema_versions() {
let file = NamedTempFile::new().unwrap();
let connection = Connection::open(file.path()).unwrap();
connection
.execute_batch(
r"
CREATE TABLE thingd_schema_migrations (
version INTEGER PRIMARY KEY,
name TEXT NOT NULL,
applied_at TEXT NOT NULL
);
INSERT INTO thingd_schema_migrations (version, name, applied_at)
VALUES (999, 'future', strftime('%Y-%m-%dT%H:%M:%fZ', 'now'));
",
)
.unwrap();
let store = SqliteThingStore::open(file.path());
assert!(store.is_ok());
}
#[test]
fn stores_objects_across_reopen() {
let file = NamedTempFile::new().unwrap();
{
let mut store = SqliteThingStore::open(file.path()).unwrap();
let object = store
.put_object(MemoryObject::new(
"decisions",
"sqlite-backend",
"{\"text\":\"Use SQLite\"}",
))
.unwrap();
assert_eq!(object.version, 1);
}
let store = SqliteThingStore::open(file.path()).unwrap();
let object = store
.get_object("decisions", "sqlite-backend")
.unwrap()
.unwrap();
assert_eq!(object.body, "{\"text\":\"Use SQLite\"}");
assert_eq!(object.version, 1);
}
#[test]
fn increments_object_versions() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
let first = store
.put_object(MemoryObject::new("decisions", "versioned", "{}"))
.unwrap();
let second = store
.put_object(MemoryObject::new("decisions", "versioned", "{\"v\":2}"))
.unwrap();
assert_eq!(first.version, 1);
assert_eq!(second.version, 2);
}
#[test]
fn lists_objects_with_optional_collection_filter() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
store
.put_object(MemoryObject::new("decisions", "sqlite-backend", "{}"))
.unwrap();
store
.put_object(MemoryObject::new("notes", "agent-guide", "{}"))
.unwrap();
let filtered = store
.list_objects(
Some(&["decisions".to_string()]),
&ListObjectsOptions::default(),
)
.unwrap();
assert_eq!(
store
.list_objects(None, &ListObjectsOptions::default())
.unwrap()
.len(),
2
);
assert_eq!(filtered.len(), 1);
assert_eq!(filtered[0].key.collection, "decisions");
}
#[test]
fn stores_events_across_reopen() {
let file = NamedTempFile::new().unwrap();
{
let mut store = SqliteThingStore::open(file.path()).unwrap();
let event = store
.append_event(MemoryEvent::new(
"project:thingd",
"decision.made",
"Use SQLite first",
))
.unwrap();
assert_eq!(event.sequence, 1);
}
let store = SqliteThingStore::open(file.path()).unwrap();
let events = store
.list_events(Some("project:thingd"), ListEventsOptions::default())
.unwrap();
assert_eq!(events.len(), 1);
assert_eq!(events[0].event_type, "decision.made");
assert_eq!(events[0].sequence, 1);
}
#[test]
fn stores_queue_jobs_across_reopen() {
let file = NamedTempFile::new().unwrap();
{
let mut store = SqliteThingStore::open(file.path()).unwrap();
let job = store
.push_job(QueueJob::new("embed", "job-1", "{\"doc\":\"doc-1\"}", 3))
.unwrap();
assert_eq!(job.status, QueueJobStatus::Ready);
}
let store = SqliteThingStore::open(file.path()).unwrap();
let jobs = store.list_jobs("embed").unwrap();
assert_eq!(jobs.len(), 1);
assert_eq!(jobs[0].id, "job-1");
assert_eq!(jobs[0].status, QueueJobStatus::Ready);
}
#[test]
fn returns_existing_queue_job_for_duplicate_push() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
let first = store
.push_job(QueueJob::new("embed", "job-1", "{\"doc\":\"doc-1\"}", 3))
.unwrap();
let second = store
.push_job(QueueJob::new("embed", "job-1", "{\"doc\":\"doc-2\"}", 3))
.unwrap();
assert_eq!(first.body, "{\"doc\":\"doc-1\"}");
assert_eq!(second.body, first.body);
assert_eq!(store.list_jobs("embed").unwrap().len(), 1);
}
#[test]
fn claims_and_acks_queue_jobs() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
store
.push_job(QueueJob::new("embed", "job-1", "{\"doc\":\"doc-1\"}", 3))
.unwrap();
let claimed = store.claim_job("embed").unwrap().unwrap();
let acked = store.ack_job("embed", "job-1").unwrap().unwrap();
assert_eq!(claimed.status, QueueJobStatus::Leased);
assert_eq!(claimed.attempts, 1);
assert!(claimed.leased_at_ms.is_some());
assert!(claimed.lease_expires_at_ms.is_some());
assert_eq!(acked.status, QueueJobStatus::Completed);
assert!(acked.completed_at_ms.is_some());
assert!(store.claim_job("embed").unwrap().is_none());
}
#[test]
fn nacks_queue_jobs_to_retry_then_dead_letter() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
store
.push_job(QueueJob::new("embed", "job-1", "{\"doc\":\"doc-1\"}", 2))
.unwrap();
store.claim_job("embed").unwrap().unwrap();
let retried = store.nack_job("embed", "job-1").unwrap().unwrap();
assert_eq!(retried.status, QueueJobStatus::Ready);
assert_eq!(retried.attempts, 1);
store.claim_job("embed").unwrap().unwrap();
let dead = store.nack_job("embed", "job-1").unwrap().unwrap();
assert_eq!(dead.status, QueueJobStatus::Dead);
assert_eq!(dead.attempts, 2);
assert!(dead.dead_at_ms.is_some());
assert_eq!(store.list_dead_jobs("embed").unwrap().len(), 1);
}
#[test]
fn does_not_claim_delayed_queue_jobs_before_available() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
store
.push_job(QueueJob::new("embed", "job-1", "{\"doc\":\"doc-1\"}", 3).delay_by_ms(60_000))
.unwrap();
assert!(store.claim_job("embed").unwrap().is_none());
}
#[test]
fn reclaims_queue_jobs_after_lease_expiration() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
store
.push_job(QueueJob::new("embed", "job-1", "{\"doc\":\"doc-1\"}", 3))
.unwrap();
let first = store
.claim_job_with_options("embed", QueueClaimOptions::new(0))
.unwrap()
.unwrap();
let second = store.claim_job("embed").unwrap().unwrap();
assert_eq!(first.status, QueueJobStatus::Leased);
assert_eq!(second.status, QueueJobStatus::Leased);
assert_eq!(second.attempts, 2);
}
#[test]
fn nacks_queue_jobs_with_retry_delay() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
store
.push_job(QueueJob::new("embed", "job-1", "{\"doc\":\"doc-1\"}", 3))
.unwrap();
store.claim_job("embed").unwrap().unwrap();
let retried = store
.nack_job_with_options("embed", "job-1", QueueNackOptions::new(60_000))
.unwrap()
.unwrap();
assert_eq!(retried.status, QueueJobStatus::Ready);
assert!(store.claim_job("embed").unwrap().is_none());
}
#[test]
fn persists_completed_queue_jobs_across_reopen() {
let file = NamedTempFile::new().unwrap();
{
let mut store = SqliteThingStore::open(file.path()).unwrap();
store
.push_job(QueueJob::new("embed", "job-1", "{\"doc\":\"doc-1\"}", 3))
.unwrap();
store.claim_job("embed").unwrap().unwrap();
store.ack_job("embed", "job-1").unwrap().unwrap();
}
let store = SqliteThingStore::open(file.path()).unwrap();
let jobs = store.list_jobs("embed").unwrap();
assert_eq!(jobs.len(), 1);
assert_eq!(jobs[0].status, QueueJobStatus::Completed);
assert_eq!(jobs[0].attempts, 1);
}
#[test]
fn test_fts5_search_indexing_and_stemming() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
store
.put_object(MemoryObject::new(
"decisions",
"choice-1",
"{\"text\":\"I choose this implementation plan because it has great benefits.\", \"status\":\"active\", \"priority\":1}",
))
.unwrap();
store
.put_object(MemoryObject::new(
"decisions",
"choice-2",
"{\"text\":\"He chooses that plan.\", \"status\":\"draft\", \"priority\":2}",
))
.unwrap();
let results = store
.search("implementation", crate::SearchOptions::default())
.unwrap();
assert_eq!(results.len(), 1);
assert_eq!(results[0].id, "choice-1");
let results_stem = store
.search("choosing", crate::SearchOptions::default())
.unwrap();
assert_eq!(results_stem.len(), 2);
let options_col = crate::SearchOptions {
collections: Some(vec!["unrelated_col".to_string()]),
..Default::default()
};
let results_col = store.search("choose", options_col).unwrap();
assert_eq!(results_col.len(), 0);
let options_filter = crate::SearchOptions {
filter: Some(serde_json::json!({"status": "active"})),
..Default::default()
};
let results_filter = store.search("choose", options_filter).unwrap();
assert_eq!(results_filter.len(), 1);
assert_eq!(results_filter[0].id, "choice-1");
store.delete_object("decisions", "choice-1").unwrap();
let results_after_del = store
.search("choose", crate::SearchOptions::default())
.unwrap();
assert_eq!(results_after_del.len(), 1);
assert_eq!(results_after_del[0].id, "choice-2");
}
#[test]
fn search_consistent_after_batch_delete() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
store
.put_object(MemoryObject::new(
"col",
"a",
r#"{"label":"target","name":"alpha"}"#,
))
.unwrap();
store
.put_object(MemoryObject::new(
"col",
"b",
r#"{"label":"target","name":"bravo"}"#,
))
.unwrap();
store
.put_object(MemoryObject::new(
"col",
"c",
r#"{"label":"target","name":"charlie"}"#,
))
.unwrap();
let results = store
.search("target", crate::SearchOptions::default())
.unwrap();
assert_eq!(results.len(), 3);
store
.delete_objects_batch(&[("col".into(), "a".into()), ("col".into(), "b".into())])
.unwrap();
let results = store
.search("target", crate::SearchOptions::default())
.unwrap();
assert_eq!(results.len(), 1);
assert_eq!(results[0].id, "c");
}
#[test]
fn search_does_not_return_orphaned_fts_entries() {
let store = SqliteThingStore::open_in_memory().unwrap();
store
.connection
.execute(
"INSERT INTO search_index (collection, id, kind, text) VALUES (?1, ?2, 'object', ?3)",
rusqlite::params!["orphan", "ghost", "this object does not exist"],
)
.unwrap();
let results = store
.search("object", crate::SearchOptions::default())
.unwrap();
assert_eq!(results.len(), 0);
}
#[test]
fn get_and_search_consistent_after_delete() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
store
.put_object(MemoryObject::new("col", "id-1", r#"{"name":"alice"}"#))
.unwrap();
assert!(store.get_object("col", "id-1").unwrap().is_some());
let results = store
.search("alice", crate::SearchOptions::default())
.unwrap();
assert_eq!(results.len(), 1);
store.delete_object("col", "id-1").unwrap();
assert!(store.get_object("col", "id-1").unwrap().is_none());
let results = store
.search("alice", crate::SearchOptions::default())
.unwrap();
assert_eq!(results.len(), 0);
}
#[test]
fn counts_objects_correctly_after_deletions() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
assert_eq!(store.count_objects().unwrap(), 0);
store
.put_object(MemoryObject::new("col1", "a", "{}"))
.unwrap();
store
.put_object(MemoryObject::new("col1", "b", "{}"))
.unwrap();
store
.put_object(MemoryObject::new("col2", "c", "{}"))
.unwrap();
assert_eq!(store.count_objects().unwrap(), 3);
store.delete_object("col1", "a").unwrap();
assert_eq!(store.count_objects().unwrap(), 2);
store.delete_object("col1", "b").unwrap();
assert_eq!(store.count_objects().unwrap(), 1);
store.delete_object("col2", "c").unwrap();
assert_eq!(store.count_objects().unwrap(), 0);
}
#[test]
fn counts_events_correctly() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
assert_eq!(store.count_events().unwrap(), 0);
store
.append_event(MemoryEvent::new("test", "a", ""))
.unwrap();
store
.append_event(MemoryEvent::new("test", "b", ""))
.unwrap();
assert_eq!(store.count_events().unwrap(), 2);
}
#[test]
fn deletes_last_event_from_stream() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
store
.append_event(MemoryEvent::new("match:1", "turn.recorded", "{}"))
.unwrap();
store
.append_event(MemoryEvent::new("match:1", "turn.recorded", "{}"))
.unwrap();
store
.append_event(MemoryEvent::new("match:2", "turn.recorded", "{}"))
.unwrap();
let deleted = store.delete_last_event("match:1").unwrap().unwrap();
assert_eq!(deleted.sequence, 2);
let remaining = store
.list_events(Some("match:1"), ListEventsOptions::default())
.unwrap();
assert_eq!(remaining.len(), 1);
assert_eq!(remaining[0].sequence, 1);
let match2 = store
.list_events(Some("match:2"), ListEventsOptions::default())
.unwrap();
assert_eq!(match2.len(), 1);
}
#[test]
fn returns_none_when_delete_last_event_on_empty_stream() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
assert!(store.delete_last_event("nonexistent").unwrap().is_none());
}
#[test]
fn deletes_stream_and_returns_count() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
store
.append_event(MemoryEvent::new("match:1", "turn.recorded", "{}"))
.unwrap();
store
.append_event(MemoryEvent::new("match:1", "turn.recorded", "{}"))
.unwrap();
store
.append_event(MemoryEvent::new("match:2", "turn.recorded", "{}"))
.unwrap();
let count = store.delete_stream("match:1").unwrap();
assert_eq!(count, 2);
let remaining = store
.list_events(Some("match:1"), ListEventsOptions::default())
.unwrap();
assert_eq!(remaining.len(), 0);
let match2 = store
.list_events(Some("match:2"), ListEventsOptions::default())
.unwrap();
assert_eq!(match2.len(), 1);
}
#[test]
fn returns_zero_for_delete_stream_on_empty_stream() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
assert_eq!(store.delete_stream("nonexistent").unwrap(), 0);
}
#[test]
fn counts_jobs_correctly() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
assert_eq!(store.count_active_jobs().unwrap(), 0);
assert_eq!(store.count_dead_jobs().unwrap(), 0);
store
.push_job(QueueJob::new("work", "j1", "p1", 3))
.unwrap();
store
.push_job(QueueJob::new("work", "j2", "p2", 3))
.unwrap();
store
.push_job(QueueJob::new("other", "j3", "p3", 1))
.unwrap();
assert_eq!(store.count_active_jobs().unwrap(), 3);
store.claim_job("other").unwrap();
store.nack_job("other", "j3").unwrap();
assert_eq!(store.count_dead_jobs().unwrap(), 1);
assert_eq!(store.count_active_jobs().unwrap(), 2);
}
#[test]
fn lists_collections_streams_and_queues() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
assert!(store.list_collections().unwrap().is_empty());
assert!(store.list_streams().unwrap().is_empty());
assert!(store.list_queues().unwrap().is_empty());
store
.put_object(MemoryObject::new("col-a", "x", "{}"))
.unwrap();
store
.put_object(MemoryObject::new("col-b", "y", "{}"))
.unwrap();
store
.put_object(MemoryObject::new("col-a", "z", "{}"))
.unwrap();
let collections = store.list_collections().unwrap();
assert_eq!(collections, vec!["col-a", "col-b"]);
store
.append_event(MemoryEvent::new("s1", "t", "e1"))
.unwrap();
store
.append_event(MemoryEvent::new("s2", "t", "e2"))
.unwrap();
let streams = store.list_streams().unwrap();
assert_eq!(streams, vec!["s1", "s2"]);
store
.push_job(QueueJob::new("work", "j1", "p1", 3))
.unwrap();
store
.push_job(QueueJob::new("jobs", "j2", "p2", 3))
.unwrap();
let queues = store.list_queues().unwrap();
assert_eq!(queues, vec!["jobs", "work"]);
}
#[test]
fn search_respects_filter_and_limit() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
store
.put_object(MemoryObject::new(
"docs",
"a",
r#"{"text":"hello world","tag":"greeting"}"#,
))
.unwrap();
store
.put_object(MemoryObject::new(
"docs",
"b",
r#"{"text":"hello there","tag":"greeting"}"#,
))
.unwrap();
store
.put_object(MemoryObject::new(
"docs",
"c",
r#"{"text":"goodbye world","tag":"farewell"}"#,
))
.unwrap();
let all = store.search("world", SearchOptions::default()).unwrap();
assert_eq!(all.len(), 2);
let limited = store
.search(
"world",
SearchOptions {
limit: Some(1),
..Default::default()
},
)
.unwrap();
assert_eq!(limited.len(), 1);
let filtered = store
.search(
"hello",
SearchOptions {
collections: Some(vec!["docs".into()]),
..Default::default()
},
)
.unwrap();
assert_eq!(filtered.len(), 2);
}
#[test]
fn list_objects_filter_returns_matching_objects() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
store
.put_object(MemoryObject::new("w", "a", r#"{"color":"red","size":1}"#))
.unwrap();
store
.put_object(MemoryObject::new("w", "b", r#"{"color":"blue","size":2}"#))
.unwrap();
store
.put_object(MemoryObject::new("w", "c", r#"{"color":"red","size":3}"#))
.unwrap();
let opts = ListObjectsOptions {
filter: vec![("color".into(), serde_json::json!("red"))],
..Default::default()
};
let results = store.list_objects(Some(&["w".to_string()]), &opts).unwrap();
assert_eq!(results.len(), 2);
assert!(results.iter().all(|o| o.body.contains("\"red\"")));
}
#[test]
fn list_objects_filter_no_match_returns_empty() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
store
.put_object(MemoryObject::new("w", "a", r#"{"color":"red"}"#))
.unwrap();
let opts = ListObjectsOptions {
filter: vec![("color".into(), serde_json::json!("green"))],
..Default::default()
};
let results = store.list_objects(Some(&["w".to_string()]), &opts).unwrap();
assert!(results.is_empty());
}
#[test]
fn list_objects_limit_truncates_results() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
for i in 0..5u32 {
store
.put_object(MemoryObject::new("col", format!("id-{i}"), "{}"))
.unwrap();
}
let opts = ListObjectsOptions {
limit: Some(3),
..Default::default()
};
let results = store
.list_objects(Some(&["col".to_string()]), &opts)
.unwrap();
assert_eq!(results.len(), 3);
}
#[test]
fn list_objects_offset_skips_results() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
for i in 0..5u32 {
store
.put_object(MemoryObject::new("col", format!("id-{i}"), "{}"))
.unwrap();
}
let opts = ListObjectsOptions {
offset: Some(3),
..Default::default()
};
let results = store
.list_objects(Some(&["col".to_string()]), &opts)
.unwrap();
assert_eq!(results.len(), 2);
}
#[test]
fn list_objects_filter_and_limit_combined() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
for i in 0..4u32 {
store
.put_object(MemoryObject::new(
"col",
format!("id-{i}"),
r#"{"status":"active"}"#,
))
.unwrap();
}
store
.put_object(MemoryObject::new("col", "id-4", r#"{"status":"inactive"}"#))
.unwrap();
let opts = ListObjectsOptions {
filter: vec![("status".into(), serde_json::json!("active"))],
limit: Some(2),
..Default::default()
};
let results = store
.list_objects(Some(&["col".to_string()]), &opts)
.unwrap();
assert_eq!(results.len(), 2);
assert!(results.iter().all(|o| o.body.contains("active")));
}
#[test]
fn list_objects_numeric_filter() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
store
.put_object(MemoryObject::new("items", "a", r#"{"score":10,"tag":"x"}"#))
.unwrap();
store
.put_object(MemoryObject::new("items", "b", r#"{"score":20,"tag":"x"}"#))
.unwrap();
store
.put_object(MemoryObject::new("items", "c", r#"{"score":10,"tag":"y"}"#))
.unwrap();
let opts = ListObjectsOptions {
filter: vec![("score".into(), serde_json::json!(10))],
..Default::default()
};
let results = store
.list_objects(Some(&["items".to_string()]), &opts)
.unwrap();
assert_eq!(results.len(), 2);
}
#[test]
fn append_event_returning_sets_sequence_and_timestamp() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
let first = store
.append_event(MemoryEvent::new("s", "ev.first", r#"{"x":1}"#))
.unwrap();
let second = store
.append_event(MemoryEvent::new("s", "ev.second", r#"{"x":2}"#))
.unwrap();
assert_eq!(first.sequence, 1);
assert_eq!(second.sequence, 2);
assert!(
!first.created_at.is_empty(),
"created_at must be set by RETURNING"
);
assert!(
!second.created_at.is_empty(),
"created_at must be set by RETURNING"
);
}
#[test]
fn append_event_sequence_monotonically_increases_across_streams() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
let a = store
.append_event(MemoryEvent::new("stream-a", "t", "{}"))
.unwrap();
let b = store
.append_event(MemoryEvent::new("stream-b", "t", "{}"))
.unwrap();
let c = store
.append_event(MemoryEvent::new("stream-a", "t", "{}"))
.unwrap();
assert_eq!(a.sequence, 1);
assert_eq!(b.sequence, 2);
assert_eq!(c.sequence, 3);
}
#[test]
fn cas_succeeds_on_matching_version() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
let stored = store
.put_object(MemoryObject::new("col", "id", r#"{"v":1}"#))
.unwrap();
assert_eq!(stored.version, 1);
let opts = crate::PutObjectOptions {
expected_version: Some(1),
..Default::default()
};
let updated = store
.put_object_with_options(MemoryObject::new("col", "id", r#"{"v":2}"#), opts)
.unwrap();
assert_eq!(updated.version, 2);
}
#[test]
fn cas_fails_on_version_mismatch() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
store
.put_object(MemoryObject::new("col", "id", r#"{"v":1}"#))
.unwrap();
let opts = crate::PutObjectOptions {
expected_version: Some(42),
..Default::default()
};
let err = store
.put_object_with_options(MemoryObject::new("col", "id", r#"{"v":2}"#), opts)
.unwrap_err();
assert!(matches!(err, crate::ThingdError::Conflict(_)));
}
#[test]
fn cas_fails_on_nonexistent_object() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
let opts = crate::PutObjectOptions {
expected_version: Some(1),
..Default::default()
};
let err = store
.put_object_with_options(MemoryObject::new("col", "id", r#"{"v":1}"#), opts)
.unwrap_err();
assert!(matches!(err, crate::ThingdError::Conflict(_)));
}
#[test]
fn cas_none_skips_check() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
let stored = store
.put_object_with_options(
MemoryObject::new("col", "id", r#"{"v":1}"#),
crate::PutObjectOptions::default(),
)
.unwrap();
assert_eq!(stored.version, 1);
}
#[test]
fn put_objects_batch_atomicity() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
store
.put_object(MemoryObject::new("col", "a", r#"{"name":"old-a"}"#))
.unwrap();
store
.put_object(MemoryObject::new("col", "b", r#"{"name":"old-b"}"#))
.unwrap();
let results = store
.put_objects_batch(vec![
MemoryObject::new("col", "a", r#"{"name":"new-a"}"#),
MemoryObject::new("col", "c", r#"{"name":"new-c"}"#),
])
.unwrap();
assert_eq!(results.len(), 2);
assert_eq!(results[0].version, 2); assert_eq!(results[1].version, 1);
assert_eq!(store.count_objects().unwrap(), 3);
}
#[test]
fn delete_objects_batch_atomicity() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
for c in ["a", "b", "c"] {
store
.put_object(MemoryObject::new("col", c, r#"{"name":"x"}"#))
.unwrap();
}
assert_eq!(store.count_objects().unwrap(), 3);
let deleted = store
.delete_objects_batch(&[("col".into(), "a".into()), ("col".into(), "b".into())])
.unwrap();
assert_eq!(deleted, 2);
assert_eq!(store.count_objects().unwrap(), 1);
assert!(store.get_object("col", "a").unwrap().is_none());
assert!(store.get_object("col", "c").unwrap().is_some());
}
#[test]
fn batch_operations_empty_input() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
let results = store.put_objects_batch(vec![]).unwrap();
assert!(results.is_empty());
let deleted = store.delete_objects_batch(&[]).unwrap();
assert_eq!(deleted, 0);
}
#[test]
fn event_idempotency_returns_existing_event() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
let mut event = MemoryEvent::new("stream", "test", r#"{"key":"val"}"#);
event.idempotency_key = "idem-1".to_string();
let first = store.append_event(event.clone()).unwrap();
assert_eq!(first.sequence, 1);
let second = store.append_event(event).unwrap();
assert_eq!(second.sequence, first.sequence);
assert_eq!(second.body, first.body);
}
#[test]
fn event_idempotency_different_keys_are_distinct() {
let mut store = SqliteThingStore::open_in_memory().unwrap();
let mut event_a = MemoryEvent::new("stream", "test", r#"{"key":"a"}"#);
event_a.idempotency_key = "idem-a".to_string();
let mut event_b = MemoryEvent::new("stream", "test", r#"{"key":"b"}"#);
event_b.idempotency_key = "idem-b".to_string();
let first = store.append_event(event_a).unwrap();
let second = store.append_event(event_b).unwrap();
assert_eq!(first.sequence, 1);
assert_eq!(second.sequence, 2);
}
}