tea_session_sqlite/
store.rs

1use std::collections::{BTreeMap, HashSet};
2use std::sync::{Mutex, mpsc};
3use std::time::Duration;
4
5use rusqlite::{Connection, OptionalExtension as _};
6use tea_policy::{ActorId, GrantId, PolicyGrant};
7use tea_protocol::SessionId;
8use tea_session::{
9    AppendOutcome, AppendTransaction, ApprovalArtifactEntry, GrantJournalEntry,
10    SessionCatalogEntry, SessionName, SessionSnapshot, SessionStore, SessionStoreError,
11    SessionStoreErrorCode, SessionStoreFuture, StoredSession, apply_transaction_in_place,
12};
13use tokio::sync::oneshot;
14
15use crate::error::SqliteSessionError;
16use crate::schema::{CURRENT_SCHEMA_VERSION, ensure_schema};
17
18/// Durable `SQLite` session store implementing `SessionStore`.
19///
20/// One dedicated blocking worker owns the synchronous connection. Async callers
21/// exchange typed commands with that worker and never execute `rusqlite` on a
22/// runtime worker thread.
23#[derive(Debug)]
24pub struct SqliteSessionStore {
25    sender: mpsc::Sender<WorkerCommand>,
26    worker: Mutex<Option<std::thread::JoinHandle<()>>>,
27}
28
29impl SqliteSessionStore {
30    /// Opens or creates a store at the supplied `SQLite` path. Use `:memory:` for
31    /// an ephemeral database.
32    ///
33    /// # Errors
34    ///
35    /// Returns an error when the database cannot be opened, initialized, or validated.
36    pub fn open(path: &str) -> Result<Self, SqliteSessionError> {
37        Self::from_connection(Connection::open(path)?)
38    }
39
40    /// Creates an in-memory store (single connection, process-local).
41    ///
42    /// # Errors
43    ///
44    /// Returns an error when the database cannot be created or validated.
45    pub fn in_memory() -> Result<Self, SqliteSessionError> {
46        Self::from_connection(Connection::open_in_memory()?)
47    }
48
49    /// Returns the installed schema version.
50    #[must_use]
51    pub const fn schema_version(&self) -> u32 {
52        CURRENT_SCHEMA_VERSION
53    }
54
55    fn from_connection(mut connection: Connection) -> Result<Self, SqliteSessionError> {
56        connection.busy_timeout(Duration::from_secs(5))?;
57        connection.execute_batch("PRAGMA foreign_keys = ON;")?;
58        ensure_schema(&mut connection)?;
59        connection.execute_batch("PRAGMA journal_mode = WAL;")?;
60        let (sender, receiver) = mpsc::channel();
61        let worker = std::thread::Builder::new()
62            .name("tea-sqlite-session".to_owned())
63            .spawn(move || Worker::new(connection).run(&receiver))
64            .map_err(|error| SqliteSessionError::Sqlite(error.to_string()))?;
65        Ok(Self {
66            sender,
67            worker: Mutex::new(Some(worker)),
68        })
69    }
70}
71
72impl Drop for SqliteSessionStore {
73    fn drop(&mut self) {
74        let _ = self.sender.send(WorkerCommand::Shutdown);
75        if let Ok(worker) = self.worker.get_mut()
76            && let Some(worker) = worker.take()
77        {
78            let _ = worker.join();
79        }
80    }
81}
82
83impl tea_session::SessionCatalog for SqliteSessionStore {
84    fn list_sessions(&self) -> SessionStoreFuture<'_, Vec<SessionCatalogEntry>> {
85        let sender = self.sender.clone();
86        Box::pin(async move {
87            let (reply, receiver) = oneshot::channel();
88            send(&sender, WorkerCommand::ListSessions { reply })?;
89            receive(receiver).await
90        })
91    }
92
93    fn set_session_name(
94        &self,
95        session_id: SessionId,
96        name: Option<SessionName>,
97    ) -> SessionStoreFuture<'_, ()> {
98        let sender = self.sender.clone();
99        Box::pin(async move {
100            let (reply, receiver) = oneshot::channel();
101            send(
102                &sender,
103                WorkerCommand::SetSessionName {
104                    session_id,
105                    name,
106                    reply,
107                },
108            )?;
109            receive(receiver).await
110        })
111    }
112
113    fn session_name(&self, session_id: SessionId) -> SessionStoreFuture<'_, Option<SessionName>> {
114        let sender = self.sender.clone();
115        Box::pin(async move {
116            let (reply, receiver) = oneshot::channel();
117            send(&sender, WorkerCommand::SessionName { session_id, reply })?;
118            receive(receiver).await
119        })
120    }
121}
122
123impl SessionStore for SqliteSessionStore {
124    fn load(&self, session_id: SessionId) -> SessionStoreFuture<'_, SessionSnapshot> {
125        let sender = self.sender.clone();
126        Box::pin(async move {
127            let (reply, receiver) = oneshot::channel();
128            send(&sender, WorkerCommand::Load { session_id, reply })?;
129            receive(receiver).await
130        })
131    }
132
133    fn append(&self, transaction: AppendTransaction) -> SessionStoreFuture<'_, AppendOutcome> {
134        let sender = self.sender.clone();
135        Box::pin(async move {
136            let (reply, receiver) = oneshot::channel();
137            send(&sender, WorkerCommand::Append { transaction, reply })?;
138            receive(receiver).await
139        })
140    }
141
142    fn active_grants_for_actor(
143        &self,
144        actor_id: ActorId,
145    ) -> SessionStoreFuture<'_, Vec<PolicyGrant>> {
146        let sender = self.sender.clone();
147        Box::pin(async move {
148            let (reply, receiver) = oneshot::channel();
149            send(&sender, WorkerCommand::ActiveGrants { actor_id, reply })?;
150            receive(receiver).await
151        })
152    }
153}
154
155fn send(
156    sender: &mpsc::Sender<WorkerCommand>,
157    command: WorkerCommand,
158) -> Result<(), SessionStoreError> {
159    sender.send(command).map_err(|_| worker_unavailable())
160}
161
162async fn receive<T>(
163    receiver: oneshot::Receiver<Result<T, SessionStoreError>>,
164) -> Result<T, SessionStoreError> {
165    receiver.await.map_err(|_| worker_unavailable())?
166}
167
168fn worker_unavailable() -> SessionStoreError {
169    SessionStoreError::new(
170        SessionStoreErrorCode::StorageUnavailable,
171        "sqlite session worker is unavailable",
172    )
173}
174
175enum WorkerCommand {
176    Load {
177        session_id: SessionId,
178        reply: oneshot::Sender<Result<SessionSnapshot, SessionStoreError>>,
179    },
180    Append {
181        transaction: AppendTransaction,
182        reply: oneshot::Sender<Result<AppendOutcome, SessionStoreError>>,
183    },
184    ActiveGrants {
185        actor_id: ActorId,
186        reply: oneshot::Sender<Result<Vec<PolicyGrant>, SessionStoreError>>,
187    },
188    ListSessions {
189        reply: oneshot::Sender<Result<Vec<SessionCatalogEntry>, SessionStoreError>>,
190    },
191    SetSessionName {
192        session_id: SessionId,
193        name: Option<SessionName>,
194        reply: oneshot::Sender<Result<(), SessionStoreError>>,
195    },
196    SessionName {
197        session_id: SessionId,
198        reply: oneshot::Sender<Result<Option<SessionName>, SessionStoreError>>,
199    },
200    Shutdown,
201}
202
203struct Worker {
204    connection: Connection,
205    sessions: BTreeMap<SessionId, StoredSession>,
206}
207
208impl Worker {
209    fn new(connection: Connection) -> Self {
210        Self {
211            connection,
212            sessions: BTreeMap::new(),
213        }
214    }
215
216    fn run(mut self, receiver: &mpsc::Receiver<WorkerCommand>) {
217        while let Ok(command) = receiver.recv() {
218            match command {
219                WorkerCommand::Load { session_id, reply } => {
220                    let _ = reply.send(self.load(session_id));
221                }
222                WorkerCommand::Append { transaction, reply } => {
223                    let _ = reply.send(self.append(&transaction));
224                }
225                WorkerCommand::ActiveGrants { actor_id, reply } => {
226                    let _ = reply.send(self.active_grants(&actor_id));
227                }
228                WorkerCommand::ListSessions { reply } => {
229                    let _ = reply.send(self.list_sessions());
230                }
231                WorkerCommand::SetSessionName {
232                    session_id,
233                    name,
234                    reply,
235                } => {
236                    let _ = reply.send(self.set_session_name(session_id, name.as_ref()));
237                }
238                WorkerCommand::SessionName { session_id, reply } => {
239                    let _ = reply.send(self.session_name(session_id));
240                }
241                WorkerCommand::Shutdown => break,
242            }
243        }
244    }
245
246    fn load(&mut self, session_id: SessionId) -> Result<SessionSnapshot, SessionStoreError> {
247        self.ensure_loaded(session_id)?;
248        self.sessions
249            .get(&session_id)
250            .map(StoredSession::snapshot)
251            .ok_or_else(session_not_found)
252    }
253
254    fn ensure_loaded(&mut self, session_id: SessionId) -> Result<(), SessionStoreError> {
255        if !self.sessions.contains_key(&session_id)
256            && let Some(stored) = load_stored(&self.connection, session_id)?
257        {
258            self.sessions.insert(session_id, stored);
259        }
260        Ok(())
261    }
262
263    fn append(
264        &mut self,
265        transaction: &AppendTransaction,
266    ) -> Result<AppendOutcome, SessionStoreError> {
267        let session_id = transaction.session_id();
268        self.ensure_loaded(session_id)?;
269        let existed = self.sessions.contains_key(&session_id);
270        let mut stored = self.sessions.remove(&session_id).unwrap_or_default();
271        let previous_grant_count = stored.grant_journal.len();
272        let tx = self
273            .connection
274            .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
275            .map_err(SqliteSessionError::from)
276            .map_err(SessionStoreError::from)?;
277        if let Err(error) = validate_persisted_expectation(&tx, transaction) {
278            self.sessions.insert(session_id, stored);
279            return Err(error);
280        }
281        let known_grant_ids = persisted_grant_ids(&tx, transaction.grant_entries())?;
282        let outcome =
283            match apply_transaction_in_place(transaction, &mut stored, existed, |grant_id| {
284                known_grant_ids.contains(&grant_id)
285            }) {
286                Ok(outcome) => outcome,
287                Err(error) => {
288                    self.sessions.insert(session_id, stored);
289                    return Err(error);
290                }
291            };
292        persist_delta(&tx, transaction, previous_grant_count)
293            .and_then(|()| persist_active_grants(&tx, session_id, transaction.grant_entries()))?;
294        tx.commit()
295            .map_err(SqliteSessionError::from)
296            .map_err(SessionStoreError::from)?;
297        self.sessions.insert(session_id, stored);
298        Ok(outcome)
299    }
300
301    fn active_grants(&self, actor_id: &ActorId) -> Result<Vec<PolicyGrant>, SessionStoreError> {
302        let mut statement = self
303            .connection
304            .prepare(
305                "SELECT grant_json FROM active_grants
306                 WHERE actor_id = ? AND revoked = 0 ORDER BY grant_id",
307            )
308            .map_err(SqliteSessionError::from)?;
309        let rows = statement
310            .query_map(rusqlite::params![actor_id.to_string()], |row| {
311                row.get::<_, String>(0)
312            })
313            .map_err(SqliteSessionError::from)?;
314        let mut grants = Vec::new();
315        for row in rows {
316            grants.push(decode_json(&row.map_err(SqliteSessionError::from)?)?);
317        }
318        Ok(grants)
319    }
320
321    fn list_sessions(&mut self) -> Result<Vec<SessionCatalogEntry>, SessionStoreError> {
322        let values = self
323            .connection
324            .prepare("SELECT DISTINCT session_id FROM records ORDER BY session_id")
325            .map_err(SqliteSessionError::from)?
326            .query_map([], |row| row.get::<_, String>(0))
327            .map_err(SqliteSessionError::from)?
328            .collect::<Result<Vec<_>, _>>()
329            .map_err(SqliteSessionError::from)?;
330        let mut entries = Vec::with_capacity(values.len());
331        for value in values {
332            let session_id = parse_session_id(&value)?;
333            self.ensure_loaded(session_id)?;
334            let stored = self.sessions.get(&session_id).ok_or_else(|| {
335                invalid_record("catalog session records disappeared during listing")
336            })?;
337            entries.push(SessionCatalogEntry::from_snapshot(
338                &stored.snapshot(),
339                query_session_name(&self.connection, session_id)?,
340            )?);
341        }
342        entries.sort_by(|left, right| {
343            right
344                .updated_at()
345                .cmp(&left.updated_at())
346                .then_with(|| left.session_id().cmp(&right.session_id()))
347        });
348        Ok(entries)
349    }
350
351    fn set_session_name(
352        &self,
353        session_id: SessionId,
354        name: Option<&SessionName>,
355    ) -> Result<(), SessionStoreError> {
356        require_session(&self.connection, session_id)?;
357        match name {
358            Some(name) => self.connection.execute(
359                "INSERT INTO session_catalog (session_id, display_name) VALUES (?, ?)
360                 ON CONFLICT(session_id) DO UPDATE SET display_name = excluded.display_name",
361                rusqlite::params![session_id.to_string(), name.as_str()],
362            ),
363            None => self.connection.execute(
364                "DELETE FROM session_catalog WHERE session_id = ?",
365                rusqlite::params![session_id.to_string()],
366            ),
367        }
368        .map_err(SqliteSessionError::from)?;
369        Ok(())
370    }
371
372    fn session_name(
373        &self,
374        session_id: SessionId,
375    ) -> Result<Option<SessionName>, SessionStoreError> {
376        require_session(&self.connection, session_id)?;
377        query_session_name(&self.connection, session_id)
378    }
379}
380
381fn load_stored(
382    connection: &Connection,
383    session_id: SessionId,
384) -> Result<Option<StoredSession>, SessionStoreError> {
385    let record_count: i64 = connection
386        .query_row(
387            "SELECT COUNT(*) FROM records WHERE session_id = ?",
388            rusqlite::params![session_id.to_string()],
389            |row| row.get(0),
390        )
391        .map_err(SqliteSessionError::from)?;
392    if record_count == 0 {
393        return Ok(None);
394    }
395    StoredSession::from_durable_facts(
396        load_records(connection, session_id)?,
397        load_approval_artifacts(connection, session_id)?,
398        load_grant_journal(connection, session_id)?,
399    )
400    .map(Some)
401}
402
403fn load_records(
404    connection: &Connection,
405    session_id: SessionId,
406) -> Result<Vec<tea_protocol::RecordEnvelope>, SessionStoreError> {
407    let mut statement = connection
408        .prepare("SELECT envelope FROM records WHERE session_id = ? ORDER BY sequence")
409        .map_err(SqliteSessionError::from)?;
410    let rows = statement
411        .query_map(rusqlite::params![session_id.to_string()], |row| {
412            row.get::<_, String>(0)
413        })
414        .map_err(SqliteSessionError::from)?;
415    let mut records = Vec::new();
416    for row in rows {
417        let value: serde_json::Value = decode_json(&row.map_err(SqliteSessionError::from)?)?;
418        records.push(
419            tea_protocol::RecordEnvelope::decode_value(value)
420                .map_err(|error| invalid_record(&error.to_string()))?,
421        );
422    }
423    Ok(records)
424}
425
426fn load_approval_artifacts(
427    connection: &Connection,
428    session_id: SessionId,
429) -> Result<Vec<ApprovalArtifactEntry>, SessionStoreError> {
430    load_json_rows(
431        connection,
432        "SELECT envelope FROM approval_artifacts WHERE session_id = ? ORDER BY record_id",
433        session_id,
434    )
435}
436
437fn load_grant_journal(
438    connection: &Connection,
439    session_id: SessionId,
440) -> Result<Vec<GrantJournalEntry>, SessionStoreError> {
441    load_json_rows(
442        connection,
443        "SELECT envelope FROM grant_journal WHERE session_id = ? ORDER BY seq",
444        session_id,
445    )
446}
447
448fn load_json_rows<T: serde::de::DeserializeOwned>(
449    connection: &Connection,
450    sql: &str,
451    session_id: SessionId,
452) -> Result<Vec<T>, SessionStoreError> {
453    let mut statement = connection.prepare(sql).map_err(SqliteSessionError::from)?;
454    let rows = statement
455        .query_map(rusqlite::params![session_id.to_string()], |row| {
456            row.get::<_, String>(0)
457        })
458        .map_err(SqliteSessionError::from)?;
459    let mut values = Vec::new();
460    for row in rows {
461        values.push(decode_json(&row.map_err(SqliteSessionError::from)?)?);
462    }
463    Ok(values)
464}
465
466fn validate_persisted_expectation(
467    tx: &rusqlite::Transaction<'_>,
468    transaction: &AppendTransaction,
469) -> Result<(), SessionStoreError> {
470    let tail: Option<i64> = tx
471        .query_row(
472            "SELECT MAX(sequence) FROM records WHERE session_id = ?",
473            rusqlite::params![transaction.session_id().to_string()],
474            |row| row.get(0),
475        )
476        .map_err(SqliteSessionError::from)?;
477    match (tail, transaction.expected_sequence()) {
478        (None, None) => {}
479        (None, Some(_)) => return Err(session_not_found()),
480        (Some(_), None) => {
481            return Err(SessionStoreError::new(
482                SessionStoreErrorCode::SessionAlreadyExists,
483                "session already exists",
484            ));
485        }
486        (Some(tail), Some(expected)) if u64::try_from(tail).ok() == Some(expected.get()) => {}
487        (Some(_), Some(_)) => return Err(sequence_conflict("expected session sequence is stale")),
488    }
489    let side_entries = transaction
490        .approval_artifacts()
491        .len()
492        .saturating_add(transaction.grant_entries().len());
493    if side_entries > 0 {
494        let revision: i64 = tx
495            .query_row(
496                "SELECT
497                    (SELECT COUNT(*) FROM approval_artifacts WHERE session_id = ?1) +
498                    (SELECT COUNT(*) FROM grant_journal WHERE session_id = ?1)",
499                rusqlite::params![transaction.session_id().to_string()],
500                |row| row.get(0),
501            )
502            .map_err(SqliteSessionError::from)?;
503        if u64::try_from(revision).ok() != transaction.expected_journal_revision() {
504            return Err(sequence_conflict(
505                "expected policy journal revision is stale",
506            ));
507        }
508    }
509    Ok(())
510}
511
512fn persisted_grant_ids(
513    tx: &rusqlite::Transaction<'_>,
514    entries: &[GrantJournalEntry],
515) -> Result<HashSet<GrantId>, SessionStoreError> {
516    let mut ids = HashSet::new();
517    for entry in entries {
518        if !matches!(entry, GrantJournalEntry::Issued { .. }) {
519            continue;
520        }
521        let grant_id = entry.grant_id();
522        let exists: bool = tx
523            .query_row(
524                "SELECT EXISTS(SELECT 1 FROM active_grants WHERE grant_id = ?)",
525                rusqlite::params![grant_id.to_string()],
526                |row| row.get(0),
527            )
528            .map_err(SqliteSessionError::from)?;
529        if exists {
530            ids.insert(grant_id);
531        }
532    }
533    Ok(ids)
534}
535
536fn persist_delta(
537    tx: &rusqlite::Transaction<'_>,
538    transaction: &AppendTransaction,
539    previous_grant_count: usize,
540) -> Result<(), SessionStoreError> {
541    let session_id = transaction.session_id().to_string();
542    for record in transaction.records() {
543        tx.execute(
544            "INSERT INTO records (session_id, sequence, record_id, envelope) VALUES (?, ?, ?, ?)",
545            rusqlite::params![
546                session_id,
547                i64::try_from(record.sequence().get()).unwrap_or(i64::MAX),
548                record.record_id().to_string(),
549                encode_json(record)?,
550            ],
551        )
552        .map_err(SqliteSessionError::from)?;
553    }
554    for artifact in transaction.approval_artifacts() {
555        tx.execute(
556            "INSERT INTO approval_artifacts (session_id, record_id, envelope) VALUES (?, ?, ?)",
557            rusqlite::params![
558                session_id,
559                artifact.record_id().to_string(),
560                encode_json(artifact)?
561            ],
562        )
563        .map_err(SqliteSessionError::from)?;
564    }
565    for (offset, entry) in transaction.grant_entries().iter().enumerate() {
566        let sequence = previous_grant_count
567            .checked_add(offset)
568            .and_then(|value| i64::try_from(value).ok())
569            .ok_or_else(|| sequence_conflict("grant journal sequence is out of range"))?;
570        tx.execute(
571            "INSERT INTO grant_journal (session_id, seq, grant_id, envelope) VALUES (?, ?, ?, ?)",
572            rusqlite::params![
573                session_id,
574                sequence,
575                entry.grant_id().to_string(),
576                encode_json(entry)?
577            ],
578        )
579        .map_err(SqliteSessionError::from)?;
580    }
581    Ok(())
582}
583
584fn persist_active_grants(
585    tx: &rusqlite::Transaction<'_>,
586    session_id: SessionId,
587    entries: &[GrantJournalEntry],
588) -> Result<(), SessionStoreError> {
589    for entry in entries {
590        match entry {
591            GrantJournalEntry::Issued { grant, .. } => {
592                tx.execute(
593                    "INSERT INTO active_grants
594                     (grant_id, session_id, actor_id, grant_json, revoked)
595                     VALUES (?, ?, ?, ?, 0)",
596                    rusqlite::params![
597                        grant.id().to_string(),
598                        session_id.to_string(),
599                        grant.actor_id().to_string(),
600                        encode_json(grant)?,
601                    ],
602                )
603                .map_err(SqliteSessionError::from)?;
604            }
605            GrantJournalEntry::Revoked { grant } => {
606                let updated = tx
607                    .execute(
608                        "UPDATE active_grants SET grant_json = ?, revoked = 1
609                         WHERE grant_id = ? AND session_id = ?",
610                        rusqlite::params![
611                            encode_json(grant)?,
612                            grant.id().to_string(),
613                            session_id.to_string(),
614                        ],
615                    )
616                    .map_err(SqliteSessionError::from)?;
617                if updated != 1 {
618                    return Err(invalid_record(
619                        "revoked grant is missing from materialized index",
620                    ));
621                }
622            }
623        }
624    }
625    Ok(())
626}
627
628fn require_session(
629    connection: &Connection,
630    session_id: SessionId,
631) -> Result<(), SessionStoreError> {
632    let exists: bool = connection
633        .query_row(
634            "SELECT EXISTS(SELECT 1 FROM records WHERE session_id = ?)",
635            rusqlite::params![session_id.to_string()],
636            |row| row.get(0),
637        )
638        .map_err(SqliteSessionError::from)?;
639    exists.then_some(()).ok_or_else(session_not_found)
640}
641
642fn query_session_name(
643    connection: &Connection,
644    session_id: SessionId,
645) -> Result<Option<SessionName>, SessionStoreError> {
646    let value = connection
647        .query_row(
648            "SELECT display_name FROM session_catalog WHERE session_id = ?",
649            rusqlite::params![session_id.to_string()],
650            |row| row.get::<_, String>(0),
651        )
652        .optional()
653        .map_err(SqliteSessionError::from)?;
654    value
655        .map(|name| name.parse())
656        .transpose()
657        .map_err(|_| invalid_record("stored session name is invalid"))
658}
659
660fn parse_session_id(value: &str) -> Result<SessionId, SessionStoreError> {
661    value
662        .parse()
663        .map_err(|_| invalid_record("stored session id is not canonical"))
664}
665
666fn encode_json(value: &impl serde::Serialize) -> Result<String, SessionStoreError> {
667    serde_json::to_string(value)
668        .map_err(|error| SqliteSessionError::Serialization(error.to_string()).into())
669}
670
671fn decode_json<T: serde::de::DeserializeOwned>(value: &str) -> Result<T, SessionStoreError> {
672    serde_json::from_str(value)
673        .map_err(|error| SqliteSessionError::Serialization(error.to_string()).into())
674}
675
676fn session_not_found() -> SessionStoreError {
677    SessionStoreError::new(
678        SessionStoreErrorCode::SessionNotFound,
679        "session does not exist",
680    )
681}
682
683fn sequence_conflict(message: &str) -> SessionStoreError {
684    SessionStoreError::new(SessionStoreErrorCode::SequenceConflict, message)
685}
686
687fn invalid_record(message: &str) -> SessionStoreError {
688    SessionStoreError::new(SessionStoreErrorCode::InvalidRecord, message)
689}
690
691#[cfg(test)]
692mod tests {
693    use std::str::FromStr as _;
694
695    use serde_json::json;
696    use tea_protocol::RecordId;
697
698    use super::*;
699
700    fn grant() -> PolicyGrant {
701        serde_json::from_value(json!({
702            "id": "0195a0b1-5e69-70ac-807e-0aa7aa000047",
703            "actorId": "user:alice",
704            "profileId": "minimal-assistant",
705            "toolName": "write_text_file",
706            "toolVersion": "1.0.0",
707            "effects": ["fs.write"],
708            "resources": [{
709                "scheme": "file",
710                "locatorPrefix": "/workspace/",
711                "access": "write"
712            }],
713            "scope": {
714                "type": "session_resource",
715                "session_id": "0195a0b1-5e3a-7d72-a902-c4e85d828bf1"
716            },
717            "issuedAt": "2026-07-23T09:30:12.006Z"
718        }))
719        .unwrap()
720    }
721
722    #[test]
723    fn active_grants_are_written_through_and_revoked_in_place() {
724        let mut connection = Connection::open_in_memory().unwrap();
725        ensure_schema(&mut connection).unwrap();
726        let session_id = SessionId::from_str("0195a0b1-5e3a-7d72-a902-c4e85d828bf1").unwrap();
727        let issued_grant = grant();
728        let issued = GrantJournalEntry::Issued {
729            approval_record_id: RecordId::from_str("0195a0b1-5e54-7c92-b8ca-0aa7aa000026").unwrap(),
730            grant: issued_grant.clone(),
731        };
732        let transaction = connection.transaction().unwrap();
733        persist_active_grants(&transaction, session_id, &[issued]).unwrap();
734        transaction.commit().unwrap();
735
736        let (stored, revoked): (String, bool) = connection
737            .query_row(
738                "SELECT grant_json, revoked FROM active_grants WHERE grant_id = ?",
739                [issued_grant.id().to_string()],
740                |row| Ok((row.get(0)?, row.get(1)?)),
741            )
742            .unwrap();
743        assert_eq!(
744            serde_json::from_str::<PolicyGrant>(&stored).unwrap(),
745            issued_grant
746        );
747        assert!(!revoked);
748
749        let revoked_grant = issued_grant
750            .revoke("2026-07-23T10:00:00.000Z".parse().unwrap())
751            .unwrap();
752        let transaction = connection.transaction().unwrap();
753        persist_active_grants(
754            &transaction,
755            session_id,
756            &[GrantJournalEntry::Revoked {
757                grant: revoked_grant.clone(),
758            }],
759        )
760        .unwrap();
761        transaction.commit().unwrap();
762
763        let (stored, revoked): (String, bool) = connection
764            .query_row(
765                "SELECT grant_json, revoked FROM active_grants WHERE grant_id = ?",
766                [revoked_grant.id().to_string()],
767                |row| Ok((row.get(0)?, row.get(1)?)),
768            )
769            .unwrap();
770        assert_eq!(
771            serde_json::from_str::<PolicyGrant>(&stored).unwrap(),
772            revoked_grant
773        );
774        assert!(revoked);
775    }
776}
tea_session_sqlite/store.rs

tea_session_sqlite/
store.rs
