paigasus_helikon_sessions_sqlite/

lib.rs

//! SQLite-backed [`Session`] implementation for the Paigasus Helikon SDK.
//!
//! Stores conversation event logs in a single SQLite database. Multiple
//! sessions share one `SqlitePool` and are isolated by `session_id`. Safe
//! for concurrent writers — appends serialize through SQLite's database-level
//! write lock; the `(session_id, sequence)` primary key is the uniqueness
//! backstop.
//!
//! ## Recommended pool configuration
//!
//! ```no_run
//! use sqlx::sqlite::{
//!     SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous,
//! };
//! use std::time::Duration;
//!
//! # async fn build() -> Result<sqlx::SqlitePool, sqlx::Error> {
//! let opts = SqliteConnectOptions::new()
//!     .filename("sessions.db")
//!     .create_if_missing(true)
//!     .journal_mode(SqliteJournalMode::Wal)
//!     .synchronous(SqliteSynchronous::Normal)
//!     .busy_timeout(Duration::from_secs(30));
//! SqlitePoolOptions::new().connect_with(opts).await
//! # }
//! ```
//!
//! `synchronous = NORMAL` is the recommended pairing with WAL for
//! multi-writer workloads: commits stop fsyncing individually (the WAL is
//! synced at checkpoints instead), which multiplies write throughput under
//! contention. The trade-off is durability, not integrity — after a power
//! loss the most recent commits may be missing, but the database cannot
//! corrupt. Keep sqlx's default `FULL` if losing any acknowledged append is
//! unacceptable.
//!
//! ## Concurrent appends under contention
//!
//! Writers serialize on SQLite's single database-level write lock.
//! `busy_timeout` caps how long one `append` waits for that lock, and under
//! sustained contention the worst-placed writer waits out the *entire
//! backlog* ahead of it — so size the timeout against
//! `(concurrent writers) × (appends per writer) × (worst-case transaction
//! latency)`, not against a single transaction. An `append` that exhausts
//! the timeout fails with [`SessionError::Backend`] wrapping `SQLITE_BUSY`
//! ("database is locked"). The failure is clean: nothing from the failed
//! call is persisted, no stored data is lost or corrupted, and the session
//! remains usable (pinned by this crate's `busy_timeout` integration test).
//! 30 seconds is a sane starting point; tune upward for heavy multi-writer
//! contention.
//!
//! ## Provider-translator caveat
//!
//! The [`project`] function in `paigasus-helikon-core` renders [`Compacted`]
//! events as `Item::System`. Both shipped provider translators reshape
//! system messages — Anthropic hoists them to the top-level `system` field,
//! OpenAI concatenates them. Compaction summaries reach the model but as
//! top-level instructions, not positional cutovers.
//!
//! [`Session`]: paigasus_helikon_core::Session
//! [`SessionError::Backend`]: paigasus_helikon_core::SessionError::Backend
//! [`project`]: paigasus_helikon_core::project
//! [`Compacted`]: paigasus_helikon_core::SessionEvent::Compacted

use async_trait::async_trait;
use paigasus_helikon_core::{
    project, ConversationSnapshot, SequenceId, Session, SessionError, SessionEvent,
};
use sqlx::SqlitePool;

static MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("./migrations");

/// SQLite-backed [`Session`] implementation. One instance is one session
/// (identified by `session_id`); pools are shared across instances.
#[derive(Debug, Clone)]
pub struct SqliteSession {
    pool: SqlitePool,
    session_id: String,
}

impl SqliteSession {
    /// Run embedded migrations on `pool`. Idempotent — safe on every startup.
    ///
    /// Optional: [`SqliteSession::open`] runs migrations internally. Call
    /// this directly if you manage many sessions and want to migrate once at
    /// process start, then use [`SqliteSession::open_without_migrate`] on the
    /// hot path to skip the per-`open` round-trip to `_sqlx_migrations`.
    ///
    /// **Editing a migration file after first deploy will fail** with a
    /// checksum mismatch against the `_sqlx_migrations` table. Add a new
    /// numbered file (e.g., `0002_…`) instead of mutating an existing one.
    pub async fn migrate(pool: &SqlitePool) -> Result<(), SessionError> {
        MIGRATOR.run(pool).await.map_err(SessionError::backend)?;
        Ok(())
    }

    /// Open (or implicitly create) a session within `pool`. Runs migrations
    /// as a side effect (one round-trip to `_sqlx_migrations`). For repeated
    /// session-opens against an already-migrated pool, prefer
    /// [`SqliteSession::open_without_migrate`].
    pub async fn open(
        pool: SqlitePool,
        session_id: impl Into<String>,
    ) -> Result<Self, SessionError> {
        Self::migrate(&pool).await?;
        Ok(Self::open_without_migrate(pool, session_id))
    }

