systemprompt_agent/repository/context/message/
mod.rs

1//! Persistence for A2A messages and their constituent parts.
2//!
3//! [`MessageRepository`] reads message history by task or context and writes
4//! messages within a transaction. The submodules split the work: `queries`
5//! handles reads and sequence-number allocation, `persistence` writes the
6//! message row, and `parts` handles the typed text/file/data parts.
7//!
8//! Copyright (c) systemprompt.io — Business Source License 1.1.
9//! See <https://systemprompt.io> for licensing details.
10
11mod parts;
12mod persistence;
13mod queries;
14
15// Why: these writes go through the generic text-parameter transaction API, so
16// jsonb columns take a serialised string bound through an explicit `::jsonb`
17// cast — binding the text OID directly is rejected by the Postgres protocol.
18pub(crate) fn jsonb_param<T: serde::Serialize>(value: &T) -> Result<String, serde_json::Error> {
19    serde_json::to_string(value)
20}
21
22use sqlx::PgPool;
23use std::sync::Arc;
24use systemprompt_database::DbPool;
25use systemprompt_identifiers::{ContextId, TaskId};
26use systemprompt_traits::RepositoryError;
27
28use crate::models::a2a::Message;
29
30pub use parts::{PersistPartSqlxParams, get_message_parts};
31pub use persistence::{
32    PersistMessageSqlxParams, PersistMessageWithTxParams, persist_message_sqlx,
33    persist_message_with_tx,
34};
35pub use queries::{
36    get_messages_by_context, get_messages_by_task, get_next_sequence_number,
37    get_next_sequence_number_in_tx, get_next_sequence_number_sqlx, message_exists,
38};
39
40#[derive(Debug, Clone)]
41pub struct MessageRepository {
42    pool: Arc<PgPool>,
43}
44
45impl MessageRepository {
46    pub fn new(db: &DbPool) -> Result<Self, RepositoryError> {
47        let pool = db.pool_arc().map_err(|e| {
48            RepositoryError::InvalidData(format!("PostgreSQL pool not available: {e}"))
49        })?;
50        Ok(Self { pool })
51    }
52
53    pub async fn get_messages_by_task(
54        &self,
55        task_id: &TaskId,
56    ) -> Result<Vec<Message>, RepositoryError> {
57        get_messages_by_task(&self.pool, task_id).await
58    }
59
60    pub async fn get_messages_by_context(
61        &self,
62        context_id: &ContextId,
63    ) -> Result<Vec<Message>, RepositoryError> {
64        get_messages_by_context(&self.pool, context_id).await
65    }
66
67    pub async fn get_next_sequence_number(&self, task_id: &TaskId) -> Result<i32, RepositoryError> {
68        get_next_sequence_number(&self.pool, task_id).await
69    }
70
71    pub async fn persist_message_sqlx(
72        &self,
73        params: PersistMessageSqlxParams<'_>,
74    ) -> Result<(), RepositoryError> {
75        persist_message_sqlx(params).await
76    }
77
78    pub async fn persist_message_with_tx(
79        &self,
80        params: PersistMessageWithTxParams<'_>,
81    ) -> Result<(), RepositoryError> {
82        persist_message_with_tx(params).await
83    }
84}
systemprompt_agent/repository/context/message/mod.rs

systemprompt_agent/repository/context/message/
mod.rs
