kernex_memory/

memory_store.rs

#![cfg_attr(test, allow(clippy::unwrap_used, clippy::expect_used))]

//! Public trait surface for the SQLite-backed memory store.
//!
//! Mirrors the inherent method surface that downstream consumers
//! (`kernex-runtime` composition, the sister-repo binary's REPL, future
//! CLI/HTTP/MCP) call today, plus three new soft-delete methods on the
//! `facts` table. Hard-delete inherent methods (`delete_fact`,
//! `delete_facts`) stay on `Store` for emergency cleanup tooling and are
//! deliberately NOT on the trait so the default consumer path uses
//! recoverable soft-delete.
//!
//! `Runtime::store_handle()` returns `Arc<dyn MemoryStore>` so a binary
//! consumer can share the runtime's composed `Store` instance instead of
//! opening a second SQLite connection against the same database file.

use std::sync::Arc;
use std::time::SystemTime;

use async_trait::async_trait;

use crate::error::MemoryError;
use crate::observation::{Observation, ObservationType, SaveEntry};
use crate::store::{DueTask, Store, TaskRunRecord, UsageSummary};
use crate::types::{HistoryRow, MessageRow};

/// Public trait surface over [`Store`].
///
/// Returned from `kernex-runtime::Runtime::store_handle()` as
/// `Arc<dyn MemoryStore>`. Consumers should prefer this trait over the
/// concrete `Store` type so future schema changes do not ripple into call
/// sites.
#[async_trait]
pub trait MemoryStore: Send + Sync {
    // --- conversations / messages ---

    /// Mark the active conversation for `(channel, sender_id, project)` as
    /// closed. Returns `true` if a row transitioned from active to closed.
    async fn close_current_conversation(
        &self,
        channel: &str,
        sender_id: &str,
        project: &str,
    ) -> Result<bool, MemoryError>;

    /// Aggregate counters:
    /// `(conversation_count, message_count, observation_count, fact_count)`.
    ///
    /// **Breaking change (kernex-memory 0.8.0):** prior versions returned
    /// a 3-tuple `(conversation, message, fact)`. The observation count
    /// joins the tuple at position 2; consumers must destructure four
    /// elements after the bump. Soft-deleted observations and facts are
    /// excluded; conversations and messages have no soft-delete column
    /// today and are counted whole.
    async fn get_memory_stats(&self, sender_id: &str) -> Result<(i64, i64, i64, i64), MemoryError>;

    /// On-disk byte size of the SQLite database file.
    async fn db_size(&self) -> Result<u64, MemoryError>;

    /// Aggregate token usage across all sessions.
    async fn get_total_usage(&self) -> Result<UsageSummary, MemoryError>;

    /// Recent closed-conversation summaries for a given channel + sender,
    /// newest first, capped at `limit`.
    async fn get_history(
        &self,
        channel: &str,
        sender_id: &str,
        limit: i64,
    ) -> Result<Vec<HistoryRow>, MemoryError>;

    /// FTS5 full-text search over user messages, excluding the live
    /// conversation. When `since` is `Some`, only rows with
    /// `timestamp >= since` are returned and `limit` applies after the
    /// recency filter.
    async fn search_messages(
        &self,
        query: &str,
        exclude_conversation_id: &str,
        sender_id: &str,
        limit: i64,
        since: Option<SystemTime>,
    ) -> Result<Vec<MessageRow>, MemoryError>;

    /// Fetch a single message row by its UUID. Returns `None` when the
    /// id is missing.
    async fn get_message_by_id(&self, id: &str) -> Result<Option<MessageRow>, MemoryError>;

    // --- facts (write paths plus soft-only delete on the trait) ---

    /// Upsert a fact for `(sender_id, key)`. If the row was previously
    /// soft-deleted, this clears `deleted_at` so the value is visible
    /// again to default-filtered reads.
    async fn store_fact(&self, sender_id: &str, key: &str, value: &str) -> Result<(), MemoryError>;

