use std::path::PathBuf;
use std::sync::Arc;
use async_trait::async_trait;
use rusqlite::{params, TransactionBehavior};
use r2d2::Pool;
use r2d2_sqlite::SqliteConnectionManager;
use crate::api::{CommitMode, EventStore, PersistedEvent, Snapshot};
pub struct SqliteEventStore {
_db_path: PathBuf,
pool: Pool<SqliteConnectionManager>,
commit_mode: CommitMode,
}
impl SqliteEventStore {
#[cfg_attr(feature = "dev-tracing", tracing::instrument(skip(db_path), fields(
crate_name = "persistence",
commit_mode = ?commit_mode
)))]
pub fn open(
db_path: impl Into<PathBuf>,
commit_mode: CommitMode,
) -> anyhow::Result<Arc<Self>> {
let db_path = db_path.into();
if let Some(parent) = db_path.parent() {
if !parent.as_os_str().is_empty() {
std::fs::create_dir_all(parent)?;
}
}
let manager = SqliteConnectionManager::file(&db_path);
let pool = Pool::new(manager)?;
pool.get()?.execute_batch(
r#"
CREATE TABLE IF NOT EXISTS events (
lsn INTEGER PRIMARY KEY AUTOINCREMENT,
tr_id TEXT NOT NULL,
doc_id TEXT NOT NULL,
ts INTEGER NOT NULL,
actor TEXT,
idempotency_key TEXT NOT NULL UNIQUE,
meta TEXT NOT NULL,
payload BLOB NOT NULL,
checksum INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS ix_events_doc_lsn ON events(doc_id, lsn);
CREATE TABLE IF NOT EXISTS snapshots (
doc_id TEXT NOT NULL,
upto_lsn INTEGER NOT NULL,
created_at INTEGER NOT NULL,
state_blob BLOB NOT NULL,
version INTEGER NOT NULL,
PRIMARY KEY (doc_id, upto_lsn)
);
CREATE INDEX IF NOT EXISTS ix_snapshots_doc_created ON snapshots(doc_id, created_at DESC);
"#,
)?;
Ok(Arc::new(Self { _db_path: db_path, pool, commit_mode }))
}
}
#[async_trait]
impl EventStore for SqliteEventStore {
#[cfg_attr(feature = "dev-tracing", tracing::instrument(skip(self, ev), fields(
crate_name = "persistence",
tr_id = %ev.tr_id,
doc_id = %ev.doc_id
)))]
async fn append(
&self,
ev: PersistedEvent,
) -> anyhow::Result<i64> {
let mut conn = self.pool.get()?;
let tx =
conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
tx.execute(
"INSERT INTO events (tr_id, doc_id, ts, actor, idempotency_key, meta, payload, checksum) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
params![
ev.tr_id.to_string(), ev.doc_id,
{ ev.ts },
ev.actor,
ev.idempotency_key,
ev.meta.to_string(),
ev.payload,
ev.checksum as i64
],
)?;
tx.commit()?;
let lsn = conn.last_insert_rowid();
if let CommitMode::SyncDurable = self.commit_mode {
conn.pragma_update(None, "wal_checkpoint", "TRUNCATE").ok();
}
Ok(lsn)
}
#[cfg_attr(feature = "dev-tracing", tracing::instrument(skip(self, evs), fields(
crate_name = "persistence",
batch_size = evs.len()
)))]
async fn append_batch(
&self,
evs: Vec<PersistedEvent>,
) -> anyhow::Result<i64> {
if evs.is_empty() {
return Ok(0);
}
let mut conn = self.pool.get()?;
let tx =
conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
{
let mut stmt = tx.prepare("INSERT INTO events (tr_id, doc_id, ts, actor, idempotency_key, meta, payload, checksum) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)")?;
for ev in evs.into_iter() {
stmt.execute(params![
ev.tr_id.to_string(), ev.doc_id,
{ ev.ts },
ev.actor,
ev.idempotency_key,
ev.meta.to_string(),
ev.payload,
ev.checksum as i64
])?;
}
}
tx.commit()?;
let lsn = conn.last_insert_rowid();
Ok(lsn)
}
#[cfg_attr(feature = "dev-tracing", tracing::instrument(skip(self), fields(
crate_name = "persistence",
doc_id = %doc_id,
from_lsn = from_lsn,
limit = limit
)))]
async fn load_since(
&self,
doc_id: &str,
from_lsn: i64,
limit: u32,
) -> anyhow::Result<Vec<PersistedEvent>> {
let conn = self.pool.get()?;
let mut stmt = conn.prepare("SELECT lsn, tr_id, doc_id, ts, actor, idempotency_key, meta, payload, checksum FROM events WHERE doc_id = ?1 AND lsn > ?2 ORDER BY lsn ASC LIMIT ?3")?;
let rows =
stmt.query_map(params![doc_id, from_lsn, limit as i64], |row| {
let meta_str: String = row.get(6)?;
let meta: serde_json::Value = serde_json::from_str(&meta_str)
.unwrap_or(serde_json::json!({}));
let tr_id_str: String = row.get(1)?;
let tr_id = uuid::Uuid::parse_str(&tr_id_str).map_err(|e| {
rusqlite::Error::FromSqlConversionFailure(
1,
rusqlite::types::Type::Text,
Box::new(e),
)
})?;
Ok(PersistedEvent {
lsn: row.get(0)?,
tr_id, doc_id: row.get(2)?,
ts: row.get(3)?,
actor: row.get(4).ok(),
idempotency_key: row.get(5)?,
meta,
payload: row.get(7)?,
checksum: row.get::<_, i64>(8)? as u32,
})
})?;
Ok(rows.filter_map(|r| r.ok()).collect())
}
#[cfg_attr(feature = "dev-tracing", tracing::instrument(skip(self), fields(
crate_name = "persistence",
doc_id = %doc_id
)))]
async fn latest_snapshot(
&self,
doc_id: &str,
) -> anyhow::Result<Option<Snapshot>> {
let conn = self.pool.get()?;
let mut stmt = conn.prepare("SELECT doc_id, upto_lsn, created_at, state_blob, version FROM snapshots WHERE doc_id = ?1 ORDER BY created_at DESC LIMIT 1")?;
let mut rows = stmt.query(params![doc_id])?;
if let Some(row) = rows.next()? {
Ok(Some(Snapshot {
doc_id: row.get(0)?,
upto_lsn: row.get(1)?,
created_at: row.get(2)?,
state_blob: row.get(3)?,
version: row.get(4)?,
}))
} else {
Ok(None)
}
}
#[cfg_attr(feature = "dev-tracing", tracing::instrument(skip(self, snap), fields(
crate_name = "persistence",
doc_id = %snap.doc_id,
upto_lsn = snap.upto_lsn
)))]
async fn write_snapshot(
&self,
snap: Snapshot,
) -> anyhow::Result<()> {
let mut conn = self.pool.get()?;
let tx =
conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
tx.execute(
"INSERT OR REPLACE INTO snapshots (doc_id, upto_lsn, created_at, state_blob, version) VALUES (?1, ?2, ?3, ?4, ?5)",
params![snap.doc_id, snap.upto_lsn, snap.created_at, snap.state_blob, snap.version],
)?;
tx.commit()?;
Ok(())
}
async fn compact(
&self,
doc_id: &str,
) -> anyhow::Result<()> {
let conn = self.pool.get()?;
let upto: Option<i64> = conn.query_row(
"SELECT upto_lsn FROM snapshots WHERE doc_id = ?1 ORDER BY upto_lsn DESC LIMIT 1",
params![doc_id],
|r| r.get(0),
).ok();
if let Some(upto_lsn) = upto {
conn.execute(
"DELETE FROM events WHERE doc_id = ?1 AND lsn <= ?2",
params![doc_id, upto_lsn],
)?;
}
Ok(())
}
}