agentic_core/storage/
conversation.rs

1//! Conversation storage operations.
2
3use std::convert::TryFrom;
4use std::sync::Arc;
5
6use super::models::{conversation, item, response};
7use super::pool::DbPool;
8use super::types::{ConversationData, InOutItem, ResponseMetadata, StorageError, StoreResult};
9use crate::utils::common::{serialize_to_string, uuid7_str};
10
11/// Conversation storage operations.
12#[derive(Clone, Debug)]
13pub struct ConversationStore {
14    pool: Option<Arc<DbPool>>,
15}
16
17impl ConversationStore {
18    /// Creates a disabled conversation store.
19    #[must_use]
20    pub fn disabled() -> Self {
21        Self { pool: None }
22    }
23
24    /// Creates a new conversation store with database pool.
25    #[must_use]
26    pub fn new(pool: Arc<DbPool>) -> Self {
27        Self { pool: Some(pool) }
28    }
29
30    /// Returns a reference to the database pool.
31    ///
32    /// # Errors
33    ///
34    /// Returns error if store is disabled (no pool configured).
35    fn pool(&self) -> StoreResult<&DbPool> {
36        self.pool.as_deref().ok_or(StorageError::NotConfigured)
37    }
38
39    /// Creates a new conversation.
40    ///
41    /// # Errors
42    ///
43    /// Returns error if database query fails.
44    pub async fn create(&self) -> StoreResult<ConversationData> {
45        let pool = self.pool()?;
46        let row = conversation::create(pool, &uuid7_str("conv_")).await?;
47        Ok(row.into())
48    }
49
50    /// Gets a conversation or creates it if it doesn't exist.
51    ///
52    /// # Errors
53    ///
54    /// Returns error if database query fails.
55    pub async fn get_or_create(&self, conversation_id: &str) -> StoreResult<ConversationData> {
56        let pool = self.pool()?;
57        let row = conversation::get_or_create(pool, conversation_id).await?;
58        Ok(row.into())
59    }
60
61    /// Gets a conversation by ID.
62    ///
63    /// # Errors
64    ///
65    /// Returns error if conversation not found or database query fails.
66    pub async fn get(&self, conversation_id: &str) -> StoreResult<ConversationData> {
67        let pool = self.pool()?;
68        let row = conversation::get(pool, conversation_id)
69            .await?
70            .ok_or_else(|| StorageError::not_found("Conversation", conversation_id))?;
71        Ok(row.into())
72    }
73
74    /// Rehydrates a conversation with all its items.
75    ///
76    /// # Errors
77    ///
78    /// Returns error if conversation not found or database query fails.
79    pub async fn rehydrate(&self, conversation_id: &str) -> StoreResult<Vec<InOutItem>> {
80        let pool = self.pool()?;
81        let rows = item::get_items_by_conversation(pool, conversation_id).await?;
82
83        Ok(rows.into_iter().filter_map(|row| row.as_inout()).collect())
84    }
85
86    /// Persists conversation turn with new items and response metadata.
87    ///
88    /// Creates items in the conversation and stores the associated response record.
89    ///
90    /// # Errors
91    ///
92    /// Returns [`StorageError`] if conversation not found or database operation fails.
93    pub async fn persist(
94        &self,
95        conversation_id: &str,
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> = Vec::new();
104        let mut items_: Vec<(String, String)> = Vec::new();
105        for any_item in new_items {
106            let item_id = uuid7_str("item_");
107            item_ids.push(item_id.clone());
108            let data_str = String::try_from(&any_item)?;
109            items_.push((item_id, data_str));
110        }
111
112        let mut tx = pool.begin().await?;
113
114        let seq_start = item::conversation_item_count(&mut tx, conversation_id)
115            .await?
116            .ok_or_else(|| StorageError::not_found("Conversation", conversation_id))?;
117
118        item::create_in_tx(&mut tx, items_, Some(conversation_id), Some(seq_start)).await?;
119
120        let history_item_ids_json = serialize_to_string(&item_ids)?;
121        let metadata_json = String::try_from(metadata)?;
122
123        response::create_in_tx(
124            &mut tx,
125            response_id,
126            Some(conversation_id),
127            previous_response_id,
128            Some(&history_item_ids_json),
129            Some(&metadata_json),
130        )
131        .await?;
132        tx.commit().await?;
133
134        Ok(())
135    }
136}
agentic_core/storage/conversation.rs

agentic_core/storage/
conversation.rs
