agentic_core/storage/
response.rs

1//! Response storage operations and queries.
2
3use std::collections::HashMap;
4use std::convert::TryFrom;
5use std::sync::Arc;
6
7use super::models::{item, response};
8use super::pool::DbPool;
9use super::types::{InOutItem, ResponseData, ResponseMetadata, StorageError, StoreResult};
10use crate::utils::common::{serialize_to_string, uuid7_str};
11
12/// Response storage operations.
13#[derive(Clone, Debug)]
14pub struct ResponseStore {
15    pool: Option<Arc<DbPool>>,
16}
17
18impl ResponseStore {
19    /// Creates a disabled response store (no persistence).
20    ///
21    /// Useful for testing or when response storage is not configured.
22    #[must_use]
23    pub fn disabled() -> Self {
24        Self { pool: None }
25    }
26
27    /// Creates a new response store with database pool.
28    ///
29    /// # Arguments
30    ///
31    /// * `pool` - Connection pool for database access
32    #[must_use]
33    pub fn new(pool: Arc<DbPool>) -> Self {
34        Self { pool: Some(pool) }
35    }
36
37    /// Returns a reference to the database pool.
38    ///
39    /// # Errors
40    ///
41    /// Returns [`StorageError::NotConfigured`] if store is disabled (no pool configured).
42    fn pool(&self) -> StoreResult<&DbPool> {
43        self.pool.as_deref().ok_or(StorageError::NotConfigured)
44    }
45
46    /// Retrieves a response by ID.
47    ///
48    /// # Errors
49    ///
50    /// Returns error if response not found, database query fails, or store is disabled.
51    pub async fn get(&self, response_id: &str) -> StoreResult<ResponseData> {
52        let pool = self.pool()?;
53        let row = response::get(pool, response_id)
54            .await?
55            .ok_or_else(|| StorageError::not_found("Response", response_id))?;
56        Ok(row.into())
57    }
58
59    /// Rehydrates a response with full history.
60    ///
61    /// Fetches all history items referenced by a response.
62    ///
63    /// # Errors
64    ///
65    /// Returns error if database query fails or store is disabled.
66    pub async fn rehydrate(&self, response_id: &str) -> StoreResult<Vec<InOutItem>> {
67        let pool = self.pool()?;
68        let response = self.get(response_id).await?;
69        let rows = item::get_items(pool, &response.history_item_ids).await?;
70        let mut items_by_id: HashMap<String, InOutItem> = rows
71            .into_iter()
72            .filter_map(|row| {
73                let id = row.id.clone();
74                row.as_inout().map(|item| (id, item))
75            })
76            .collect();
77
78        let ordered_items = response
79            .history_item_ids
80            .iter()
81            .filter_map(|id| items_by_id.remove(id))
82            .collect();
83
84        Ok(ordered_items)
85    }
86
87    /// Persists a response with its items and metadata.
88    ///
89    /// Creates items and stores the associated response record.
90    ///
91    /// # Errors
92    ///
93    /// Returns [`StorageError`] if database operation fails or store is disabled.
94    pub async fn persist(
95        &self,
96        response_id: &str,
97        previous_response_id: Option<&str>,
98        new_items: Vec<InOutItem>,
99        metadata: &ResponseMetadata,
100    ) -> StoreResult<()> {
101        let pool = self.pool()?;
102
103        let mut item_ids: Vec<String> = match previous_response_id {
104            Some(prev_id) => self.get(prev_id).await?.history_item_ids,
105            None => Vec::new(),
106        };
107        let mut items_: Vec<(String, String)> = Vec::new();
108        for any_item in new_items {
109            let item_id = uuid7_str("item_");
110            item_ids.push(item_id.clone());
111            let data_str = String::try_from(&any_item)?;
112            items_.push((item_id, data_str));
113        }
114
115        let mut tx = pool.begin().await?;
116
117        item::create_in_tx(&mut tx, items_, None, None).await?;
118
119        let history_item_ids_json = serialize_to_string(&item_ids)?;
120        let metadata_json = String::try_from(metadata)?;
121
122        response::create_in_tx(
123            &mut tx,
124            response_id,
125            None,
126            previous_response_id,
127            Some(&history_item_ids_json),
128            Some(&metadata_json),
129        )
130        .await?;
131        tx.commit().await?;
132
133        Ok(())
134    }
135}
136
137#[cfg(test)]
138mod tests {
139    use super::super::types::ResponseMetadata;
140    use super::*;
141
142    #[test]
143    fn test_response_store_disabled() {
144        let store = ResponseStore::disabled();
145        assert!(store.pool().is_err());
146    }
147
148    #[test]
149    fn test_response_metadata_default() {
150        let meta = ResponseMetadata::default();
151        assert!(meta.model.is_empty());
152        assert!(meta.previous_response_id.is_none());
153    }
154}
agentic_core/storage/response.rs

agentic_core/storage/
response.rs
