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

systemprompt_agent/repository/content/
push_notification.rs