    /// Read a single active fact by `(sender_id, key)`. Returns `None` if
    /// the row is soft-deleted, missing, or never existed.
    async fn get_fact(&self, sender_id: &str, key: &str) -> Result<Option<String>, MemoryError>;

    /// Active (not soft-deleted) facts for `sender_id`.
    async fn get_facts(&self, sender_id: &str) -> Result<Vec<(String, String)>, MemoryError>;

    /// Soft-delete a single fact by setting its `deleted_at` timestamp.
    /// Returns `true` if a row transitioned from active to deleted; `false`
    /// if the row was already deleted, missing, or never existed.
    async fn soft_delete_fact(&self, sender_id: &str, key: &str) -> Result<bool, MemoryError>;

    /// Soft-delete multiple facts. With `Some(key)`, soft-deletes that
    /// specific key. With `None`, soft-deletes every active fact for the
    /// sender. Returns the count of rows that transitioned from active to
    /// deleted.
    async fn soft_delete_facts(
        &self,
        sender_id: &str,
        key: Option<&str>,
    ) -> Result<u64, MemoryError>;

    /// Read soft-deleted facts (debug / recovery helper). Returns
    /// `(key, value, deleted_at)` rows for `sender_id`.
    async fn list_soft_deleted_facts(
        &self,
        sender_id: &str,
    ) -> Result<Vec<(String, String, String)>, MemoryError>;

    // --- observations (typed write surface introduced in 0.8.0) ---

    /// Persist a typed observation and return the saved row. Generates
    /// a fresh UUIDv4 id; sets `created_at == updated_at == now`. The
    /// DB enforces `length(title) > 0` and the seven-value `type` CHECK
    /// constraint; violations surface as `MemoryError::Sqlite`.
    async fn save_observation(&self, entry: SaveEntry) -> Result<Observation, MemoryError>;

    /// Fetch an active observation by id. Returns `None` when the id is
    /// missing OR the row is soft-deleted (CC-9 invariant). Mirrors the
    /// `get_message_by_id` shape introduced in 0.7.0.
    async fn get_observation_by_id(&self, id: &str) -> Result<Option<Observation>, MemoryError>;

    /// FTS5 search across observation title + structured fields.
    /// Optional `since` filters by `created_at >=`; optional `kind`
    /// narrows to a single type. Soft-deleted rows never appear.
    async fn search_observations(
        &self,
        query: &str,
        sender_id: &str,
        limit: i64,
        since: Option<SystemTime>,
        kind: Option<ObservationType>,
    ) -> Result<Vec<Observation>, MemoryError>;

    /// Soft-delete an observation by id. Returns `Ok(true)` on
    /// transition from active to deleted; `Ok(false)` when the row was
    /// already deleted, missing, or never existed (matches the
    /// `soft_delete_fact` contract).
    async fn soft_delete_observation(&self, id: &str) -> Result<bool, MemoryError>;

    /// Read soft-deleted observations for a sender. Recovery helper;
    /// surfaced on the trait so future tooling can offer an "undelete"
    /// command without dropping back to the inherent `Store`.
    async fn list_soft_deleted_observations(
        &self,
        sender_id: &str,
    ) -> Result<Vec<Observation>, MemoryError>;

    // --- scheduled tasks ---

    /// Insert a new scheduled task. Returns the new task id.
    #[allow(clippy::too_many_arguments)]
    async fn create_task(
        &self,
        channel: &str,
        sender_id: &str,
        reply_target: &str,
        description: &str,
        due_at: &str,
        repeat: Option<&str>,
        task_type: &str,
        project: &str,
    ) -> Result<String, MemoryError>;

    /// Pending tasks for `sender_id` as raw `(id, description, due_at,
    /// repeat, task_type, project)` rows, ordered by `due_at` ascending.
    async fn get_tasks_for_sender(
        &self,
        sender_id: &str,
    ) -> Result<Vec<(String, String, String, Option<String>, String, String)>, MemoryError>;

    /// Mark a task as completed. With `Some("daily")` / `Some("weekly")` /
    /// etc., reschedules the next occurrence; with `None` or `Some("once")`,
    /// the task transitions to a terminal status.
    async fn complete_task(&self, id: &str, repeat: Option<&str>) -> Result<(), MemoryError>;

