zopp_store_postgres/
lib.rs

1use sqlx::{postgres::PgPoolOptions, PgPool};
2use uuid::Uuid;
3use zopp_storage::{
4    AddWorkspacePrincipalParams, CreateEnvParams, CreateInviteParams, CreatePrincipalParams,
5    CreateProjectParams, CreateUserParams, CreateWorkspaceParams, EnvName, Environment,
6    EnvironmentId, Invite, InviteId, Principal, PrincipalId, ProjectName, SecretRow, Store,
7    StoreError, User, UserId, Workspace, WorkspaceId, WorkspacePrincipal,
8};
9
10static MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("./migrations");
11
12pub struct PostgresStore {
13    pool: PgPool,
14}
15
16impl PostgresStore {
17    pub async fn open(url: &str) -> Result<Self, StoreError> {
18        let pool = PgPoolOptions::new()
19            .max_connections(10)
20            .connect(url)
21            .await
22            .map_err(|e| StoreError::Backend(e.to_string()))?;
23
24        MIGRATOR
25            .run(&pool)
26            .await
27            .map_err(|e| StoreError::Backend(e.to_string()))?;
28
29        Ok(Self { pool })
30    }
31}
32
33#[async_trait::async_trait]
34impl Store for PostgresStore {
35    // ───────────────────────────── Users ─────────────────────────────
36
37    async fn create_user(
38        &self,
39        params: &CreateUserParams,
40    ) -> Result<(UserId, Option<PrincipalId>), StoreError> {
41        let needs_tx = params.principal.is_some() || !params.workspace_ids.is_empty();
42
43        if needs_tx {
44            let mut tx = self
45                .pool
46                .begin()
47                .await
48                .map_err(|e| StoreError::Backend(e.to_string()))?;
49
50            let existing_user = sqlx::query!("SELECT id FROM users WHERE email = $1", params.email)
51                .fetch_optional(&mut *tx)
52                .await
53                .map_err(|e| StoreError::Backend(e.to_string()))?;
54
55            let actual_user_id = if let Some(existing) = existing_user {
56                existing.id
57            } else {
58                let user_id = Uuid::now_v7();
59                sqlx::query!(
60                    "INSERT INTO users(id, email) VALUES($1, $2)",
61                    user_id,
62                    params.email
63                )
64                .execute(&mut *tx)
65                .await
66                .map_err(|e| StoreError::Backend(e.to_string()))?;
67                user_id
68            };
69
70            let principal_id = if let Some(principal_data) = &params.principal {
71                let principal_id = Uuid::now_v7();
72
73                let user_id_for_principal = if principal_data.is_service {
74                    None
75                } else {
76                    Some(actual_user_id)
77                };
78
79                sqlx::query!(
80                    "INSERT INTO principals(id, user_id, name, public_key, x25519_public_key) VALUES($1, $2, $3, $4, $5)",
81                    principal_id,
82                    user_id_for_principal,
83                    principal_data.name,
84                    principal_data.public_key,
85                    principal_data.x25519_public_key
86                )
87                .execute(&mut *tx)
88                .await
89                .map_err(|e| StoreError::Backend(e.to_string()))?;
90
91                Some(PrincipalId(principal_id))
92            } else {
93                None
94            };
95
96            for workspace_id in &params.workspace_ids {
97                sqlx::query!(
98                    "INSERT INTO workspace_members(workspace_id, user_id) VALUES($1, $2)",
99                    workspace_id.0,
100                    actual_user_id
101                )
102                .execute(&mut *tx)
103                .await
104                .map_err(|e| StoreError::Backend(e.to_string()))?;
105            }
106
107            tx.commit()
108                .await
109                .map_err(|e| StoreError::Backend(e.to_string()))?;
110
111            Ok((UserId(actual_user_id), principal_id))
112        } else {
113            let user_id = Uuid::now_v7();
114
115            sqlx::query!(
116                "INSERT INTO users(id, email) VALUES($1, $2)",
117                user_id,
118                params.email
119            )
120            .execute(&self.pool)
121            .await
122            .map_err(|e| {
123                let s = e.to_string();
124                if s.contains("duplicate key") || s.contains("unique constraint") {
125                    StoreError::AlreadyExists
126                } else {
127                    StoreError::Backend(s)
128                }
129            })?;
130
131            Ok((UserId(user_id), None))
132        }
133    }
134
135    async fn get_user_by_email(&self, email: &str) -> Result<User, StoreError> {
136        let row = sqlx::query!(
137            r#"SELECT id, email, created_at, updated_at FROM users WHERE email = $1"#,
138            email
139        )
140        .fetch_optional(&self.pool)
141        .await
142        .map_err(|e| StoreError::Backend(e.to_string()))?
143        .ok_or(StoreError::NotFound)?;
144
145        Ok(User {
146            id: UserId(row.id),
147            email: row.email,
148            created_at: row.created_at,
149            updated_at: row.updated_at,
150        })
151    }
152
153    async fn get_user_by_id(&self, user_id: &UserId) -> Result<User, StoreError> {
154        let row = sqlx::query!(
155            r#"SELECT id, email, created_at, updated_at FROM users WHERE id = $1"#,
156            user_id.0
157        )
158        .fetch_optional(&self.pool)
159        .await
160        .map_err(|e| StoreError::Backend(e.to_string()))?
161        .ok_or(StoreError::NotFound)?;
162
163        Ok(User {
164            id: UserId(row.id),
165            email: row.email,
166            created_at: row.created_at,
167            updated_at: row.updated_at,
168        })
169    }
170
171    // ───────────────────────────── Principals ─────────────────────────────
172
173    async fn create_principal(
174        &self,
175        params: &CreatePrincipalParams,
176    ) -> Result<PrincipalId, StoreError> {
177        let principal_id = Uuid::now_v7();
178        let user_id = params.user_id.as_ref().map(|id| id.0);
179
180        sqlx::query!(
181            "INSERT INTO principals(id, user_id, name, public_key, x25519_public_key) VALUES($1, $2, $3, $4, $5)",
182            principal_id,
183            user_id,
184            params.name,
185            params.public_key,
186            params.x25519_public_key
187        )
188        .execute(&self.pool)
189        .await
190        .map_err(|e| {
191            let s = e.to_string();
192            if s.contains("duplicate key") || s.contains("unique constraint") {
193                StoreError::AlreadyExists
194            } else {
195                StoreError::Backend(s)
196            }
197        })?;
198        Ok(PrincipalId(principal_id))
199    }
200
201    async fn get_principal(&self, principal_id: &PrincipalId) -> Result<Principal, StoreError> {
202        let row = sqlx::query!(
203            r#"SELECT id, user_id, name, public_key, x25519_public_key, created_at, updated_at
204               FROM principals WHERE id = $1"#,
205            principal_id.0
206        )
207        .fetch_optional(&self.pool)
208        .await
209        .map_err(|e| StoreError::Backend(e.to_string()))?
210        .ok_or(StoreError::NotFound)?;
211
212        Ok(Principal {
213            id: PrincipalId(row.id),
214            user_id: row.user_id.map(UserId),
215            name: row.name,
216            public_key: row.public_key,
217            x25519_public_key: row.x25519_public_key,
218            created_at: row.created_at,
219            updated_at: row.updated_at,
220        })
221    }
222
223    async fn rename_principal(
224        &self,
225        principal_id: &PrincipalId,
226        new_name: &str,
227    ) -> Result<(), StoreError> {
228        let result = sqlx::query!(
229            "UPDATE principals SET name = $1 WHERE id = $2",
230            new_name,
231            principal_id.0
232        )
233        .execute(&self.pool)
234        .await
235        .map_err(|e| StoreError::Backend(e.to_string()))?;
236
237        if result.rows_affected() == 0 {
238            Err(StoreError::NotFound)
239        } else {
240            Ok(())
241        }
242    }
243
244    async fn list_principals(&self, user_id: &UserId) -> Result<Vec<Principal>, StoreError> {
245        let rows = sqlx::query!(
246            r#"SELECT id, user_id, name, public_key, x25519_public_key, created_at, updated_at
247               FROM principals WHERE user_id = $1"#,
248            user_id.0
249        )
250        .fetch_all(&self.pool)
251        .await
252        .map_err(|e| StoreError::Backend(e.to_string()))?;
253
254        Ok(rows
255            .into_iter()
256            .map(|row| Principal {
257                id: PrincipalId(row.id),
258                user_id: row.user_id.map(UserId),
259                name: row.name,
260                public_key: row.public_key,
261                x25519_public_key: row.x25519_public_key,
262                created_at: row.created_at,
263                updated_at: row.updated_at,
264            })
265            .collect())
266    }
267
268    // ───────────────────────────── Invites ─────────────────────────────
269
270    async fn create_invite(&self, params: &CreateInviteParams) -> Result<Invite, StoreError> {
271        let invite_id = Uuid::now_v7();
272        let created_by_user_id = params.created_by_user_id.as_ref().map(|id| id.0);
273
274        let row = sqlx::query!(
275            r#"INSERT INTO invites(id, token, expires_at, created_by_user_id, kek_encrypted, kek_nonce)
276               VALUES($1, $2, $3, $4, $5, $6)
277               RETURNING created_at, updated_at"#,
278            invite_id,
279            params.token,
280            params.expires_at,
281            created_by_user_id,
282            params.kek_encrypted,
283            params.kek_nonce
284        )
285        .fetch_one(&self.pool)
286        .await
287        .map_err(|e| StoreError::Backend(e.to_string()))?;
288
289        for ws_id in &params.workspace_ids {
290            sqlx::query!(
291                "INSERT INTO invite_workspaces(invite_id, workspace_id) VALUES($1, $2)",
292                invite_id,
293                ws_id.0
294            )
295            .execute(&self.pool)
296            .await
297            .map_err(|e| StoreError::Backend(e.to_string()))?;
298        }
299
300        Ok(Invite {
301            id: InviteId(invite_id),
302            token: params.token.clone(),
303            workspace_ids: params.workspace_ids.clone(),
304            kek_encrypted: params.kek_encrypted.clone(),
305            kek_nonce: params.kek_nonce.clone(),
306            created_at: row.created_at,
307            updated_at: row.updated_at,
308            expires_at: params.expires_at,
309            created_by_user_id: params.created_by_user_id.clone(),
310        })
311    }
312
313    async fn get_invite_by_token(&self, token: &str) -> Result<Invite, StoreError> {
314        let row = sqlx::query!(
315            r#"SELECT id, token, created_at, updated_at, expires_at, created_by_user_id, revoked, kek_encrypted, kek_nonce
316               FROM invites WHERE token = $1"#,
317            token
318        )
319        .fetch_optional(&self.pool)
320        .await
321        .map_err(|e| StoreError::Backend(e.to_string()))?
322        .ok_or(StoreError::NotFound)?;
323
324        if row.revoked {
325            return Err(StoreError::NotFound);
326        }
327
328        let ws_rows = sqlx::query!(
329            "SELECT workspace_id FROM invite_workspaces WHERE invite_id = $1",
330            row.id
331        )
332        .fetch_all(&self.pool)
333        .await
334        .map_err(|e| StoreError::Backend(e.to_string()))?;
335
336        let workspace_ids = ws_rows
337            .into_iter()
338            .map(|r| WorkspaceId(r.workspace_id))
339            .collect();
340
341        Ok(Invite {
342            id: InviteId(row.id),
343            token: row.token,
344            workspace_ids,
345            kek_encrypted: row.kek_encrypted,
346            kek_nonce: row.kek_nonce,
347            created_at: row.created_at,
348            updated_at: row.updated_at,
349            expires_at: row.expires_at,
350            created_by_user_id: row.created_by_user_id.map(UserId),
351        })
352    }
353
354    async fn list_invites(&self, user_id: Option<&UserId>) -> Result<Vec<Invite>, StoreError> {
355        let user_id_opt = user_id.map(|id| id.0);
356
357        let rows = sqlx::query!(
358            r#"SELECT id, token, created_at, updated_at, expires_at, created_by_user_id, revoked, kek_encrypted, kek_nonce
359               FROM invites
360               WHERE revoked = FALSE AND (
361                   ($1::UUID IS NOT NULL AND created_by_user_id = $1) OR
362                   ($1::UUID IS NULL AND created_by_user_id IS NULL)
363               )"#,
364            user_id_opt
365        )
366        .fetch_all(&self.pool)
367        .await
368        .map_err(|e| StoreError::Backend(e.to_string()))?;
369
370        let mut invites = Vec::new();
371        for row in rows {
372            let ws_rows = sqlx::query!(
373                "SELECT workspace_id FROM invite_workspaces WHERE invite_id = $1",
374                row.id
375            )
376            .fetch_all(&self.pool)
377            .await
378            .map_err(|e| StoreError::Backend(e.to_string()))?;
379
380            let workspace_ids = ws_rows
381                .into_iter()
382                .map(|r| WorkspaceId(r.workspace_id))
383                .collect();
384
385            invites.push(Invite {
386                id: InviteId(row.id),
387                token: row.token,
388                workspace_ids,
389                kek_encrypted: row.kek_encrypted,
390                kek_nonce: row.kek_nonce,
391                created_at: row.created_at,
392                updated_at: row.updated_at,
393                expires_at: row.expires_at,
394                created_by_user_id: row.created_by_user_id.map(UserId),
395            });
396        }
397        Ok(invites)
398    }
399
400    async fn revoke_invite(&self, invite_id: &InviteId) -> Result<(), StoreError> {
401        let result = sqlx::query!(
402            "UPDATE invites SET revoked = TRUE WHERE id = $1",
403            invite_id.0
404        )
405        .execute(&self.pool)
406        .await
407        .map_err(|e| StoreError::Backend(e.to_string()))?;
408
409        if result.rows_affected() == 0 {
410            Err(StoreError::NotFound)
411        } else {
412            Ok(())
413        }
414    }
415
416    // ───────────────────────────── Workspaces ─────────────────────────────
417
418    async fn create_workspace(
419        &self,
420        params: &CreateWorkspaceParams,
421    ) -> Result<WorkspaceId, StoreError> {
422        let m_cost = params.m_cost_kib as i32;
423        let t_cost = params.t_cost as i32;
424        let p_cost = params.p_cost as i32;
425
426        sqlx::query!(
427            "INSERT INTO workspaces(id, name, owner_user_id, kdf_salt, kdf_m_cost_kib, kdf_t_cost, kdf_p_cost)
428             VALUES($1, $2, $3, $4, $5, $6, $7)",
429            params.id.0,
430            params.name,
431            params.owner_user_id.0,
432            params.kdf_salt,
433            m_cost,
434            t_cost,
435            p_cost
436        )
437        .execute(&self.pool)
438        .await
439        .map_err(|e| StoreError::Backend(e.to_string()))?;
440        Ok(params.id.clone())
441    }
442
443    async fn list_workspaces(&self, user_id: &UserId) -> Result<Vec<Workspace>, StoreError> {
444        let rows = sqlx::query!(
445            r#"SELECT w.id, w.name, w.owner_user_id, w.kdf_salt, w.kdf_m_cost_kib, w.kdf_t_cost, w.kdf_p_cost,
446               w.created_at, w.updated_at
447               FROM workspaces w
448               JOIN workspace_members wm ON w.id = wm.workspace_id
449               WHERE wm.user_id = $1"#,
450            user_id.0
451        )
452        .fetch_all(&self.pool)
453        .await
454        .map_err(|e| StoreError::Backend(e.to_string()))?;
455
456        Ok(rows
457            .into_iter()
458            .map(|row| Workspace {
459                id: WorkspaceId(row.id),
460                name: row.name,
461                owner_user_id: UserId(row.owner_user_id),
462                kdf_salt: row.kdf_salt,
463                m_cost_kib: row.kdf_m_cost_kib as u32,
464                t_cost: row.kdf_t_cost as u32,
465                p_cost: row.kdf_p_cost as u32,
466                created_at: row.created_at,
467                updated_at: row.updated_at,
468            })
469            .collect())
470    }
471
472    async fn get_workspace(&self, ws: &WorkspaceId) -> Result<Workspace, StoreError> {
473        let row = sqlx::query!(
474            r#"SELECT id, name, owner_user_id, kdf_salt, kdf_m_cost_kib, kdf_t_cost, kdf_p_cost,
475               created_at, updated_at
476               FROM workspaces WHERE id = $1"#,
477            ws.0
478        )
479        .fetch_optional(&self.pool)
480        .await
481        .map_err(|e| StoreError::Backend(e.to_string()))?
482        .ok_or(StoreError::NotFound)?;
483
484        Ok(Workspace {
485            id: WorkspaceId(row.id),
486            name: row.name,
487            owner_user_id: UserId(row.owner_user_id),
488            kdf_salt: row.kdf_salt,
489            m_cost_kib: row.kdf_m_cost_kib as u32,
490            t_cost: row.kdf_t_cost as u32,
491            p_cost: row.kdf_p_cost as u32,
492            created_at: row.created_at,
493            updated_at: row.updated_at,
494        })
495    }
496
497    async fn get_workspace_by_name(
498        &self,
499        user_id: &UserId,
500        name: &str,
501    ) -> Result<Workspace, StoreError> {
502        let row = sqlx::query!(
503            r#"SELECT w.id, w.name, w.owner_user_id, w.kdf_salt, w.kdf_m_cost_kib, w.kdf_t_cost, w.kdf_p_cost,
504               w.created_at, w.updated_at
505               FROM workspaces w
506               LEFT JOIN workspace_members wm ON w.id = wm.workspace_id
507               WHERE w.name = $1 AND (w.owner_user_id = $2 OR wm.user_id = $2)
508               LIMIT 1"#,
509            name,
510            user_id.0
511        )
512        .fetch_optional(&self.pool)
513        .await
514        .map_err(|e| StoreError::Backend(e.to_string()))?
515        .ok_or(StoreError::NotFound)?;
516
517        Ok(Workspace {
518            id: WorkspaceId(row.id),
519            name: row.name,
520            owner_user_id: UserId(row.owner_user_id),
521            kdf_salt: row.kdf_salt,
522            m_cost_kib: row.kdf_m_cost_kib as u32,
523            t_cost: row.kdf_t_cost as u32,
524            p_cost: row.kdf_p_cost as u32,
525            created_at: row.created_at,
526            updated_at: row.updated_at,
527        })
528    }
529
530    async fn get_workspace_by_name_for_principal(
531        &self,
532        principal_id: &PrincipalId,
533        name: &str,
534    ) -> Result<Workspace, StoreError> {
535        let row = sqlx::query!(
536            r#"SELECT w.id, w.name, w.owner_user_id, w.kdf_salt, w.kdf_m_cost_kib, w.kdf_t_cost, w.kdf_p_cost,
537               w.created_at, w.updated_at
538               FROM workspaces w
539               INNER JOIN workspace_principals wp ON w.id = wp.workspace_id
540               WHERE w.name = $1 AND wp.principal_id = $2
541               LIMIT 1"#,
542            name,
543            principal_id.0
544        )
545        .fetch_optional(&self.pool)
546        .await
547        .map_err(|e| StoreError::Backend(e.to_string()))?
548        .ok_or(StoreError::NotFound)?;
549
550        Ok(Workspace {
551            id: WorkspaceId(row.id),
552            name: row.name,
553            owner_user_id: UserId(row.owner_user_id),
554            kdf_salt: row.kdf_salt,
555            m_cost_kib: row.kdf_m_cost_kib as u32,
556            t_cost: row.kdf_t_cost as u32,
557            p_cost: row.kdf_p_cost as u32,
558            created_at: row.created_at,
559            updated_at: row.updated_at,
560        })
561    }
562
563    async fn add_workspace_principal(
564        &self,
565        params: &AddWorkspacePrincipalParams,
566    ) -> Result<(), StoreError> {
567        sqlx::query!(
568            "INSERT INTO workspace_principals(workspace_id, principal_id, ephemeral_pub, kek_wrapped, kek_nonce) VALUES($1, $2, $3, $4, $5)",
569            params.workspace_id.0,
570            params.principal_id.0,
571            params.ephemeral_pub,
572            params.kek_wrapped,
573            params.kek_nonce
574        )
575        .execute(&self.pool)
576        .await
577        .map_err(|e| {
578            let s = e.to_string();
579            if s.contains("duplicate key") || s.contains("unique constraint") {
580                StoreError::AlreadyExists
581            } else {
582                StoreError::Backend(s)
583            }
584        })?;
585        Ok(())
586    }
587
588    async fn get_workspace_principal(
589        &self,
590        workspace_id: &WorkspaceId,
591        principal_id: &PrincipalId,
592    ) -> Result<WorkspacePrincipal, StoreError> {
593        let row = sqlx::query!(
594            r#"SELECT workspace_id, principal_id, ephemeral_pub, kek_wrapped, kek_nonce, created_at
595               FROM workspace_principals WHERE workspace_id = $1 AND principal_id = $2"#,
596            workspace_id.0,
597            principal_id.0
598        )
599        .fetch_optional(&self.pool)
600        .await
601        .map_err(|e| StoreError::Backend(e.to_string()))?
602        .ok_or(StoreError::NotFound)?;
603
604        Ok(WorkspacePrincipal {
605            workspace_id: WorkspaceId(row.workspace_id),
606            principal_id: PrincipalId(row.principal_id),
607            ephemeral_pub: row.ephemeral_pub,
608            kek_wrapped: row.kek_wrapped,
609            kek_nonce: row.kek_nonce,
610            created_at: row.created_at,
611        })
612    }
613
614    async fn list_workspace_principals(
615        &self,
616        workspace_id: &WorkspaceId,
617    ) -> Result<Vec<WorkspacePrincipal>, StoreError> {
618        let rows = sqlx::query!(
619            r#"SELECT workspace_id, principal_id, ephemeral_pub, kek_wrapped, kek_nonce, created_at
620               FROM workspace_principals WHERE workspace_id = $1"#,
621            workspace_id.0
622        )
623        .fetch_all(&self.pool)
624        .await
625        .map_err(|e| StoreError::Backend(e.to_string()))?;
626
627        Ok(rows
628            .into_iter()
629            .map(|row| WorkspacePrincipal {
630                workspace_id: WorkspaceId(row.workspace_id),
631                principal_id: PrincipalId(row.principal_id),
632                ephemeral_pub: row.ephemeral_pub,
633                kek_wrapped: row.kek_wrapped,
634                kek_nonce: row.kek_nonce,
635                created_at: row.created_at,
636            })
637            .collect())
638    }
639
640    async fn add_user_to_workspace(
641        &self,
642        workspace_id: &WorkspaceId,
643        user_id: &UserId,
644    ) -> Result<(), StoreError> {
645        sqlx::query!(
646            "INSERT INTO workspace_members(workspace_id, user_id) VALUES($1, $2)",
647            workspace_id.0,
648            user_id.0
649        )
650        .execute(&self.pool)
651        .await
652        .map_err(|e| {
653            let s = e.to_string();
654            if s.contains("duplicate key") || s.contains("unique constraint") {
655                StoreError::AlreadyExists
656            } else {
657                StoreError::Backend(s)
658            }
659        })?;
660        Ok(())
661    }
662
663    // ───────────────────────────── Projects ───────────────────────────────
664
665    async fn create_project(
666        &self,
667        params: &CreateProjectParams,
668    ) -> Result<zopp_storage::ProjectId, StoreError> {
669        let id = Uuid::now_v7();
670
671        sqlx::query!(
672            "INSERT INTO projects(id, workspace_id, name) VALUES($1, $2, $3)",
673            id,
674            params.workspace_id.0,
675            params.name
676        )
677        .execute(&self.pool)
678        .await
679        .map_err(|e| {
680            let s = e.to_string();
681            if s.contains("duplicate key") || s.contains("unique constraint") {
682                StoreError::AlreadyExists
683            } else {
684                StoreError::Backend(s)
685            }
686        })?;
687        Ok(zopp_storage::ProjectId(id))
688    }
689
690    async fn list_projects(
691        &self,
692        workspace_id: &WorkspaceId,
693    ) -> Result<Vec<zopp_storage::Project>, StoreError> {
694        let rows = sqlx::query!(
695            r#"SELECT id, workspace_id, name, created_at, updated_at
696               FROM projects WHERE workspace_id = $1 ORDER BY created_at DESC"#,
697            workspace_id.0
698        )
699        .fetch_all(&self.pool)
700        .await
701        .map_err(|e| StoreError::Backend(e.to_string()))?;
702
703        Ok(rows
704            .into_iter()
705            .map(|row| zopp_storage::Project {
706                id: zopp_storage::ProjectId(row.id),
707                workspace_id: WorkspaceId(row.workspace_id),
708                name: row.name,
709                created_at: row.created_at,
710                updated_at: row.updated_at,
711            })
712            .collect())
713    }
714
715    async fn get_project(
716        &self,
717        project_id: &zopp_storage::ProjectId,
718    ) -> Result<zopp_storage::Project, StoreError> {
719        let row = sqlx::query!(
720            r#"SELECT id, workspace_id, name, created_at, updated_at
721               FROM projects WHERE id = $1"#,
722            project_id.0
723        )
724        .fetch_optional(&self.pool)
725        .await
726        .map_err(|e| StoreError::Backend(e.to_string()))?
727        .ok_or(StoreError::NotFound)?;
728
729        Ok(zopp_storage::Project {
730            id: zopp_storage::ProjectId(row.id),
731            workspace_id: WorkspaceId(row.workspace_id),
732            name: row.name,
733            created_at: row.created_at,
734            updated_at: row.updated_at,
735        })
736    }
737
738    async fn get_project_by_name(
739        &self,
740        workspace_id: &WorkspaceId,
741        name: &str,
742    ) -> Result<zopp_storage::Project, StoreError> {
743        let row = sqlx::query!(
744            r#"SELECT id, workspace_id, name, created_at, updated_at
745               FROM projects WHERE workspace_id = $1 AND name = $2"#,
746            workspace_id.0,
747            name
748        )
749        .fetch_optional(&self.pool)
750        .await
751        .map_err(|e| StoreError::Backend(e.to_string()))?
752        .ok_or(StoreError::NotFound)?;
753
754        Ok(zopp_storage::Project {
755            id: zopp_storage::ProjectId(row.id),
756            workspace_id: WorkspaceId(row.workspace_id),
757            name: row.name,
758            created_at: row.created_at,
759            updated_at: row.updated_at,
760        })
761    }
762
763    async fn delete_project(&self, project_id: &zopp_storage::ProjectId) -> Result<(), StoreError> {
764        let result = sqlx::query!("DELETE FROM projects WHERE id = $1", project_id.0)
765            .execute(&self.pool)
766            .await
767            .map_err(|e| StoreError::Backend(e.to_string()))?;
768
769        if result.rows_affected() == 0 {
770            return Err(StoreError::NotFound);
771        }
772
773        Ok(())
774    }
775
776    // ──────────────────────────── Environments ────────────────────────────
777
778    async fn create_env(&self, params: &CreateEnvParams) -> Result<EnvironmentId, StoreError> {
779        let env_id = Uuid::now_v7();
780
781        let proj_row = sqlx::query!(
782            "SELECT workspace_id FROM projects WHERE id = $1",
783            params.project_id.0
784        )
785        .fetch_optional(&self.pool)
786        .await
787        .map_err(|e| StoreError::Backend(e.to_string()))?
788        .ok_or(StoreError::NotFound)?;
789
790        sqlx::query!(
791            "INSERT INTO environments(id, workspace_id, project_id, name, dek_wrapped, dek_nonce)
792             VALUES($1, $2, $3, $4, $5, $6)",
793            env_id,
794            proj_row.workspace_id,
795            params.project_id.0,
796            params.name,
797            params.dek_wrapped,
798            params.dek_nonce
799        )
800        .execute(&self.pool)
801        .await
802        .map_err(|e| {
803            let s = e.to_string();
804            if s.contains("duplicate key") || s.contains("unique constraint") {
805                StoreError::AlreadyExists
806            } else {
807                StoreError::Backend(s)
808            }
809        })?;
810
811        Ok(EnvironmentId(env_id))
812    }
813
814    async fn list_environments(
815        &self,
816        project_id: &zopp_storage::ProjectId,
817    ) -> Result<Vec<Environment>, StoreError> {
818        let rows = sqlx::query!(
819            r#"SELECT id, project_id, name, dek_wrapped, dek_nonce, version, created_at, updated_at
820               FROM environments WHERE project_id = $1 ORDER BY created_at DESC"#,
821            project_id.0
822        )
823        .fetch_all(&self.pool)
824        .await
825        .map_err(|e| StoreError::Backend(e.to_string()))?;
826
827        Ok(rows
828            .into_iter()
829            .map(|row| Environment {
830                id: EnvironmentId(row.id),
831                project_id: zopp_storage::ProjectId(row.project_id),
832                name: row.name,
833                dek_wrapped: row.dek_wrapped,
834                dek_nonce: row.dek_nonce,
835                version: row.version,
836                created_at: row.created_at,
837                updated_at: row.updated_at,
838            })
839            .collect())
840    }
841
842    async fn get_environment(&self, env_id: &EnvironmentId) -> Result<Environment, StoreError> {
843        let row = sqlx::query!(
844            r#"SELECT id, project_id, name, dek_wrapped, dek_nonce, version, created_at, updated_at
845               FROM environments WHERE id = $1"#,
846            env_id.0
847        )
848        .fetch_optional(&self.pool)
849        .await
850        .map_err(|e| StoreError::Backend(e.to_string()))?
851        .ok_or(StoreError::NotFound)?;
852
853        Ok(Environment {
854            id: EnvironmentId(row.id),
855            project_id: zopp_storage::ProjectId(row.project_id),
856            name: row.name,
857            dek_wrapped: row.dek_wrapped,
858            dek_nonce: row.dek_nonce,
859            version: row.version,
860            created_at: row.created_at,
861            updated_at: row.updated_at,
862        })
863    }
864
865    async fn get_environment_by_name(
866        &self,
867        project_id: &zopp_storage::ProjectId,
868        name: &str,
869    ) -> Result<Environment, StoreError> {
870        let row = sqlx::query!(
871            r#"SELECT id, project_id, name, dek_wrapped, dek_nonce, version, created_at, updated_at
872               FROM environments WHERE project_id = $1 AND name = $2"#,
873            project_id.0,
874            name
875        )
876        .fetch_optional(&self.pool)
877        .await
878        .map_err(|e| StoreError::Backend(e.to_string()))?
879        .ok_or(StoreError::NotFound)?;
880
881        Ok(Environment {
882            id: EnvironmentId(row.id),
883            project_id: zopp_storage::ProjectId(row.project_id),
884            name: row.name,
885            dek_wrapped: row.dek_wrapped,
886            dek_nonce: row.dek_nonce,
887            version: row.version,
888            created_at: row.created_at,
889            updated_at: row.updated_at,
890        })
891    }
892
893    async fn delete_environment(&self, env_id: &EnvironmentId) -> Result<(), StoreError> {
894        let result = sqlx::query!("DELETE FROM environments WHERE id = $1", env_id.0)
895            .execute(&self.pool)
896            .await
897            .map_err(|e| StoreError::Backend(e.to_string()))?;
898
899        if result.rows_affected() == 0 {
900            return Err(StoreError::NotFound);
901        }
902
903        Ok(())
904    }
905
906    async fn get_env_wrap(
907        &self,
908        ws: &WorkspaceId,
909        project: &ProjectName,
910        env: &EnvName,
911    ) -> Result<(Vec<u8>, Vec<u8>), StoreError> {
912        let row = sqlx::query!(
913            "SELECT e.dek_wrapped, e.dek_nonce
914             FROM environments e
915             JOIN projects p ON p.id = e.project_id
916             WHERE e.workspace_id = $1 AND p.workspace_id = $1 AND p.name = $2 AND e.name = $3",
917            ws.0,
918            project.0,
919            env.0
920        )
921        .fetch_optional(&self.pool)
922        .await
923        .map_err(|e| StoreError::Backend(e.to_string()))?
924        .ok_or(StoreError::NotFound)?;
925
926        Ok((row.dek_wrapped, row.dek_nonce))
927    }
928
929    // ────────────────────────────── Secrets ───────────────────────────────
930
931    async fn upsert_secret(
932        &self,
933        env_id: &EnvironmentId,
934        key: &str,
935        nonce: &[u8],
936        ciphertext: &[u8],
937    ) -> Result<i64, StoreError> {
938        let mut tx = self
939            .pool
940            .begin()
941            .await
942            .map_err(|e| StoreError::Backend(e.to_string()))?;
943
944        let env_row = sqlx::query!(
945            "SELECT workspace_id FROM environments WHERE id = $1",
946            env_id.0
947        )
948        .fetch_optional(&mut *tx)
949        .await
950        .map_err(|e| StoreError::Backend(e.to_string()))?
951        .ok_or(StoreError::NotFound)?;
952
953        let secret_id = Uuid::now_v7();
954
955        sqlx::query!(
956            "INSERT INTO secrets(id, workspace_id, env_id, key_name, nonce, ciphertext)
957             VALUES($1, $2, $3, $4, $5, $6)
958             ON CONFLICT(workspace_id, env_id, key_name)
959             DO UPDATE SET nonce = EXCLUDED.nonce, ciphertext = EXCLUDED.ciphertext",
960            secret_id,
961            env_row.workspace_id,
962            env_id.0,
963            key,
964            nonce,
965            ciphertext
966        )
967        .execute(&mut *tx)
968        .await
969        .map_err(|e| StoreError::Backend(e.to_string()))?;
970
971        let result = sqlx::query!(
972            "UPDATE environments SET version = version + 1 WHERE id = $1 RETURNING version",
973            env_id.0
974        )
975        .fetch_one(&mut *tx)
976        .await
977        .map_err(|e| StoreError::Backend(e.to_string()))?;
978
979        tx.commit()
980            .await
981            .map_err(|e| StoreError::Backend(e.to_string()))?;
982
983        Ok(result.version)
984    }
985
986    async fn get_secret(&self, env_id: &EnvironmentId, key: &str) -> Result<SecretRow, StoreError> {
987        let row = sqlx::query!(
988            "SELECT nonce, ciphertext FROM secrets WHERE env_id = $1 AND key_name = $2",
989            env_id.0,
990            key
991        )
992        .fetch_optional(&self.pool)
993        .await
994        .map_err(|e| StoreError::Backend(e.to_string()))?
995        .ok_or(StoreError::NotFound)?;
996
997        Ok(SecretRow {
998            nonce: row.nonce,
999            ciphertext: row.ciphertext,
1000        })
1001    }
1002
1003    async fn list_secret_keys(&self, env_id: &EnvironmentId) -> Result<Vec<String>, StoreError> {
1004        let rows = sqlx::query!(
1005            "SELECT key_name FROM secrets WHERE env_id = $1 ORDER BY key_name",
1006            env_id.0
1007        )
1008        .fetch_all(&self.pool)
1009        .await
1010        .map_err(|e| StoreError::Backend(e.to_string()))?;
1011
1012        Ok(rows.into_iter().map(|row| row.key_name).collect())
1013    }
1014
1015    async fn delete_secret(&self, env_id: &EnvironmentId, key: &str) -> Result<i64, StoreError> {
1016        let mut tx = self
1017            .pool
1018            .begin()
1019            .await
1020            .map_err(|e| StoreError::Backend(e.to_string()))?;
1021
1022        let result = sqlx::query!(
1023            "DELETE FROM secrets WHERE env_id = $1 AND key_name = $2",
1024            env_id.0,
1025            key
1026        )
1027        .execute(&mut *tx)
1028        .await
1029        .map_err(|e| StoreError::Backend(e.to_string()))?;
1030
1031        if result.rows_affected() == 0 {
1032            return Err(StoreError::NotFound);
1033        }
1034
1035        let version_result = sqlx::query!(
1036            "UPDATE environments SET version = version + 1 WHERE id = $1 RETURNING version",
1037            env_id.0
1038        )
1039        .fetch_one(&mut *tx)
1040        .await
1041        .map_err(|e| StoreError::Backend(e.to_string()))?;
1042
1043        tx.commit()
1044            .await
1045            .map_err(|e| StoreError::Backend(e.to_string()))?;
1046
1047        Ok(version_result.version)
1048    }
1049}
1050
1051#[cfg(test)]
1052mod tests;
zopp_store_postgres/lib.rs

zopp_store_postgres/
lib.rs