    /// Open a session without running migrations. The caller must have
    /// already invoked [`SqliteSession::migrate`] on this pool; otherwise
    /// the first [`Session::append`] fails with `SessionError::Backend`
    /// wrapping a `no such table` error.
    pub fn open_without_migrate(pool: SqlitePool, session_id: impl Into<String>) -> Self {
        Self {
            pool,
            session_id: session_id.into(),
        }
    }

    /// The `session_id` this instance reads and writes.
    pub fn session_id(&self) -> &str {
        &self.session_id
    }
}

#[async_trait]
impl Session for SqliteSession {
    async fn append(&self, events: &[SessionEvent]) -> Result<(), SessionError> {
        if events.is_empty() {
            return Ok(());
        }

        // `BEGIN IMMEDIATE` acquires SQLite's RESERVED lock up-front, so the
        // SELECT-MAX-then-INSERT sequence can't race two concurrent writers
        // into a UNIQUE-constraint collision on `(session_id, sequence)`.
        // sqlx's plain `pool.begin()` issues `BEGIN` (DEFERRED), which would
        // let two readers compute the same `next` value before either locks
        // for write. See `begin_ansi_transaction_sql` in sqlx-core.
        let mut tx = self
            .pool
            .begin_with("BEGIN IMMEDIATE")
            .await
            .map_err(SessionError::backend)?;

        // Find next sequence number for this session. COALESCE handles the
        // first-append case (MAX returns NULL on an empty result set).
        let row: (i64,) = sqlx::query_as(
            "SELECT COALESCE(MAX(sequence), -1) + 1 FROM session_events WHERE session_id = ?",
        )
        .bind(&self.session_id)
        .fetch_one(&mut *tx)
        .await
        .map_err(SessionError::backend)?;
        let start: i64 = row.0;

        for (offset, ev) in events.iter().enumerate() {
            let next = start + offset as i64;
            let (kind, ts_nanos) = event_metadata(ev);
            let payload = serde_json::to_string(ev).map_err(SessionError::backend)?;

            sqlx::query(
                "INSERT INTO session_events (session_id, sequence, ts_nanos, kind, payload) \
                 VALUES (?, ?, ?, ?, ?)",
            )
            .bind(&self.session_id)
            .bind(next)
            .bind(ts_nanos)
            .bind(kind)
            .bind(&payload)
            .execute(&mut *tx)
            .await
            .map_err(SessionError::backend)?;
        }

        tx.commit().await.map_err(SessionError::backend)?;
        Ok(())
    }

    async fn events(&self, since: Option<SequenceId>) -> Result<Vec<SessionEvent>, SessionError> {
        // `since` is exclusive ("those after"). Default to -1 so the
        // `sequence > ?` filter is a no-op when None.
        let watermark: i64 = match since {
            // `s.0` is u64; `> i64::MAX` is unreachable in practice but means "after
            // every possible sequence." Saturating to i64::MAX makes the filter
            // return empty rather than erroring — semantically correct.
            Some(s) => i64::try_from(s.0).unwrap_or(i64::MAX),
            None => -1,
        };

        let rows: Vec<(String,)> = sqlx::query_as(
            "SELECT payload FROM session_events \
             WHERE session_id = ? AND sequence > ? \
             ORDER BY sequence",
        )
        .bind(&self.session_id)
        .bind(watermark)
        .fetch_all(&self.pool)
        .await
        .map_err(SessionError::backend)?;

        rows.into_iter()
            .map(|(payload,)| {
                serde_json::from_str::<SessionEvent>(&payload).map_err(SessionError::backend)
            })
            .collect()
    }

    async fn snapshot(&self) -> Result<ConversationSnapshot, SessionError> {
        let events = self.events(None).await?;
        Ok(project(&events))
    }
}

/// Extract the `(kind, ts_nanos)` denormalized columns for an event. `kind`
/// matches the serde tag of the variant; `ts_nanos` is the timestamp in
/// i64 nanoseconds since the Unix epoch (covers ±292 years from 1970).
///
/// Delegates to [`SessionEvent::kind`] and [`SessionEvent::ts_nanos_saturating`]
/// so new variants added to core are handled automatically without requiring
/// updates here.
fn event_metadata(ev: &SessionEvent) -> (&'static str, i64) {
    (ev.kind(), ev.ts_nanos_saturating())
}