    /// Record a task failure. Increments retry counter; transitions to a
    /// terminal failed status when retries exhaust. Returns `true` if the
    /// task transitioned to a terminal state.
    async fn fail_task(&self, id: &str, error: &str, max_retries: u32)
        -> Result<bool, MemoryError>;

    /// Cancel a pending task whose id starts with `id_prefix`, scoped to
    /// `sender_id`. Returns `true` if a row was cancelled.
    async fn cancel_task(&self, id_prefix: &str, sender_id: &str) -> Result<bool, MemoryError>;

    /// All pending tasks whose `due_at` is in the past.
    async fn get_due_tasks(&self) -> Result<Vec<DueTask>, MemoryError>;

    /// Atomically claim every due task (status 'pending' -> 'claimed' in a
    /// single statement) and return the claimed rows. When several pollers
    /// share a store, each due task is handed to exactly one of them; a
    /// claim abandoned by a dead claimer becomes reclaimable after a
    /// timeout. The claim is released by `complete_task` (recurring tasks
    /// return to 'pending' at the next due time) or `fail_task` (retries
    /// return to 'pending').
    async fn claim_due_tasks(&self) -> Result<Vec<DueTask>, MemoryError>;

    /// Record one execution of a scheduled task. `status` must be
    /// `"completed"` or `"failed"`. Returns the new run id.
    #[allow(clippy::too_many_arguments)]
    async fn record_task_run(
        &self,
        task_id: &str,
        started_at: &str,
        status: &str,
        result: Option<&str>,
        error: Option<&str>,
        tokens_used: Option<u64>,
    ) -> Result<String, MemoryError>;

    /// Recorded runs for tasks whose id starts with `task_id_prefix`,
    /// newest first, capped at `limit`.
    async fn list_task_runs(
        &self,
        task_id_prefix: &str,
        limit: u32,
    ) -> Result<Vec<TaskRunRecord>, MemoryError>;
}

#[async_trait]
impl MemoryStore for Store {
    async fn close_current_conversation(
        &self,
        channel: &str,
        sender_id: &str,
        project: &str,
    ) -> Result<bool, MemoryError> {
        Store::close_current_conversation(self, channel, sender_id, project).await
    }

    async fn get_memory_stats(&self, sender_id: &str) -> Result<(i64, i64, i64, i64), MemoryError> {
        Store::get_memory_stats(self, sender_id).await
    }

    async fn db_size(&self) -> Result<u64, MemoryError> {
        Store::db_size(self).await
    }

    async fn get_total_usage(&self) -> Result<UsageSummary, MemoryError> {
        Store::get_total_usage(self).await
    }

    async fn get_history(
        &self,
        channel: &str,
        sender_id: &str,
        limit: i64,
    ) -> Result<Vec<HistoryRow>, MemoryError> {
        Store::get_history(self, channel, sender_id, limit).await
    }

    async fn search_messages(
        &self,
        query: &str,
        exclude_conversation_id: &str,
        sender_id: &str,
        limit: i64,
        since: Option<SystemTime>,
    ) -> Result<Vec<MessageRow>, MemoryError> {
        Store::search_messages(
            self,
            query,
            exclude_conversation_id,
            sender_id,
            limit,
            since,
        )
        .await
    }

    async fn get_message_by_id(&self, id: &str) -> Result<Option<MessageRow>, MemoryError> {
        Store::get_message_by_id(self, id).await
    }

    async fn store_fact(&self, sender_id: &str, key: &str, value: &str) -> Result<(), MemoryError> {
        Store::store_fact(self, sender_id, key, value).await
    }

    async fn get_fact(&self, sender_id: &str, key: &str) -> Result<Option<String>, MemoryError> {
        Store::get_fact(self, sender_id, key).await
    }

    async fn get_facts(&self, sender_id: &str) -> Result<Vec<(String, String)>, MemoryError> {
        Store::get_facts(self, sender_id).await
    }

