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::retention::{RetentionPolicy, SharedArchiveSink, Tombstone};
11use super::signing::SessionSigner;
12use harn_vm::redact::RedactionPolicy;
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#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
85pub struct ListFilter {
86    #[serde(default)]
87    pub tenant_id: Option<String>,
88    #[serde(default)]
89    pub persona: Option<String>,
90    #[serde(default)]
91    pub status: Option<SessionStatus>,
92    #[serde(default)]
93    pub tag: Option<String>,
94    /// Inclusive lower bound on `created_at_ms`.
95    #[serde(default)]
96    pub created_after_ms: Option<i64>,
97    /// Inclusive upper bound on `created_at_ms`.
98    #[serde(default)]
99    pub created_before_ms: Option<i64>,
100    #[serde(default)]
101    pub limit: Option<usize>,
102    #[serde(default)]
103    pub cursor: Option<String>,
104}
105
106#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
107pub struct ReadRange {
108    /// Inclusive lower bound; `None` means start from the genesis event.
109    #[serde(default)]
110    pub from_event_id: Option<EventId>,
111    /// Inclusive upper bound; `None` means up to the latest event.
112    #[serde(default)]
113    pub to_event_id: Option<EventId>,
114    /// Maximum number of events to return. Capped at
115    /// [`MAX_READ_BATCH`] by the store; callers iterate by advancing
116    /// `from_event_id` on the returned cursor.
117    #[serde(default)]
118    pub limit: Option<usize>,
119}
120
121/// Page of events plus the cursor needed to continue reading.
122#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
123pub struct EventPage {
124    pub events: Vec<StoredEvent>,
125    /// Inclusive `from_event_id` to pass into the next read to resume.
126    /// `None` when the requested range was fully drained.
127    pub next_cursor: Option<EventId>,
128}
129
130#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
131pub struct SnapshotId(pub String);
132
133#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
134pub struct Snapshot {
135    pub id: SnapshotId,
136    pub session: SessionMeta,
137    pub events: Vec<StoredEvent>,
138    pub captured_at_ms: i64,
139    pub captured_at: String,
140}
141
142#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
143pub struct VerifyReport {
144    pub session_id: SessionId,
145    pub chain_root_hash: String,
146    pub event_count: usize,
147    pub signed_event_count: usize,
148    pub failures: Vec<VerifyFailure>,
149}
150
151#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
152pub struct VerifyFailure {
153    pub event_id: EventId,
154    pub reason: String,
155}
156
157/// Errors returned by every backend.
158#[derive(Clone, Debug, PartialEq, Eq)]
159pub enum StoreError {
160    NotFound(String),
161    AlreadyExists(String),
162    Conflict(String),
163    InvalidInput(String),
164    Tenant(String),
165    Backend(String),
166}
167
168impl std::fmt::Display for StoreError {
169    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
170        match self {
171            Self::NotFound(message) => write!(f, "not found: {message}"),
172            Self::AlreadyExists(message) => write!(f, "already exists: {message}"),
173            Self::Conflict(message) => write!(f, "conflict: {message}"),
174            Self::InvalidInput(message) => write!(f, "invalid input: {message}"),
175            Self::Tenant(message) => write!(f, "tenant: {message}"),
176            Self::Backend(message) => write!(f, "backend error: {message}"),
177        }
178    }
179}
180
181impl std::error::Error for StoreError {}
182
183pub type StoreResult<T> = Result<T, StoreError>;
184
185/// Soft cap on a single page of events to keep response sizes bounded.
186/// Memory + sqlite backends apply this; callers iterate via cursors.
187pub const MAX_READ_BATCH: usize = 1_000;
188
189/// Optional processors a host can plug in. They run inline on append
190/// and finalisation; backends call these hooks from the `SessionStore`
191/// mutation points.
192#[derive(Default, Clone)]
193pub struct StoreHooks {
194    /// Applied to event payloads, headers, and tags before persistence.
195    pub redaction: Option<RedactionPolicy>,
196    /// If set, every event is signed at append time. Without a signer
197    /// only the `Receipt` event minted by [`SessionStore::close`] is
198    /// signed (which is enough to verify the chain end-to-end).
199    pub event_signer: Option<SessionSigner>,
200    /// Required to mint receipts on `close`. Without this, `close`
201    /// still finalises the chain root hash but emits an unsigned
202    /// `Receipt` event.
203    pub receipt_signer: Option<SessionSigner>,
204    /// Default retention policy applied to new sessions when their
205    /// meta does not override it.
206    pub retention: RetentionPolicy,
207    /// Optional durable archive destination. The default
208    /// [`SessionStore::sweep_retention`] writes archived sessions and
209    /// tombstones here before the rows leave primary storage; see
210    /// [`super::retention::ArchiveSink`].
211    pub archive_sink: Option<SharedArchiveSink>,
212}
213
214#[async_trait]
215pub trait SessionStore: Send + Sync {
216    /// Plug-in processors configured for this store. The default
217    /// [`Self::sweep_retention`] reads `hooks.archive_sink` so the
218    /// retention loop can hand archived sessions to durable storage
219    /// without callers wiring the sink explicitly.
220    fn hooks(&self) -> &StoreHooks;
221
222    async fn create(&self, request: CreateSession) -> StoreResult<SessionMeta>;
223    async fn describe(&self, session_id: &str) -> StoreResult<SessionMeta>;
224    async fn list(&self, filter: ListFilter) -> StoreResult<Vec<SessionMeta>>;
225    async fn append(&self, session_id: &str, event: AppendEvent) -> StoreResult<StoredEvent>;
226    async fn read(&self, session_id: &str, range: ReadRange) -> StoreResult<EventPage>;
227    async fn fork(
228        &self,
229        session_id: &str,
230        at_event_id: EventId,
231        child_id: Option<SessionId>,
232    ) -> StoreResult<ForkResult>;
233    async fn truncate(&self, session_id: &str, at_event_id: EventId)
234        -> StoreResult<TruncateResult>;
235    async fn snapshot(&self, session_id: &str) -> StoreResult<Snapshot>;
236    async fn replay(&self, snapshot_id: &SnapshotId) -> StoreResult<Snapshot>;
237    async fn close(&self, session_id: &str) -> StoreResult<StoredEvent>;
238    async fn soft_delete(&self, session_id: &str) -> StoreResult<SessionMeta>;
239    async fn hard_delete(&self, session_id: &str) -> StoreResult<()>;
240    async fn verify(&self, session_id: &str) -> StoreResult<VerifyReport>;
241
242    /// Sweep retention. Backends with native scheduling can override
243    /// to skip the default loop; the default sweeps all sessions
244    /// against the configured [`RetentionPolicy`] and routes archived
245    /// sessions + tombstones through `hooks().archive_sink` when set.
246    async fn sweep_retention(
247        &self,
248        policy: &RetentionPolicy,
249        now_ms: i64,
250    ) -> StoreResult<SweepReport> {
251        use tracing::Instrument as _;
252        let span = tracing::info_span!(
253            "harn.session.sweep_retention",
254            harn.session.sweep.archive_sink_configured = self.hooks().archive_sink.is_some(),
255            harn.session.sweep.archived = tracing::field::Empty,
256            harn.session.sweep.soft_deleted = tracing::field::Empty,
257            harn.session.sweep.hard_deleted = tracing::field::Empty,
258        );
259        let span_for_record = span.clone();
260        let sink = self.hooks().archive_sink.clone();
261        let result = async move {
262            let mut report = SweepReport::default();
263            let sessions = self.list(ListFilter::default()).await?;
264            for session in sessions {
265                if policy.should_hard_delete(&session, now_ms) {
266                    if let Some(sink) = sink.as_ref() {
267                        let tombstone = Tombstone {
268                            session_id: session.id.clone(),
269                            tenant_id: session.tenant_id.clone(),
270                            deleted_at_ms: now_ms,
271                            deleted_at: super::event::ms_to_rfc3339(now_ms),
272                            final_chain_root_hash: session.chain_root_hash.clone(),
273                            final_event_id: session.last_event_id,
274                        };
275                        sink.tombstone(&tombstone).await?;
276                        report.tombstoned += 1;
277                    }
278                    self.hard_delete(&session.id).await?;
279                    report.hard_deleted += 1;
280                } else if policy.should_soft_delete(&session, now_ms) {
281                    if policy.should_archive(&session, now_ms) {
282                        if let Some(sink) = sink.as_ref() {
283                            let events = read_all_events(self, &session.id).await?;
284                            sink.archive(&session, &events).await?;
285                            report.archived += 1;
286                        }
287                    }
288                    self.soft_delete(&session.id).await?;
289                    report.soft_deleted += 1;
290                }
291            }
292            Ok::<_, StoreError>(report)
293        }
294        .instrument(span)
295        .await?;
296        span_for_record.record("harn.session.sweep.archived", result.archived as i64);
297        span_for_record.record(
298            "harn.session.sweep.soft_deleted",
299            result.soft_deleted as i64,
300        );
301        span_for_record.record(
302            "harn.session.sweep.hard_deleted",
303            result.hard_deleted as i64,
304        );
305        Ok(result)
306    }
307}
308
309/// Drain every event for a session via repeated paginated reads. Used
310/// by [`SessionStore::sweep_retention`] when shipping a session to the
311/// [`super::retention::ArchiveSink`].
312async fn read_all_events<S: SessionStore + ?Sized>(
313    store: &S,
314    session_id: &str,
315) -> StoreResult<Vec<StoredEvent>> {
316    let mut all = Vec::new();
317    let mut cursor: Option<EventId> = None;
318    loop {
319        let page = store
320            .read(
321                session_id,
322                ReadRange {
323                    from_event_id: cursor,
324                    to_event_id: None,
325                    limit: Some(MAX_READ_BATCH),
326                },
327            )
328            .await?;
329        let next = page.next_cursor;
330        all.extend(page.events);
331        match next {
332            Some(next_cursor) => cursor = Some(next_cursor),
333            None => break,
334        }
335    }
336    Ok(all)
337}
338
339#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
340pub struct SweepReport {
341    pub soft_deleted: usize,
342    pub hard_deleted: usize,
343    /// Sessions handed to [`super::retention::ArchiveSink::archive`]
344    /// because they crossed `min_age_before_archive_seconds`.
345    pub archived: usize,
346    /// Hard-deleted sessions whose final state was emitted as a
347    /// [`super::retention::Tombstone`] to the archive sink.
348    pub tombstoned: usize,
349}
350
351/// Dyn-dispatch alias so adapters can keep one `Arc<dyn SessionStore>`
352/// in their state without naming the concrete backend everywhere.
353pub type SharedSessionStore = Arc<dyn SessionStore>;
harn_session_store/store.rs

harn_session_store/
store.rs
