turnframe_store_postgres/
replay.rs

1//! Replay records over `tf_replay` (spec §23.1, I20).
2//!
3//! One row per turn, rewritten as the turn advances: the runtime writes the
4//! record at `Received` and again with more detail at every step, so the write
5//! is an upsert and the table never grows a second row for one turn. The record
6//! is stored whole as `jsonb`; the columns beside it are the keys the lookups
7//! need, extracted so that "the records of this conversation, most recent first"
8//! is an index scan rather than a scan of every document.
9
10use async_trait::async_trait;
11use sqlx::PgConnection;
12use sqlx::postgres::PgRow;
13use turnframe_core::ids::{AccountId, ConversationId, TurnId};
14use turnframe_core::replay::ReplayRecord;
15use turnframe_store::error::StoreError;
16use turnframe_store::replay::{ReplayReader, ReplayWriter};
17
18use crate::codec::{column, from_json, label, limit_to_sql, to_json};
19use crate::error::store_error;
20use crate::store::{PgStores, commit};
21
22/// Inserts or replaces the record of one turn.
23pub(crate) async fn put(conn: &mut PgConnection, record: ReplayRecord) -> Result<(), StoreError> {
24    sqlx::query(
25        "INSERT INTO tf_replay (account_id, turn_id, conversation_id, phase, recorded_at, record)
26         VALUES ($1, $2, $3, $4, $5, $6)
27         ON CONFLICT (account_id, turn_id) DO UPDATE
28         SET conversation_id = EXCLUDED.conversation_id,
29             phase = EXCLUDED.phase,
30             recorded_at = EXCLUDED.recorded_at,
31             record = EXCLUDED.record",
32    )
33    .bind(record.account_id.as_str())
34    .bind(record.turn_id.as_uuid())
35    .bind(record.conversation_id.as_uuid())
36    .bind(label(&record.phase)?)
37    .bind(record.recorded_at)
38    .bind(to_json(&record)?)
39    .execute(conn)
40    .await
41    .map_err(|error| store_error(&error))?;
42    Ok(())
43}
44
45/// Loads the record of a turn.
46pub(crate) async fn get(
47    conn: &mut PgConnection,
48    account: &AccountId,
49    turn_id: &TurnId,
50) -> Result<ReplayRecord, StoreError> {
51    let row = sqlx::query("SELECT record FROM tf_replay WHERE account_id = $1 AND turn_id = $2")
52        .bind(account.as_str())
53        .bind(turn_id.as_uuid())
54        .fetch_optional(conn)
55        .await
56        .map_err(|error| store_error(&error))?
57        .ok_or(StoreError::NotFound)?;
58    decode_record(&row)
59}
60
61/// The most recent `limit` records of a conversation, oldest of them first.
62pub(crate) async fn list_for_conversation(
63    conn: &mut PgConnection,
64    account: &AccountId,
65    conversation: &ConversationId,
66    limit: usize,
67) -> Result<Vec<ReplayRecord>, StoreError> {
68    let rows = sqlx::query(
69        "SELECT record FROM tf_replay
70         WHERE account_id = $1 AND conversation_id = $2
71         ORDER BY recorded_at DESC, turn_id DESC
72         LIMIT $3",
73    )
74    .bind(account.as_str())
75    .bind(conversation.as_uuid())
76    .bind(limit_to_sql(limit))
77    .fetch_all(conn)
78    .await
79    .map_err(|error| store_error(&error))?;
80    let mut records = rows
81        .iter()
82        .map(decode_record)
83        .collect::<Result<Vec<ReplayRecord>, StoreError>>()?;
84    records.reverse();
85    Ok(records)
86}
87
88/// Rebuilds a record from its row.
89fn decode_record(row: &PgRow) -> Result<ReplayRecord, StoreError> {
90    from_json(column(row, "record")?)
91}
92
93#[async_trait]
94impl ReplayReader for PgStores {
95    async fn get(&self, account: &AccountId, turn_id: &TurnId) -> Result<ReplayRecord, StoreError> {
96        let mut conn = self.connection().await?;
97        get(&mut conn, account, turn_id).await
98    }
99
100    async fn list_for_conversation(
101        &self,
102        account: &AccountId,
103        conversation: &ConversationId,
104        limit: usize,
105    ) -> Result<Vec<ReplayRecord>, StoreError> {
106        let mut conn = self.connection().await?;
107        list_for_conversation(&mut conn, account, conversation, limit).await
108    }
109}
110
111#[async_trait]
112impl ReplayWriter for PgStores {
113    async fn put(&self, record: ReplayRecord) -> Result<(), StoreError> {
114        let mut transaction = self.transaction().await?;
115        put(&mut transaction, record).await?;
116        commit(transaction).await
117    }
118}
turnframe_store_postgres/replay.rs

turnframe_store_postgres/
replay.rs
