harn_session_store/
store.rs

1//! `SessionStore` trait + the shared types every backend speaks.
2
3use std::collections::BTreeMap;
4use std::sync::Arc;
5
6use async_trait::async_trait;
7use serde::{Deserialize, Serialize};
8
9use super::event::{AppendEvent, EventId, StoredEvent};
10use super::redaction::SharedEventRedactor;
11use super::retention::{RetentionPolicy, SharedArchiveSink, Tombstone};
12use super::signing::SessionSigner;
13
14pub type SessionId = String;
15
16/// Result of a fork operation.
17#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
18pub struct ForkResult {
19    pub child_session_id: SessionId,
20    pub forked_from_event_id: EventId,
21    pub copied_event_count: usize,
22}
23
24#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
25pub struct TruncateResult {
26    pub kept_event_count: usize,
27    pub removed_event_count: usize,
28    pub new_tip_event_id: Option<EventId>,
29}
30
31/// Per-session retention/lifecycle state.
32#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
33#[serde(rename_all = "snake_case")]
34pub enum SessionStatus {
35    Open,
36    Closed,
37    /// Soft-deleted; will become `HardDeleted` once the grace window
38    /// elapses (enforced by retention sweeps; see [`crate::retention`]).
39    SoftDeleted,
40    HardDeleted,
41}
42
43#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
44pub struct SessionMeta {
45    pub id: SessionId,
46    pub tenant_id: Option<String>,
47    pub persona: Option<String>,
48    pub parent_session_id: Option<SessionId>,
49    pub created_at_ms: i64,
50    pub created_at: String,
51    /// Last `append`/`fork` timestamp; refreshed on every mutation.
52    pub updated_at_ms: i64,
53    pub updated_at: String,
54    pub status: SessionStatus,
55    pub event_count: usize,
56    pub last_event_id: Option<EventId>,
57    pub chain_root_hash: Option<String>,
58    pub closed_at_ms: Option<i64>,
59    pub closed_at: Option<String>,
60    pub soft_deleted_at_ms: Option<i64>,
61    pub ttl_seconds: Option<u64>,
62    pub tags: Vec<String>,
63    pub attributes: BTreeMap<String, serde_json::Value>,
64}
65
66#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
67pub struct CreateSession {
68    #[serde(default)]
69    pub id: Option<SessionId>,
70    #[serde(default)]
71    pub tenant_id: Option<String>,
72    #[serde(default)]
73    pub persona: Option<String>,
74    #[serde(default)]
75    pub parent_session_id: Option<SessionId>,
76    #[serde(default)]
77    pub ttl_seconds: Option<u64>,
78    #[serde(default)]
79    pub tags: Vec<String>,
80    #[serde(default)]
81    pub attributes: BTreeMap<String, serde_json::Value>,
82}
83
84/// One atomic, idempotent import into a new canonical session.
85///
86/// `source_id` names the external source independently of the target session;
87/// its receipt survives session deletion so retired sources cannot resurrect
88/// data. Reusing a source id with a different digest is a conflict.
89#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
90pub struct ImportSession {
91    pub source_id: String,
92    pub source_digest: String,
93    pub session: CreateSession,
94    #[serde(default)]
95    pub events: Vec<AppendEvent>,
96}
97
98impl ImportSession {
99    /// Validate the backend-independent import contract.
100    pub fn validate(&self) -> StoreResult<()> {
101        if self.source_id.trim().is_empty() || self.source_digest.trim().is_empty() {
102            return Err(StoreError::InvalidInput(
103                "import source_id and source_digest must be non-empty".to_string(),
104            ));
105        }
106        if self
107            .session
108            .id
109            .as_deref()
110            .is_none_or(|session_id| session_id.trim().is_empty())
111        {
112            return Err(StoreError::InvalidInput(
113                "import session id must be explicit and non-empty".to_string(),
114            ));
115        }
116        Ok(())
117    }
118}
119
120#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
121pub struct ImportResult {
122    pub source_id: String,
123    pub source_digest: String,
124    pub session_id: SessionId,
125    pub event_count: usize,
126    /// True only for the call that committed the import.
127    pub imported: bool,
128}
129
130#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
131pub struct ListFilter {
132    #[serde(default)]
133    pub tenant_id: Option<String>,
134    #[serde(default)]
135    pub persona: Option<String>,
136    #[serde(default)]
137    pub status: Option<SessionStatus>,
138    #[serde(default)]
139    pub tag: Option<String>,
140    /// Inclusive lower bound on `created_at_ms`.
141    #[serde(default)]
142    pub created_after_ms: Option<i64>,
143    /// Inclusive upper bound on `created_at_ms`.
144    #[serde(default)]
145    pub created_before_ms: Option<i64>,
146    #[serde(default)]
147    pub limit: Option<usize>,
148    #[serde(default)]
149    pub cursor: Option<String>,
150}
151
152#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
153pub struct ReadRange {
154    /// Inclusive lower bound; `None` means start from the genesis event.
155    #[serde(default)]
156    pub from_event_id: Option<EventId>,
157    /// Inclusive upper bound; `None` means up to the latest event.
158    #[serde(default)]
159    pub to_event_id: Option<EventId>,
160    /// Maximum number of events to return. Capped at
161    /// [`MAX_READ_BATCH`] by the store; callers iterate by advancing
162    /// `from_event_id` on the returned cursor.
163    #[serde(default)]
164    pub limit: Option<usize>,
165}
166
167/// Page of events plus the cursor needed to continue reading.
168#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
169pub struct EventPage {
170    pub events: Vec<StoredEvent>,
171    /// Inclusive `from_event_id` to pass into the next read to resume.
172    /// `None` when the requested range was fully drained.
173    pub next_cursor: Option<EventId>,
174}
175
176#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
177pub struct SnapshotId(pub String);
178
179#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
180pub struct Snapshot {
181    pub id: SnapshotId,
182    pub session: SessionMeta,
183    pub events: Vec<StoredEvent>,
184    pub captured_at_ms: i64,
185    pub captured_at: String,
186}
187
188#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
189pub struct VerifyReport {
190    pub session_id: SessionId,
191    pub chain_root_hash: String,
192    pub event_count: usize,
193    pub signed_event_count: usize,
194    pub failures: Vec<VerifyFailure>,
195}
196
197#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
198pub struct VerifyFailure {
199    pub event_id: EventId,
200    pub reason: String,
201}
202
203/// Typed transient contention returned by a persistent backend.
204#[derive(Clone, Copy, Debug, PartialEq, Eq)]
205pub enum StoreContention {
206    /// Another SQLite connection owns the database write lock.
207    DatabaseBusy,
208    /// A shared-cache SQLite table lock blocks the operation.
209    DatabaseLocked,
210}
211
212impl std::fmt::Display for StoreContention {
213    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
214        match self {
215            Self::DatabaseBusy => f.write_str("database_busy"),
216            Self::DatabaseLocked => f.write_str("database_locked"),
217        }
218    }
219}
220
221/// Errors returned by every backend.
222#[derive(Clone, Debug, PartialEq, Eq)]
223pub enum StoreError {
224    NotFound(String),
225    AlreadyExists(String),
226    Conflict(String),
227    InvalidInput(String),
228    Tenant(String),
229    /// SQLite could not acquire a lock after applying its busy policy.
230    Contention {
231        /// Machine-readable lock-contention reason.
232        kind: StoreContention,
233        /// Backend diagnostic retained for operators.
234        message: String,
235    },
236    Backend(String),
237}
238
239impl std::fmt::Display for StoreError {
240    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
241        match self {
242            Self::NotFound(message) => write!(f, "not found: {message}"),
243            Self::AlreadyExists(message) => write!(f, "already exists: {message}"),
244            Self::Conflict(message) => write!(f, "conflict: {message}"),
245            Self::InvalidInput(message) => write!(f, "invalid input: {message}"),
246            Self::Tenant(message) => write!(f, "tenant: {message}"),
247            Self::Contention { kind, message } => {
248                write!(f, "retryable backend contention ({kind}): {message}")
249            }
250            Self::Backend(message) => write!(f, "backend error: {message}"),
251        }
252    }
253}
254
255impl std::error::Error for StoreError {}
256
257pub type StoreResult<T> = Result<T, StoreError>;
258
259/// Soft cap on a single page of events to keep response sizes bounded.
260/// Memory + sqlite backends apply this; callers iterate via cursors.
261pub const MAX_READ_BATCH: usize = 1_000;
262
263/// Optional processors a host can plug in. Mutation hooks run inline before
264/// persistence; redaction is also reapplied to public retrieval projections
265/// as defense in depth for older stored data.
266#[derive(Default, Clone)]
267pub struct StoreHooks {
268    /// Applied to event payloads and headers before persistence and again
269    /// when events are read, snapshotted, or replayed.
270    pub redaction: Option<SharedEventRedactor>,
271    /// If set, every event is signed at append time. Without a signer
272    /// only the `Receipt` event minted by [`SessionStore::close`] is
273    /// signed (which is enough to verify the chain end-to-end).
274    pub event_signer: Option<SessionSigner>,
275    /// Required to mint receipts on `close`. Without this, `close`
276    /// still finalises the chain root hash but emits an unsigned
277    /// `Receipt` event.
278    pub receipt_signer: Option<SessionSigner>,
279    /// Default retention policy applied to new sessions when their
280    /// meta does not override it.
281    pub retention: RetentionPolicy,
282    /// Optional durable archive destination. The default
283    /// [`SessionStore::sweep_retention`] writes archived sessions and
284    /// tombstones here before the rows leave primary storage; see
285    /// [`super::retention::ArchiveSink`].
286    pub archive_sink: Option<SharedArchiveSink>,
287}
288
289#[async_trait]
290pub trait SessionStore: Send + Sync {
291    /// Plug-in processors configured for this store. The default
292    /// [`Self::sweep_retention`] reads `hooks.archive_sink` so the
293    /// retention loop can hand archived sessions to durable storage
294    /// without callers wiring the sink explicitly.
295    fn hooks(&self) -> &StoreHooks;
296
297    async fn create(&self, request: CreateSession) -> StoreResult<SessionMeta>;
298    async fn describe(&self, session_id: &str) -> StoreResult<SessionMeta>;
299    async fn list(&self, filter: ListFilter) -> StoreResult<Vec<SessionMeta>>;
300    async fn append(&self, session_id: &str, event: AppendEvent) -> StoreResult<StoredEvent>;
301    async fn read(&self, session_id: &str, range: ReadRange) -> StoreResult<EventPage>;
302    async fn fork(
303        &self,
304        session_id: &str,
305        at_event_id: EventId,
306        child_id: Option<SessionId>,
307    ) -> StoreResult<ForkResult>;
308    async fn truncate(&self, session_id: &str, at_event_id: EventId)
309        -> StoreResult<TruncateResult>;
310    async fn snapshot(&self, session_id: &str) -> StoreResult<Snapshot>;
311    async fn replay(&self, snapshot_id: &SnapshotId) -> StoreResult<Snapshot>;
312    async fn close(&self, session_id: &str) -> StoreResult<StoredEvent>;
313    async fn soft_delete(&self, session_id: &str) -> StoreResult<SessionMeta>;
314    async fn hard_delete(&self, session_id: &str) -> StoreResult<()>;
315    async fn verify(&self, session_id: &str) -> StoreResult<VerifyReport>;
316
317    /// Sweep retention. Backends with native scheduling can override
318    /// to skip the default loop; the default sweeps all sessions
319    /// against the configured [`RetentionPolicy`] and routes archived
320    /// sessions + tombstones through `hooks().archive_sink` when set.
321    async fn sweep_retention(
322        &self,
323        policy: &RetentionPolicy,
324        now_ms: i64,
325    ) -> StoreResult<SweepReport> {
326        use tracing::Instrument as _;
327        let span = tracing::info_span!(
328            "harn.session.sweep_retention",
329            harn.session.sweep.archive_sink_configured = self.hooks().archive_sink.is_some(),
330            harn.session.sweep.archived = tracing::field::Empty,
331            harn.session.sweep.soft_deleted = tracing::field::Empty,
332            harn.session.sweep.hard_deleted = tracing::field::Empty,
333        );
334        let span_for_record = span.clone();
335        let sink = self.hooks().archive_sink.clone();
336        let result = async move {
337            let mut report = SweepReport::default();
338            let sessions = self.list(ListFilter::default()).await?;
339            for session in sessions {
340                if policy.should_hard_delete(&session, now_ms) {
341                    if let Some(sink) = sink.as_ref() {
342                        let tombstone = Tombstone {
343                            session_id: session.id.clone(),
344                            tenant_id: session.tenant_id.clone(),
345                            deleted_at_ms: now_ms,
346                            deleted_at: super::event::ms_to_rfc3339(now_ms),
347                            final_chain_root_hash: session.chain_root_hash.clone(),
348                            final_event_id: session.last_event_id,
349                        };
350                        sink.tombstone(&tombstone).await?;
351                        report.tombstoned += 1;
352                    }
353                    self.hard_delete(&session.id).await?;
354                    report.hard_deleted += 1;
355                } else if policy.should_soft_delete(&session, now_ms) {
356                    if policy.should_archive(&session, now_ms) {
357                        if let Some(sink) = sink.as_ref() {
358                            let events = read_all_events(self, &session.id).await?;
359                            sink.archive(&session, &events).await?;
360                            report.archived += 1;
361                        }
362                    }
363                    self.soft_delete(&session.id).await?;
364                    report.soft_deleted += 1;
365                }
366            }
367            Ok::<_, StoreError>(report)
368        }
369        .instrument(span)
370        .await?;
371        span_for_record.record("harn.session.sweep.archived", result.archived as i64);
372        span_for_record.record(
373            "harn.session.sweep.soft_deleted",
374            result.soft_deleted as i64,
375        );
376        span_for_record.record(
377            "harn.session.sweep.hard_deleted",
378            result.hard_deleted as i64,
379        );
380        Ok(result)
381    }
382}
383
384/// Atomic, idempotent ingestion for stores that accept external session data.
385///
386/// This remains separate from [`SessionStore`] so downstream backends do not
387/// need to implement migration semantics unless they expose import support.
388#[async_trait]
389pub trait SessionImporter: SessionStore {
390    async fn import(&self, request: ImportSession) -> StoreResult<ImportResult>;
391}
392
393/// Drain every event for a session via repeated paginated reads. Used
394/// by [`SessionStore::sweep_retention`] when shipping a session to the
395/// [`super::retention::ArchiveSink`].
396async fn read_all_events<S: SessionStore + ?Sized>(
397    store: &S,
398    session_id: &str,
399) -> StoreResult<Vec<StoredEvent>> {
400    let mut all = Vec::new();
401    let mut cursor: Option<EventId> = None;
402    loop {
403        let page = store
404            .read(
405                session_id,
406                ReadRange {
407                    from_event_id: cursor,
408                    to_event_id: None,
409                    limit: Some(MAX_READ_BATCH),
410                },
411            )
412            .await?;
413        let next = page.next_cursor;
414        all.extend(page.events);
415        match next {
416            Some(next_cursor) => cursor = Some(next_cursor),
417            None => break,
418        }
419    }
420    Ok(all)
421}
422
423#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
424pub struct SweepReport {
425    pub soft_deleted: usize,
426    pub hard_deleted: usize,
427    /// Sessions handed to [`super::retention::ArchiveSink::archive`]
428    /// because they crossed `min_age_before_archive_seconds`.
429    pub archived: usize,
430    /// Hard-deleted sessions whose final state was emitted as a
431    /// [`super::retention::Tombstone`] to the archive sink.
432    pub tombstoned: usize,
433}
434
435/// Dyn-dispatch alias so adapters can keep one `Arc<dyn SessionStore>`
436/// in their state without naming the concrete backend everywhere.
437pub type SharedSessionStore = Arc<dyn SessionStore>;
harn_session_store/store.rs

harn_session_store/
store.rs
