runifold_store_postgres/
conversation.rs

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

runifold_store_postgres/
conversation.rs
