runifold_store_postgres/
conversation.rs

1//! `PostgreSQL` conversation, summary, and semantic-memory adapter.
2
3mod durable;
4mod memory;
5mod schema;
6mod store;
7mod support;
8mod transcript;
9
10use support::validate_identifier;
11
12use std::{fmt, sync::Arc};
13
14use runifold_retrieval::EmbeddingModel;
15use thiserror::Error;
16use tokio_postgres::{Client, NoTls};
17
18use crate::blocking::PostgresBlockingClient;
19
20const MAX_CONTENT_BYTES: usize = 262_144;
21
22/// `PostgreSQL` conversation-store configuration or schema failure.
23#[derive(Debug, Error)]
24#[non_exhaustive]
25pub enum PostgresConversationStoreError {
26    /// The configured table name is unsafe for SQL interpolation.
27    #[error("conversation table must be a portable PostgreSQL identifier of at most 48 bytes")]
28    InvalidTable,
29    /// `PostgreSQL` connection or explicit schema setup failed.
30    #[error("PostgreSQL conversation store operation failed: {0}")]
31    Database(#[from] tokio_postgres::Error),
32    /// The blocking connection worker could not be joined.
33    #[error("PostgreSQL blocking connection task failed: {0}")]
34    ConnectionTask(String),
35}
36
37/// PostgreSQL-backed append-only conversation and semantic-memory store.
38#[derive(Clone)]
39pub struct PostgresConversationStore {
40    client: Arc<Client>,
41    transaction_client: Arc<tokio::sync::Mutex<Client>>,
42    blocking: PostgresBlockingClient,
43    table: String,
44    semantic_embedder: Option<Arc<dyn EmbeddingModel>>,
45}
46
47impl fmt::Debug for PostgresConversationStore {
48    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
49        formatter
50            .debug_struct("PostgresConversationStore")
51            .field("table", &self.table)
52            .field(
53                "semantic_embedding",
54                &self.semantic_embedder.as_ref().map(|_| "configured"),
55            )
56            .finish_non_exhaustive()
57    }
58}
59
60impl PostgresConversationStore {
61    /// Connects without creating or changing schema.
62    ///
63    /// # Errors
64    ///
65    /// Rejects unsafe table identifiers and propagates connection failures.
66    pub async fn connect(
67        connection: &str,
68        table: &str,
69    ) -> Result<Self, PostgresConversationStoreError> {
70        validate_identifier(table)?;
71        let sync_connection = connection.to_owned();
72        let blocking =
73            tokio::task::spawn_blocking(move || PostgresBlockingClient::connect(&sync_connection))
74                .await
75                .map_err(|error| PostgresConversationStoreError::ConnectionTask(error.to_string()))?
76                .map_err(PostgresConversationStoreError::ConnectionTask)?;
77        let (client, primary_connection) = tokio_postgres::connect(connection, NoTls).await?;
78        tokio::spawn(async move {
79            let _ = primary_connection.await;
80        });
81        let (transaction_client, transaction_connection) =
82            tokio_postgres::connect(connection, NoTls).await?;
83        tokio::spawn(async move {
84            let _ = transaction_connection.await;
85        });
86        Ok(Self {
87            client: Arc::new(client),
88            transaction_client: Arc::new(tokio::sync::Mutex::new(transaction_client)),
89            blocking,
90            table: table.to_owned(),
91            semantic_embedder: None,
92        })
93    }
94
95    /// Enables vector semantic-memory writes and searches for scoped operations.
96    #[must_use]
97    pub fn with_semantic_memory_embedder(mut self, embedder: Arc<dyn EmbeddingModel>) -> Self {
98        self.semantic_embedder = Some(embedder);
99        self
100    }
101
102    pub(crate) const fn blocking(&self) -> &PostgresBlockingClient {
103        &self.blocking
104    }
105
106    pub(crate) fn table(&self) -> &str {
107        &self.table
108    }
109}
110
111#[cfg(test)]
112mod tests {
113    use super::*;
114
115    #[test]
116    fn table_identifiers_are_restricted_before_sql_construction() {
117        assert!(validate_identifier("runifold_conversations").is_ok());
118        assert!(matches!(
119            validate_identifier("bad-name"),
120            Err(PostgresConversationStoreError::InvalidTable)
121        ));
122        assert!(matches!(
123            validate_identifier("1bad"),
124            Err(PostgresConversationStoreError::InvalidTable)
125        ));
126    }
127
128    #[test]
129    fn schema_preserves_append_only_transcript_and_search_indexes() {
130        let schema = PostgresConversationStore::schema_sql("runifold_conversations");
131
132        assert!(schema.contains("PRIMARY KEY (conversation_id, sequence)"));
133        assert!(schema.contains("to_tsvector('simple', content)"));
134        assert!(!schema.contains("CREATE TRIGGER"));
135    }
136}
runifold_store_postgres/conversation.rs

runifold_store_postgres/
conversation.rs
