systemprompt_agent/repository/content/
push_notification.rs

1//! Persistence for per-task push-notification delivery configuration.
2//!
3//! [`PushNotificationConfigRepository`] stores, retrieves, and deletes the
4//! webhook [`PushNotificationConfig`] entries attached to a task, mapping the
5//! stored row's JSON header/auth columns back into the typed config.
6//!
7//! Copyright (c) systemprompt.io — Business Source License 1.1.
8//! See <https://systemprompt.io> for licensing details.
9
10use chrono::Utc;
11use sqlx::PgPool;
12use std::sync::Arc;
13use systemprompt_database::DbPool;
14use systemprompt_identifiers::{ConfigId, TaskId};
15use systemprompt_traits::RepositoryError;
16
17use crate::models::a2a::protocol::PushNotificationConfig;
18use crate::models::database_rows::PushNotificationConfigRow;
19
20#[derive(Clone)]
21pub struct PushNotificationConfigRepository {
22    pool: Arc<PgPool>,
23    write_pool: Arc<PgPool>,
24}
25
26impl std::fmt::Debug for PushNotificationConfigRepository {
27    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
28        f.debug_struct("PushNotificationConfigRepository")
29            .field("pool", &"<PgPool>")
30            .finish()
31    }
32}
33
34impl PushNotificationConfigRepository {
35    pub fn new(db: &DbPool) -> Result<Self, crate::error::AgentError> {
36        let pool = db
37            .pool_arc()
38            .map_err(|e| crate::error::AgentError::Init(e.to_string()))?;
39        let write_pool = db
40            .write_pool_arc()
41            .map_err(|e| crate::error::AgentError::Init(e.to_string()))?;
42        Ok(Self { pool, write_pool })
43    }
44
45    pub async fn add_config(
46        &self,
47        task_id: &TaskId,
48        config: &PushNotificationConfig,
49    ) -> Result<String, RepositoryError> {
50        let config_id = uuid::Uuid::new_v4().to_string();
51        let headers_json = config
52            .headers
53            .as_ref()
54            .map(serde_json::to_value)
55            .transpose()?;
56        let auth_json = config
57            .authentication
58            .as_ref()
59            .map(serde_json::to_value)
60            .transpose()?;
61        let now = Utc::now();
62        let task_id_str = task_id.as_str();
63
64        sqlx::query!(
65            r#"INSERT INTO task_push_notification_configs
66                (id, task_id, url, endpoint, token, headers, authentication, created_at, updated_at)
67            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)"#,
68            config_id,
69            task_id_str,
70            config.url,
71            config.endpoint,
72            config.token,
73            headers_json,
74            auth_json,
75            now,
76            now
77        )
78        .execute(&*self.write_pool)
79        .await
80        .map_err(RepositoryError::database)?;
81
82        Ok(config_id)
83    }
84
85    pub async fn get_config(
86        &self,
87        task_id: &TaskId,
88        config_id: &ConfigId,
89    ) -> Result<Option<PushNotificationConfig>, RepositoryError> {
90        let task_id_str = task_id.as_str();
91        let config_id_str = config_id.as_str();
92        let row = sqlx::query_as!(
93            PushNotificationConfigRow,
94            r#"SELECT
95                id,
96                task_id,
97                url,
98                endpoint,
99                token,
100                headers,
101                authentication,
102                created_at,
103                updated_at
104            FROM task_push_notification_configs
105            WHERE task_id = $1 AND id = $2"#,
106            task_id_str,
107            config_id_str
108        )
109        .fetch_optional(&*self.pool)
110        .await
111        .map_err(RepositoryError::database)?;
112
113        row.map(|r| Self::row_to_config(&r)).transpose()
114    }
115
116    pub async fn list_configs(
117        &self,
118        task_id: &TaskId,
119    ) -> Result<Vec<PushNotificationConfig>, RepositoryError> {
120        let task_id_str = task_id.as_str();
121        let rows: Vec<PushNotificationConfigRow> = sqlx::query_as!(
122            PushNotificationConfigRow,
123            r#"SELECT
124                id,
125                task_id,
126                url,
127                endpoint,
128                token,
129                headers,
130                authentication,
131                created_at,
132                updated_at
133            FROM task_push_notification_configs
134            WHERE task_id = $1"#,
135            task_id_str
136        )
137        .fetch_all(&*self.pool)
138        .await
139        .map_err(RepositoryError::database)?;
140
141        rows.iter()
142            .map(Self::row_to_config)
143            .collect::<Result<Vec<_>, RepositoryError>>()
144    }
145
146    pub async fn delete_config(
147        &self,
148        task_id: &TaskId,
149        config_id: &ConfigId,
150    ) -> Result<bool, RepositoryError> {
151        let task_id_str = task_id.as_str();
152        let config_id_str = config_id.as_str();
153        let result = sqlx::query!(
154            "DELETE FROM task_push_notification_configs WHERE task_id = $1 AND id = $2",
155            task_id_str,
156            config_id_str
157        )
158        .execute(&*self.write_pool)
159        .await
160        .map_err(RepositoryError::database)?;
161
162        Ok(result.rows_affected() > 0)
163    }
164
165    pub async fn delete_all_for_task(&self, task_id: &TaskId) -> Result<u64, RepositoryError> {
166        let task_id_str = task_id.as_str();
167        let result = sqlx::query!(
168            "DELETE FROM task_push_notification_configs WHERE task_id = $1",
169            task_id_str
170        )
171        .execute(&*self.write_pool)
172        .await
173        .map_err(RepositoryError::database)?;
174
175        Ok(result.rows_affected())
176    }
177
178    fn row_to_config(
179        row: &PushNotificationConfigRow,
180    ) -> Result<PushNotificationConfig, RepositoryError> {
181        let headers = row
182            .headers
183            .as_ref()
184            .map(|v| serde_json::from_value(v.clone()))
185            .transpose()?;
186        let authentication = row
187            .authentication
188            .as_ref()
189            .map(|v| serde_json::from_value(v.clone()))
190            .transpose()?;
191
192        Ok(PushNotificationConfig {
193            url: row.url.clone(),
194            endpoint: row.endpoint.clone(),
195            token: row.token.clone(),
196            headers,
197            authentication,
198        })
199    }
200}
systemprompt_agent/repository/content/push_notification.rs

systemprompt_agent/repository/content/
push_notification.rs
