agentic_core/storage/models/
conversation.rs

1//! Conversation context and history.
2
3use super::super::pool::{DbPool, DbResult, DbTransaction};
4use crate::storage::backend::DatabaseBackend;
5use crate::utils::common::utcnow_str;
6
7/// Conversation context and history.
8///
9/// Maps to the `conversations` table and represents a logical conversation
10/// containing multiple responses and items.
11#[derive(Debug, Clone, sqlx::FromRow)]
12pub struct Conversation {
13    /// Unique conversation identifier.
14    pub id: String,
15
16    /// Optional metadata as JSON string.
17    pub metadata: Option<String>,
18
19    /// Creation timestamp as Unix timestamp in seconds.
20    pub created_at: i64,
21}
22
23/// Create a new conversation.
24///
25/// # Errors
26/// Returns `DbResult::Err` if the database insertion fails.
27pub async fn create(pool: &DbPool, id: &str) -> DbResult<Conversation> {
28    let now = utcnow_str();
29    sqlx::query_as::<_, Conversation>(
30        "INSERT INTO conversations (id, created_at) \
31         VALUES ($1, $2) RETURNING *",
32    )
33    .bind(id)
34    .bind(now)
35    .fetch_one(pool)
36    .await
37}
38
39/// Get or create a conversation.
40///
41/// # Errors
42/// Returns `DbResult::Err` if the database query fails.
43pub async fn get_or_create(pool: &DbPool, id: &str) -> DbResult<Conversation> {
44    let now = utcnow_str();
45    sqlx::query_as::<_, Conversation>(
46        "INSERT INTO conversations (id, created_at) \
47         VALUES ($1, $2) \
48         ON CONFLICT (id) DO UPDATE SET created_at = created_at \
49         RETURNING *",
50    )
51    .bind(id)
52    .bind(now)
53    .fetch_one(pool)
54    .await
55}
56
57/// Get a conversation by ID.
58///
59/// # Errors
60/// Returns `DbResult::Err` if the database query fails.
61pub async fn get(pool: &DbPool, id: &str) -> DbResult<Option<Conversation>> {
62    sqlx::query_as::<_, Conversation>("SELECT * FROM conversations WHERE id = $1")
63        .bind(id)
64        .fetch_optional(pool)
65        .await
66}
67
68/// Locks an existing conversation for the lifetime of the transaction.
69///
70/// `PostgreSQL` takes a row lock without writing the row. `SQLite` uses a no-op
71/// update to acquire its database-wide write lock, which serializes persistence
72/// across all conversations. Both protect sequence allocation when multiple
73/// gateway replicas persist turns concurrently, but with different lock granularity.
74///
75/// # Errors
76/// Returns `DbResult::Err` if the database query fails or the conversation does not exist.
77pub async fn lock_in_tx(tx: &mut DbTransaction<'_>, id: &str) -> DbResult<()> {
78    if DatabaseBackend::from_connection(tx.as_mut()) == DatabaseBackend::Postgres {
79        let locked_id = sqlx::query_scalar::<_, String>("SELECT id FROM conversations WHERE id = $1 FOR UPDATE")
80            .bind(id)
81            .fetch_optional(&mut **tx)
82            .await?;
83        return locked_id.map(|_| ()).ok_or(sqlx::Error::RowNotFound);
84    }
85
86    let result = sqlx::query("UPDATE conversations SET created_at = created_at WHERE id = $1")
87        .bind(id)
88        .execute(&mut **tx)
89        .await?;
90    if result.rows_affected() == 0 {
91        return Err(sqlx::Error::RowNotFound);
92    }
93    Ok(())
94}
95
96#[cfg(test)]
97mod tests {
98    use super::*;
99
100    #[test]
101    fn test_conversation_basic() {
102        let conversation = Conversation {
103            id: "conv_1".to_string(),
104            metadata: None,
105            created_at: 1_704_067_200,
106        };
107
108        assert_eq!(conversation.id, "conv_1");
109        assert!(conversation.metadata.is_none());
110        assert_eq!(conversation.created_at, 1_704_067_200);
111    }
112}
agentic_core/storage/models/conversation.rs

agentic_core/storage/models/
conversation.rs
