use std::path::Path;
use std::sync::Mutex;
use rusqlite::{params, Connection, OptionalExtension};
use crate::{
LocalProviderBinding, RuntimeError, RuntimeRelease, RuntimeSnapshot, RuntimeStore,
RuntimeTraceRecord,
};
const MAX_TRACE_OUTBOX_RECORDS: i64 = 1_000;
pub struct SqliteRuntimeStore {
connection: Mutex<Connection>,
}
impl SqliteRuntimeStore {
pub fn open(path: impl AsRef<Path>) -> Result<Self, RuntimeError> {
let path = path.as_ref();
if let Some(parent) = path
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
{
std::fs::create_dir_all(parent)
.map_err(|error| RuntimeError::store(error.to_string()))?;
}
let connection =
Connection::open(path).map_err(|error| RuntimeError::store(error.to_string()))?;
Self::from_connection(connection)
}
pub fn in_memory() -> Result<Self, RuntimeError> {
let connection =
Connection::open_in_memory().map_err(|error| RuntimeError::store(error.to_string()))?;
Self::from_connection(connection)
}
fn from_connection(connection: Connection) -> Result<Self, RuntimeError> {
connection
.execute_batch(
"PRAGMA foreign_keys = ON;
CREATE TABLE IF NOT EXISTS runtime_sessions (
project_id TEXT NOT NULL,
session_id TEXT NOT NULL,
revision INTEGER NOT NULL,
state_json TEXT NOT NULL,
PRIMARY KEY (project_id, session_id)
);
CREATE TABLE IF NOT EXISTS runtime_releases (
project_id TEXT NOT NULL,
version INTEGER NOT NULL,
content_hash TEXT NOT NULL,
manifest_json TEXT NOT NULL,
installed_at_ms INTEGER NOT NULL,
PRIMARY KEY (project_id, version),
UNIQUE (project_id, content_hash)
);
CREATE TABLE IF NOT EXISTS runtime_active_releases (
project_id TEXT PRIMARY KEY,
version INTEGER NOT NULL,
FOREIGN KEY (project_id, version)
REFERENCES runtime_releases(project_id, version)
);
CREATE TABLE IF NOT EXISTS runtime_provider_bindings (
project_id TEXT NOT NULL,
provider_id TEXT NOT NULL,
binding_json TEXT NOT NULL,
PRIMARY KEY (project_id, provider_id)
);
CREATE TABLE IF NOT EXISTS runtime_trace_outbox (
trace_id TEXT PRIMARY KEY,
project_id TEXT NOT NULL,
payload_json TEXT NOT NULL,
created_at_ms INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS runtime_trace_outbox_created_idx
ON runtime_trace_outbox(created_at_ms, trace_id);",
)
.map_err(|error| RuntimeError::store(error.to_string()))?;
Ok(Self {
connection: Mutex::new(connection),
})
}
fn connection(&self) -> Result<std::sync::MutexGuard<'_, Connection>, RuntimeError> {
self.connection.lock().map_err(|_| RuntimeError::Internal)
}
}
impl RuntimeStore for SqliteRuntimeStore {
fn load(
&self,
project_id: &str,
session_id: &str,
) -> Result<Option<RuntimeSnapshot>, RuntimeError> {
let connection = self.connection()?;
connection
.query_row(
"SELECT revision, state_json
FROM runtime_sessions
WHERE project_id = ?1 AND session_id = ?2",
params![project_id, session_id],
|row| {
let revision = row.get::<_, i64>(0)?;
let state_json = row.get::<_, String>(1)?;
Ok((revision, state_json))
},
)
.optional()
.map_err(|error| RuntimeError::store(error.to_string()))?
.map(|(revision, state_json)| {
Ok(RuntimeSnapshot {
revision: u64::try_from(revision)
.map_err(|error| RuntimeError::store(error.to_string()))?,
state: serde_json::from_str(&state_json)
.map_err(|error| RuntimeError::store(error.to_string()))?,
})
})
.transpose()
}
fn save(
&self,
project_id: &str,
session_id: &str,
snapshot: &RuntimeSnapshot,
) -> Result<(), RuntimeError> {
let state_json = serde_json::to_string(&snapshot.state)
.map_err(|error| RuntimeError::store(error.to_string()))?;
let revision = i64::try_from(snapshot.revision)
.map_err(|error| RuntimeError::store(error.to_string()))?;
self.connection()?
.execute(
"INSERT INTO runtime_sessions(project_id, session_id, revision, state_json)
VALUES (?1, ?2, ?3, ?4)
ON CONFLICT(project_id, session_id) DO UPDATE SET
revision = excluded.revision,
state_json = excluded.state_json",
params![project_id, session_id, revision, state_json],
)
.map_err(|error| RuntimeError::store(error.to_string()))?;
Ok(())
}
fn save_release(&self, release: &RuntimeRelease) -> Result<(), RuntimeError> {
release.validate()?;
let version = i64::try_from(release.version)
.map_err(|error| RuntimeError::store(error.to_string()))?;
let manifest_json = serde_json::to_string(&release.manifest)
.map_err(|error| RuntimeError::store(error.to_string()))?;
let installed_at_ms = i64::try_from(crate::unix_time_ms())
.map_err(|error| RuntimeError::store(error.to_string()))?;
let connection = self.connection()?;
let inserted = connection
.execute(
"INSERT OR IGNORE INTO runtime_releases(
project_id, version, content_hash, manifest_json, installed_at_ms
) VALUES (?1, ?2, ?3, ?4, ?5)",
params![
release.manifest.project_id,
version,
release.content_hash,
manifest_json,
installed_at_ms
],
)
.map_err(|error| RuntimeError::store(error.to_string()))?;
if inserted == 0 {
let existing =
load_release(&connection, &release.manifest.project_id, release.version)?;
if existing.as_ref() != Some(release) {
return Err(RuntimeError::store(
"runtime release versions are immutable".to_string(),
));
}
}
Ok(())
}
fn load_release(
&self,
project_id: &str,
version: u64,
) -> Result<Option<RuntimeRelease>, RuntimeError> {
let connection = self.connection()?;
load_release(&connection, project_id, version)
}
fn list_releases(&self, project_id: &str) -> Result<Vec<RuntimeRelease>, RuntimeError> {
let connection = self.connection()?;
let mut statement = connection
.prepare(
"SELECT version, content_hash, manifest_json
FROM runtime_releases
WHERE project_id = ?1
ORDER BY version DESC",
)
.map_err(|error| RuntimeError::store(error.to_string()))?;
let rows = statement
.query_map(params![project_id], release_from_row)
.map_err(|error| RuntimeError::store(error.to_string()))?;
rows.map(|row| row.map_err(|error| RuntimeError::store(error.to_string())))
.collect()
}
fn active_release(&self, project_id: &str) -> Result<Option<u64>, RuntimeError> {
self.connection()?
.query_row(
"SELECT version FROM runtime_active_releases WHERE project_id = ?1",
params![project_id],
|row| row.get::<_, i64>(0),
)
.optional()
.map_err(|error| RuntimeError::store(error.to_string()))?
.map(|version| {
u64::try_from(version).map_err(|error| RuntimeError::store(error.to_string()))
})
.transpose()
}
fn set_active_release(&self, project_id: &str, version: u64) -> Result<(), RuntimeError> {
let version =
i64::try_from(version).map_err(|error| RuntimeError::store(error.to_string()))?;
self.connection()?
.execute(
"INSERT INTO runtime_active_releases(project_id, version)
VALUES (?1, ?2)
ON CONFLICT(project_id) DO UPDATE SET version = excluded.version",
params![project_id, version],
)
.map_err(|error| RuntimeError::store(error.to_string()))?;
Ok(())
}
fn save_local_provider_binding(
&self,
project_id: &str,
binding: &LocalProviderBinding,
) -> Result<(), RuntimeError> {
let binding_json = serde_json::to_string(binding)
.map_err(|error| RuntimeError::store(error.to_string()))?;
self.connection()?
.execute(
"INSERT INTO runtime_provider_bindings(project_id, provider_id, binding_json)
VALUES (?1, ?2, ?3)
ON CONFLICT(project_id, provider_id) DO UPDATE SET
binding_json = excluded.binding_json",
params![project_id, binding.provider_id, binding_json],
)
.map_err(|error| RuntimeError::store(error.to_string()))?;
Ok(())
}
fn local_provider_bindings(
&self,
project_id: &str,
) -> Result<Vec<LocalProviderBinding>, RuntimeError> {
let connection = self.connection()?;
let mut statement = connection
.prepare(
"SELECT binding_json
FROM runtime_provider_bindings
WHERE project_id = ?1
ORDER BY provider_id",
)
.map_err(|error| RuntimeError::store(error.to_string()))?;
let rows = statement
.query_map(params![project_id], |row| row.get::<_, String>(0))
.map_err(|error| RuntimeError::store(error.to_string()))?;
rows.map(|row| {
let json = row.map_err(|error| RuntimeError::store(error.to_string()))?;
serde_json::from_str(&json).map_err(|error| RuntimeError::store(error.to_string()))
})
.collect()
}
fn enqueue_trace(&self, trace: &RuntimeTraceRecord) -> Result<(), RuntimeError> {
let payload_json =
serde_json::to_string(trace).map_err(|error| RuntimeError::store(error.to_string()))?;
let created_at_ms = i64::try_from(trace.created_at_ms)
.map_err(|error| RuntimeError::store(error.to_string()))?;
let mut connection = self.connection()?;
let transaction = connection
.transaction()
.map_err(|error| RuntimeError::store(error.to_string()))?;
transaction
.execute(
"INSERT OR IGNORE INTO runtime_trace_outbox(
trace_id, project_id, payload_json, created_at_ms
) VALUES (?1, ?2, ?3, ?4)",
params![trace.id, trace.project_id, payload_json, created_at_ms],
)
.map_err(|error| RuntimeError::store(error.to_string()))?;
transaction
.execute(
"DELETE FROM runtime_trace_outbox
WHERE trace_id IN (
SELECT trace_id FROM runtime_trace_outbox
ORDER BY created_at_ms DESC, trace_id DESC
LIMIT -1 OFFSET ?1
)",
params![MAX_TRACE_OUTBOX_RECORDS],
)
.map_err(|error| RuntimeError::store(error.to_string()))?;
transaction
.commit()
.map_err(|error| RuntimeError::store(error.to_string()))?;
Ok(())
}
fn pending_traces(&self, limit: usize) -> Result<Vec<RuntimeTraceRecord>, RuntimeError> {
let limit = i64::try_from(limit.min(MAX_TRACE_OUTBOX_RECORDS as usize))
.map_err(|error| RuntimeError::store(error.to_string()))?;
let connection = self.connection()?;
let mut statement = connection
.prepare(
"SELECT payload_json
FROM runtime_trace_outbox
ORDER BY created_at_ms, trace_id
LIMIT ?1",
)
.map_err(|error| RuntimeError::store(error.to_string()))?;
let rows = statement
.query_map(params![limit], |row| row.get::<_, String>(0))
.map_err(|error| RuntimeError::store(error.to_string()))?;
rows.map(|row| {
let json = row.map_err(|error| RuntimeError::store(error.to_string()))?;
serde_json::from_str(&json).map_err(|error| RuntimeError::store(error.to_string()))
})
.collect()
}
fn acknowledge_traces(&self, trace_ids: &[String]) -> Result<(), RuntimeError> {
let mut connection = self.connection()?;
let transaction = connection
.transaction()
.map_err(|error| RuntimeError::store(error.to_string()))?;
for trace_id in trace_ids {
transaction
.execute(
"DELETE FROM runtime_trace_outbox WHERE trace_id = ?1",
params![trace_id],
)
.map_err(|error| RuntimeError::store(error.to_string()))?;
}
transaction
.commit()
.map_err(|error| RuntimeError::store(error.to_string()))?;
Ok(())
}
}
fn load_release(
connection: &Connection,
project_id: &str,
version: u64,
) -> Result<Option<RuntimeRelease>, RuntimeError> {
let version = i64::try_from(version).map_err(|error| RuntimeError::store(error.to_string()))?;
connection
.query_row(
"SELECT version, content_hash, manifest_json
FROM runtime_releases
WHERE project_id = ?1 AND version = ?2",
params![project_id, version],
release_from_row,
)
.optional()
.map_err(|error| RuntimeError::store(error.to_string()))
}
fn release_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<RuntimeRelease> {
let version = row.get::<_, i64>(0)?;
let content_hash = row.get::<_, String>(1)?;
let manifest_json = row.get::<_, String>(2)?;
let manifest = serde_json::from_str(&manifest_json).map_err(|error| {
rusqlite::Error::FromSqlConversionFailure(
manifest_json.len(),
rusqlite::types::Type::Text,
Box::new(error),
)
})?;
let version = u64::try_from(version).map_err(|error| {
rusqlite::Error::FromSqlConversionFailure(
8,
rusqlite::types::Type::Integer,
Box::new(error),
)
})?;
Ok(RuntimeRelease {
version,
content_hash,
manifest,
})
}
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;
use serde_json::json;
use super::*;
use crate::{
AgentDefinition, EndpointDefinition, ProviderRequirement, RuntimeManifest,
RUNTIME_MANIFEST_SCHEMA_VERSION,
};
fn release(version: u64) -> RuntimeRelease {
RuntimeRelease::new(
version,
RuntimeManifest {
schema_version: RUNTIME_MANIFEST_SCHEMA_VERSION,
project_id: "test-project".to_string(),
providers: vec![ProviderRequirement {
id: "native".to_string(),
provider_type: "native".to_string(),
capabilities: vec!["chat".to_string()],
settings: json!({}),
resources: BTreeMap::new(),
}],
agents: vec![AgentDefinition {
id: "guide".to_string(),
name: "Guide".to_string(),
provider: "native".to_string(),
capabilities: vec!["chat".to_string()],
metadata: json!({}),
}],
endpoints: vec![EndpointDefinition {
name: "guide".to_string(),
agent: "guide".to_string(),
capability: "chat".to_string(),
timeout_ms: 30_000,
}],
metadata: json!({}),
},
)
.unwrap()
}
#[test]
fn sqlite_store_persists_release_session_binding_and_trace() {
let store = SqliteRuntimeStore::in_memory().unwrap();
let release = release(1);
store.save_release(&release).unwrap();
store.set_active_release("test-project", 1).unwrap();
store
.save(
"test-project",
"player",
&RuntimeSnapshot {
revision: 2,
state: json!({ "chapter": 3 }),
},
)
.unwrap();
store
.save_local_provider_binding(
"test-project",
&LocalProviderBinding {
provider_id: "native".to_string(),
configuration: json!({ "credentialRef": "keychain:vifu/native" }),
},
)
.unwrap();
store
.enqueue_trace(&RuntimeTraceRecord {
id: "trace-1".to_string(),
project_id: "test-project".to_string(),
invocation_id: "invocation-1".to_string(),
endpoint: "guide".to_string(),
agent: Some("guide".to_string()),
provider: Some("native".to_string()),
capability: Some("chat".to_string()),
status: "completed".to_string(),
duration_ms: 4,
created_at_ms: 10,
})
.unwrap();
assert_eq!(store.active_release("test-project").unwrap(), Some(1));
assert_eq!(
store
.load("test-project", "player")
.unwrap()
.unwrap()
.revision,
2
);
assert_eq!(
store.local_provider_bindings("test-project").unwrap().len(),
1
);
assert_eq!(store.pending_traces(10).unwrap().len(), 1);
store.acknowledge_traces(&["trace-1".to_string()]).unwrap();
assert!(store.pending_traces(10).unwrap().is_empty());
}
#[test]
fn sqlite_store_rejects_mutating_an_existing_release_version() {
let store = SqliteRuntimeStore::in_memory().unwrap();
store.save_release(&release(1)).unwrap();
let mut changed = release(1);
changed.manifest.metadata = json!({ "changed": true });
changed.content_hash = changed.manifest.content_hash().unwrap();
assert!(store.save_release(&changed).is_err());
}
}