turnframe_store/
replay.rs

1//! Replay records: one per turn, rewritten as the turn progresses (spec §23.1, I20).
2//!
3//! # Contract
4//!
5//! * A replay record is keyed by `(account_id, turn_id)` and **upserted**: the
6//!   runtime writes it at `Received` and rewrites it with more detail as the
7//!   turn advances, so [`ReplayWriter::put`] replaces any previous version.
8//! * Lookups are account-scoped; another tenant's turn is `NotFound`.
9
10use async_trait::async_trait;
11use turnframe_core::ids::{AccountId, ConversationId, TurnId};
12use turnframe_core::replay::ReplayRecord;
13
14use crate::error::StoreError;
15
16/// The read half of the replay record contract (spec §22.1).
17#[async_trait]
18pub trait ReplayReader: Send + Sync {
19    /// Loads the record of a turn.
20    ///
21    /// # Errors
22    /// * `NotFound` when it does not exist for `account`.
23    async fn get(&self, account: &AccountId, turn_id: &TurnId) -> Result<ReplayRecord, StoreError>;
24
25    /// The most recent `limit` records of a conversation in chronological order
26    /// (by `recorded_at`, then `turn_id`).
27    ///
28    /// # Errors
29    /// * [`StoreError`] when the listing could not be read.
30    async fn list_for_conversation(
31        &self,
32        account: &AccountId,
33        conversation: &ConversationId,
34        limit: usize,
35    ) -> Result<Vec<ReplayRecord>, StoreError>;
36}
37
38/// The write half of the replay record contract (spec §22.1).
39#[async_trait]
40pub trait ReplayWriter: Send + Sync {
41    /// Inserts or replaces the record of `(record.account_id, record.turn_id)`.
42    ///
43    /// # Errors
44    /// * [`StoreError`] when the record could not be written.
45    async fn put(&self, record: ReplayRecord) -> Result<(), StoreError>;
46}
47
48/// Persistence of replay records (spec §22.1): both halves.
49///
50/// There is nothing to implement here: write [`ReplayReader`] and
51/// [`ReplayWriter`] and the blanket implementation below supplies this trait.
52pub trait ReplayStore: ReplayReader + ReplayWriter {}
53
54impl<T: ReplayReader + ReplayWriter + ?Sized> ReplayStore for T {}
55
56#[cfg(test)]
57mod tests {
58    use chrono::{DateTime, TimeDelta, Utc};
59    use turnframe_core::ids::EventId;
60    use turnframe_core::replay::TurnPhase;
61
62    use super::*;
63    use crate::memory::MemoryStores;
64
65    fn record(
66        account: &AccountId,
67        conversation: ConversationId,
68        turn: TurnId,
69        at: DateTime<Utc>,
70    ) -> ReplayRecord {
71        ReplayRecord::received(turn, conversation, account.clone(), at)
72    }
73
74    #[tokio::test]
75    async fn put_replaces_and_get_is_account_scoped() {
76        let store = MemoryStores::new();
77        let account = AccountId::from("a");
78        let conversation = ConversationId::new();
79        let turn = TurnId::new();
80        let epoch = DateTime::<Utc>::UNIX_EPOCH;
81
82        store
83            .put(record(&account, conversation, turn, epoch))
84            .await
85            .unwrap();
86
87        // The runtime rewrites the record as the turn advances; the store keeps
88        // one version per turn, not a history.
89        let mut advanced = record(&account, conversation, turn, epoch + TimeDelta::seconds(1));
90        advanced.phase = TurnPhase::Composed;
91        advanced.event_ids = vec![EventId::new()];
92        store.put(advanced.clone()).await.unwrap();
93
94        assert_eq!(
95            ReplayReader::get(&store, &account, &turn).await.unwrap(),
96            advanced
97        );
98        assert_eq!(
99            store
100                .list_for_conversation(&account, &conversation, 10)
101                .await
102                .unwrap(),
103            vec![advanced]
104        );
105        assert_eq!(
106            ReplayReader::get(&store, &AccountId::from("b"), &turn).await,
107            Err(StoreError::NotFound)
108        );
109        assert!(
110            store
111                .list_for_conversation(&AccountId::from("b"), &conversation, 10)
112                .await
113                .unwrap()
114                .is_empty()
115        );
116    }
117
118    #[tokio::test]
119    async fn list_returns_the_most_recent_records_chronologically() {
120        let store = MemoryStores::new();
121        let account = AccountId::from("a");
122        let conversation = ConversationId::new();
123        let epoch = DateTime::<Utc>::UNIX_EPOCH;
124        let mut turns = Vec::new();
125        for seconds in 0..4 {
126            let turn = TurnId::new();
127            turns.push(turn);
128            store
129                .put(record(
130                    &account,
131                    conversation,
132                    turn,
133                    epoch + TimeDelta::seconds(seconds),
134                ))
135                .await
136                .unwrap();
137        }
138        let recent = store
139            .list_for_conversation(&account, &conversation, 2)
140            .await
141            .unwrap();
142        assert_eq!(
143            recent
144                .iter()
145                .map(|record| record.turn_id)
146                .collect::<Vec<_>>(),
147            vec![turns[2], turns[3]]
148        );
149    }
150}
turnframe_store/replay.rs

turnframe_store/
replay.rs
