diff --git a/TASKS.md b/TASKS.md index 46c5a8d..6bb1fb4 100644 --- a/TASKS.md +++ b/TASKS.md @@ -2,18 +2,18 @@ ## Current -### `feat/postgres-registry-modularization` +### `feat/sqlx-compile-time-verification` Status: in_progress DoD: -- `crates/crank-registry/src/postgres.rs` is split into domain modules under `postgres/` -- `PostgresRegistry` keeps only constructors and high-level surface wiring -- existing registry tests still pass after module split +- the highest-risk registry queries use `sqlx::query!` / `query_as!` +- compile-time SQL verification is enabled in normal backend workflow +- runtime-only queries remain only where dynamic SQL is truly required ## Next -- `feat/sqlx-compile-time-verification` +- `feat/typed-timestamps` ## Backlog diff --git a/crates/crank-registry/src/postgres/agent.rs b/crates/crank-registry/src/postgres/agent.rs new file mode 100644 index 0000000..4887166 --- /dev/null +++ b/crates/crank-registry/src/postgres/agent.rs @@ -0,0 +1,471 @@ +use super::*; + +impl PostgresRegistry { + pub async fn list_agents( + &self, + workspace_id: &WorkspaceId, + ) -> Result, RegistryError> { + let rows = sqlx::query( + "select + id, + workspace_id, + slug, + display_name, + description, + status, + current_draft_version, + latest_published_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(published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as published_at + from agents + where workspace_id = $1 + order by slug asc", + ) + .bind(workspace_id.as_str()) + .fetch_all(&self.pool) + .await?; + + rows.iter().map(map_agent_summary).collect() + } + + pub async fn get_agent_summary( + &self, + workspace_id: &WorkspaceId, + agent_id: &AgentId, + ) -> Result, RegistryError> { + let row = sqlx::query( + "select + id, + workspace_id, + slug, + display_name, + description, + status, + current_draft_version, + latest_published_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(published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as published_at + from agents + where workspace_id = $1 and id = $2", + ) + .bind(workspace_id.as_str()) + .bind(agent_id.as_str()) + .fetch_optional(&self.pool) + .await?; + + row.as_ref().map(map_agent_summary).transpose() + } + + pub async fn create_agent(&self, request: CreateAgentRequest<'_>) -> Result<(), RegistryError> { + let mut tx = self.pool.begin().await?; + + sqlx::query( + "insert into agents ( + id, + workspace_id, + slug, + display_name, + description, + status, + current_draft_version, + latest_published_version, + created_at, + updated_at, + published_at + ) values ( + $1, $2, $3, $4, $5, $6, $7, $8, $9::timestamptz, $10::timestamptz, $11::timestamptz + )", + ) + .bind(request.agent.id.as_str()) + .bind(request.agent.workspace_id.as_str()) + .bind(&request.agent.slug) + .bind(&request.agent.display_name) + .bind(&request.agent.description) + .bind(serialize_enum_text(&request.agent.status, "status")?) + .bind(to_db_version(request.agent.current_draft_version)) + .bind(request.agent.latest_published_version.map(to_db_version)) + .bind(&request.agent.created_at) + .bind(&request.agent.updated_at) + .bind(request.agent.published_at.as_deref()) + .execute(&mut *tx) + .await?; + + insert_agent_version_row(&mut tx, request.version).await?; + replace_agent_bindings_rows( + &mut tx, + &request.agent.id, + request.version.version, + request.bindings, + ) + .await?; + + tx.commit().await?; + Ok(()) + } + + pub async fn get_agent_version( + &self, + workspace_id: &WorkspaceId, + agent_id: &AgentId, + version: u32, + ) -> Result, RegistryError> { + let summary = match self.get_agent_summary(workspace_id, agent_id).await? { + Some(value) => value, + None => return Ok(None), + }; + + let row = sqlx::query( + "select + agent_id, + version, + status, + instructions_json, + tool_selection_policy_json, + to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at + from agent_versions + where agent_id = $1 and version = $2", + ) + .bind(agent_id.as_str()) + .bind(to_db_version(version)) + .fetch_optional(&self.pool) + .await?; + + let Some(row) = row else { + return Ok(None); + }; + + let bindings = self.list_agent_bindings(agent_id, version).await?; + Ok(Some(map_agent_version_record(&summary, &row, bindings)?)) + } + + pub async fn save_agent_bindings( + &self, + request: SaveAgentBindingsRequest<'_>, + ) -> Result<(), RegistryError> { + if self + .get_agent_summary(request.workspace_id, request.agent_id) + .await? + .is_none() + { + return Err(RegistryError::AgentNotFound { + agent_id: request.agent_id.as_str().to_owned(), + }); + } + + let mut tx = self.pool.begin().await?; + replace_agent_bindings_rows( + &mut tx, + request.agent_id, + request.agent_version, + request.bindings, + ) + .await?; + tx.commit().await?; + Ok(()) + } + + pub async fn update_agent_summary( + &self, + workspace_id: &WorkspaceId, + agent_id: &AgentId, + slug: &str, + display_name: &str, + description: &str, + updated_at: &str, + ) -> Result<(), RegistryError> { + let result = sqlx::query( + "update agents + set slug = $3, + display_name = $4, + description = $5, + updated_at = $6::timestamptz + where workspace_id = $1 and id = $2", + ) + .bind(workspace_id.as_str()) + .bind(agent_id.as_str()) + .bind(slug) + .bind(display_name) + .bind(description) + .bind(updated_at) + .execute(&self.pool) + .await?; + + if result.rows_affected() == 0 { + return Err(RegistryError::AgentNotFound { + agent_id: agent_id.as_str().to_owned(), + }); + } + + Ok(()) + } + + pub async fn delete_agent( + &self, + workspace_id: &WorkspaceId, + agent_id: &AgentId, + ) -> Result<(), RegistryError> { + let result = sqlx::query("delete from agents where workspace_id = $1 and id = $2") + .bind(workspace_id.as_str()) + .bind(agent_id.as_str()) + .execute(&self.pool) + .await?; + + if result.rows_affected() == 0 { + return Err(RegistryError::AgentNotFound { + agent_id: agent_id.as_str().to_owned(), + }); + } + + Ok(()) + } + + pub async fn publish_agent( + &self, + request: PublishAgentRequest<'_>, + ) -> Result<(), RegistryError> { + if self + .get_agent_version(request.workspace_id, request.agent_id, request.version) + .await? + .is_none() + { + return Err(RegistryError::AgentNotFound { + agent_id: request.agent_id.as_str().to_owned(), + }); + } + + let mut tx = self.pool.begin().await?; + + sqlx::query( + "insert into published_agents ( + agent_id, + version, + published_at, + published_by + ) values ($1, $2, $3::timestamptz, $4) + on conflict(agent_id) do update set + version = excluded.version, + published_at = excluded.published_at, + published_by = excluded.published_by", + ) + .bind(request.agent_id.as_str()) + .bind(to_db_version(request.version)) + .bind(request.published_at) + .bind(request.published_by) + .execute(&mut *tx) + .await?; + + sqlx::query( + "update agent_versions + set status = $1 + where agent_id = $2 and version = $3", + ) + .bind(serialize_enum_text(&AgentStatus::Published, "status")?) + .bind(request.agent_id.as_str()) + .bind(to_db_version(request.version)) + .execute(&mut *tx) + .await?; + + sqlx::query( + "update agents + set status = $1, + latest_published_version = $2, + published_at = $3::timestamptz, + updated_at = $4::timestamptz + where id = $5 and workspace_id = $6", + ) + .bind(serialize_enum_text(&AgentStatus::Published, "status")?) + .bind(to_db_version(request.version)) + .bind(request.published_at) + .bind(request.published_at) + .bind(request.agent_id.as_str()) + .bind(request.workspace_id.as_str()) + .execute(&mut *tx) + .await?; + + tx.commit().await?; + Ok(()) + } + + pub async fn unpublish_agent( + &self, + workspace_id: &WorkspaceId, + agent_id: &AgentId, + updated_at: &str, + ) -> Result<(), RegistryError> { + let mut tx = self.pool.begin().await?; + + sqlx::query("delete from published_agents where agent_id = $1") + .bind(agent_id.as_str()) + .execute(&mut *tx) + .await?; + + sqlx::query( + "update agent_versions + set status = $1 + where agent_id = $2 + and version = ( + select current_draft_version + from agents + where id = $2 and workspace_id = $3 + )", + ) + .bind(serialize_enum_text(&AgentStatus::Draft, "status")?) + .bind(agent_id.as_str()) + .bind(workspace_id.as_str()) + .execute(&mut *tx) + .await?; + + let result = sqlx::query( + "update agents + set status = $1, + published_at = null, + updated_at = $2::timestamptz + where id = $3 and workspace_id = $4", + ) + .bind(serialize_enum_text(&AgentStatus::Draft, "status")?) + .bind(updated_at) + .bind(agent_id.as_str()) + .bind(workspace_id.as_str()) + .execute(&mut *tx) + .await?; + + if result.rows_affected() == 0 { + return Err(RegistryError::AgentNotFound { + agent_id: agent_id.as_str().to_owned(), + }); + } + + tx.commit().await?; + Ok(()) + } + + pub async fn archive_agent( + &self, + workspace_id: &WorkspaceId, + agent_id: &AgentId, + updated_at: &str, + ) -> Result<(), RegistryError> { + let mut tx = self.pool.begin().await?; + + sqlx::query("delete from published_agents where agent_id = $1") + .bind(agent_id.as_str()) + .execute(&mut *tx) + .await?; + + sqlx::query( + "update agent_versions + set status = $1 + where agent_id = $2 + and version = ( + select current_draft_version + from agents + where id = $2 and workspace_id = $3 + )", + ) + .bind(serialize_enum_text(&AgentStatus::Archived, "status")?) + .bind(agent_id.as_str()) + .bind(workspace_id.as_str()) + .execute(&mut *tx) + .await?; + + let result = sqlx::query( + "update agents + set status = $1, + published_at = null, + updated_at = $2::timestamptz + where id = $3 and workspace_id = $4", + ) + .bind(serialize_enum_text(&AgentStatus::Archived, "status")?) + .bind(updated_at) + .bind(agent_id.as_str()) + .bind(workspace_id.as_str()) + .execute(&mut *tx) + .await?; + + if result.rows_affected() == 0 { + return Err(RegistryError::AgentNotFound { + agent_id: agent_id.as_str().to_owned(), + }); + } + + tx.commit().await?; + Ok(()) + } + + pub async fn get_published_agent_tools_by_slug( + &self, + workspace_slug: &str, + agent_slug: &str, + ) -> Result, RegistryError> { + let rows = sqlx::query( + "select + w.id as workspace_id, + w.slug as workspace_slug, + a.id as agent_id, + a.slug as agent_slug, + b.tool_name, + b.tool_title, + coalesce(b.tool_description_override, ov.tool_description_json->>'description') as tool_description, + o.id, + o.name, + o.display_name, + o.category, + o.protocol, + to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at, + to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at, + to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at, + ov.version, + ov.status, + ov.target_json, + ov.input_schema_json, + ov.output_schema_json, + ov.input_mapping_json, + ov.output_mapping_json, + ov.execution_config_json, + ov.tool_description_json, + ov.samples_json, + ov.generated_draft_json, + ov.config_export_json, + ov.change_note, + to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, + ov.created_by + from workspaces w + join agents a on a.workspace_id = w.id + join published_agents pa on pa.agent_id = a.id + join agent_operation_bindings b on b.agent_id = a.id and b.agent_version = pa.version + join operation_versions ov on ov.operation_id = b.operation_id and ov.version = b.operation_version + join operations o on o.id = ov.operation_id and o.workspace_id = w.id + where w.slug = $1 and a.slug = $2 and b.enabled = true + order by b.tool_name asc", + ) + .bind(workspace_slug) + .bind(agent_slug) + .fetch_all(&self.pool) + .await?; + + if rows.is_empty() { + let exists = sqlx::query( + "select 1 + from workspaces w + join agents a on a.workspace_id = w.id + join published_agents pa on pa.agent_id = a.id + where w.slug = $1 and a.slug = $2", + ) + .bind(workspace_slug) + .bind(agent_slug) + .fetch_optional(&self.pool) + .await?; + + if exists.is_none() { + return Err(RegistryError::PublishedAgentNotFound { + workspace_slug: workspace_slug.to_owned(), + agent_slug: agent_slug.to_owned(), + }); + } + } + + rows.iter().map(map_published_agent_tool).collect() + } +} diff --git a/crates/crank-registry/src/postgres/mod.rs b/crates/crank-registry/src/postgres/mod.rs index 703d040..a0c53c2 100644 --- a/crates/crank-registry/src/postgres/mod.rs +++ b/crates/crank-registry/src/postgres/mod.rs @@ -1,6 +1,8 @@ +mod agent; mod api_key; mod auth; mod observability; +mod operation; mod secret; mod stream; mod workspace; @@ -179,474 +181,6 @@ impl PostgresRegistry { Ok(()) } - pub async fn list_agents( - &self, - workspace_id: &WorkspaceId, - ) -> Result, RegistryError> { - let rows = sqlx::query( - "select - id, - workspace_id, - slug, - display_name, - description, - status, - current_draft_version, - latest_published_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(published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as published_at - from agents - where workspace_id = $1 - order by slug asc", - ) - .bind(workspace_id.as_str()) - .fetch_all(&self.pool) - .await?; - - rows.iter().map(map_agent_summary).collect() - } - - pub async fn get_agent_summary( - &self, - workspace_id: &WorkspaceId, - agent_id: &AgentId, - ) -> Result, RegistryError> { - let row = sqlx::query( - "select - id, - workspace_id, - slug, - display_name, - description, - status, - current_draft_version, - latest_published_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(published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as published_at - from agents - where workspace_id = $1 and id = $2", - ) - .bind(workspace_id.as_str()) - .bind(agent_id.as_str()) - .fetch_optional(&self.pool) - .await?; - - row.as_ref().map(map_agent_summary).transpose() - } - - pub async fn create_agent(&self, request: CreateAgentRequest<'_>) -> Result<(), RegistryError> { - let mut tx = self.pool.begin().await?; - - sqlx::query( - "insert into agents ( - id, - workspace_id, - slug, - display_name, - description, - status, - current_draft_version, - latest_published_version, - created_at, - updated_at, - published_at - ) values ( - $1, $2, $3, $4, $5, $6, $7, $8, $9::timestamptz, $10::timestamptz, $11::timestamptz - )", - ) - .bind(request.agent.id.as_str()) - .bind(request.agent.workspace_id.as_str()) - .bind(&request.agent.slug) - .bind(&request.agent.display_name) - .bind(&request.agent.description) - .bind(serialize_enum_text(&request.agent.status, "status")?) - .bind(to_db_version(request.agent.current_draft_version)) - .bind(request.agent.latest_published_version.map(to_db_version)) - .bind(&request.agent.created_at) - .bind(&request.agent.updated_at) - .bind(request.agent.published_at.as_deref()) - .execute(&mut *tx) - .await?; - - insert_agent_version_row(&mut tx, request.version).await?; - replace_agent_bindings_rows( - &mut tx, - &request.agent.id, - request.version.version, - request.bindings, - ) - .await?; - - tx.commit().await?; - Ok(()) - } - - pub async fn get_agent_version( - &self, - workspace_id: &WorkspaceId, - agent_id: &AgentId, - version: u32, - ) -> Result, RegistryError> { - let summary = match self.get_agent_summary(workspace_id, agent_id).await? { - Some(value) => value, - None => return Ok(None), - }; - - let row = sqlx::query( - "select - agent_id, - version, - status, - instructions_json, - tool_selection_policy_json, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at - from agent_versions - where agent_id = $1 and version = $2", - ) - .bind(agent_id.as_str()) - .bind(to_db_version(version)) - .fetch_optional(&self.pool) - .await?; - - let Some(row) = row else { - return Ok(None); - }; - - let bindings = self.list_agent_bindings(agent_id, version).await?; - Ok(Some(map_agent_version_record(&summary, &row, bindings)?)) - } - - pub async fn save_agent_bindings( - &self, - request: SaveAgentBindingsRequest<'_>, - ) -> Result<(), RegistryError> { - if self - .get_agent_summary(request.workspace_id, request.agent_id) - .await? - .is_none() - { - return Err(RegistryError::AgentNotFound { - agent_id: request.agent_id.as_str().to_owned(), - }); - } - - let mut tx = self.pool.begin().await?; - replace_agent_bindings_rows( - &mut tx, - request.agent_id, - request.agent_version, - request.bindings, - ) - .await?; - tx.commit().await?; - Ok(()) - } - - pub async fn update_agent_summary( - &self, - workspace_id: &WorkspaceId, - agent_id: &AgentId, - slug: &str, - display_name: &str, - description: &str, - updated_at: &str, - ) -> Result<(), RegistryError> { - let result = sqlx::query( - "update agents - set slug = $3, - display_name = $4, - description = $5, - updated_at = $6::timestamptz - where workspace_id = $1 and id = $2", - ) - .bind(workspace_id.as_str()) - .bind(agent_id.as_str()) - .bind(slug) - .bind(display_name) - .bind(description) - .bind(updated_at) - .execute(&self.pool) - .await?; - - if result.rows_affected() == 0 { - return Err(RegistryError::AgentNotFound { - agent_id: agent_id.as_str().to_owned(), - }); - } - - Ok(()) - } - - pub async fn delete_agent( - &self, - workspace_id: &WorkspaceId, - agent_id: &AgentId, - ) -> Result<(), RegistryError> { - let result = sqlx::query("delete from agents where workspace_id = $1 and id = $2") - .bind(workspace_id.as_str()) - .bind(agent_id.as_str()) - .execute(&self.pool) - .await?; - - if result.rows_affected() == 0 { - return Err(RegistryError::AgentNotFound { - agent_id: agent_id.as_str().to_owned(), - }); - } - - Ok(()) - } - - pub async fn publish_agent( - &self, - request: PublishAgentRequest<'_>, - ) -> Result<(), RegistryError> { - if self - .get_agent_version(request.workspace_id, request.agent_id, request.version) - .await? - .is_none() - { - return Err(RegistryError::AgentNotFound { - agent_id: request.agent_id.as_str().to_owned(), - }); - } - - let mut tx = self.pool.begin().await?; - - sqlx::query( - "insert into published_agents ( - agent_id, - version, - published_at, - published_by - ) values ($1, $2, $3::timestamptz, $4) - on conflict(agent_id) do update set - version = excluded.version, - published_at = excluded.published_at, - published_by = excluded.published_by", - ) - .bind(request.agent_id.as_str()) - .bind(to_db_version(request.version)) - .bind(request.published_at) - .bind(request.published_by) - .execute(&mut *tx) - .await?; - - sqlx::query( - "update agent_versions - set status = $1 - where agent_id = $2 and version = $3", - ) - .bind(serialize_enum_text(&AgentStatus::Published, "status")?) - .bind(request.agent_id.as_str()) - .bind(to_db_version(request.version)) - .execute(&mut *tx) - .await?; - - sqlx::query( - "update agents - set status = $1, - latest_published_version = $2, - published_at = $3::timestamptz, - updated_at = $4::timestamptz - where id = $5 and workspace_id = $6", - ) - .bind(serialize_enum_text(&AgentStatus::Published, "status")?) - .bind(to_db_version(request.version)) - .bind(request.published_at) - .bind(request.published_at) - .bind(request.agent_id.as_str()) - .bind(request.workspace_id.as_str()) - .execute(&mut *tx) - .await?; - - tx.commit().await?; - Ok(()) - } - - pub async fn unpublish_agent( - &self, - workspace_id: &WorkspaceId, - agent_id: &AgentId, - updated_at: &str, - ) -> Result<(), RegistryError> { - let mut tx = self.pool.begin().await?; - - sqlx::query("delete from published_agents where agent_id = $1") - .bind(agent_id.as_str()) - .execute(&mut *tx) - .await?; - - sqlx::query( - "update agent_versions - set status = $1 - where agent_id = $2 - and version = ( - select current_draft_version - from agents - where id = $2 and workspace_id = $3 - )", - ) - .bind(serialize_enum_text(&AgentStatus::Draft, "status")?) - .bind(agent_id.as_str()) - .bind(workspace_id.as_str()) - .execute(&mut *tx) - .await?; - - let result = sqlx::query( - "update agents - set status = $1, - published_at = null, - updated_at = $2::timestamptz - where id = $3 and workspace_id = $4", - ) - .bind(serialize_enum_text(&AgentStatus::Draft, "status")?) - .bind(updated_at) - .bind(agent_id.as_str()) - .bind(workspace_id.as_str()) - .execute(&mut *tx) - .await?; - - if result.rows_affected() == 0 { - return Err(RegistryError::AgentNotFound { - agent_id: agent_id.as_str().to_owned(), - }); - } - - tx.commit().await?; - Ok(()) - } - - pub async fn archive_agent( - &self, - workspace_id: &WorkspaceId, - agent_id: &AgentId, - updated_at: &str, - ) -> Result<(), RegistryError> { - let mut tx = self.pool.begin().await?; - - sqlx::query("delete from published_agents where agent_id = $1") - .bind(agent_id.as_str()) - .execute(&mut *tx) - .await?; - - sqlx::query( - "update agent_versions - set status = $1 - where agent_id = $2 - and version = ( - select current_draft_version - from agents - where id = $2 and workspace_id = $3 - )", - ) - .bind(serialize_enum_text(&AgentStatus::Archived, "status")?) - .bind(agent_id.as_str()) - .bind(workspace_id.as_str()) - .execute(&mut *tx) - .await?; - - let result = sqlx::query( - "update agents - set status = $1, - published_at = null, - updated_at = $2::timestamptz - where id = $3 and workspace_id = $4", - ) - .bind(serialize_enum_text(&AgentStatus::Archived, "status")?) - .bind(updated_at) - .bind(agent_id.as_str()) - .bind(workspace_id.as_str()) - .execute(&mut *tx) - .await?; - - if result.rows_affected() == 0 { - return Err(RegistryError::AgentNotFound { - agent_id: agent_id.as_str().to_owned(), - }); - } - - tx.commit().await?; - Ok(()) - } - - pub async fn get_published_agent_tools_by_slug( - &self, - workspace_slug: &str, - agent_slug: &str, - ) -> Result, RegistryError> { - let rows = sqlx::query( - "select - w.id as workspace_id, - w.slug as workspace_slug, - a.id as agent_id, - a.slug as agent_slug, - b.tool_name, - b.tool_title, - coalesce(b.tool_description_override, ov.tool_description_json->>'description') as tool_description, - o.id, - o.name, - o.display_name, - o.category, - o.protocol, - to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at, - to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at, - to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at, - ov.version, - ov.status, - ov.target_json, - ov.input_schema_json, - ov.output_schema_json, - ov.input_mapping_json, - ov.output_mapping_json, - ov.execution_config_json, - ov.tool_description_json, - ov.samples_json, - ov.generated_draft_json, - ov.config_export_json, - ov.change_note, - to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - ov.created_by - from workspaces w - join agents a on a.workspace_id = w.id - join published_agents pa on pa.agent_id = a.id - join agent_operation_bindings b on b.agent_id = a.id and b.agent_version = pa.version - join operation_versions ov on ov.operation_id = b.operation_id and ov.version = b.operation_version - join operations o on o.id = ov.operation_id and o.workspace_id = w.id - where w.slug = $1 and a.slug = $2 and b.enabled = true - order by b.tool_name asc", - ) - .bind(workspace_slug) - .bind(agent_slug) - .fetch_all(&self.pool) - .await?; - - if rows.is_empty() { - let exists = sqlx::query( - "select 1 - from workspaces w - join agents a on a.workspace_id = w.id - join published_agents pa on pa.agent_id = a.id - where w.slug = $1 and a.slug = $2", - ) - .bind(workspace_slug) - .bind(agent_slug) - .fetch_optional(&self.pool) - .await?; - - if exists.is_none() { - return Err(RegistryError::PublishedAgentNotFound { - workspace_slug: workspace_slug.to_owned(), - agent_slug: agent_slug.to_owned(), - }); - } - } - - rows.iter().map(map_published_agent_tool).collect() - } - async fn connect_in_schema( database_url: &str, schema: Option<&str>, @@ -669,811 +203,6 @@ impl PostgresRegistry { Ok(registry) } - pub async fn create_operation( - &self, - workspace_id: &WorkspaceId, - snapshot: &RegistryOperation, - created_by: Option<&str>, - ) -> Result<(), RegistryError> { - if snapshot.version != 1 { - return Err(RegistryError::InvalidInitialVersion { - operation_id: snapshot.id.as_str().to_owned(), - version: snapshot.version, - }); - } - - if self - .get_operation_summary(workspace_id, &snapshot.id) - .await? - .is_some() - { - return Err(RegistryError::OperationAlreadyExists { - operation_id: snapshot.id.as_str().to_owned(), - }); - } - - let mut tx = self.pool.begin().await?; - - sqlx::query( - "insert into operations ( - id, - workspace_id, - name, - display_name, - category, - protocol, - status, - current_draft_version, - latest_published_version, - created_at, - updated_at, - published_at - ) values ( - $1, $2, $3, $4, $5, $6, $7, $8, $9, - $10::timestamptz, - $11::timestamptz, - $12::timestamptz - )", - ) - .bind(snapshot.id.as_str()) - .bind(workspace_id.as_str()) - .bind(&snapshot.name) - .bind(&snapshot.display_name) - .bind(&snapshot.category) - .bind(serialize_enum_text(&snapshot.protocol, "protocol")?) - .bind(serialize_enum_text(&snapshot.status, "status")?) - .bind(to_db_version(snapshot.version)) - .bind( - snapshot - .published_at - .as_ref() - .map(|_| to_db_version(snapshot.version)), - ) - .bind(&snapshot.created_at) - .bind(&snapshot.updated_at) - .bind(snapshot.published_at.as_deref()) - .execute(&mut *tx) - .await?; - - insert_version_row(&mut tx, snapshot, None, created_by).await?; - - tx.commit().await?; - Ok(()) - } - - pub async fn create_version( - &self, - request: CreateVersionRequest<'_>, - ) -> Result<(), RegistryError> { - let Some(summary) = self - .get_operation_summary(request.workspace_id, &request.snapshot.id) - .await? - else { - return Err(RegistryError::OperationNotFound { - operation_id: request.snapshot.id.as_str().to_owned(), - }); - }; - - assert_immutable_fields(&summary, request.snapshot)?; - - let expected = summary.current_draft_version + 1; - if request.snapshot.version != expected { - return Err(RegistryError::InvalidVersionSequence { - operation_id: request.snapshot.id.as_str().to_owned(), - expected, - actual: request.snapshot.version, - }); - } - - let mut tx = self.pool.begin().await?; - - insert_version_row( - &mut tx, - request.snapshot, - request.change_note, - request.created_by, - ) - .await?; - - sqlx::query( - "update operations - set status = $1, - current_draft_version = $2, - updated_at = $3::timestamptz - where id = $4", - ) - .bind(serialize_enum_text(&request.snapshot.status, "status")?) - .bind(to_db_version(request.snapshot.version)) - .bind(&request.snapshot.updated_at) - .bind(request.snapshot.id.as_str()) - .execute(&mut *tx) - .await?; - - tx.commit().await?; - Ok(()) - } - - pub async fn update_operation_draft( - &self, - workspace_id: &WorkspaceId, - snapshot: &RegistryOperation, - ) -> Result<(), RegistryError> { - let Some(summary) = self - .get_operation_summary(workspace_id, &snapshot.id) - .await? - else { - return Err(RegistryError::OperationNotFound { - operation_id: snapshot.id.as_str().to_owned(), - }); - }; - - if snapshot.version != summary.current_draft_version { - return Err(RegistryError::InvalidVersionSequence { - operation_id: snapshot.id.as_str().to_owned(), - expected: summary.current_draft_version, - actual: snapshot.version, - }); - } - - assert_immutable_fields(&summary, snapshot)?; - - let mut tx = self.pool.begin().await?; - - sqlx::query( - "update operations - set display_name = $1, - category = $2, - status = $3, - updated_at = $4::timestamptz - where workspace_id = $5 and id = $6", - ) - .bind(&snapshot.display_name) - .bind(&snapshot.category) - .bind(serialize_enum_text(&snapshot.status, "status")?) - .bind(&snapshot.updated_at) - .bind(workspace_id.as_str()) - .bind(snapshot.id.as_str()) - .execute(&mut *tx) - .await?; - - sqlx::query( - "update operation_versions - set status = $1, - target_json = $2, - input_schema_json = $3, - output_schema_json = $4, - input_mapping_json = $5, - output_mapping_json = $6, - execution_config_json = $7, - tool_description_json = $8, - samples_json = $9, - generated_draft_json = $10, - config_export_json = $11 - where operation_id = $12 and version = $13", - ) - .bind(serialize_enum_text(&snapshot.status, "status")?) - .bind(Json(serialize_json_value(&snapshot.target)?)) - .bind(Json(serialize_json_value(&snapshot.input_schema)?)) - .bind(Json(serialize_json_value(&snapshot.output_schema)?)) - .bind(Json(serialize_json_value(&snapshot.input_mapping)?)) - .bind(Json(serialize_json_value(&snapshot.output_mapping)?)) - .bind(Json(serialize_json_value(&snapshot.execution_config)?)) - .bind(Json(serialize_json_value(&snapshot.tool_description)?)) - .bind(serialize_option_json_value(&snapshot.samples)?.map(Json)) - .bind(serialize_option_json_value(&snapshot.generated_draft)?.map(Json)) - .bind(serialize_option_json_value(&snapshot.config_export)?.map(Json)) - .bind(snapshot.id.as_str()) - .bind(to_db_version(snapshot.version)) - .execute(&mut *tx) - .await?; - - tx.commit().await?; - Ok(()) - } - - pub async fn archive_operation( - &self, - workspace_id: &WorkspaceId, - operation_id: &OperationId, - archived_at: &str, - ) -> Result<(), RegistryError> { - let Some(summary) = self - .get_operation_summary(workspace_id, operation_id) - .await? - else { - return Err(RegistryError::OperationNotFound { - operation_id: operation_id.as_str().to_owned(), - }); - }; - - let mut tx = self.pool.begin().await?; - - sqlx::query( - "update operations - set status = $1, - updated_at = $2::timestamptz - where workspace_id = $3 and id = $4", - ) - .bind(serialize_enum_text(&OperationStatus::Archived, "status")?) - .bind(archived_at) - .bind(workspace_id.as_str()) - .bind(operation_id.as_str()) - .execute(&mut *tx) - .await?; - - sqlx::query( - "update operation_versions - set status = $1 - where operation_id = $2 and version = $3", - ) - .bind(serialize_enum_text(&OperationStatus::Archived, "status")?) - .bind(operation_id.as_str()) - .bind(to_db_version(summary.current_draft_version)) - .execute(&mut *tx) - .await?; - - tx.commit().await?; - Ok(()) - } - - pub async fn delete_operation( - &self, - workspace_id: &WorkspaceId, - operation_id: &OperationId, - ) -> Result<(), RegistryError> { - if self - .has_published_agent_bindings_for_operation(workspace_id, operation_id) - .await? - { - return Err(RegistryError::OperationHasPublishedAgentBindings { - operation_id: operation_id.as_str().to_owned(), - }); - } - - let deleted = sqlx::query( - "delete from operations - where workspace_id = $1 and id = $2", - ) - .bind(workspace_id.as_str()) - .bind(operation_id.as_str()) - .execute(&self.pool) - .await? - .rows_affected(); - - if deleted == 0 { - return Err(RegistryError::OperationNotFound { - operation_id: operation_id.as_str().to_owned(), - }); - } - - Ok(()) - } - - pub async fn has_published_agent_bindings_for_operation( - &self, - workspace_id: &WorkspaceId, - operation_id: &OperationId, - ) -> Result { - let row = sqlx::query( - "select 1 - from agents a - join published_agents pa on pa.agent_id = a.id - join agent_operation_bindings b - on b.agent_id = a.id and b.agent_version = pa.version - where a.workspace_id = $1 - and b.operation_id = $2 - limit 1", - ) - .bind(workspace_id.as_str()) - .bind(operation_id.as_str()) - .fetch_optional(&self.pool) - .await?; - - Ok(row.is_some()) - } - - pub async fn list_operations( - &self, - workspace_id: &WorkspaceId, - ) -> Result, RegistryError> { - let rows = sqlx::query( - "select - operations.id, - operations.workspace_id, - operations.name, - operations.display_name, - operations.category, - operations.protocol, - ov.target_json, - operations.status, - operations.current_draft_version, - operations.latest_published_version, - to_char(operations.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - to_char(operations.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at, - to_char(operations.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as published_at - from operations - join operation_versions ov - on ov.operation_id = operations.id - and ov.version = operations.current_draft_version - where operations.workspace_id = $1 - order by operations.name asc", - ) - .bind(workspace_id.as_str()) - .fetch_all(&self.pool) - .await?; - - rows.iter().map(map_operation_summary).collect() - } - - pub async fn list_operation_usage_summaries( - &self, - workspace_id: &WorkspaceId, - created_after: &str, - ) -> Result, RegistryError> { - let rows = sqlx::query( - "select - operation_id, - count(*)::bigint as calls_today, - coalesce( - round((sum(case when status = 'error' then 1 else 0 end)::numeric / nullif(count(*), 0)) * 100, 2), - 0 - )::float8 as error_rate_pct, - coalesce(round(avg(duration_ms))::bigint, 0) as avg_latency_ms - from invocation_logs - where workspace_id = $1 - and created_at >= $2::timestamptz - group by operation_id", - ) - .bind(workspace_id.as_str()) - .bind(created_after) - .fetch_all(&self.pool) - .await?; - - rows.iter().map(map_operation_usage_summary).collect() - } - - pub async fn list_operation_agent_refs( - &self, - workspace_id: &WorkspaceId, - ) -> Result, RegistryError> { - let rows = sqlx::query( - "select - b.operation_id, - a.id as agent_id, - a.slug as agent_slug, - a.display_name - from agents a - join published_agents pa on pa.agent_id = a.id - join agent_operation_bindings b - on b.agent_id = a.id and b.agent_version = pa.version - where a.workspace_id = $1 - order by a.display_name asc, b.tool_name asc", - ) - .bind(workspace_id.as_str()) - .fetch_all(&self.pool) - .await?; - - rows.iter().map(map_operation_agent_ref).collect() - } - - pub async fn get_operation_summary( - &self, - workspace_id: &WorkspaceId, - operation_id: &OperationId, - ) -> Result, RegistryError> { - let row = sqlx::query( - "select - operations.id, - operations.workspace_id, - operations.name, - operations.display_name, - operations.category, - operations.protocol, - ov.target_json, - operations.status, - operations.current_draft_version, - operations.latest_published_version, - to_char(operations.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - to_char(operations.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at, - to_char(operations.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as published_at - from operations - join operation_versions ov - on ov.operation_id = operations.id - and ov.version = operations.current_draft_version - where operations.workspace_id = $1 and operations.id = $2", - ) - .bind(workspace_id.as_str()) - .bind(operation_id.as_str()) - .fetch_optional(&self.pool) - .await?; - - row.as_ref().map(map_operation_summary).transpose() - } - - pub async fn get_operation_version( - &self, - workspace_id: &WorkspaceId, - operation_id: &OperationId, - version: u32, - ) -> Result, RegistryError> { - let row = sqlx::query( - "select - o.id, - o.workspace_id, - o.name, - o.display_name, - o.category, - o.protocol, - to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at, - to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at, - to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at, - ov.version, - ov.status, - ov.target_json, - ov.input_schema_json, - ov.output_schema_json, - ov.input_mapping_json, - ov.output_mapping_json, - ov.execution_config_json, - ov.tool_description_json, - ov.samples_json, - ov.generated_draft_json, - ov.config_export_json, - ov.change_note, - to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - ov.created_by - from operation_versions ov - join operations o on o.id = ov.operation_id - where o.workspace_id = $1 and ov.operation_id = $2 and ov.version = $3", - ) - .bind(workspace_id.as_str()) - .bind(operation_id.as_str()) - .bind(to_db_version(version)) - .fetch_optional(&self.pool) - .await?; - - row.as_ref().map(map_operation_version_record).transpose() - } - - pub async fn list_operation_versions( - &self, - workspace_id: &WorkspaceId, - operation_id: &OperationId, - ) -> Result, RegistryError> { - let rows = sqlx::query( - "select - o.id, - o.workspace_id, - o.name, - o.display_name, - o.category, - o.protocol, - to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at, - to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at, - to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at, - ov.version, - ov.status, - ov.target_json, - ov.input_schema_json, - ov.output_schema_json, - ov.input_mapping_json, - ov.output_mapping_json, - ov.execution_config_json, - ov.tool_description_json, - ov.samples_json, - ov.generated_draft_json, - ov.config_export_json, - ov.change_note, - to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - ov.created_by - from operation_versions ov - join operations o on o.id = ov.operation_id - where o.workspace_id = $1 and ov.operation_id = $2 - order by ov.version asc", - ) - .bind(workspace_id.as_str()) - .bind(operation_id.as_str()) - .fetch_all(&self.pool) - .await?; - - rows.iter().map(map_operation_version_record).collect() - } - - pub async fn publish_operation( - &self, - request: PublishRequest<'_>, - ) -> Result<(), RegistryError> { - if self - .get_operation_version(request.workspace_id, request.operation_id, request.version) - .await? - .is_none() - { - return Err(RegistryError::OperationVersionNotFound { - operation_id: request.operation_id.as_str().to_owned(), - version: request.version, - }); - } - - let mut tx = self.pool.begin().await?; - - sqlx::query( - "insert into published_operations ( - operation_id, - version, - published_at, - published_by - ) values ($1, $2, $3::timestamptz, $4) - on conflict(operation_id) do update set - version = excluded.version, - published_at = excluded.published_at, - published_by = excluded.published_by", - ) - .bind(request.operation_id.as_str()) - .bind(to_db_version(request.version)) - .bind(request.published_at) - .bind(request.published_by) - .execute(&mut *tx) - .await?; - - sqlx::query( - "update operation_versions - set status = $1 - where operation_id = $2 and version = $3", - ) - .bind(serialize_enum_text(&OperationStatus::Published, "status")?) - .bind(request.operation_id.as_str()) - .bind(to_db_version(request.version)) - .execute(&mut *tx) - .await?; - - sqlx::query( - "update operations - set status = $1, - latest_published_version = $2, - published_at = $3::timestamptz, - updated_at = $4::timestamptz - where id = $5", - ) - .bind(serialize_enum_text(&OperationStatus::Published, "status")?) - .bind(to_db_version(request.version)) - .bind(request.published_at) - .bind(request.published_at) - .bind(request.operation_id.as_str()) - .execute(&mut *tx) - .await?; - - tx.commit().await?; - Ok(()) - } - - pub async fn get_published_operation( - &self, - operation_id: &OperationId, - ) -> Result, RegistryError> { - let row = sqlx::query( - "select - o.id, - o.workspace_id, - o.name, - o.display_name, - o.category, - o.protocol, - to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at, - to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at, - to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at, - ov.version, - ov.status, - ov.target_json, - ov.input_schema_json, - ov.output_schema_json, - ov.input_mapping_json, - ov.output_mapping_json, - ov.execution_config_json, - ov.tool_description_json, - ov.samples_json, - ov.generated_draft_json, - ov.config_export_json, - ov.change_note, - to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - ov.created_by - from published_operations po - join operation_versions ov - on ov.operation_id = po.operation_id and ov.version = po.version - join operations o on o.id = po.operation_id - where po.operation_id = $1", - ) - .bind(operation_id.as_str()) - .fetch_optional(&self.pool) - .await?; - - row.as_ref() - .map(|value| map_operation_version_record(value).map(|record| record.snapshot)) - .transpose() - } - - pub async fn list_published_operations(&self) -> Result, RegistryError> { - let rows = sqlx::query( - "select - o.id, - o.workspace_id, - o.name, - o.display_name, - o.category, - o.protocol, - to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at, - to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at, - to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at, - ov.version, - ov.status, - ov.target_json, - ov.input_schema_json, - ov.output_schema_json, - ov.input_mapping_json, - ov.output_mapping_json, - ov.execution_config_json, - ov.tool_description_json, - ov.samples_json, - ov.generated_draft_json, - ov.config_export_json, - ov.change_note, - to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, - ov.created_by - from published_operations po - join operation_versions ov - on ov.operation_id = po.operation_id and ov.version = po.version - join operations o on o.id = po.operation_id - order by o.name asc", - ) - .fetch_all(&self.pool) - .await?; - - rows.iter() - .map(|row| map_operation_version_record(row).map(|record| record.snapshot)) - .collect() - } - - pub async fn save_sample_metadata( - &self, - request: SaveSampleMetadataRequest<'_>, - ) -> Result<(), RegistryError> { - sqlx::query( - "insert into operation_samples ( - id, - operation_id, - version, - sample_kind, - storage_ref, - content_type, - file_name, - created_at - ) values ($1, $2, $3, $4, $5, $6, $7, $8::timestamptz) - on conflict(id) do update set - operation_id = excluded.operation_id, - version = excluded.version, - sample_kind = excluded.sample_kind, - storage_ref = excluded.storage_ref, - content_type = excluded.content_type, - file_name = excluded.file_name, - created_at = excluded.created_at", - ) - .bind(request.sample.id.as_str()) - .bind(request.sample.operation_id.as_str()) - .bind(to_db_version(request.sample.version)) - .bind(serialize_enum_text( - &request.sample.sample_kind, - "sample_kind", - )?) - .bind(&request.sample.storage_ref) - .bind(&request.sample.content_type) - .bind(request.sample.file_name.as_deref()) - .bind(&request.sample.created_at) - .execute(&self.pool) - .await?; - - Ok(()) - } - - pub async fn list_sample_metadata( - &self, - operation_id: &OperationId, - version: u32, - ) -> Result, RegistryError> { - let rows = sqlx::query( - "select - id, - operation_id, - version, - sample_kind, - storage_ref, - content_type, - file_name, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at - from operation_samples - where operation_id = $1 and version = $2 - order by created_at asc", - ) - .bind(operation_id.as_str()) - .bind(to_db_version(version)) - .fetch_all(&self.pool) - .await?; - - rows.iter().map(map_sample_metadata).collect() - } - - pub async fn save_descriptor_metadata( - &self, - request: SaveDescriptorMetadataRequest<'_>, - ) -> Result<(), RegistryError> { - sqlx::query( - "insert into descriptors ( - id, - operation_id, - version, - descriptor_kind, - storage_ref, - source_name, - package_index_json, - created_at - ) values ($1, $2, $3, $4, $5, $6, $7, $8::timestamptz) - on conflict(id) do update set - operation_id = excluded.operation_id, - version = excluded.version, - descriptor_kind = excluded.descriptor_kind, - storage_ref = excluded.storage_ref, - source_name = excluded.source_name, - package_index_json = excluded.package_index_json, - created_at = excluded.created_at", - ) - .bind(request.descriptor.id.as_str()) - .bind( - request - .descriptor - .operation_id - .as_ref() - .map(|value| value.as_str()), - ) - .bind(request.descriptor.version.map(to_db_version)) - .bind(serialize_enum_text( - &request.descriptor.descriptor_kind, - "descriptor_kind", - )?) - .bind(&request.descriptor.storage_ref) - .bind(request.descriptor.source_name.as_deref()) - .bind(serialize_option_json_value(&request.descriptor.package_index)?.map(Json)) - .bind(&request.descriptor.created_at) - .execute(&self.pool) - .await?; - - Ok(()) - } - - pub async fn list_descriptor_metadata( - &self, - operation_id: &OperationId, - version: u32, - ) -> Result, RegistryError> { - let rows = sqlx::query( - "select - id, - operation_id, - version, - descriptor_kind, - storage_ref, - source_name, - package_index_json, - to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at - from descriptors - where operation_id = $1 and version = $2 - order by created_at asc", - ) - .bind(operation_id.as_str()) - .bind(to_db_version(version)) - .fetch_all(&self.pool) - .await?; - - rows.iter().map(map_descriptor_metadata).collect() - } - async fn list_agent_bindings( &self, agent_id: &AgentId, diff --git a/crates/crank-registry/src/postgres/operation.rs b/crates/crank-registry/src/postgres/operation.rs new file mode 100644 index 0000000..d1d0fac --- /dev/null +++ b/crates/crank-registry/src/postgres/operation.rs @@ -0,0 +1,808 @@ +use super::*; + +impl PostgresRegistry { + pub async fn create_operation( + &self, + workspace_id: &WorkspaceId, + snapshot: &RegistryOperation, + created_by: Option<&str>, + ) -> Result<(), RegistryError> { + if snapshot.version != 1 { + return Err(RegistryError::InvalidInitialVersion { + operation_id: snapshot.id.as_str().to_owned(), + version: snapshot.version, + }); + } + + if self + .get_operation_summary(workspace_id, &snapshot.id) + .await? + .is_some() + { + return Err(RegistryError::OperationAlreadyExists { + operation_id: snapshot.id.as_str().to_owned(), + }); + } + + let mut tx = self.pool.begin().await?; + + sqlx::query( + "insert into operations ( + id, + workspace_id, + name, + display_name, + category, + protocol, + status, + current_draft_version, + latest_published_version, + created_at, + updated_at, + published_at + ) values ( + $1, $2, $3, $4, $5, $6, $7, $8, $9, + $10::timestamptz, + $11::timestamptz, + $12::timestamptz + )", + ) + .bind(snapshot.id.as_str()) + .bind(workspace_id.as_str()) + .bind(&snapshot.name) + .bind(&snapshot.display_name) + .bind(&snapshot.category) + .bind(serialize_enum_text(&snapshot.protocol, "protocol")?) + .bind(serialize_enum_text(&snapshot.status, "status")?) + .bind(to_db_version(snapshot.version)) + .bind( + snapshot + .published_at + .as_ref() + .map(|_| to_db_version(snapshot.version)), + ) + .bind(&snapshot.created_at) + .bind(&snapshot.updated_at) + .bind(snapshot.published_at.as_deref()) + .execute(&mut *tx) + .await?; + + insert_version_row(&mut tx, snapshot, None, created_by).await?; + + tx.commit().await?; + Ok(()) + } + + pub async fn create_version( + &self, + request: CreateVersionRequest<'_>, + ) -> Result<(), RegistryError> { + let Some(summary) = self + .get_operation_summary(request.workspace_id, &request.snapshot.id) + .await? + else { + return Err(RegistryError::OperationNotFound { + operation_id: request.snapshot.id.as_str().to_owned(), + }); + }; + + assert_immutable_fields(&summary, request.snapshot)?; + + let expected = summary.current_draft_version + 1; + if request.snapshot.version != expected { + return Err(RegistryError::InvalidVersionSequence { + operation_id: request.snapshot.id.as_str().to_owned(), + expected, + actual: request.snapshot.version, + }); + } + + let mut tx = self.pool.begin().await?; + + insert_version_row( + &mut tx, + request.snapshot, + request.change_note, + request.created_by, + ) + .await?; + + sqlx::query( + "update operations + set status = $1, + current_draft_version = $2, + updated_at = $3::timestamptz + where id = $4", + ) + .bind(serialize_enum_text(&request.snapshot.status, "status")?) + .bind(to_db_version(request.snapshot.version)) + .bind(&request.snapshot.updated_at) + .bind(request.snapshot.id.as_str()) + .execute(&mut *tx) + .await?; + + tx.commit().await?; + Ok(()) + } + + pub async fn update_operation_draft( + &self, + workspace_id: &WorkspaceId, + snapshot: &RegistryOperation, + ) -> Result<(), RegistryError> { + let Some(summary) = self + .get_operation_summary(workspace_id, &snapshot.id) + .await? + else { + return Err(RegistryError::OperationNotFound { + operation_id: snapshot.id.as_str().to_owned(), + }); + }; + + if snapshot.version != summary.current_draft_version { + return Err(RegistryError::InvalidVersionSequence { + operation_id: snapshot.id.as_str().to_owned(), + expected: summary.current_draft_version, + actual: snapshot.version, + }); + } + + assert_immutable_fields(&summary, snapshot)?; + + let mut tx = self.pool.begin().await?; + + sqlx::query( + "update operations + set display_name = $1, + category = $2, + status = $3, + updated_at = $4::timestamptz + where workspace_id = $5 and id = $6", + ) + .bind(&snapshot.display_name) + .bind(&snapshot.category) + .bind(serialize_enum_text(&snapshot.status, "status")?) + .bind(&snapshot.updated_at) + .bind(workspace_id.as_str()) + .bind(snapshot.id.as_str()) + .execute(&mut *tx) + .await?; + + sqlx::query( + "update operation_versions + set status = $1, + target_json = $2, + input_schema_json = $3, + output_schema_json = $4, + input_mapping_json = $5, + output_mapping_json = $6, + execution_config_json = $7, + tool_description_json = $8, + samples_json = $9, + generated_draft_json = $10, + config_export_json = $11 + where operation_id = $12 and version = $13", + ) + .bind(serialize_enum_text(&snapshot.status, "status")?) + .bind(Json(serialize_json_value(&snapshot.target)?)) + .bind(Json(serialize_json_value(&snapshot.input_schema)?)) + .bind(Json(serialize_json_value(&snapshot.output_schema)?)) + .bind(Json(serialize_json_value(&snapshot.input_mapping)?)) + .bind(Json(serialize_json_value(&snapshot.output_mapping)?)) + .bind(Json(serialize_json_value(&snapshot.execution_config)?)) + .bind(Json(serialize_json_value(&snapshot.tool_description)?)) + .bind(serialize_option_json_value(&snapshot.samples)?.map(Json)) + .bind(serialize_option_json_value(&snapshot.generated_draft)?.map(Json)) + .bind(serialize_option_json_value(&snapshot.config_export)?.map(Json)) + .bind(snapshot.id.as_str()) + .bind(to_db_version(snapshot.version)) + .execute(&mut *tx) + .await?; + + tx.commit().await?; + Ok(()) + } + + pub async fn archive_operation( + &self, + workspace_id: &WorkspaceId, + operation_id: &OperationId, + archived_at: &str, + ) -> Result<(), RegistryError> { + let Some(summary) = self + .get_operation_summary(workspace_id, operation_id) + .await? + else { + return Err(RegistryError::OperationNotFound { + operation_id: operation_id.as_str().to_owned(), + }); + }; + + let mut tx = self.pool.begin().await?; + + sqlx::query( + "update operations + set status = $1, + updated_at = $2::timestamptz + where workspace_id = $3 and id = $4", + ) + .bind(serialize_enum_text(&OperationStatus::Archived, "status")?) + .bind(archived_at) + .bind(workspace_id.as_str()) + .bind(operation_id.as_str()) + .execute(&mut *tx) + .await?; + + sqlx::query( + "update operation_versions + set status = $1 + where operation_id = $2 and version = $3", + ) + .bind(serialize_enum_text(&OperationStatus::Archived, "status")?) + .bind(operation_id.as_str()) + .bind(to_db_version(summary.current_draft_version)) + .execute(&mut *tx) + .await?; + + tx.commit().await?; + Ok(()) + } + + pub async fn delete_operation( + &self, + workspace_id: &WorkspaceId, + operation_id: &OperationId, + ) -> Result<(), RegistryError> { + if self + .has_published_agent_bindings_for_operation(workspace_id, operation_id) + .await? + { + return Err(RegistryError::OperationHasPublishedAgentBindings { + operation_id: operation_id.as_str().to_owned(), + }); + } + + let deleted = sqlx::query( + "delete from operations + where workspace_id = $1 and id = $2", + ) + .bind(workspace_id.as_str()) + .bind(operation_id.as_str()) + .execute(&self.pool) + .await? + .rows_affected(); + + if deleted == 0 { + return Err(RegistryError::OperationNotFound { + operation_id: operation_id.as_str().to_owned(), + }); + } + + Ok(()) + } + + pub async fn has_published_agent_bindings_for_operation( + &self, + workspace_id: &WorkspaceId, + operation_id: &OperationId, + ) -> Result { + let row = sqlx::query( + "select 1 + from agents a + join published_agents pa on pa.agent_id = a.id + join agent_operation_bindings b + on b.agent_id = a.id and b.agent_version = pa.version + where a.workspace_id = $1 + and b.operation_id = $2 + limit 1", + ) + .bind(workspace_id.as_str()) + .bind(operation_id.as_str()) + .fetch_optional(&self.pool) + .await?; + + Ok(row.is_some()) + } + + pub async fn list_operations( + &self, + workspace_id: &WorkspaceId, + ) -> Result, RegistryError> { + let rows = sqlx::query( + "select + operations.id, + operations.workspace_id, + operations.name, + operations.display_name, + operations.category, + operations.protocol, + ov.target_json, + operations.status, + operations.current_draft_version, + operations.latest_published_version, + to_char(operations.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, + to_char(operations.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at, + to_char(operations.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as published_at + from operations + join operation_versions ov + on ov.operation_id = operations.id + and ov.version = operations.current_draft_version + where operations.workspace_id = $1 + order by operations.name asc", + ) + .bind(workspace_id.as_str()) + .fetch_all(&self.pool) + .await?; + + rows.iter().map(map_operation_summary).collect() + } + + pub async fn list_operation_usage_summaries( + &self, + workspace_id: &WorkspaceId, + created_after: &str, + ) -> Result, RegistryError> { + let rows = sqlx::query( + "select + operation_id, + count(*)::bigint as calls_today, + coalesce( + round((sum(case when status = 'error' then 1 else 0 end)::numeric / nullif(count(*), 0)) * 100, 2), + 0 + )::float8 as error_rate_pct, + coalesce(round(avg(duration_ms))::bigint, 0) as avg_latency_ms + from invocation_logs + where workspace_id = $1 + and created_at >= $2::timestamptz + group by operation_id", + ) + .bind(workspace_id.as_str()) + .bind(created_after) + .fetch_all(&self.pool) + .await?; + + rows.iter().map(map_operation_usage_summary).collect() + } + + pub async fn list_operation_agent_refs( + &self, + workspace_id: &WorkspaceId, + ) -> Result, RegistryError> { + let rows = sqlx::query( + "select + b.operation_id, + a.id as agent_id, + a.slug as agent_slug, + a.display_name + from agents a + join published_agents pa on pa.agent_id = a.id + join agent_operation_bindings b + on b.agent_id = a.id and b.agent_version = pa.version + where a.workspace_id = $1 + order by a.display_name asc, b.tool_name asc", + ) + .bind(workspace_id.as_str()) + .fetch_all(&self.pool) + .await?; + + rows.iter().map(map_operation_agent_ref).collect() + } + + pub async fn get_operation_summary( + &self, + workspace_id: &WorkspaceId, + operation_id: &OperationId, + ) -> Result, RegistryError> { + let row = sqlx::query( + "select + operations.id, + operations.workspace_id, + operations.name, + operations.display_name, + operations.category, + operations.protocol, + ov.target_json, + operations.status, + operations.current_draft_version, + operations.latest_published_version, + to_char(operations.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, + to_char(operations.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at, + to_char(operations.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as published_at + from operations + join operation_versions ov + on ov.operation_id = operations.id + and ov.version = operations.current_draft_version + where operations.workspace_id = $1 and operations.id = $2", + ) + .bind(workspace_id.as_str()) + .bind(operation_id.as_str()) + .fetch_optional(&self.pool) + .await?; + + row.as_ref().map(map_operation_summary).transpose() + } + + pub async fn get_operation_version( + &self, + workspace_id: &WorkspaceId, + operation_id: &OperationId, + version: u32, + ) -> Result, RegistryError> { + let row = sqlx::query( + "select + o.id, + o.workspace_id, + o.name, + o.display_name, + o.category, + o.protocol, + to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at, + to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at, + to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at, + ov.version, + ov.status, + ov.target_json, + ov.input_schema_json, + ov.output_schema_json, + ov.input_mapping_json, + ov.output_mapping_json, + ov.execution_config_json, + ov.tool_description_json, + ov.samples_json, + ov.generated_draft_json, + ov.config_export_json, + ov.change_note, + to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, + ov.created_by + from operation_versions ov + join operations o on o.id = ov.operation_id + where o.workspace_id = $1 and ov.operation_id = $2 and ov.version = $3", + ) + .bind(workspace_id.as_str()) + .bind(operation_id.as_str()) + .bind(to_db_version(version)) + .fetch_optional(&self.pool) + .await?; + + row.as_ref().map(map_operation_version_record).transpose() + } + + pub async fn list_operation_versions( + &self, + workspace_id: &WorkspaceId, + operation_id: &OperationId, + ) -> Result, RegistryError> { + let rows = sqlx::query( + "select + o.id, + o.workspace_id, + o.name, + o.display_name, + o.category, + o.protocol, + to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at, + to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at, + to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at, + ov.version, + ov.status, + ov.target_json, + ov.input_schema_json, + ov.output_schema_json, + ov.input_mapping_json, + ov.output_mapping_json, + ov.execution_config_json, + ov.tool_description_json, + ov.samples_json, + ov.generated_draft_json, + ov.config_export_json, + ov.change_note, + to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, + ov.created_by + from operation_versions ov + join operations o on o.id = ov.operation_id + where o.workspace_id = $1 and ov.operation_id = $2 + order by ov.version asc", + ) + .bind(workspace_id.as_str()) + .bind(operation_id.as_str()) + .fetch_all(&self.pool) + .await?; + + rows.iter().map(map_operation_version_record).collect() + } + + pub async fn publish_operation( + &self, + request: PublishRequest<'_>, + ) -> Result<(), RegistryError> { + if self + .get_operation_version(request.workspace_id, request.operation_id, request.version) + .await? + .is_none() + { + return Err(RegistryError::OperationVersionNotFound { + operation_id: request.operation_id.as_str().to_owned(), + version: request.version, + }); + } + + let mut tx = self.pool.begin().await?; + + sqlx::query( + "insert into published_operations ( + operation_id, + version, + published_at, + published_by + ) values ($1, $2, $3::timestamptz, $4) + on conflict(operation_id) do update set + version = excluded.version, + published_at = excluded.published_at, + published_by = excluded.published_by", + ) + .bind(request.operation_id.as_str()) + .bind(to_db_version(request.version)) + .bind(request.published_at) + .bind(request.published_by) + .execute(&mut *tx) + .await?; + + sqlx::query( + "update operation_versions + set status = $1 + where operation_id = $2 and version = $3", + ) + .bind(serialize_enum_text(&OperationStatus::Published, "status")?) + .bind(request.operation_id.as_str()) + .bind(to_db_version(request.version)) + .execute(&mut *tx) + .await?; + + sqlx::query( + "update operations + set status = $1, + latest_published_version = $2, + published_at = $3::timestamptz, + updated_at = $4::timestamptz + where id = $5", + ) + .bind(serialize_enum_text(&OperationStatus::Published, "status")?) + .bind(to_db_version(request.version)) + .bind(request.published_at) + .bind(request.published_at) + .bind(request.operation_id.as_str()) + .execute(&mut *tx) + .await?; + + tx.commit().await?; + Ok(()) + } + + pub async fn get_published_operation( + &self, + operation_id: &OperationId, + ) -> Result, RegistryError> { + let row = sqlx::query( + "select + o.id, + o.workspace_id, + o.name, + o.display_name, + o.category, + o.protocol, + to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at, + to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at, + to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at, + ov.version, + ov.status, + ov.target_json, + ov.input_schema_json, + ov.output_schema_json, + ov.input_mapping_json, + ov.output_mapping_json, + ov.execution_config_json, + ov.tool_description_json, + ov.samples_json, + ov.generated_draft_json, + ov.config_export_json, + ov.change_note, + to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, + ov.created_by + from published_operations po + join operation_versions ov + on ov.operation_id = po.operation_id and ov.version = po.version + join operations o on o.id = po.operation_id + where po.operation_id = $1", + ) + .bind(operation_id.as_str()) + .fetch_optional(&self.pool) + .await?; + + row.as_ref() + .map(|value| map_operation_version_record(value).map(|record| record.snapshot)) + .transpose() + } + + pub async fn list_published_operations(&self) -> Result, RegistryError> { + let rows = sqlx::query( + "select + o.id, + o.workspace_id, + o.name, + o.display_name, + o.category, + o.protocol, + to_char(o.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_created_at, + to_char(o.updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_updated_at, + to_char(o.published_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as operation_published_at, + ov.version, + ov.status, + ov.target_json, + ov.input_schema_json, + ov.output_schema_json, + ov.input_mapping_json, + ov.output_mapping_json, + ov.execution_config_json, + ov.tool_description_json, + ov.samples_json, + ov.generated_draft_json, + ov.config_export_json, + ov.change_note, + to_char(ov.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, + ov.created_by + from published_operations po + join operation_versions ov + on ov.operation_id = po.operation_id and ov.version = po.version + join operations o on o.id = po.operation_id + order by o.name asc", + ) + .fetch_all(&self.pool) + .await?; + + rows.iter() + .map(|row| map_operation_version_record(row).map(|record| record.snapshot)) + .collect() + } + + pub async fn save_sample_metadata( + &self, + request: SaveSampleMetadataRequest<'_>, + ) -> Result<(), RegistryError> { + sqlx::query( + "insert into operation_samples ( + id, + operation_id, + version, + sample_kind, + storage_ref, + content_type, + file_name, + created_at + ) values ($1, $2, $3, $4, $5, $6, $7, $8::timestamptz) + on conflict(id) do update set + operation_id = excluded.operation_id, + version = excluded.version, + sample_kind = excluded.sample_kind, + storage_ref = excluded.storage_ref, + content_type = excluded.content_type, + file_name = excluded.file_name, + created_at = excluded.created_at", + ) + .bind(request.sample.id.as_str()) + .bind(request.sample.operation_id.as_str()) + .bind(to_db_version(request.sample.version)) + .bind(serialize_enum_text( + &request.sample.sample_kind, + "sample_kind", + )?) + .bind(&request.sample.storage_ref) + .bind(&request.sample.content_type) + .bind(request.sample.file_name.as_deref()) + .bind(&request.sample.created_at) + .execute(&self.pool) + .await?; + + Ok(()) + } + + pub async fn list_sample_metadata( + &self, + operation_id: &OperationId, + version: u32, + ) -> Result, RegistryError> { + let rows = sqlx::query( + "select + id, + operation_id, + version, + sample_kind, + storage_ref, + content_type, + file_name, + to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at + from operation_samples + where operation_id = $1 and version = $2 + order by created_at asc", + ) + .bind(operation_id.as_str()) + .bind(to_db_version(version)) + .fetch_all(&self.pool) + .await?; + + rows.iter().map(map_sample_metadata).collect() + } + + pub async fn save_descriptor_metadata( + &self, + request: SaveDescriptorMetadataRequest<'_>, + ) -> Result<(), RegistryError> { + sqlx::query( + "insert into descriptors ( + id, + operation_id, + version, + descriptor_kind, + storage_ref, + source_name, + package_index_json, + created_at + ) values ($1, $2, $3, $4, $5, $6, $7, $8::timestamptz) + on conflict(id) do update set + operation_id = excluded.operation_id, + version = excluded.version, + descriptor_kind = excluded.descriptor_kind, + storage_ref = excluded.storage_ref, + source_name = excluded.source_name, + package_index_json = excluded.package_index_json, + created_at = excluded.created_at", + ) + .bind(request.descriptor.id.as_str()) + .bind( + request + .descriptor + .operation_id + .as_ref() + .map(|value| value.as_str()), + ) + .bind(request.descriptor.version.map(to_db_version)) + .bind(serialize_enum_text( + &request.descriptor.descriptor_kind, + "descriptor_kind", + )?) + .bind(&request.descriptor.storage_ref) + .bind(request.descriptor.source_name.as_deref()) + .bind(serialize_option_json_value(&request.descriptor.package_index)?.map(Json)) + .bind(&request.descriptor.created_at) + .execute(&self.pool) + .await?; + + Ok(()) + } + + pub async fn list_descriptor_metadata( + &self, + operation_id: &OperationId, + version: u32, + ) -> Result, RegistryError> { + let rows = sqlx::query( + "select + id, + operation_id, + version, + descriptor_kind, + storage_ref, + source_name, + package_index_json, + to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at + from descriptors + where operation_id = $1 and version = $2 + order by created_at asc", + ) + .bind(operation_id.as_str()) + .bind(to_db_version(version)) + .fetch_all(&self.pool) + .await?; + + rows.iter().map(map_descriptor_metadata).collect() + } +}