From 77c449f455766903ce25cc906f74105270652f28 Mon Sep 17 00:00:00 2001 From: "a.tolmachev" Date: Sun, 12 Apr 2026 11:54:18 +0300 Subject: [PATCH] registry: extract storage domain modules --- crates/crank-registry/src/postgres/auth.rs | 437 +++++ .../src/{postgres.rs => postgres/mod.rs} | 1488 +---------------- crates/crank-registry/src/postgres/secret.rs | 261 +++ crates/crank-registry/src/postgres/stream.rs | 478 ++++++ .../crank-registry/src/postgres/workspace.rs | 319 ++++ 5 files changed, 1500 insertions(+), 1483 deletions(-) create mode 100644 crates/crank-registry/src/postgres/auth.rs rename crates/crank-registry/src/{postgres.rs => postgres/mod.rs} (74%) create mode 100644 crates/crank-registry/src/postgres/secret.rs create mode 100644 crates/crank-registry/src/postgres/stream.rs create mode 100644 crates/crank-registry/src/postgres/workspace.rs diff --git a/crates/crank-registry/src/postgres/auth.rs b/crates/crank-registry/src/postgres/auth.rs new file mode 100644 index 0000000..8a11da2 --- /dev/null +++ b/crates/crank-registry/src/postgres/auth.rs @@ -0,0 +1,437 @@ +use super::*; + +impl PostgresRegistry { + pub async fn upsert_bootstrap_user( + &self, + email: &str, + display_name: &str, + password_hash: &str, + ) -> Result { + let row = sqlx::query( + "insert into users ( + id, + email, + display_name, + password_hash, + status, + created_at + ) values ( + $1, $2, $3, $4, 'active', now() + ) + on conflict (email) do update + set display_name = excluded.display_name, + password_hash = excluded.password_hash, + status = 'active' + returning id", + ) + .bind(format!("user_{}", uuid::Uuid::now_v7().simple())) + .bind(email) + .bind(display_name) + .bind(password_hash) + .fetch_one(&self.pool) + .await?; + + Ok(UserId::new(row.try_get::("id")?)) + } + + pub async fn ensure_membership( + &self, + workspace_id: &WorkspaceId, + user_id: &UserId, + role: MembershipRole, + ) -> Result<(), RegistryError> { + sqlx::query( + "insert into memberships ( + workspace_id, + user_id, + role, + created_at + ) values ( + $1, $2, $3, now() + ) + on conflict (workspace_id, user_id) do update + set role = excluded.role", + ) + .bind(workspace_id.as_str()) + .bind(user_id.as_str()) + .bind(serialize_enum_text(&role, "role")?) + .execute(&self.pool) + .await?; + + Ok(()) + } + + pub async fn get_auth_user_by_email( + &self, + email: &str, + ) -> Result, RegistryError> { + let row = sqlx::query( + "select + id, + email, + display_name, + password_hash, + status, + to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at + from users + where email = $1 + limit 1", + ) + .bind(email) + .fetch_optional(&self.pool) + .await?; + + row.as_ref().map(map_auth_user_record).transpose() + } + + pub async fn get_auth_user_by_id( + &self, + user_id: &UserId, + ) -> Result, RegistryError> { + let row = sqlx::query( + "select + id, + email, + display_name, + password_hash, + status, + to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at + from users + where id = $1 + limit 1", + ) + .bind(user_id.as_str()) + .fetch_optional(&self.pool) + .await?; + + row.as_ref().map(map_auth_user_record).transpose() + } + + pub async fn update_user_profile( + &self, + user_id: &UserId, + email: &str, + display_name: &str, + ) -> Result { + let row = sqlx::query( + "update users + set email = $2, + display_name = $3 + where id = $1 + returning + id, + email, + display_name, + status, + to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at", + ) + .bind(user_id.as_str()) + .bind(email) + .bind(display_name) + .fetch_one(&self.pool) + .await + .map_err(|error| map_user_update_error(error, user_id, email))?; + + map_user(&row) + } + + pub async fn update_user_password( + &self, + user_id: &UserId, + password_hash: &str, + ) -> Result<(), RegistryError> { + let result = sqlx::query( + "update users + set password_hash = $2 + where id = $1", + ) + .bind(user_id.as_str()) + .bind(password_hash) + .execute(&self.pool) + .await?; + + if result.rows_affected() == 0 { + return Err(RegistryError::UserNotFound { + user_id: user_id.as_str().to_owned(), + }); + } + + Ok(()) + } + + pub async fn create_user_session( + &self, + session_id: &UserSessionId, + user_id: &UserId, + current_workspace_id: Option<&WorkspaceId>, + secret_hash: &str, + expires_at: &str, + ) -> Result<(), RegistryError> { + sqlx::query( + "insert into user_sessions ( + id, + user_id, + current_workspace_id, + secret_hash, + status, + expires_at, + last_seen_at, + created_at + ) values ( + $1, + $2, + $3, + $4, + 'active', + $5::timestamptz, + now(), + now() + )", + ) + .bind(session_id.as_str()) + .bind(user_id.as_str()) + .bind(current_workspace_id.map(|id| id.as_str())) + .bind(secret_hash) + .bind(expires_at) + .execute(&self.pool) + .await?; + + Ok(()) + } + + pub async fn get_user_session( + &self, + session_id: &UserSessionId, + secret_hash: &str, + ) -> Result, RegistryError> { + let row = sqlx::query( + "select + s.id, + s.user_id, + s.current_workspace_id, + u.email, + u.display_name, + u.status, + to_char(u.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at + from user_sessions s + join users u on u.id = s.user_id + where s.id = $1 + and s.secret_hash = $2 + and s.status = 'active' + and s.expires_at > now() + limit 1", + ) + .bind(session_id.as_str()) + .bind(secret_hash) + .fetch_optional(&self.pool) + .await?; + + let Some(row) = row else { + return Ok(None); + }; + + let user_id = UserId::new(row.try_get::("user_id")?); + let user = User { + id: user_id.clone(), + email: row.try_get("email")?, + display_name: row.try_get("display_name")?, + status: deserialize_enum_text(&row.try_get::("status")?, "status")?, + created_at: row.try_get("created_at")?, + }; + let memberships = self.list_workspaces_for_user(&user_id).await?; + let stored_workspace_id = row + .try_get::, _>("current_workspace_id")? + .map(WorkspaceId::new); + let current_workspace_id = stored_workspace_id + .filter(|workspace_id| { + memberships + .iter() + .any(|membership| membership.workspace.id == *workspace_id) + }) + .or_else(|| { + memberships + .first() + .map(|membership| membership.workspace.id.clone()) + }); + + Ok(Some(SessionRecord { + session_id: UserSessionId::new(row.try_get::("id")?), + user, + memberships, + current_workspace_id, + })) + } + + pub async fn touch_user_session( + &self, + session_id: &UserSessionId, + ) -> Result<(), RegistryError> { + sqlx::query( + "update user_sessions + set last_seen_at = now() + where id = $1", + ) + .bind(session_id.as_str()) + .execute(&self.pool) + .await?; + + Ok(()) + } + + pub async fn revoke_user_session( + &self, + session_id: &UserSessionId, + ) -> Result<(), RegistryError> { + sqlx::query( + "update user_sessions + set status = 'revoked' + where id = $1", + ) + .bind(session_id.as_str()) + .execute(&self.pool) + .await?; + + Ok(()) + } + + pub async fn set_user_session_current_workspace( + &self, + session_id: &UserSessionId, + workspace_id: &WorkspaceId, + ) -> Result<(), RegistryError> { + sqlx::query( + "update user_sessions + set current_workspace_id = $2 + where id = $1", + ) + .bind(session_id.as_str()) + .bind(workspace_id.as_str()) + .execute(&self.pool) + .await?; + + Ok(()) + } + + pub async fn user_has_workspace_access( + &self, + user_id: &UserId, + workspace_id: &WorkspaceId, + ) -> Result { + let row = sqlx::query( + "select exists( + select 1 + from memberships + where user_id = $1 + and workspace_id = $2 + ) as allowed", + ) + .bind(user_id.as_str()) + .bind(workspace_id.as_str()) + .fetch_one(&self.pool) + .await?; + + Ok(row.try_get("allowed")?) + } + + pub async fn save_auth_profile( + &self, + request: SaveAuthProfileRequest<'_>, + ) -> Result<(), RegistryError> { + sqlx::query( + "insert into auth_profiles ( + id, + workspace_id, + name, + kind, + config_json, + created_at, + updated_at + ) values ($1, $2, $3, $4, $5, $6::timestamptz, $7::timestamptz) + on conflict(id) do update set + workspace_id = excluded.workspace_id, + name = excluded.name, + kind = excluded.kind, + config_json = excluded.config_json, + updated_at = excluded.updated_at", + ) + .bind(request.profile.id.as_str()) + .bind(request.workspace_id.as_str()) + .bind(&request.profile.name) + .bind(serialize_enum_text(&request.profile.kind, "kind")?) + .bind(Json(serialize_json_value(&request.profile.config)?)) + .bind(&request.profile.created_at) + .bind(&request.profile.updated_at) + .execute(&self.pool) + .await?; + + Ok(()) + } + + pub async fn get_auth_profile( + &self, + workspace_id: &WorkspaceId, + auth_profile_id: &crank_core::AuthProfileId, + ) -> Result, RegistryError> { + let row = sqlx::query( + "select + id, + workspace_id, + name, + kind, + config_json, + to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, + to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at + from auth_profiles + where workspace_id = $1 and id = $2", + ) + .bind(workspace_id.as_str()) + .bind(auth_profile_id.as_str()) + .fetch_optional(&self.pool) + .await?; + + row.as_ref().map(map_auth_profile).transpose() + } + + pub async fn list_auth_profiles( + &self, + workspace_id: &WorkspaceId, + ) -> Result, RegistryError> { + let rows = sqlx::query( + "select + id, + workspace_id, + name, + kind, + config_json, + to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, + to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at + from auth_profiles + where workspace_id = $1 + order by name asc", + ) + .bind(workspace_id.as_str()) + .fetch_all(&self.pool) + .await?; + + rows.iter().map(map_auth_profile).collect() + } + + pub async fn list_auth_profiles_referencing_secret( + &self, + workspace_id: &WorkspaceId, + secret_id: &SecretId, + ) -> Result, RegistryError> { + let profiles = self.list_auth_profiles(workspace_id).await?; + + Ok(profiles + .into_iter() + .filter(|profile| { + profile + .config + .secret_ids() + .into_iter() + .any(|candidate| candidate == secret_id) + }) + .collect()) + } +} diff --git a/crates/crank-registry/src/postgres.rs b/crates/crank-registry/src/postgres/mod.rs similarity index 74% rename from crates/crank-registry/src/postgres.rs rename to crates/crank-registry/src/postgres/mod.rs index e91d0ca..92f60b2 100644 --- a/crates/crank-registry/src/postgres.rs +++ b/crates/crank-registry/src/postgres/mod.rs @@ -1,3 +1,8 @@ +mod auth; +mod secret; +mod stream; +mod workspace; + use crank_core::{ AgentId, AgentOperationBinding, AgentStatus, AgentVersion, AsyncJobHandle, AsyncJobId, AuthProfile, GraphqlOperationType, HttpMethod, InvitationId, InvitationToken, InvocationLog, @@ -171,553 +176,6 @@ impl PostgresRegistry { Ok(()) } - pub async fn list_workspaces(&self) -> Result, RegistryError> { - let rows = sqlx::query( - "select - id, - slug, - display_name, - status, - settings_json, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at - from workspaces - order by slug asc", - ) - .fetch_all(&self.pool) - .await?; - - rows.iter().map(map_workspace_record).collect() - } - - pub async fn list_workspaces_for_user( - &self, - user_id: &UserId, - ) -> Result, RegistryError> { - let rows = sqlx::query( - "select - w.id, - w.slug, - w.display_name, - w.status, - w.settings_json, - to_char(w.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - to_char(w.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at, - m.role - from memberships m - join workspaces w on w.id = m.workspace_id - where m.user_id = $1 - order by w.slug asc", - ) - .bind(user_id.as_str()) - .fetch_all(&self.pool) - .await?; - - rows.iter().map(map_workspace_membership_record).collect() - } - - pub async fn upsert_bootstrap_user( - &self, - email: &str, - display_name: &str, - password_hash: &str, - ) -> Result { - let row = sqlx::query( - "insert into users ( - id, - email, - display_name, - password_hash, - status, - created_at - ) values ( - $1, $2, $3, $4, 'active', now() - ) - on conflict (email) do update - set display_name = excluded.display_name, - password_hash = excluded.password_hash, - status = 'active' - returning id", - ) - .bind(format!("user_{}", uuid::Uuid::now_v7().simple())) - .bind(email) - .bind(display_name) - .bind(password_hash) - .fetch_one(&self.pool) - .await?; - - Ok(UserId::new(row.try_get::("id")?)) - } - - pub async fn ensure_membership( - &self, - workspace_id: &WorkspaceId, - user_id: &UserId, - role: MembershipRole, - ) -> Result<(), RegistryError> { - sqlx::query( - "insert into memberships ( - workspace_id, - user_id, - role, - created_at - ) values ( - $1, $2, $3, now() - ) - on conflict (workspace_id, user_id) do update - set role = excluded.role", - ) - .bind(workspace_id.as_str()) - .bind(user_id.as_str()) - .bind(serialize_enum_text(&role, "role")?) - .execute(&self.pool) - .await?; - - Ok(()) - } - - pub async fn get_auth_user_by_email( - &self, - email: &str, - ) -> Result, RegistryError> { - let row = sqlx::query( - "select - id, - email, - display_name, - password_hash, - status, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at - from users - where email = $1 - limit 1", - ) - .bind(email) - .fetch_optional(&self.pool) - .await?; - - row.as_ref().map(map_auth_user_record).transpose() - } - - pub async fn get_auth_user_by_id( - &self, - user_id: &UserId, - ) -> Result, RegistryError> { - let row = sqlx::query( - "select - id, - email, - display_name, - password_hash, - status, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at - from users - where id = $1 - limit 1", - ) - .bind(user_id.as_str()) - .fetch_optional(&self.pool) - .await?; - - row.as_ref().map(map_auth_user_record).transpose() - } - - pub async fn update_user_profile( - &self, - user_id: &UserId, - email: &str, - display_name: &str, - ) -> Result { - let row = sqlx::query( - "update users - set email = $2, - display_name = $3 - where id = $1 - returning - id, - email, - display_name, - status, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at", - ) - .bind(user_id.as_str()) - .bind(email) - .bind(display_name) - .fetch_one(&self.pool) - .await - .map_err(|error| map_user_update_error(error, user_id, email))?; - - map_user(&row) - } - - pub async fn update_user_password( - &self, - user_id: &UserId, - password_hash: &str, - ) -> Result<(), RegistryError> { - let result = sqlx::query( - "update users - set password_hash = $2 - where id = $1", - ) - .bind(user_id.as_str()) - .bind(password_hash) - .execute(&self.pool) - .await?; - - if result.rows_affected() == 0 { - return Err(RegistryError::UserNotFound { - user_id: user_id.as_str().to_owned(), - }); - } - - Ok(()) - } - - pub async fn create_user_session( - &self, - session_id: &UserSessionId, - user_id: &UserId, - current_workspace_id: Option<&WorkspaceId>, - secret_hash: &str, - expires_at: &str, - ) -> Result<(), RegistryError> { - sqlx::query( - "insert into user_sessions ( - id, - user_id, - current_workspace_id, - secret_hash, - status, - expires_at, - last_seen_at, - created_at - ) values ( - $1, - $2, - $3, - $4, - 'active', - $5::timestamptz, - now(), - now() - )", - ) - .bind(session_id.as_str()) - .bind(user_id.as_str()) - .bind(current_workspace_id.map(|id| id.as_str())) - .bind(secret_hash) - .bind(expires_at) - .execute(&self.pool) - .await?; - - Ok(()) - } - - pub async fn get_user_session( - &self, - session_id: &UserSessionId, - secret_hash: &str, - ) -> Result, RegistryError> { - let row = sqlx::query( - "select - s.id, - s.user_id, - s.current_workspace_id, - u.email, - u.display_name, - u.status, - to_char(u.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at - from user_sessions s - join users u on u.id = s.user_id - where s.id = $1 - and s.secret_hash = $2 - and s.status = 'active' - and s.expires_at > now() - limit 1", - ) - .bind(session_id.as_str()) - .bind(secret_hash) - .fetch_optional(&self.pool) - .await?; - - let Some(row) = row else { - return Ok(None); - }; - - let user_id = UserId::new(row.try_get::("user_id")?); - let user = User { - id: user_id.clone(), - email: row.try_get("email")?, - display_name: row.try_get("display_name")?, - status: deserialize_enum_text(&row.try_get::("status")?, "status")?, - created_at: row.try_get("created_at")?, - }; - let memberships = self.list_workspaces_for_user(&user_id).await?; - let stored_workspace_id = row - .try_get::, _>("current_workspace_id")? - .map(WorkspaceId::new); - let current_workspace_id = stored_workspace_id - .filter(|workspace_id| { - memberships - .iter() - .any(|membership| membership.workspace.id == *workspace_id) - }) - .or_else(|| { - memberships - .first() - .map(|membership| membership.workspace.id.clone()) - }); - - Ok(Some(SessionRecord { - session_id: UserSessionId::new(row.try_get::("id")?), - user, - memberships, - current_workspace_id, - })) - } - - pub async fn touch_user_session( - &self, - session_id: &UserSessionId, - ) -> Result<(), RegistryError> { - sqlx::query( - "update user_sessions - set last_seen_at = now() - where id = $1", - ) - .bind(session_id.as_str()) - .execute(&self.pool) - .await?; - - Ok(()) - } - - pub async fn revoke_user_session( - &self, - session_id: &UserSessionId, - ) -> Result<(), RegistryError> { - sqlx::query( - "update user_sessions - set status = 'revoked' - where id = $1", - ) - .bind(session_id.as_str()) - .execute(&self.pool) - .await?; - - Ok(()) - } - - pub async fn set_user_session_current_workspace( - &self, - session_id: &UserSessionId, - workspace_id: &WorkspaceId, - ) -> Result<(), RegistryError> { - sqlx::query( - "update user_sessions - set current_workspace_id = $2 - where id = $1", - ) - .bind(session_id.as_str()) - .bind(workspace_id.as_str()) - .execute(&self.pool) - .await?; - - Ok(()) - } - - pub async fn user_has_workspace_access( - &self, - user_id: &UserId, - workspace_id: &WorkspaceId, - ) -> Result { - let row = sqlx::query( - "select exists( - select 1 - from memberships - where user_id = $1 - and workspace_id = $2 - ) as allowed", - ) - .bind(user_id.as_str()) - .bind(workspace_id.as_str()) - .fetch_one(&self.pool) - .await?; - - Ok(row.try_get("allowed")?) - } - - pub async fn list_memberships( - &self, - workspace_id: &WorkspaceId, - ) -> Result, RegistryError> { - let rows = sqlx::query( - "select - m.workspace_id, - m.user_id, - m.role, - to_char(m.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - u.email, - u.display_name, - u.status, - to_char(u.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as user_created_at - from memberships m - join users u on u.id = m.user_id - where m.workspace_id = $1 - order by u.email asc", - ) - .bind(workspace_id.as_str()) - .fetch_all(&self.pool) - .await?; - - rows.iter().map(map_membership_record).collect() - } - - pub async fn update_membership_role( - &self, - workspace_id: &WorkspaceId, - user_id: &UserId, - role: MembershipRole, - ) -> Result<(), RegistryError> { - let result = sqlx::query( - "update memberships - set role = $3 - where workspace_id = $1 and user_id = $2", - ) - .bind(workspace_id.as_str()) - .bind(user_id.as_str()) - .bind(serialize_enum_text(&role, "role")?) - .execute(&self.pool) - .await?; - - if result.rows_affected() == 0 { - return Err(RegistryError::MembershipNotFound { - workspace_id: workspace_id.as_str().to_owned(), - user_id: user_id.as_str().to_owned(), - }); - } - - Ok(()) - } - - pub async fn delete_membership( - &self, - workspace_id: &WorkspaceId, - user_id: &UserId, - ) -> Result<(), RegistryError> { - let result = sqlx::query( - "delete from memberships - where workspace_id = $1 and user_id = $2", - ) - .bind(workspace_id.as_str()) - .bind(user_id.as_str()) - .execute(&self.pool) - .await?; - - if result.rows_affected() == 0 { - return Err(RegistryError::MembershipNotFound { - workspace_id: workspace_id.as_str().to_owned(), - user_id: user_id.as_str().to_owned(), - }); - } - - Ok(()) - } - - pub async fn list_invitations( - &self, - workspace_id: &WorkspaceId, - ) -> Result, RegistryError> { - let rows = sqlx::query( - "select - id, - workspace_id, - email, - role, - status, - token_hash, - to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at - from invitation_tokens - where workspace_id = $1 - order by created_at desc", - ) - .bind(workspace_id.as_str()) - .fetch_all(&self.pool) - .await?; - - rows.iter().map(map_invitation_record).collect() - } - - pub async fn create_invitation( - &self, - request: CreateInvitationRequest<'_>, - ) -> Result<(), RegistryError> { - sqlx::query( - "insert into invitation_tokens ( - id, - workspace_id, - email, - role, - status, - token_hash, - expires_at, - created_at - ) values ( - $1, $2, $3, $4, $5, $6, $7::timestamptz, $8::timestamptz - )", - ) - .bind(request.invitation.id.as_str()) - .bind(request.invitation.workspace_id.as_str()) - .bind(&request.invitation.email) - .bind(serialize_enum_text(&request.invitation.role, "role")?) - .bind(serialize_enum_text(&request.invitation.status, "status")?) - .bind(&request.invitation.token_hash) - .bind(&request.invitation.expires_at) - .bind(&request.invitation.created_at) - .execute(&self.pool) - .await?; - - Ok(()) - } - - pub async fn delete_invitation( - &self, - workspace_id: &WorkspaceId, - invitation_id: &InvitationId, - ) -> Result<(), RegistryError> { - let result = - sqlx::query("delete from invitation_tokens where workspace_id = $1 and id = $2") - .bind(workspace_id.as_str()) - .bind(invitation_id.as_str()) - .execute(&self.pool) - .await?; - - if result.rows_affected() == 0 { - return Err(RegistryError::InvitationNotFound { - invitation_id: invitation_id.as_str().to_owned(), - }); - } - - Ok(()) - } - - pub async fn delete_workspace(&self, workspace_id: &WorkspaceId) -> Result<(), RegistryError> { - let result = sqlx::query("delete from workspaces where id = $1") - .bind(workspace_id.as_str()) - .execute(&self.pool) - .await?; - - if result.rows_affected() == 0 { - return Err(RegistryError::WorkspaceNotFound { - workspace_id: workspace_id.as_str().to_owned(), - }); - } - - Ok(()) - } - pub async fn list_platform_api_keys( &self, workspace_id: &WorkspaceId, @@ -773,739 +231,6 @@ impl PostgresRegistry { row.as_ref().map(map_platform_api_key_record).transpose() } - pub async fn list_secrets( - &self, - workspace_id: &WorkspaceId, - ) -> Result, RegistryError> { - let rows = sqlx::query( - "select - id, - workspace_id, - name, - kind, - status, - current_version, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at, - to_char(last_used_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as last_used_at - from secrets - where workspace_id = $1 - order by name asc", - ) - .bind(workspace_id.as_str()) - .fetch_all(&self.pool) - .await?; - - rows.iter().map(map_secret_record).collect() - } - - pub async fn get_secret( - &self, - workspace_id: &WorkspaceId, - secret_id: &SecretId, - ) -> Result, RegistryError> { - let row = sqlx::query( - "select - id, - workspace_id, - name, - kind, - status, - current_version, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at, - to_char(last_used_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as last_used_at - from secrets - where workspace_id = $1 and id = $2", - ) - .bind(workspace_id.as_str()) - .bind(secret_id.as_str()) - .fetch_optional(&self.pool) - .await?; - - row.as_ref().map(map_secret_record).transpose() - } - - pub async fn get_current_secret_version( - &self, - workspace_id: &WorkspaceId, - secret_id: &SecretId, - ) -> Result, RegistryError> { - let row = sqlx::query( - "select - sv.secret_id, - sv.version, - sv.ciphertext, - sv.key_version, - to_char(sv.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - sv.created_by - from secrets s - join secret_versions sv - on sv.secret_id = s.id and sv.version = s.current_version - where s.workspace_id = $1 and s.id = $2", - ) - .bind(workspace_id.as_str()) - .bind(secret_id.as_str()) - .fetch_optional(&self.pool) - .await?; - - row.as_ref().map(map_secret_version_record).transpose() - } - - pub async fn create_secret( - &self, - request: CreateSecretRequest<'_>, - ) -> Result<(), RegistryError> { - let mut tx = self.pool.begin().await?; - let result = sqlx::query( - "insert into secrets ( - id, - workspace_id, - name, - kind, - status, - current_version, - last_used_at, - created_at, - updated_at - ) values ( - $1, $2, $3, $4, $5, $6, $7::timestamptz, $8::timestamptz, $9::timestamptz - )", - ) - .bind(request.secret.id.as_str()) - .bind(request.secret.workspace_id.as_str()) - .bind(&request.secret.name) - .bind(serialize_enum_text(&request.secret.kind, "kind")?) - .bind(serialize_enum_text(&request.secret.status, "status")?) - .bind(to_db_version(request.secret.current_version)) - .bind(request.secret.last_used_at.as_deref()) - .bind(&request.secret.created_at) - .bind(&request.secret.updated_at) - .execute(&mut *tx) - .await; - - match result { - Ok(_) => { - sqlx::query( - "insert into secret_versions ( - secret_id, - version, - ciphertext, - key_version, - created_at, - created_by - ) values ( - $1, $2, $3, $4, $5::timestamptz, $6 - )", - ) - .bind(request.secret.id.as_str()) - .bind(to_db_version(request.secret.current_version)) - .bind(request.ciphertext) - .bind(request.key_version) - .bind(&request.secret.created_at) - .bind(request.created_by.map(|value| value.as_str())) - .execute(&mut *tx) - .await?; - tx.commit().await?; - Ok(()) - } - Err(sqlx::Error::Database(error)) - if error.constraint() == Some("secrets_workspace_name_idx") => - { - Err(RegistryError::SecretNameAlreadyExists { - workspace_id: request.secret.workspace_id.as_str().to_owned(), - name: request.secret.name.clone(), - }) - } - Err(error) => Err(RegistryError::Storage(error)), - } - } - - pub async fn rotate_secret( - &self, - request: RotateSecretRequest<'_>, - ) -> Result { - let existing = self - .get_secret(request.workspace_id, request.secret_id) - .await? - .ok_or_else(|| RegistryError::SecretNotFound { - secret_id: request.secret_id.as_str().to_owned(), - })?; - let next_version = existing.secret.current_version + 1; - - let mut tx = self.pool.begin().await?; - sqlx::query( - "insert into secret_versions ( - secret_id, - version, - ciphertext, - key_version, - created_at, - created_by - ) values ( - $1, $2, $3, $4, $5::timestamptz, $6 - )", - ) - .bind(request.secret_id.as_str()) - .bind(to_db_version(next_version)) - .bind(request.ciphertext) - .bind(request.key_version) - .bind(request.created_at) - .bind(request.created_by.map(|value| value.as_str())) - .execute(&mut *tx) - .await?; - - sqlx::query( - "update secrets - set current_version = $3, - updated_at = $4::timestamptz - where workspace_id = $1 and id = $2", - ) - .bind(request.workspace_id.as_str()) - .bind(request.secret_id.as_str()) - .bind(to_db_version(next_version)) - .bind(request.updated_at) - .execute(&mut *tx) - .await?; - - tx.commit().await?; - - Ok(SecretVersionRecord { - secret_version: SecretVersion { - secret_id: request.secret_id.clone(), - version: next_version, - ciphertext: request.ciphertext.to_owned(), - key_version: request.key_version.to_owned(), - created_at: request.created_at.to_owned(), - created_by: request.created_by.cloned(), - }, - }) - } - - pub async fn delete_secret( - &self, - workspace_id: &WorkspaceId, - secret_id: &SecretId, - ) -> Result<(), RegistryError> { - let result = sqlx::query( - "delete from secrets - where workspace_id = $1 and id = $2", - ) - .bind(workspace_id.as_str()) - .bind(secret_id.as_str()) - .execute(&self.pool) - .await?; - - if result.rows_affected() == 0 { - return Err(RegistryError::SecretNotFound { - secret_id: secret_id.as_str().to_owned(), - }); - } - - Ok(()) - } - - pub async fn touch_secret( - &self, - workspace_id: &WorkspaceId, - secret_id: &SecretId, - used_at: &str, - ) -> Result<(), RegistryError> { - let result = sqlx::query( - "update secrets - set last_used_at = $3::timestamptz - where workspace_id = $1 and id = $2", - ) - .bind(workspace_id.as_str()) - .bind(secret_id.as_str()) - .bind(used_at) - .execute(&self.pool) - .await?; - - if result.rows_affected() == 0 { - return Err(RegistryError::SecretNotFound { - secret_id: secret_id.as_str().to_owned(), - }); - } - - Ok(()) - } - - pub async fn create_stream_session( - &self, - request: CreateStreamSessionRequest<'_>, - ) -> Result { - sqlx::query( - "insert into stream_sessions ( - id, - workspace_id, - agent_id, - operation_id, - protocol, - mode, - status, - cursor_json, - state_json, - expires_at, - last_poll_at, - created_at, - closed_at - ) values ( - $1, $2, $3, $4, $5, $6, $7, $8::jsonb, $9::jsonb, $10::timestamptz, - $11::timestamptz, $12::timestamptz, $13::timestamptz - )", - ) - .bind(request.session.id.as_str()) - .bind(request.session.workspace_id.as_str()) - .bind( - request - .session - .agent_id - .as_ref() - .map(|value| value.as_str()), - ) - .bind(request.session.operation_id.as_str()) - .bind(serialize_enum_text(&request.session.protocol, "protocol")?) - .bind(serialize_enum_text(&request.session.mode, "mode")?) - .bind(serialize_enum_text(&request.session.status, "status")?) - .bind(request.session.cursor.clone().map(Json)) - .bind(Json(request.session.state.clone())) - .bind(&request.session.expires_at) - .bind(request.session.last_poll_at.as_deref()) - .bind(&request.session.created_at) - .bind(request.session.closed_at.as_deref()) - .execute(&self.pool) - .await?; - - Ok(request.session.clone()) - } - - pub async fn get_stream_session( - &self, - id: &StreamSessionId, - ) -> Result, RegistryError> { - let row = sqlx::query( - "select - id, - workspace_id, - agent_id, - operation_id, - protocol, - mode, - status, - cursor_json, - state_json, - to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at, - to_char(last_poll_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as last_poll_at, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - to_char(closed_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as closed_at - from stream_sessions - where id = $1", - ) - .bind(id.as_str()) - .fetch_optional(&self.pool) - .await?; - - row.as_ref().map(map_stream_session).transpose() - } - - pub async fn update_stream_session_state( - &self, - request: UpdateStreamSessionStateRequest<'_>, - ) -> Result { - validate_stream_session_transition( - request.session_id, - request.current_status, - request.next_status, - )?; - - let row = sqlx::query( - "update stream_sessions - set status = $3, - cursor_json = $4::jsonb, - state_json = $5::jsonb, - expires_at = coalesce($6::timestamptz, expires_at), - last_poll_at = coalesce($7::timestamptz, last_poll_at), - closed_at = case - when $8::timestamptz is null then closed_at - else $8::timestamptz - end - where id = $1 - and status = $2 - returning - id, - workspace_id, - agent_id, - operation_id, - protocol, - mode, - status, - cursor_json, - state_json, - to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at, - to_char(last_poll_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as last_poll_at, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - to_char(closed_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as closed_at", - ) - .bind(request.session_id.as_str()) - .bind(serialize_enum_text(&request.current_status, "status")?) - .bind(serialize_enum_text(&request.next_status, "status")?) - .bind(request.cursor.cloned().map(Json)) - .bind(Json(request.state.clone())) - .bind(request.expires_at) - .bind(request.last_poll_at) - .bind(request.closed_at) - .fetch_optional(&self.pool) - .await?; - - match row.as_ref() { - Some(row) => map_stream_session(row), - None => { - if self.get_stream_session(request.session_id).await?.is_some() { - Err(RegistryError::InvalidStreamSessionTransition { - session_id: request.session_id.as_str().to_owned(), - from: serialize_enum_text(&request.current_status, "status")?, - to: serialize_enum_text(&request.next_status, "status")?, - }) - } else { - Err(RegistryError::StreamSessionNotFound { - session_id: request.session_id.as_str().to_owned(), - }) - } - } - } - } - - pub async fn close_stream_session( - &self, - id: &StreamSessionId, - now: &str, - ) -> Result<(), RegistryError> { - let result = sqlx::query( - "update stream_sessions - set status = 'stopped', - last_poll_at = $2::timestamptz, - closed_at = $2::timestamptz - where id = $1 - and status in ('created', 'running')", - ) - .bind(id.as_str()) - .bind(now) - .execute(&self.pool) - .await?; - - if result.rows_affected() > 0 { - return Ok(()); - } - - match self.get_stream_session(id).await? { - Some(existing) => Err(RegistryError::InvalidStreamSessionTransition { - session_id: id.as_str().to_owned(), - from: serialize_enum_text(&existing.status, "status")?, - to: "stopped".to_owned(), - }), - None => Err(RegistryError::StreamSessionNotFound { - session_id: id.as_str().to_owned(), - }), - } - } - - pub async fn list_stream_sessions( - &self, - filter: StreamSessionFilter<'_>, - ) -> Result, RegistryError> { - let status = filter - .status - .as_ref() - .map(|value| serialize_enum_text(value, "status")) - .transpose()?; - let mode = filter - .mode - .as_ref() - .map(|value| serialize_enum_text(value, "mode")) - .transpose()?; - let limit = i64::from(filter.limit); - - let rows = sqlx::query( - "select - id, - workspace_id, - agent_id, - operation_id, - protocol, - mode, - status, - cursor_json, - state_json, - to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at, - to_char(last_poll_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as last_poll_at, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - to_char(closed_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as closed_at - from stream_sessions - where workspace_id = $1 - and ($2::text is null or agent_id = $2) - and ($3::text is null or operation_id = $3) - and ($4::text is null or status = $4) - and ($5::text is null or mode = $5) - order by created_at desc - limit $6", - ) - .bind(filter.workspace_id.as_str()) - .bind(filter.agent_id.map(|value| value.as_str())) - .bind(filter.operation_id.map(|value| value.as_str())) - .bind(status) - .bind(mode) - .bind(limit) - .fetch_all(&self.pool) - .await?; - - let items = rows - .iter() - .map(map_stream_session) - .collect::, _>>()?; - let total = items.len() as u64; - - Ok(Page { items, total }) - } - - pub async fn delete_expired_stream_sessions(&self, now: &str) -> Result { - let result = sqlx::query( - "delete from stream_sessions - where expires_at <= $1::timestamptz", - ) - .bind(now) - .execute(&self.pool) - .await?; - - Ok(result.rows_affected()) - } - - pub async fn create_async_job( - &self, - request: CreateAsyncJobRequest<'_>, - ) -> Result { - sqlx::query( - "insert into async_jobs ( - id, - workspace_id, - agent_id, - operation_id, - status, - progress_json, - result_json, - error_json, - expires_at, - created_at, - updated_at, - finished_at - ) values ( - $1, $2, $3, $4, $5, $6::jsonb, $7::jsonb, $8::jsonb, $9::timestamptz, - $10::timestamptz, $11::timestamptz, $12::timestamptz - )", - ) - .bind(request.job.id.as_str()) - .bind(request.job.workspace_id.as_str()) - .bind(request.job.agent_id.as_ref().map(|value| value.as_str())) - .bind(request.job.operation_id.as_str()) - .bind(serialize_enum_text(&request.job.status, "status")?) - .bind(Json(request.job.progress.clone())) - .bind(request.job.result.clone().map(Json)) - .bind(request.job.error.clone().map(Json)) - .bind(request.job.expires_at.as_deref()) - .bind(&request.job.created_at) - .bind(&request.job.updated_at) - .bind(request.job.finished_at.as_deref()) - .execute(&self.pool) - .await?; - - Ok(request.job.clone()) - } - - pub async fn get_async_job( - &self, - id: &AsyncJobId, - ) -> Result, RegistryError> { - let row = sqlx::query( - "select - id, - workspace_id, - agent_id, - operation_id, - status, - progress_json, - result_json, - error_json, - to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at, - to_char(finished_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as finished_at - from async_jobs - where id = $1", - ) - .bind(id.as_str()) - .fetch_optional(&self.pool) - .await?; - - row.as_ref().map(map_async_job).transpose() - } - - pub async fn update_async_job_status( - &self, - request: UpdateAsyncJobStatusRequest<'_>, - ) -> Result { - validate_async_job_transition(request.job_id, request.current_status, request.next_status)?; - - let row = sqlx::query( - "update async_jobs - set status = $3, - progress_json = $4::jsonb, - result_json = $5::jsonb, - error_json = $6::jsonb, - expires_at = coalesce($7::timestamptz, expires_at), - updated_at = $8::timestamptz, - finished_at = case - when $9::timestamptz is null then finished_at - else $9::timestamptz - end - where id = $1 - and status = $2 - returning - id, - workspace_id, - agent_id, - operation_id, - status, - progress_json, - result_json, - error_json, - to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at, - to_char(finished_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as finished_at", - ) - .bind(request.job_id.as_str()) - .bind(serialize_enum_text(&request.current_status, "status")?) - .bind(serialize_enum_text(&request.next_status, "status")?) - .bind(Json(request.progress.clone())) - .bind(request.result.cloned().map(Json)) - .bind(request.error.cloned().map(Json)) - .bind(request.expires_at) - .bind(request.updated_at) - .bind(request.finished_at) - .fetch_optional(&self.pool) - .await?; - - match row.as_ref() { - Some(row) => map_async_job(row), - None => { - if self.get_async_job(request.job_id).await?.is_some() { - Err(RegistryError::InvalidAsyncJobTransition { - job_id: request.job_id.as_str().to_owned(), - from: serialize_enum_text(&request.current_status, "status")?, - to: serialize_enum_text(&request.next_status, "status")?, - }) - } else { - Err(RegistryError::AsyncJobNotFound { - job_id: request.job_id.as_str().to_owned(), - }) - } - } - } - } - - pub async fn cancel_async_job(&self, id: &AsyncJobId, now: &str) -> Result<(), RegistryError> { - let result = sqlx::query( - "update async_jobs - set status = 'cancelled', - updated_at = $2::timestamptz, - finished_at = $2::timestamptz - where id = $1 - and status in ('created', 'running')", - ) - .bind(id.as_str()) - .bind(now) - .execute(&self.pool) - .await?; - - if result.rows_affected() > 0 { - return Ok(()); - } - - match self.get_async_job(id).await? { - Some(existing) => Err(RegistryError::InvalidAsyncJobTransition { - job_id: id.as_str().to_owned(), - from: serialize_enum_text(&existing.status, "status")?, - to: "cancelled".to_owned(), - }), - None => Err(RegistryError::AsyncJobNotFound { - job_id: id.as_str().to_owned(), - }), - } - } - - pub async fn list_async_jobs( - &self, - filter: AsyncJobFilter<'_>, - ) -> Result, RegistryError> { - let status = filter - .status - .as_ref() - .map(|value| serialize_enum_text(value, "status")) - .transpose()?; - let limit = i64::from(filter.limit); - - let rows = sqlx::query( - "select - id, - workspace_id, - agent_id, - operation_id, - status, - progress_json, - result_json, - error_json, - to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at, - to_char(finished_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as finished_at - from async_jobs - where workspace_id = $1 - and ($2::text is null or agent_id = $2) - and ($3::text is null or operation_id = $3) - and ($4::text is null or status = $4) - order by updated_at desc - limit $5", - ) - .bind(filter.workspace_id.as_str()) - .bind(filter.agent_id.map(|value| value.as_str())) - .bind(filter.operation_id.map(|value| value.as_str())) - .bind(status) - .bind(limit) - .fetch_all(&self.pool) - .await?; - - let items = rows - .iter() - .map(map_async_job) - .collect::, _>>()?; - let total = items.len() as u64; - - Ok(Page { items, total }) - } - - pub async fn delete_expired_async_jobs(&self, now: &str) -> Result { - let result = sqlx::query( - "delete from async_jobs - where expires_at is not null - and expires_at <= $1::timestamptz", - ) - .bind(now) - .execute(&self.pool) - .await?; - - Ok(result.rows_affected()) - } - pub async fn create_platform_api_key( &self, request: CreatePlatformApiKeyRequest<'_>, @@ -2071,107 +796,6 @@ impl PostgresRegistry { })) } - pub async fn get_workspace( - &self, - workspace_id: &WorkspaceId, - ) -> Result, RegistryError> { - let row = sqlx::query( - "select - id, - slug, - display_name, - status, - settings_json, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at - from workspaces - where id = $1", - ) - .bind(workspace_id.as_str()) - .fetch_optional(&self.pool) - .await?; - - row.as_ref().map(map_workspace_record).transpose() - } - - pub async fn create_workspace( - &self, - request: CreateWorkspaceRequest<'_>, - ) -> Result<(), RegistryError> { - let result = sqlx::query( - "insert into workspaces ( - id, - slug, - display_name, - status, - settings_json, - created_at, - updated_at - ) values ( - $1, $2, $3, $4, $5, $6::timestamptz, $7::timestamptz - )", - ) - .bind(request.workspace.id.as_str()) - .bind(&request.workspace.slug) - .bind(&request.workspace.display_name) - .bind(serialize_enum_text(&request.workspace.status, "status")?) - .bind(Json(request.workspace.settings.clone())) - .bind(&request.workspace.created_at) - .bind(&request.workspace.updated_at) - .execute(&self.pool) - .await; - - match result { - Ok(_) => Ok(()), - Err(sqlx::Error::Database(error)) - if error.constraint() == Some("workspaces_slug_key") => - { - Err(RegistryError::WorkspaceSlugAlreadyExists { - slug: request.workspace.slug.clone(), - }) - } - Err(error) => Err(RegistryError::Storage(error)), - } - } - - pub async fn update_workspace( - &self, - request: UpdateWorkspaceRequest<'_>, - ) -> Result<(), RegistryError> { - let result = sqlx::query( - "update workspaces - set slug = $1, - display_name = $2, - status = $3, - settings_json = $4, - updated_at = $5::timestamptz - where id = $6", - ) - .bind(&request.workspace.slug) - .bind(&request.workspace.display_name) - .bind(serialize_enum_text(&request.workspace.status, "status")?) - .bind(Json(request.workspace.settings.clone())) - .bind(&request.workspace.updated_at) - .bind(request.workspace.id.as_str()) - .execute(&self.pool) - .await; - - match result { - Ok(result) if result.rows_affected() == 0 => Err(RegistryError::WorkspaceNotFound { - workspace_id: request.workspace.id.as_str().to_owned(), - }), - Ok(_) => Ok(()), - Err(sqlx::Error::Database(error)) - if error.constraint() == Some("workspaces_slug_key") => - { - Err(RegistryError::WorkspaceSlugAlreadyExists { - slug: request.workspace.slug.clone(), - }) - } - Err(error) => Err(RegistryError::Storage(error)), - } - } - pub async fn list_agents( &self, workspace_id: &WorkspaceId, @@ -3325,108 +1949,6 @@ impl PostgresRegistry { .collect() } - pub async fn save_auth_profile( - &self, - request: SaveAuthProfileRequest<'_>, - ) -> Result<(), RegistryError> { - sqlx::query( - "insert into auth_profiles ( - id, - workspace_id, - name, - kind, - config_json, - created_at, - updated_at - ) values ($1, $2, $3, $4, $5, $6::timestamptz, $7::timestamptz) - on conflict(id) do update set - workspace_id = excluded.workspace_id, - name = excluded.name, - kind = excluded.kind, - config_json = excluded.config_json, - updated_at = excluded.updated_at", - ) - .bind(request.profile.id.as_str()) - .bind(request.workspace_id.as_str()) - .bind(&request.profile.name) - .bind(serialize_enum_text(&request.profile.kind, "kind")?) - .bind(Json(serialize_json_value(&request.profile.config)?)) - .bind(&request.profile.created_at) - .bind(&request.profile.updated_at) - .execute(&self.pool) - .await?; - - Ok(()) - } - - pub async fn get_auth_profile( - &self, - workspace_id: &WorkspaceId, - auth_profile_id: &crank_core::AuthProfileId, - ) -> Result, RegistryError> { - let row = sqlx::query( - "select - id, - workspace_id, - name, - kind, - config_json, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at - from auth_profiles - where workspace_id = $1 and id = $2", - ) - .bind(workspace_id.as_str()) - .bind(auth_profile_id.as_str()) - .fetch_optional(&self.pool) - .await?; - - row.as_ref().map(map_auth_profile).transpose() - } - - pub async fn list_auth_profiles( - &self, - workspace_id: &WorkspaceId, - ) -> Result, RegistryError> { - let rows = sqlx::query( - "select - id, - workspace_id, - name, - kind, - config_json, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at - from auth_profiles - where workspace_id = $1 - order by name asc", - ) - .bind(workspace_id.as_str()) - .fetch_all(&self.pool) - .await?; - - rows.iter().map(map_auth_profile).collect() - } - - pub async fn list_auth_profiles_referencing_secret( - &self, - workspace_id: &WorkspaceId, - secret_id: &SecretId, - ) -> Result, RegistryError> { - let profiles = self.list_auth_profiles(workspace_id).await?; - - Ok(profiles - .into_iter() - .filter(|profile| { - profile - .config - .secret_ids() - .into_iter() - .any(|candidate| candidate == secret_id) - }) - .collect()) - } - pub async fn save_sample_metadata( &self, request: SaveSampleMetadataRequest<'_>, diff --git a/crates/crank-registry/src/postgres/secret.rs b/crates/crank-registry/src/postgres/secret.rs new file mode 100644 index 0000000..582f0ee --- /dev/null +++ b/crates/crank-registry/src/postgres/secret.rs @@ -0,0 +1,261 @@ +use super::*; + +impl PostgresRegistry { + pub async fn list_secrets( + &self, + workspace_id: &WorkspaceId, + ) -> Result, RegistryError> { + let rows = sqlx::query( + "select + id, + workspace_id, + name, + kind, + status, + current_version, + to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, + to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at, + to_char(last_used_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as last_used_at + from secrets + where workspace_id = $1 + order by name asc", + ) + .bind(workspace_id.as_str()) + .fetch_all(&self.pool) + .await?; + + rows.iter().map(map_secret_record).collect() + } + + pub async fn get_secret( + &self, + workspace_id: &WorkspaceId, + secret_id: &SecretId, + ) -> Result, RegistryError> { + let row = sqlx::query( + "select + id, + workspace_id, + name, + kind, + status, + current_version, + to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, + to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at, + to_char(last_used_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as last_used_at + from secrets + where workspace_id = $1 and id = $2", + ) + .bind(workspace_id.as_str()) + .bind(secret_id.as_str()) + .fetch_optional(&self.pool) + .await?; + + row.as_ref().map(map_secret_record).transpose() + } + + pub async fn get_current_secret_version( + &self, + workspace_id: &WorkspaceId, + secret_id: &SecretId, + ) -> Result, RegistryError> { + let row = sqlx::query( + "select + sv.secret_id, + sv.version, + sv.ciphertext, + sv.key_version, + to_char(sv.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, + sv.created_by + from secrets s + join secret_versions sv + on sv.secret_id = s.id and sv.version = s.current_version + where s.workspace_id = $1 and s.id = $2", + ) + .bind(workspace_id.as_str()) + .bind(secret_id.as_str()) + .fetch_optional(&self.pool) + .await?; + + row.as_ref().map(map_secret_version_record).transpose() + } + + pub async fn create_secret( + &self, + request: CreateSecretRequest<'_>, + ) -> Result<(), RegistryError> { + let mut tx = self.pool.begin().await?; + let result = sqlx::query( + "insert into secrets ( + id, + workspace_id, + name, + kind, + status, + current_version, + last_used_at, + created_at, + updated_at + ) values ( + $1, $2, $3, $4, $5, $6, $7::timestamptz, $8::timestamptz, $9::timestamptz + )", + ) + .bind(request.secret.id.as_str()) + .bind(request.secret.workspace_id.as_str()) + .bind(&request.secret.name) + .bind(serialize_enum_text(&request.secret.kind, "kind")?) + .bind(serialize_enum_text(&request.secret.status, "status")?) + .bind(to_db_version(request.secret.current_version)) + .bind(request.secret.last_used_at.as_deref()) + .bind(&request.secret.created_at) + .bind(&request.secret.updated_at) + .execute(&mut *tx) + .await; + + match result { + Ok(_) => { + sqlx::query( + "insert into secret_versions ( + secret_id, + version, + ciphertext, + key_version, + created_at, + created_by + ) values ( + $1, $2, $3, $4, $5::timestamptz, $6 + )", + ) + .bind(request.secret.id.as_str()) + .bind(to_db_version(request.secret.current_version)) + .bind(request.ciphertext) + .bind(request.key_version) + .bind(&request.secret.created_at) + .bind(request.created_by.map(|value| value.as_str())) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(()) + } + Err(sqlx::Error::Database(error)) + if error.constraint() == Some("secrets_workspace_name_idx") => + { + Err(RegistryError::SecretNameAlreadyExists { + workspace_id: request.secret.workspace_id.as_str().to_owned(), + name: request.secret.name.clone(), + }) + } + Err(error) => Err(RegistryError::Storage(error)), + } + } + + pub async fn rotate_secret( + &self, + request: RotateSecretRequest<'_>, + ) -> Result { + let existing = self + .get_secret(request.workspace_id, request.secret_id) + .await? + .ok_or_else(|| RegistryError::SecretNotFound { + secret_id: request.secret_id.as_str().to_owned(), + })?; + let next_version = existing.secret.current_version + 1; + + let mut tx = self.pool.begin().await?; + sqlx::query( + "insert into secret_versions ( + secret_id, + version, + ciphertext, + key_version, + created_at, + created_by + ) values ( + $1, $2, $3, $4, $5::timestamptz, $6 + )", + ) + .bind(request.secret_id.as_str()) + .bind(to_db_version(next_version)) + .bind(request.ciphertext) + .bind(request.key_version) + .bind(request.created_at) + .bind(request.created_by.map(|value| value.as_str())) + .execute(&mut *tx) + .await?; + + sqlx::query( + "update secrets + set current_version = $3, + updated_at = $4::timestamptz + where workspace_id = $1 and id = $2", + ) + .bind(request.workspace_id.as_str()) + .bind(request.secret_id.as_str()) + .bind(to_db_version(next_version)) + .bind(request.updated_at) + .execute(&mut *tx) + .await?; + + tx.commit().await?; + + Ok(SecretVersionRecord { + secret_version: SecretVersion { + secret_id: request.secret_id.clone(), + version: next_version, + ciphertext: request.ciphertext.to_owned(), + key_version: request.key_version.to_owned(), + created_at: request.created_at.to_owned(), + created_by: request.created_by.cloned(), + }, + }) + } + + pub async fn delete_secret( + &self, + workspace_id: &WorkspaceId, + secret_id: &SecretId, + ) -> Result<(), RegistryError> { + let result = sqlx::query( + "delete from secrets + where workspace_id = $1 and id = $2", + ) + .bind(workspace_id.as_str()) + .bind(secret_id.as_str()) + .execute(&self.pool) + .await?; + + if result.rows_affected() == 0 { + return Err(RegistryError::SecretNotFound { + secret_id: secret_id.as_str().to_owned(), + }); + } + + Ok(()) + } + + pub async fn touch_secret( + &self, + workspace_id: &WorkspaceId, + secret_id: &SecretId, + used_at: &str, + ) -> Result<(), RegistryError> { + let result = sqlx::query( + "update secrets + set last_used_at = $3::timestamptz + where workspace_id = $1 and id = $2", + ) + .bind(workspace_id.as_str()) + .bind(secret_id.as_str()) + .bind(used_at) + .execute(&self.pool) + .await?; + + if result.rows_affected() == 0 { + return Err(RegistryError::SecretNotFound { + secret_id: secret_id.as_str().to_owned(), + }); + } + + Ok(()) + } +} diff --git a/crates/crank-registry/src/postgres/stream.rs b/crates/crank-registry/src/postgres/stream.rs new file mode 100644 index 0000000..a1acdae --- /dev/null +++ b/crates/crank-registry/src/postgres/stream.rs @@ -0,0 +1,478 @@ +use super::*; + +impl PostgresRegistry { + pub async fn create_stream_session( + &self, + request: CreateStreamSessionRequest<'_>, + ) -> Result { + sqlx::query( + "insert into stream_sessions ( + id, + workspace_id, + agent_id, + operation_id, + protocol, + mode, + status, + cursor_json, + state_json, + expires_at, + last_poll_at, + created_at, + closed_at + ) values ( + $1, $2, $3, $4, $5, $6, $7, $8::jsonb, $9::jsonb, $10::timestamptz, + $11::timestamptz, $12::timestamptz, $13::timestamptz + )", + ) + .bind(request.session.id.as_str()) + .bind(request.session.workspace_id.as_str()) + .bind( + request + .session + .agent_id + .as_ref() + .map(|value| value.as_str()), + ) + .bind(request.session.operation_id.as_str()) + .bind(serialize_enum_text(&request.session.protocol, "protocol")?) + .bind(serialize_enum_text(&request.session.mode, "mode")?) + .bind(serialize_enum_text(&request.session.status, "status")?) + .bind(request.session.cursor.clone().map(Json)) + .bind(Json(request.session.state.clone())) + .bind(&request.session.expires_at) + .bind(request.session.last_poll_at.as_deref()) + .bind(&request.session.created_at) + .bind(request.session.closed_at.as_deref()) + .execute(&self.pool) + .await?; + + Ok(request.session.clone()) + } + + pub async fn get_stream_session( + &self, + id: &StreamSessionId, + ) -> Result, RegistryError> { + let row = sqlx::query( + "select + id, + workspace_id, + agent_id, + operation_id, + protocol, + mode, + status, + cursor_json, + state_json, + to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at, + to_char(last_poll_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as last_poll_at, + to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, + to_char(closed_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as closed_at + from stream_sessions + where id = $1", + ) + .bind(id.as_str()) + .fetch_optional(&self.pool) + .await?; + + row.as_ref().map(map_stream_session).transpose() + } + + pub async fn update_stream_session_state( + &self, + request: UpdateStreamSessionStateRequest<'_>, + ) -> Result { + validate_stream_session_transition( + request.session_id, + request.current_status, + request.next_status, + )?; + + let row = sqlx::query( + "update stream_sessions + set status = $3, + cursor_json = $4::jsonb, + state_json = $5::jsonb, + expires_at = coalesce($6::timestamptz, expires_at), + last_poll_at = coalesce($7::timestamptz, last_poll_at), + closed_at = case + when $8::timestamptz is null then closed_at + else $8::timestamptz + end + where id = $1 + and status = $2 + returning + id, + workspace_id, + agent_id, + operation_id, + protocol, + mode, + status, + cursor_json, + state_json, + to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at, + to_char(last_poll_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as last_poll_at, + to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, + to_char(closed_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as closed_at", + ) + .bind(request.session_id.as_str()) + .bind(serialize_enum_text(&request.current_status, "status")?) + .bind(serialize_enum_text(&request.next_status, "status")?) + .bind(request.cursor.cloned().map(Json)) + .bind(Json(request.state.clone())) + .bind(request.expires_at) + .bind(request.last_poll_at) + .bind(request.closed_at) + .fetch_optional(&self.pool) + .await?; + + match row.as_ref() { + Some(row) => map_stream_session(row), + None => { + if self.get_stream_session(request.session_id).await?.is_some() { + Err(RegistryError::InvalidStreamSessionTransition { + session_id: request.session_id.as_str().to_owned(), + from: serialize_enum_text(&request.current_status, "status")?, + to: serialize_enum_text(&request.next_status, "status")?, + }) + } else { + Err(RegistryError::StreamSessionNotFound { + session_id: request.session_id.as_str().to_owned(), + }) + } + } + } + } + + pub async fn close_stream_session( + &self, + id: &StreamSessionId, + now: &str, + ) -> Result<(), RegistryError> { + let result = sqlx::query( + "update stream_sessions + set status = 'stopped', + last_poll_at = $2::timestamptz, + closed_at = $2::timestamptz + where id = $1 + and status in ('created', 'running')", + ) + .bind(id.as_str()) + .bind(now) + .execute(&self.pool) + .await?; + + if result.rows_affected() > 0 { + return Ok(()); + } + + match self.get_stream_session(id).await? { + Some(existing) => Err(RegistryError::InvalidStreamSessionTransition { + session_id: id.as_str().to_owned(), + from: serialize_enum_text(&existing.status, "status")?, + to: "stopped".to_owned(), + }), + None => Err(RegistryError::StreamSessionNotFound { + session_id: id.as_str().to_owned(), + }), + } + } + + pub async fn list_stream_sessions( + &self, + filter: StreamSessionFilter<'_>, + ) -> Result, RegistryError> { + let status = filter + .status + .as_ref() + .map(|value| serialize_enum_text(value, "status")) + .transpose()?; + let mode = filter + .mode + .as_ref() + .map(|value| serialize_enum_text(value, "mode")) + .transpose()?; + let limit = i64::from(filter.limit); + + let rows = sqlx::query( + "select + id, + workspace_id, + agent_id, + operation_id, + protocol, + mode, + status, + cursor_json, + state_json, + to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at, + to_char(last_poll_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as last_poll_at, + to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, + to_char(closed_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as closed_at + from stream_sessions + where workspace_id = $1 + and ($2::text is null or agent_id = $2) + and ($3::text is null or operation_id = $3) + and ($4::text is null or status = $4) + and ($5::text is null or mode = $5) + order by created_at desc + limit $6", + ) + .bind(filter.workspace_id.as_str()) + .bind(filter.agent_id.map(|value| value.as_str())) + .bind(filter.operation_id.map(|value| value.as_str())) + .bind(status) + .bind(mode) + .bind(limit) + .fetch_all(&self.pool) + .await?; + + let items = rows + .iter() + .map(map_stream_session) + .collect::, _>>()?; + let total = items.len() as u64; + + Ok(Page { items, total }) + } + + pub async fn delete_expired_stream_sessions(&self, now: &str) -> Result { + let result = sqlx::query( + "delete from stream_sessions + where expires_at <= $1::timestamptz", + ) + .bind(now) + .execute(&self.pool) + .await?; + + Ok(result.rows_affected()) + } + + pub async fn create_async_job( + &self, + request: CreateAsyncJobRequest<'_>, + ) -> Result { + sqlx::query( + "insert into async_jobs ( + id, + workspace_id, + agent_id, + operation_id, + status, + progress_json, + result_json, + error_json, + expires_at, + created_at, + updated_at, + finished_at + ) values ( + $1, $2, $3, $4, $5, $6::jsonb, $7::jsonb, $8::jsonb, $9::timestamptz, + $10::timestamptz, $11::timestamptz, $12::timestamptz + )", + ) + .bind(request.job.id.as_str()) + .bind(request.job.workspace_id.as_str()) + .bind(request.job.agent_id.as_ref().map(|value| value.as_str())) + .bind(request.job.operation_id.as_str()) + .bind(serialize_enum_text(&request.job.status, "status")?) + .bind(Json(request.job.progress.clone())) + .bind(request.job.result.clone().map(Json)) + .bind(request.job.error.clone().map(Json)) + .bind(request.job.expires_at.as_deref()) + .bind(&request.job.created_at) + .bind(&request.job.updated_at) + .bind(request.job.finished_at.as_deref()) + .execute(&self.pool) + .await?; + + Ok(request.job.clone()) + } + + pub async fn get_async_job( + &self, + id: &AsyncJobId, + ) -> Result, RegistryError> { + let row = sqlx::query( + "select + id, + workspace_id, + agent_id, + operation_id, + status, + progress_json, + result_json, + error_json, + to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at, + to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, + to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at, + to_char(finished_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as finished_at + from async_jobs + where id = $1", + ) + .bind(id.as_str()) + .fetch_optional(&self.pool) + .await?; + + row.as_ref().map(map_async_job).transpose() + } + + pub async fn update_async_job_status( + &self, + request: UpdateAsyncJobStatusRequest<'_>, + ) -> Result { + validate_async_job_transition(request.job_id, request.current_status, request.next_status)?; + + let row = sqlx::query( + "update async_jobs + set status = $3, + progress_json = $4::jsonb, + result_json = $5::jsonb, + error_json = $6::jsonb, + expires_at = coalesce($7::timestamptz, expires_at), + updated_at = $8::timestamptz, + finished_at = case + when $9::timestamptz is null then finished_at + else $9::timestamptz + end + where id = $1 + and status = $2 + returning + id, + workspace_id, + agent_id, + operation_id, + status, + progress_json, + result_json, + error_json, + to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at, + to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, + to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at, + to_char(finished_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as finished_at", + ) + .bind(request.job_id.as_str()) + .bind(serialize_enum_text(&request.current_status, "status")?) + .bind(serialize_enum_text(&request.next_status, "status")?) + .bind(Json(request.progress.clone())) + .bind(request.result.cloned().map(Json)) + .bind(request.error.cloned().map(Json)) + .bind(request.expires_at) + .bind(request.updated_at) + .bind(request.finished_at) + .fetch_optional(&self.pool) + .await?; + + match row.as_ref() { + Some(row) => map_async_job(row), + None => { + if self.get_async_job(request.job_id).await?.is_some() { + Err(RegistryError::InvalidAsyncJobTransition { + job_id: request.job_id.as_str().to_owned(), + from: serialize_enum_text(&request.current_status, "status")?, + to: serialize_enum_text(&request.next_status, "status")?, + }) + } else { + Err(RegistryError::AsyncJobNotFound { + job_id: request.job_id.as_str().to_owned(), + }) + } + } + } + } + + pub async fn cancel_async_job(&self, id: &AsyncJobId, now: &str) -> Result<(), RegistryError> { + let result = sqlx::query( + "update async_jobs + set status = 'cancelled', + updated_at = $2::timestamptz, + finished_at = $2::timestamptz + where id = $1 + and status in ('created', 'running')", + ) + .bind(id.as_str()) + .bind(now) + .execute(&self.pool) + .await?; + + if result.rows_affected() > 0 { + return Ok(()); + } + + match self.get_async_job(id).await? { + Some(existing) => Err(RegistryError::InvalidAsyncJobTransition { + job_id: id.as_str().to_owned(), + from: serialize_enum_text(&existing.status, "status")?, + to: "cancelled".to_owned(), + }), + None => Err(RegistryError::AsyncJobNotFound { + job_id: id.as_str().to_owned(), + }), + } + } + + pub async fn list_async_jobs( + &self, + filter: AsyncJobFilter<'_>, + ) -> Result, RegistryError> { + let status = filter + .status + .as_ref() + .map(|value| serialize_enum_text(value, "status")) + .transpose()?; + let limit = i64::from(filter.limit); + + let rows = sqlx::query( + "select + id, + workspace_id, + agent_id, + operation_id, + status, + progress_json, + result_json, + error_json, + to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at, + to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, + to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at, + to_char(finished_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as finished_at + from async_jobs + where workspace_id = $1 + and ($2::text is null or agent_id = $2) + and ($3::text is null or operation_id = $3) + and ($4::text is null or status = $4) + order by updated_at desc + limit $5", + ) + .bind(filter.workspace_id.as_str()) + .bind(filter.agent_id.map(|value| value.as_str())) + .bind(filter.operation_id.map(|value| value.as_str())) + .bind(status) + .bind(limit) + .fetch_all(&self.pool) + .await?; + + let items = rows + .iter() + .map(map_async_job) + .collect::, _>>()?; + let total = items.len() as u64; + + Ok(Page { items, total }) + } + + pub async fn delete_expired_async_jobs(&self, now: &str) -> Result { + let result = sqlx::query( + "delete from async_jobs + where expires_at is not null + and expires_at <= $1::timestamptz", + ) + .bind(now) + .execute(&self.pool) + .await?; + + Ok(result.rows_affected()) + } +} diff --git a/crates/crank-registry/src/postgres/workspace.rs b/crates/crank-registry/src/postgres/workspace.rs new file mode 100644 index 0000000..7e3c4e4 --- /dev/null +++ b/crates/crank-registry/src/postgres/workspace.rs @@ -0,0 +1,319 @@ +use super::*; + +impl PostgresRegistry { + pub async fn list_workspaces(&self) -> Result, RegistryError> { + let rows = sqlx::query( + "select + id, + slug, + display_name, + status, + settings_json, + to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, + to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at + from workspaces + order by slug asc", + ) + .fetch_all(&self.pool) + .await?; + + rows.iter().map(map_workspace_record).collect() + } + + pub async fn list_workspaces_for_user( + &self, + user_id: &UserId, + ) -> Result, RegistryError> { + let rows = sqlx::query( + "select + w.id, + w.slug, + w.display_name, + w.status, + w.settings_json, + to_char(w.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, + to_char(w.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at, + m.role + from memberships m + join workspaces w on w.id = m.workspace_id + where m.user_id = $1 + order by w.slug asc", + ) + .bind(user_id.as_str()) + .fetch_all(&self.pool) + .await?; + + rows.iter().map(map_workspace_membership_record).collect() + } + + pub async fn list_memberships( + &self, + workspace_id: &WorkspaceId, + ) -> Result, RegistryError> { + let rows = sqlx::query( + "select + m.workspace_id, + m.user_id, + m.role, + to_char(m.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, + u.email, + u.display_name, + u.status, + to_char(u.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as user_created_at + from memberships m + join users u on u.id = m.user_id + where m.workspace_id = $1 + order by u.email asc", + ) + .bind(workspace_id.as_str()) + .fetch_all(&self.pool) + .await?; + + rows.iter().map(map_membership_record).collect() + } + + pub async fn update_membership_role( + &self, + workspace_id: &WorkspaceId, + user_id: &UserId, + role: MembershipRole, + ) -> Result<(), RegistryError> { + let result = sqlx::query( + "update memberships + set role = $3 + where workspace_id = $1 and user_id = $2", + ) + .bind(workspace_id.as_str()) + .bind(user_id.as_str()) + .bind(serialize_enum_text(&role, "role")?) + .execute(&self.pool) + .await?; + + if result.rows_affected() == 0 { + return Err(RegistryError::MembershipNotFound { + workspace_id: workspace_id.as_str().to_owned(), + user_id: user_id.as_str().to_owned(), + }); + } + + Ok(()) + } + + pub async fn delete_membership( + &self, + workspace_id: &WorkspaceId, + user_id: &UserId, + ) -> Result<(), RegistryError> { + let result = sqlx::query( + "delete from memberships + where workspace_id = $1 and user_id = $2", + ) + .bind(workspace_id.as_str()) + .bind(user_id.as_str()) + .execute(&self.pool) + .await?; + + if result.rows_affected() == 0 { + return Err(RegistryError::MembershipNotFound { + workspace_id: workspace_id.as_str().to_owned(), + user_id: user_id.as_str().to_owned(), + }); + } + + Ok(()) + } + + pub async fn list_invitations( + &self, + workspace_id: &WorkspaceId, + ) -> Result, RegistryError> { + let rows = sqlx::query( + "select + id, + workspace_id, + email, + role, + status, + token_hash, + to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at, + to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at + from invitation_tokens + where workspace_id = $1 + order by created_at desc", + ) + .bind(workspace_id.as_str()) + .fetch_all(&self.pool) + .await?; + + rows.iter().map(map_invitation_record).collect() + } + + pub async fn create_invitation( + &self, + request: CreateInvitationRequest<'_>, + ) -> Result<(), RegistryError> { + sqlx::query( + "insert into invitation_tokens ( + id, + workspace_id, + email, + role, + status, + token_hash, + expires_at, + created_at + ) values ( + $1, $2, $3, $4, $5, $6, $7::timestamptz, $8::timestamptz + )", + ) + .bind(request.invitation.id.as_str()) + .bind(request.invitation.workspace_id.as_str()) + .bind(&request.invitation.email) + .bind(serialize_enum_text(&request.invitation.role, "role")?) + .bind(serialize_enum_text(&request.invitation.status, "status")?) + .bind(&request.invitation.token_hash) + .bind(&request.invitation.expires_at) + .bind(&request.invitation.created_at) + .execute(&self.pool) + .await?; + + Ok(()) + } + + pub async fn delete_invitation( + &self, + workspace_id: &WorkspaceId, + invitation_id: &InvitationId, + ) -> Result<(), RegistryError> { + let result = + sqlx::query("delete from invitation_tokens where workspace_id = $1 and id = $2") + .bind(workspace_id.as_str()) + .bind(invitation_id.as_str()) + .execute(&self.pool) + .await?; + + if result.rows_affected() == 0 { + return Err(RegistryError::InvitationNotFound { + invitation_id: invitation_id.as_str().to_owned(), + }); + } + + Ok(()) + } + + pub async fn delete_workspace(&self, workspace_id: &WorkspaceId) -> Result<(), RegistryError> { + let result = sqlx::query("delete from workspaces where id = $1") + .bind(workspace_id.as_str()) + .execute(&self.pool) + .await?; + + if result.rows_affected() == 0 { + return Err(RegistryError::WorkspaceNotFound { + workspace_id: workspace_id.as_str().to_owned(), + }); + } + + Ok(()) + } + + pub async fn get_workspace( + &self, + workspace_id: &WorkspaceId, + ) -> Result, RegistryError> { + let row = sqlx::query( + "select + id, + slug, + display_name, + status, + settings_json, + to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, + to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at + from workspaces + where id = $1", + ) + .bind(workspace_id.as_str()) + .fetch_optional(&self.pool) + .await?; + + row.as_ref().map(map_workspace_record).transpose() + } + + pub async fn create_workspace( + &self, + request: CreateWorkspaceRequest<'_>, + ) -> Result<(), RegistryError> { + let result = sqlx::query( + "insert into workspaces ( + id, + slug, + display_name, + status, + settings_json, + created_at, + updated_at + ) values ( + $1, $2, $3, $4, $5, $6::timestamptz, $7::timestamptz + )", + ) + .bind(request.workspace.id.as_str()) + .bind(&request.workspace.slug) + .bind(&request.workspace.display_name) + .bind(serialize_enum_text(&request.workspace.status, "status")?) + .bind(Json(request.workspace.settings.clone())) + .bind(&request.workspace.created_at) + .bind(&request.workspace.updated_at) + .execute(&self.pool) + .await; + + match result { + Ok(_) => Ok(()), + Err(sqlx::Error::Database(error)) + if error.constraint() == Some("workspaces_slug_key") => + { + Err(RegistryError::WorkspaceSlugAlreadyExists { + slug: request.workspace.slug.clone(), + }) + } + Err(error) => Err(RegistryError::Storage(error)), + } + } + + pub async fn update_workspace( + &self, + request: UpdateWorkspaceRequest<'_>, + ) -> Result<(), RegistryError> { + let result = sqlx::query( + "update workspaces + set slug = $1, + display_name = $2, + status = $3, + settings_json = $4, + updated_at = $5::timestamptz + where id = $6", + ) + .bind(&request.workspace.slug) + .bind(&request.workspace.display_name) + .bind(serialize_enum_text(&request.workspace.status, "status")?) + .bind(Json(request.workspace.settings.clone())) + .bind(&request.workspace.updated_at) + .bind(request.workspace.id.as_str()) + .execute(&self.pool) + .await; + + match result { + Ok(result) if result.rows_affected() == 0 => Err(RegistryError::WorkspaceNotFound { + workspace_id: request.workspace.id.as_str().to_owned(), + }), + Ok(_) => Ok(()), + Err(sqlx::Error::Database(error)) + if error.constraint() == Some("workspaces_slug_key") => + { + Err(RegistryError::WorkspaceSlugAlreadyExists { + slug: request.workspace.slug.clone(), + }) + } + Err(error) => Err(RegistryError::Storage(error)), + } + } +}