    async fn soft_delete_fact(&self, sender_id: &str, key: &str) -> Result<bool, MemoryError> {
        Store::soft_delete_fact(self, sender_id, key).await
    }

    async fn soft_delete_facts(
        &self,
        sender_id: &str,
        key: Option<&str>,
    ) -> Result<u64, MemoryError> {
        Store::soft_delete_facts(self, sender_id, key).await
    }

    async fn list_soft_deleted_facts(
        &self,
        sender_id: &str,
    ) -> Result<Vec<(String, String, String)>, MemoryError> {
        Store::list_soft_deleted_facts(self, sender_id).await
    }

    async fn save_observation(&self, entry: SaveEntry) -> Result<Observation, MemoryError> {
        Store::save_observation(self, entry).await
    }

    async fn get_observation_by_id(&self, id: &str) -> Result<Option<Observation>, MemoryError> {
        Store::get_observation_by_id(self, id).await
    }

    async fn search_observations(
        &self,
        query: &str,
        sender_id: &str,
        limit: i64,
        since: Option<SystemTime>,
        kind: Option<ObservationType>,
    ) -> Result<Vec<Observation>, MemoryError> {
        Store::search_observations(self, query, sender_id, limit, since, kind).await
    }

    async fn soft_delete_observation(&self, id: &str) -> Result<bool, MemoryError> {
        Store::soft_delete_observation(self, id).await
    }

    async fn list_soft_deleted_observations(
        &self,
        sender_id: &str,
    ) -> Result<Vec<Observation>, MemoryError> {
        Store::list_soft_deleted_observations(self, sender_id).await
    }

    async fn create_task(
        &self,
        channel: &str,
        sender_id: &str,
        reply_target: &str,
        description: &str,
        due_at: &str,
        repeat: Option<&str>,
        task_type: &str,
        project: &str,
    ) -> Result<String, MemoryError> {
        Store::create_task(
            self,
            channel,
            sender_id,
            reply_target,
            description,
            due_at,
            repeat,
            task_type,
            project,
        )
        .await
    }

    async fn get_tasks_for_sender(
        &self,
        sender_id: &str,
    ) -> Result<Vec<(String, String, String, Option<String>, String, String)>, MemoryError> {
        Store::get_tasks_for_sender(self, sender_id).await
    }

    async fn complete_task(&self, id: &str, repeat: Option<&str>) -> Result<(), MemoryError> {
        Store::complete_task(self, id, repeat).await
    }

    async fn fail_task(
        &self,
        id: &str,
        error: &str,
        max_retries: u32,
    ) -> Result<bool, MemoryError> {
        Store::fail_task(self, id, error, max_retries).await
    }

    async fn cancel_task(&self, id_prefix: &str, sender_id: &str) -> Result<bool, MemoryError> {
        Store::cancel_task(self, id_prefix, sender_id).await
    }

    async fn get_due_tasks(&self) -> Result<Vec<DueTask>, MemoryError> {
        Store::get_due_tasks(self).await
    }

    async fn claim_due_tasks(&self) -> Result<Vec<DueTask>, MemoryError> {
        Store::claim_due_tasks(self).await
    }

    async fn record_task_run(
        &self,
        task_id: &str,
        started_at: &str,
        status: &str,
        result: Option<&str>,
        error: Option<&str>,
        tokens_used: Option<u64>,
    ) -> Result<String, MemoryError> {
        Store::record_task_run(
            self,
            task_id,
            started_at,
            status,
            result,
            error,
            tokens_used,
        )
        .await
    }

    async fn list_task_runs(
        &self,
        task_id_prefix: &str,
        limit: u32,
    ) -> Result<Vec<TaskRunRecord>, MemoryError> {
        Store::list_task_runs(self, task_id_prefix, limit).await
    }
}

/// Expose a [`Store`] through the [`MemoryStore`] trait surface.
///
/// `Store` already implements `Clone` (its `SqlitePool` is internally
/// reference-counted); cloning here shares the same connection pool.
pub fn into_handle(store: Store) -> Arc<dyn MemoryStore> {
    Arc::new(store)
}