nyx_agent_core/store/
integration.rs

1//! `project_integrations` table - outbound delivery settings scoped to
2//! one project.
3
4use sqlx::{Row, SqlitePool};
5
6pub use nyx_agent_types::integration::{
7    CreateProjectIntegrationRequest, PatchProjectIntegrationRequest, ProjectIntegrationEvent,
8    ProjectIntegrationKind, ProjectIntegrationRecord,
9};
10
11use crate::store::StoreError;
12
13#[derive(Debug, Clone, PartialEq, Eq)]
14pub struct ProjectIntegrationStoredRecord {
15    pub public: ProjectIntegrationRecord,
16    pub config_json: String,
17}
18
19#[derive(Debug, Clone)]
20pub struct ProjectIntegrationInsert {
21    pub id: String,
22    pub project_id: String,
23    pub kind: ProjectIntegrationKind,
24    pub name: String,
25    pub enabled: bool,
26    pub events: Vec<ProjectIntegrationEvent>,
27    pub min_severity: Option<String>,
28    pub config_json: String,
29    pub target: String,
30    pub now_ms: i64,
31}
32
33#[derive(Debug, Clone, Default)]
34pub struct ProjectIntegrationPatch {
35    pub name: Option<String>,
36    pub enabled: Option<bool>,
37    pub events: Option<Vec<ProjectIntegrationEvent>>,
38    pub min_severity: Option<Option<String>>,
39    pub config_json: Option<String>,
40    pub target: Option<String>,
41    pub updated_at: i64,
42}
43
44pub struct ProjectIntegrationStore<'a> {
45    pool: &'a SqlitePool,
46}
47
48impl<'a> ProjectIntegrationStore<'a> {
49    pub fn new(pool: &'a SqlitePool) -> Self {
50        Self { pool }
51    }
52
53    pub async fn create(
54        &self,
55        rec: ProjectIntegrationInsert,
56    ) -> Result<ProjectIntegrationRecord, StoreError> {
57        sqlx::query(
58            r#"
59            INSERT INTO project_integrations (
60                id, project_id, kind, name, enabled, events_json, min_severity,
61                config_json, target, created_at, updated_at
62            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
63            "#,
64        )
65        .bind(&rec.id)
66        .bind(&rec.project_id)
67        .bind(rec.kind.as_str())
68        .bind(&rec.name)
69        .bind(i64::from(rec.enabled))
70        .bind(serde_json::to_string(&rec.events).map_err(StoreError::IntegrationJson)?)
71        .bind(&rec.min_severity)
72        .bind(&rec.config_json)
73        .bind(&rec.target)
74        .bind(rec.now_ms)
75        .bind(rec.now_ms)
76        .execute(self.pool)
77        .await?;
78        self.get(&rec.id).await?.ok_or(StoreError::Sqlx(sqlx::Error::RowNotFound))
79    }
80
81    pub async fn get(&self, id: &str) -> Result<Option<ProjectIntegrationRecord>, StoreError> {
82        Ok(self.get_stored(id).await?.map(|r| r.public))
83    }
84
85    pub async fn get_stored(
86        &self,
87        id: &str,
88    ) -> Result<Option<ProjectIntegrationStoredRecord>, StoreError> {
89        let row = sqlx::query(
90            r#"
91            SELECT id, project_id, kind, name, enabled, events_json, min_severity,
92                   config_json, target, created_at, updated_at, last_delivery_at,
93                   last_delivery_status, last_delivery_error
94            FROM project_integrations
95            WHERE id = ?
96            "#,
97        )
98        .bind(id)
99        .fetch_optional(self.pool)
100        .await?;
101        row.map(row_to_integration).transpose()
102    }
103
104    pub async fn list_by_project(
105        &self,
106        project_id: &str,
107    ) -> Result<Vec<ProjectIntegrationRecord>, StoreError> {
108        let rows = sqlx::query(
109            r#"
110            SELECT id, project_id, kind, name, enabled, events_json, min_severity,
111                   config_json, target, created_at, updated_at, last_delivery_at,
112                   last_delivery_status, last_delivery_error
113            FROM project_integrations
114            WHERE project_id = ?
115            ORDER BY created_at DESC, name
116            "#,
117        )
118        .bind(project_id)
119        .fetch_all(self.pool)
120        .await?;
121        rows.into_iter().map(row_to_integration_public).collect()
122    }
123
124    pub async fn list_enabled_by_project(
125        &self,
126        project_id: &str,
127    ) -> Result<Vec<ProjectIntegrationStoredRecord>, StoreError> {
128        let rows = sqlx::query(
129            r#"
130            SELECT id, project_id, kind, name, enabled, events_json, min_severity,
131                   config_json, target, created_at, updated_at, last_delivery_at,
132                   last_delivery_status, last_delivery_error
133            FROM project_integrations
134            WHERE project_id = ? AND enabled = 1
135            ORDER BY created_at ASC
136            "#,
137        )
138        .bind(project_id)
139        .fetch_all(self.pool)
140        .await?;
141        rows.into_iter().map(row_to_integration).collect()
142    }
143
144    pub async fn update(
145        &self,
146        id: &str,
147        patch: ProjectIntegrationPatch,
148    ) -> Result<Option<ProjectIntegrationRecord>, StoreError> {
149        let Some(existing) = self.get_stored(id).await? else {
150            return Ok(None);
151        };
152        let public = existing.public;
153        let name = patch.name.unwrap_or(public.name);
154        let enabled = patch.enabled.unwrap_or(public.enabled);
155        let events = patch.events.unwrap_or(public.events);
156        let min_severity = patch.min_severity.unwrap_or(public.min_severity);
157        let config_json = patch.config_json.unwrap_or(existing.config_json);
158        let target = patch.target.unwrap_or(public.target);
159        sqlx::query(
160            r#"
161            UPDATE project_integrations SET
162                name = ?,
163                enabled = ?,
164                events_json = ?,
165                min_severity = ?,
166                config_json = ?,
167                target = ?,
168                updated_at = ?
169            WHERE id = ?
170            "#,
171        )
172        .bind(name)
173        .bind(i64::from(enabled))
174        .bind(serde_json::to_string(&events).map_err(StoreError::IntegrationJson)?)
175        .bind(min_severity)
176        .bind(config_json)
177        .bind(target)
178        .bind(patch.updated_at)
179        .bind(id)
180        .execute(self.pool)
181        .await?;
182        self.get(id).await
183    }
184
185    pub async fn delete(&self, id: &str) -> Result<u64, StoreError> {
186        let res = sqlx::query("DELETE FROM project_integrations WHERE id = ?")
187            .bind(id)
188            .execute(self.pool)
189            .await?;
190        Ok(res.rows_affected())
191    }
192
193    pub async fn record_delivery(
194        &self,
195        id: &str,
196        at_ms: i64,
197        status: &str,
198        error: Option<&str>,
199    ) -> Result<(), StoreError> {
200        sqlx::query(
201            r#"
202            UPDATE project_integrations SET
203                last_delivery_at = ?,
204                last_delivery_status = ?,
205                last_delivery_error = ?
206            WHERE id = ?
207            "#,
208        )
209        .bind(at_ms)
210        .bind(status)
211        .bind(error)
212        .bind(id)
213        .execute(self.pool)
214        .await?;
215        Ok(())
216    }
217}
218
219fn row_to_integration_public(
220    row: sqlx::sqlite::SqliteRow,
221) -> Result<ProjectIntegrationRecord, StoreError> {
222    row_to_integration(row).map(|r| r.public)
223}
224
225fn row_to_integration(
226    row: sqlx::sqlite::SqliteRow,
227) -> Result<ProjectIntegrationStoredRecord, StoreError> {
228    let kind_raw: String = row.try_get("kind")?;
229    let kind = kind_raw
230        .parse::<ProjectIntegrationKind>()
231        .map_err(|()| StoreError::InvalidIntegrationKind(kind_raw.clone()))?;
232    let events_json: String = row.try_get("events_json")?;
233    let events = serde_json::from_str(&events_json).map_err(StoreError::IntegrationJson)?;
234    Ok(ProjectIntegrationStoredRecord {
235        public: ProjectIntegrationRecord {
236            id: row.try_get("id")?,
237            project_id: row.try_get("project_id")?,
238            kind,
239            name: row.try_get("name")?,
240            enabled: row.try_get::<i64, _>("enabled")? != 0,
241            events,
242            min_severity: row.try_get("min_severity")?,
243            target: row.try_get("target")?,
244            created_at: row.try_get::<i64, _>("created_at")?,
245            updated_at: row.try_get::<i64, _>("updated_at")?,
246            last_delivery_at: row.try_get("last_delivery_at")?,
247            last_delivery_status: row.try_get("last_delivery_status")?,
248            last_delivery_error: row.try_get("last_delivery_error")?,
249        },
250        config_json: row.try_get("config_json")?,
251    })
252}
253
254#[cfg(test)]
255mod tests {
256    use super::*;
257    use crate::store::testutil::fresh_store;
258
259    fn insert(project_id: &str, id: &str, enabled: bool) -> ProjectIntegrationInsert {
260        ProjectIntegrationInsert {
261            id: id.to_string(),
262            project_id: project_id.to_string(),
263            kind: ProjectIntegrationKind::Webhook,
264            name: id.to_string(),
265            enabled,
266            events: vec![ProjectIntegrationEvent::RunFinished],
267            min_severity: Some("High".to_string()),
268            config_json: r#"{"kind":"webhook","url":"https://example.invalid"}"#.to_string(),
269            target: "example.invalid".to_string(),
270            now_ms: 1_000,
271        }
272    }
273
274    #[tokio::test]
275    async fn create_then_list_by_project_roundtrips() {
276        let (_tmp, s) = fresh_store().await;
277        s.integrations()
278            .create(insert(crate::store::DEFAULT_PROJECT_ID, "int-1", true))
279            .await
280            .expect("create");
281        let rows =
282            s.integrations().list_by_project(crate::store::DEFAULT_PROJECT_ID).await.unwrap();
283        assert_eq!(rows.len(), 1);
284        assert_eq!(rows[0].target, "example.invalid");
285        assert_eq!(rows[0].events, vec![ProjectIntegrationEvent::RunFinished]);
286    }
287
288    #[tokio::test]
289    async fn enabled_list_excludes_disabled() {
290        let (_tmp, s) = fresh_store().await;
291        s.integrations()
292            .create(insert(crate::store::DEFAULT_PROJECT_ID, "on", true))
293            .await
294            .expect("on");
295        s.integrations()
296            .create(insert(crate::store::DEFAULT_PROJECT_ID, "off", false))
297            .await
298            .expect("off");
299        let rows = s
300            .integrations()
301            .list_enabled_by_project(crate::store::DEFAULT_PROJECT_ID)
302            .await
303            .unwrap();
304        assert_eq!(rows.iter().map(|r| r.public.id.as_str()).collect::<Vec<_>>(), vec!["on"]);
305    }
306
307    #[tokio::test]
308    async fn project_delete_cascades() {
309        let (_tmp, s) = fresh_store().await;
310        s.integrations()
311            .create(insert(crate::store::DEFAULT_PROJECT_ID, "int-1", true))
312            .await
313            .expect("create");
314        s.projects().delete(crate::store::DEFAULT_PROJECT_ID).await.expect("delete");
315        assert!(s.integrations().get("int-1").await.expect("get").is_none());
316    }
317}
nyx_agent_core/store/integration.rs

nyx_agent_core/store/
integration.rs
