use crank_core::{ AgentId, AgentOperationBinding, AgentStatus, AgentVersion, AuthProfile, GraphqlOperationType, HttpMethod, InvitationId, InvitationToken, InvocationLog, InvocationLogId, MembershipRole, OperationId, OperationStatus, PlatformApiKey, PlatformApiKeyId, PlatformApiKeyStatus, Target, UsageRollup, User, UserId, UserSessionId, Workspace, WorkspaceId, }; use serde::{Serialize, de::DeserializeOwned}; use serde_json::Value; use sqlx::{ PgPool, Postgres, Row, Transaction, postgres::{PgPoolOptions, PgRow}, types::Json, }; use crate::{ error::RegistryError, migrations, model::{ AgentSummary, AgentVersionRecord, AuthUserRecord, CreateAgentRequest, CreateInvitationRequest, CreateInvocationLogRequest, CreatePlatformApiKeyRequest, CreateVersionRequest, CreateWorkspaceRequest, CreateYamlImportJobRequest, DescriptorMetadata, InvitationRecord, InvocationLogRecord, ListInvocationLogsQuery, MembershipRecord, OperationAgentRef, OperationSampleMetadata, OperationSummary, OperationUsageSummary, OperationVersionRecord, PlatformApiKeyRecord, PublishAgentRequest, PublishRequest, PublishedAgentTool, RegistryOperation, SaveAgentBindingsRequest, SaveAuthProfileRequest, SaveDescriptorMetadataRequest, SaveSampleMetadataRequest, SessionRecord, UpdateWorkspaceRequest, UsageAgentBreakdown, UsageOperationBreakdown, UsageQuery, UsageRollupRecord, UsageSummary, UsageTimelinePoint, WorkspaceMembershipRecord, WorkspaceRecord, YamlImportJob, YamlImportJobCompletion, YamlImportJobId, YamlImportJobStatus, }, }; #[derive(Clone, Debug)] pub struct PostgresRegistry { pool: PgPool, } impl PostgresRegistry { pub async fn connect(database_url: &str) -> Result { Self::connect_in_schema(database_url, None).await } pub fn pool(&self) -> &PgPool { &self.pool } pub async fn migrate(&self) -> Result<(), RegistryError> { migrations::apply_postgres(&self.pool).await?; 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 create_user_session( &self, session_id: &UserSessionId, user_id: &UserId, secret_hash: &str, expires_at: &str, ) -> Result<(), RegistryError> { sqlx::query( "insert into user_sessions ( id, user_id, secret_hash, status, expires_at, last_seen_at, created_at ) values ( $1, $2, $3, 'active', $4::timestamptz, now(), now() )", ) .bind(session_id.as_str()) .bind(user_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, 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?; Ok(Some(SessionRecord { session_id: UserSessionId::new(row.try_get::("id")?), user, memberships, })) } 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 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 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 list_platform_api_keys( &self, workspace_id: &WorkspaceId, ) -> Result, RegistryError> { let rows = sqlx::query( "select id, workspace_id, name, prefix, scopes_json, status, to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, to_char(last_used_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as last_used_at from platform_api_keys where workspace_id = $1 order by created_at desc", ) .bind(workspace_id.as_str()) .fetch_all(&self.pool) .await?; rows.iter().map(map_platform_api_key_record).collect() } pub async fn create_platform_api_key( &self, request: CreatePlatformApiKeyRequest<'_>, ) -> Result<(), RegistryError> { let result = sqlx::query( "insert into platform_api_keys ( id, workspace_id, name, prefix, secret_hash, scopes_json, status, created_at, last_used_at, revoked_at ) values ( $1, $2, $3, $4, $5, $6, $7, $8::timestamptz, $9::timestamptz, $10::timestamptz )", ) .bind(request.api_key.id.as_str()) .bind(request.api_key.workspace_id.as_str()) .bind(&request.api_key.name) .bind(&request.api_key.prefix) .bind(request.secret_hash) .bind(Json(serialize_json_value(&request.api_key.scopes)?)) .bind(serialize_enum_text(&request.api_key.status, "status")?) .bind(&request.api_key.created_at) .bind(request.api_key.last_used_at.as_deref()) .bind(Option::<&str>::None) .execute(&self.pool) .await; match result { Ok(_) => Ok(()), Err(sqlx::Error::Database(error)) if error.constraint() == Some("platform_api_keys_workspace_name_idx") => { Err(RegistryError::Storage(sqlx::Error::Database(error))) } Err(error) => Err(RegistryError::Storage(error)), } } pub async fn revoke_platform_api_key( &self, workspace_id: &WorkspaceId, key_id: &PlatformApiKeyId, revoked_at: &str, ) -> Result<(), RegistryError> { let result = sqlx::query( "update platform_api_keys set status = $1, revoked_at = $2::timestamptz where workspace_id = $3 and id = $4", ) .bind(serialize_enum_text( &PlatformApiKeyStatus::Revoked, "status", )?) .bind(revoked_at) .bind(workspace_id.as_str()) .bind(key_id.as_str()) .execute(&self.pool) .await?; if result.rows_affected() == 0 { return Err(RegistryError::PlatformApiKeyNotFound { key_id: key_id.as_str().to_owned(), }); } Ok(()) } pub async fn delete_platform_api_key( &self, workspace_id: &WorkspaceId, key_id: &PlatformApiKeyId, ) -> Result<(), RegistryError> { let result = sqlx::query("delete from platform_api_keys where workspace_id = $1 and id = $2") .bind(workspace_id.as_str()) .bind(key_id.as_str()) .execute(&self.pool) .await?; if result.rows_affected() == 0 { return Err(RegistryError::PlatformApiKeyNotFound { key_id: key_id.as_str().to_owned(), }); } Ok(()) } pub async fn create_invocation_log( &self, request: CreateInvocationLogRequest<'_>, ) -> Result<(), RegistryError> { sqlx::query( "insert into invocation_logs ( id, workspace_id, agent_id, operation_id, source, level, status, tool_name, message, request_id, status_code, duration_ms, error_kind, request_preview_json, response_preview_json, created_at ) values ( $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16::timestamptz )", ) .bind(request.log.id.as_str()) .bind(request.log.workspace_id.as_str()) .bind(request.log.agent_id.as_ref().map(|value| value.as_str())) .bind(request.log.operation_id.as_str()) .bind(serialize_enum_text(&request.log.source, "source")?) .bind(serialize_enum_text(&request.log.level, "level")?) .bind(serialize_enum_text(&request.log.status, "status")?) .bind(&request.log.tool_name) .bind(&request.log.message) .bind(&request.log.request_id) .bind(request.log.status_code.map(i32::from)) .bind(i64::try_from(request.log.duration_ms).map_err(|_| { RegistryError::InvalidNumericValue { field: "duration_ms", value: i64::MAX, } })?) .bind(&request.log.error_kind) .bind(Json(request.log.request_preview.clone())) .bind(Json(request.log.response_preview.clone())) .bind(&request.log.created_at) .execute(&self.pool) .await?; Ok(()) } pub async fn list_invocation_logs( &self, query: ListInvocationLogsQuery<'_>, ) -> Result, RegistryError> { let rows = sqlx::query( "select l.id, l.workspace_id, l.agent_id, l.operation_id, l.source, l.level, l.status, l.tool_name, l.message, l.request_id, l.status_code, l.duration_ms, l.error_kind, l.request_preview_json, l.response_preview_json, to_char(l.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS.MS\"Z\"') as created_at, o.name as operation_name, o.display_name as operation_display_name, a.slug as agent_slug, a.display_name as agent_display_name from invocation_logs l join operations o on o.id = l.operation_id left join agents a on a.id = l.agent_id where l.workspace_id = $1 and ($2::text is null or l.level = $2) and ($3::text is null or l.source = $3) and ($4::text is null or l.operation_id = $4) and ($5::text is null or l.agent_id = $5) and ($6::timestamptz is null or l.created_at >= $6::timestamptz) and ( $7::text is null or l.tool_name ilike '%' || $7 || '%' or l.message ilike '%' || $7 || '%' or o.name ilike '%' || $7 || '%' or o.display_name ilike '%' || $7 || '%' ) order by l.created_at desc limit $8", ) .bind(query.workspace_id.as_str()) .bind( query .level .as_ref() .map(|value| serialize_enum_text(value, "level")) .transpose()?, ) .bind( query .source .as_ref() .map(|value| serialize_enum_text(value, "source")) .transpose()?, ) .bind(query.operation_id.map(|value| value.as_str())) .bind(query.agent_id.map(|value| value.as_str())) .bind(query.created_after) .bind(query.search_text) .bind(i64::from(query.limit)) .fetch_all(&self.pool) .await?; rows.iter().map(map_invocation_log_record).collect() } pub async fn get_invocation_log( &self, workspace_id: &WorkspaceId, log_id: &InvocationLogId, ) -> Result, RegistryError> { let row = sqlx::query( "select l.id, l.workspace_id, l.agent_id, l.operation_id, l.source, l.level, l.status, l.tool_name, l.message, l.request_id, l.status_code, l.duration_ms, l.error_kind, l.request_preview_json, l.response_preview_json, to_char(l.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS.MS\"Z\"') as created_at, o.name as operation_name, o.display_name as operation_display_name, a.slug as agent_slug, a.display_name as agent_display_name from invocation_logs l join operations o on o.id = l.operation_id left join agents a on a.id = l.agent_id where l.workspace_id = $1 and l.id = $2", ) .bind(workspace_id.as_str()) .bind(log_id.as_str()) .fetch_optional(&self.pool) .await?; row.as_ref().map(map_invocation_log_record).transpose() } pub async fn summarize_usage( &self, query: UsageQuery<'_>, ) -> Result { let row = sqlx::query( "select count(*)::bigint as calls_total, count(*) filter (where status = 'ok')::bigint as calls_ok, count(*) filter (where status = 'error')::bigint as calls_error, coalesce(percentile_cont(0.5) within group (order by duration_ms), 0)::bigint as p50_ms, coalesce(percentile_cont(0.95) within group (order by duration_ms), 0)::bigint as p95_ms, coalesce(percentile_cont(0.99) within group (order by duration_ms), 0)::bigint as p99_ms from invocation_logs where workspace_id = $1 and created_at >= $2::timestamptz and ($3::text is null or source = $3)", ) .bind(query.workspace_id.as_str()) .bind(query.created_after) .bind( query .source .as_ref() .map(|value| serialize_enum_text(value, "source")) .transpose()?, ) .fetch_one(&self.pool) .await?; let calls_total = row.try_get::("calls_total")?; let calls_ok = row.try_get::("calls_ok")?; let calls_error = row.try_get::("calls_error")?; let rollup = UsageRollup { workspace_id: query.workspace_id.clone(), agent_id: None, operation_id: None, period: query.period, calls_total: to_u64(calls_total, "calls_total")?, calls_ok: to_u64(calls_ok, "calls_ok")?, calls_error: to_u64(calls_error, "calls_error")?, p50_ms: to_u64(row.try_get::("p50_ms")?, "p50_ms")?, p95_ms: to_u64(row.try_get::("p95_ms")?, "p95_ms")?, p99_ms: to_u64(row.try_get::("p99_ms")?, "p99_ms")?, }; let success_rate = if rollup.calls_total == 0 { 0.0 } else { (rollup.calls_ok as f64 / rollup.calls_total as f64) * 100.0 }; Ok(UsageSummary { rollup, success_rate, }) } pub async fn list_usage_timeline( &self, query: UsageQuery<'_>, ) -> Result, RegistryError> { let bucket = usage_bucket_sql(query.bucket); let sql = format!( "select to_char(date_trunc('{bucket}', created_at at time zone 'UTC'), 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as bucket_start, count(*) filter (where status = 'ok')::bigint as calls_ok, count(*) filter (where status = 'error')::bigint as calls_error from invocation_logs where workspace_id = $1 and created_at >= $2::timestamptz and ($3::text is null or source = $3) group by 1 order by 1 asc" ); let rows = sqlx::query(&sql) .bind(query.workspace_id.as_str()) .bind(query.created_after) .bind( query .source .as_ref() .map(|value| serialize_enum_text(value, "source")) .transpose()?, ) .fetch_all(&self.pool) .await?; rows.iter() .map(|row| { Ok(UsageTimelinePoint { bucket_start: row.try_get("bucket_start")?, calls_ok: to_u64(row.try_get::("calls_ok")?, "calls_ok")?, calls_error: to_u64(row.try_get::("calls_error")?, "calls_error")?, }) }) .collect() } pub async fn list_usage_by_operation( &self, query: UsageQuery<'_>, ) -> Result, RegistryError> { let rows = sqlx::query( "select o.id as operation_id, o.name as operation_name, o.display_name as operation_display_name, o.protocol, count(*)::bigint as calls_total, count(*) filter (where l.status = 'error')::bigint as calls_error, coalesce(percentile_cont(0.5) within group (order by l.duration_ms), 0)::bigint as p50_ms, coalesce(percentile_cont(0.95) within group (order by l.duration_ms), 0)::bigint as p95_ms, coalesce(percentile_cont(0.99) within group (order by l.duration_ms), 0)::bigint as p99_ms from invocation_logs l join operations o on o.id = l.operation_id where l.workspace_id = $1 and l.created_at >= $2::timestamptz and ($3::text is null or l.source = $3) group by o.id, o.name, o.display_name, o.protocol order by calls_total desc, o.name asc", ) .bind(query.workspace_id.as_str()) .bind(query.created_after) .bind( query .source .as_ref() .map(|value| serialize_enum_text(value, "source")) .transpose()?, ) .fetch_all(&self.pool) .await?; rows.iter().map(map_usage_operation_breakdown).collect() } pub async fn get_usage_for_operation( &self, query: UsageQuery<'_>, operation_id: &OperationId, ) -> Result, RegistryError> { let row = sqlx::query( "select count(*)::bigint as calls_total, count(*) filter (where status = 'ok')::bigint as calls_ok, count(*) filter (where status = 'error')::bigint as calls_error, coalesce(percentile_cont(0.5) within group (order by duration_ms), 0)::bigint as p50_ms, coalesce(percentile_cont(0.95) within group (order by duration_ms), 0)::bigint as p95_ms, coalesce(percentile_cont(0.99) within group (order by duration_ms), 0)::bigint as p99_ms from invocation_logs where workspace_id = $1 and operation_id = $2 and created_at >= $3::timestamptz and ($4::text is null or source = $4)", ) .bind(query.workspace_id.as_str()) .bind(operation_id.as_str()) .bind(query.created_after) .bind( query .source .as_ref() .map(|value| serialize_enum_text(value, "source")) .transpose()?, ) .fetch_one(&self.pool) .await?; let calls_total = to_u64(row.try_get::("calls_total")?, "calls_total")?; if calls_total == 0 { return Ok(None); } Ok(Some(UsageRollupRecord { rollup: UsageRollup { workspace_id: query.workspace_id.clone(), agent_id: None, operation_id: Some(operation_id.clone()), period: query.period, calls_total, calls_ok: to_u64(row.try_get::("calls_ok")?, "calls_ok")?, calls_error: to_u64(row.try_get::("calls_error")?, "calls_error")?, p50_ms: to_u64(row.try_get::("p50_ms")?, "p50_ms")?, p95_ms: to_u64(row.try_get::("p95_ms")?, "p95_ms")?, p99_ms: to_u64(row.try_get::("p99_ms")?, "p99_ms")?, }, })) } pub async fn list_usage_by_agent( &self, query: UsageQuery<'_>, ) -> Result, RegistryError> { let rows = sqlx::query( "select a.id as agent_id, a.slug as agent_slug, a.display_name as agent_display_name, count(*)::bigint as calls_total, count(*) filter (where l.status = 'error')::bigint as calls_error, coalesce(percentile_cont(0.5) within group (order by l.duration_ms), 0)::bigint as p50_ms, coalesce(percentile_cont(0.95) within group (order by l.duration_ms), 0)::bigint as p95_ms, coalesce(percentile_cont(0.99) within group (order by l.duration_ms), 0)::bigint as p99_ms from invocation_logs l join agents a on a.id = l.agent_id where l.workspace_id = $1 and l.created_at >= $2::timestamptz and ($3::text is null or l.source = $3) group by a.id, a.slug, a.display_name order by calls_total desc, a.slug asc", ) .bind(query.workspace_id.as_str()) .bind(query.created_after) .bind( query .source .as_ref() .map(|value| serialize_enum_text(value, "source")) .transpose()?, ) .fetch_all(&self.pool) .await?; rows.iter().map(map_usage_agent_breakdown).collect() } pub async fn get_usage_for_agent( &self, query: UsageQuery<'_>, agent_id: &AgentId, ) -> Result, RegistryError> { let row = sqlx::query( "select count(*)::bigint as calls_total, count(*) filter (where status = 'ok')::bigint as calls_ok, count(*) filter (where status = 'error')::bigint as calls_error, coalesce(percentile_cont(0.5) within group (order by duration_ms), 0)::bigint as p50_ms, coalesce(percentile_cont(0.95) within group (order by duration_ms), 0)::bigint as p95_ms, coalesce(percentile_cont(0.99) within group (order by duration_ms), 0)::bigint as p99_ms from invocation_logs where workspace_id = $1 and agent_id = $2 and created_at >= $3::timestamptz and ($4::text is null or source = $4)", ) .bind(query.workspace_id.as_str()) .bind(agent_id.as_str()) .bind(query.created_after) .bind( query .source .as_ref() .map(|value| serialize_enum_text(value, "source")) .transpose()?, ) .fetch_one(&self.pool) .await?; let calls_total = to_u64(row.try_get::("calls_total")?, "calls_total")?; if calls_total == 0 { return Ok(None); } Ok(Some(UsageRollupRecord { rollup: UsageRollup { workspace_id: query.workspace_id.clone(), agent_id: Some(agent_id.clone()), operation_id: None, period: query.period, calls_total, calls_ok: to_u64(row.try_get::("calls_ok")?, "calls_ok")?, calls_error: to_u64(row.try_get::("calls_error")?, "calls_error")?, p50_ms: to_u64(row.try_get::("p50_ms")?, "p50_ms")?, p95_ms: to_u64(row.try_get::("p95_ms")?, "p95_ms")?, p99_ms: to_u64(row.try_get::("p99_ms")?, "p99_ms")?, }, })) } 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, ) -> 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 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>, ) -> Result { let pool = PgPoolOptions::new() .min_connections(0) .max_connections(1) .idle_timeout(std::time::Duration::from_secs(1)) .connect(database_url) .await?; if let Some(schema) = schema { sqlx::query(&format!("set search_path to {schema}")) .execute(&pool) .await?; } let registry = Self { pool }; registry.migrate().await?; 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(Json( snapshot .samples .clone() .map(|value| serialize_json_value(&value)) .transpose()?, )) .bind(Json( snapshot .generated_draft .clone() .map(|value| serialize_json_value(&value)) .transpose()?, )) .bind(Json( snapshot .config_export .clone() .map(|value| serialize_json_value(&value)) .transpose()?, )) .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_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 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() } pub async fn create_yaml_import_job( &self, request: CreateYamlImportJobRequest<'_>, ) -> Result<(), RegistryError> { sqlx::query( "insert into yaml_import_jobs ( id, source_sample_id, status, format_version, mode, result_operation_id, result_version, error_text, created_at, finished_at ) values ($1, $2, $3, $4, $5, null, null, null, $6::timestamptz, null)", ) .bind(request.id.as_str()) .bind(request.source_sample_id.map(|value| value.as_str())) .bind(serialize_enum_text( &YamlImportJobStatus::Pending, "status", )?) .bind(request.format_version) .bind(serialize_enum_text(&request.mode, "mode")?) .bind(request.created_at) .execute(&self.pool) .await?; Ok(()) } pub async fn finish_yaml_import_job( &self, job_id: &YamlImportJobId, completion: &YamlImportJobCompletion, ) -> Result<(), RegistryError> { let result = sqlx::query( "update yaml_import_jobs set status = $1, result_operation_id = $2, result_version = $3, error_text = $4, finished_at = $5::timestamptz where id = $6", ) .bind(serialize_enum_text(&completion.status, "status")?) .bind( completion .result_operation_id .as_ref() .map(|value| value.as_str()), ) .bind(completion.result_version.map(to_db_version)) .bind(completion.error_text.as_deref()) .bind(&completion.finished_at) .bind(job_id.as_str()) .execute(&self.pool) .await?; if result.rows_affected() == 0 { return Err(RegistryError::YamlImportJobNotFound { job_id: job_id.as_str().to_owned(), }); } Ok(()) } pub async fn get_yaml_import_job( &self, job_id: &YamlImportJobId, ) -> Result, RegistryError> { let row = sqlx::query( "select id, source_sample_id, status, format_version, mode, result_operation_id, result_version, error_text, to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at, to_char(finished_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as finished_at from yaml_import_jobs where id = $1", ) .bind(job_id.as_str()) .fetch_optional(&self.pool) .await?; row.as_ref().map(map_yaml_import_job).transpose() } async fn list_agent_bindings( &self, agent_id: &AgentId, version: u32, ) -> Result, RegistryError> { let rows = sqlx::query( "select operation_id, operation_version, tool_name, tool_title, tool_description_override, enabled from agent_operation_bindings where agent_id = $1 and agent_version = $2 order by tool_name asc", ) .bind(agent_id.as_str()) .bind(to_db_version(version)) .fetch_all(&self.pool) .await?; rows.iter().map(map_agent_binding).collect() } } async fn insert_version_row( tx: &mut Transaction<'_, Postgres>, snapshot: &RegistryOperation, change_note: Option<&str>, created_by: Option<&str>, ) -> Result<(), RegistryError> { sqlx::query( "insert into operation_versions ( operation_id, version, status, target_json, input_schema_json, output_schema_json, input_mapping_json, output_mapping_json, execution_config_json, tool_description_json, samples_json, generated_draft_json, config_export_json, change_note, created_at, created_by ) values ( $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15::timestamptz, $16 )", ) .bind(snapshot.id.as_str()) .bind(to_db_version(snapshot.version)) .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(change_note) .bind(&snapshot.updated_at) .bind(created_by) .execute(&mut **tx) .await?; Ok(()) } async fn insert_agent_version_row( tx: &mut Transaction<'_, Postgres>, version: &AgentVersion, ) -> Result<(), RegistryError> { sqlx::query( "insert into agent_versions ( agent_id, version, status, instructions_json, tool_selection_policy_json, created_at ) values ( $1, $2, $3, $4, $5, $6::timestamptz )", ) .bind(version.agent_id.as_str()) .bind(to_db_version(version.version)) .bind(serialize_enum_text(&version.status, "status")?) .bind(Json(version.instructions.clone())) .bind(Json(version.tool_selection_policy.clone())) .bind(&version.created_at) .execute(&mut **tx) .await?; Ok(()) } async fn replace_agent_bindings_rows( tx: &mut Transaction<'_, Postgres>, agent_id: &AgentId, version: u32, bindings: &[AgentOperationBinding], ) -> Result<(), RegistryError> { sqlx::query( "delete from agent_operation_bindings where agent_id = $1 and agent_version = $2", ) .bind(agent_id.as_str()) .bind(to_db_version(version)) .execute(&mut **tx) .await?; for binding in bindings { sqlx::query( "insert into agent_operation_bindings ( agent_id, agent_version, operation_id, operation_version, tool_name, tool_title, tool_description_override, enabled ) values ($1, $2, $3, $4, $5, $6, $7, $8)", ) .bind(agent_id.as_str()) .bind(to_db_version(version)) .bind(binding.operation_id.as_str()) .bind(to_db_version(binding.operation_version)) .bind(&binding.tool_name) .bind(&binding.tool_title) .bind(binding.tool_description_override.as_deref()) .bind(binding.enabled) .execute(&mut **tx) .await?; } Ok(()) } fn assert_immutable_fields( summary: &OperationSummary, snapshot: &RegistryOperation, ) -> Result<(), RegistryError> { if summary.name != snapshot.name { return Err(RegistryError::ImmutableOperationFieldChanged { operation_id: snapshot.id.as_str().to_owned(), field: "name", }); } if summary.protocol != snapshot.protocol { return Err(RegistryError::ImmutableOperationFieldChanged { operation_id: snapshot.id.as_str().to_owned(), field: "protocol", }); } Ok(()) } fn map_workspace_record(row: &PgRow) -> Result { Ok(WorkspaceRecord { workspace: Workspace { id: WorkspaceId::new(row.try_get::("id")?), slug: row.try_get("slug")?, display_name: row.try_get("display_name")?, status: deserialize_enum_text(&row.try_get::("status")?, "status")?, settings: row.try_get::, _>("settings_json")?.0, created_at: row.try_get("created_at")?, updated_at: row.try_get("updated_at")?, }, }) } fn map_workspace_membership_record( row: &PgRow, ) -> Result { Ok(WorkspaceMembershipRecord { workspace: Workspace { id: WorkspaceId::new(row.try_get::("id")?), slug: row.try_get("slug")?, display_name: row.try_get("display_name")?, status: deserialize_enum_text(&row.try_get::("status")?, "status")?, settings: row.try_get::, _>("settings_json")?.0, created_at: row.try_get("created_at")?, updated_at: row.try_get("updated_at")?, }, role: deserialize_enum_text(&row.try_get::("role")?, "role")?, }) } fn map_membership_record(row: &PgRow) -> Result { Ok(MembershipRecord { workspace_id: WorkspaceId::new(row.try_get::("workspace_id")?), user: User { id: UserId::new(row.try_get::("user_id")?), 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("user_created_at")?, }, role: deserialize_enum_text(&row.try_get::("role")?, "role")?, created_at: row.try_get("created_at")?, }) } fn map_auth_user_record(row: &PgRow) -> Result { Ok(AuthUserRecord { user: User { id: UserId::new(row.try_get::("id")?), 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")?, }, password_hash: row.try_get("password_hash")?, }) } fn map_invitation_record(row: &PgRow) -> Result { Ok(InvitationRecord { invitation: InvitationToken { id: InvitationId::new(row.try_get::("id")?), workspace_id: WorkspaceId::new(row.try_get::("workspace_id")?), email: row.try_get("email")?, role: deserialize_enum_text(&row.try_get::("role")?, "role")?, status: deserialize_enum_text(&row.try_get::("status")?, "status")?, token_hash: row.try_get("token_hash")?, expires_at: row.try_get("expires_at")?, created_at: row.try_get("created_at")?, }, }) } fn map_platform_api_key_record(row: &PgRow) -> Result { Ok(PlatformApiKeyRecord { api_key: PlatformApiKey { id: PlatformApiKeyId::new(row.try_get::("id")?), workspace_id: WorkspaceId::new(row.try_get::("workspace_id")?), name: row.try_get("name")?, prefix: row.try_get("prefix")?, scopes: deserialize_json_value(row.try_get::, _>("scopes_json")?.0)?, status: deserialize_enum_text(&row.try_get::("status")?, "status")?, created_at: row.try_get("created_at")?, last_used_at: row.try_get("last_used_at")?, }, }) } fn map_invocation_log_record(row: &PgRow) -> Result { Ok(InvocationLogRecord { log: InvocationLog { id: InvocationLogId::new(row.try_get::("id")?), workspace_id: WorkspaceId::new(row.try_get::("workspace_id")?), agent_id: row .try_get::, _>("agent_id")? .map(AgentId::new), operation_id: OperationId::new(row.try_get::("operation_id")?), source: deserialize_enum_text(&row.try_get::("source")?, "source")?, level: deserialize_enum_text(&row.try_get::("level")?, "level")?, status: deserialize_enum_text(&row.try_get::("status")?, "status")?, tool_name: row.try_get("tool_name")?, message: row.try_get("message")?, request_id: row.try_get("request_id")?, status_code: match row.try_get::, _>("status_code")? { Some(value) => { Some( u16::try_from(value).map_err(|_| RegistryError::InvalidNumericValue { field: "status_code", value: i64::from(value), })?, ) } None => None, }, duration_ms: to_u64(row.try_get::("duration_ms")?, "duration_ms")?, error_kind: row.try_get("error_kind")?, request_preview: row.try_get::, _>("request_preview_json")?.0, response_preview: row.try_get::, _>("response_preview_json")?.0, created_at: row.try_get("created_at")?, }, operation_name: row.try_get("operation_name")?, operation_display_name: row.try_get("operation_display_name")?, agent_slug: row.try_get("agent_slug")?, agent_display_name: row.try_get("agent_display_name")?, }) } fn map_agent_summary(row: &PgRow) -> Result { Ok(AgentSummary { id: AgentId::new(row.try_get::("id")?), workspace_id: WorkspaceId::new(row.try_get::("workspace_id")?), slug: row.try_get("slug")?, display_name: row.try_get("display_name")?, description: row.try_get("description")?, status: deserialize_enum_text(&row.try_get::("status")?, "status")?, current_draft_version: from_db_version( row.try_get("current_draft_version")?, "current_draft_version", )?, latest_published_version: row .try_get::, _>("latest_published_version")? .map(|value| from_db_version(value, "latest_published_version")) .transpose()?, created_at: row.try_get("created_at")?, updated_at: row.try_get("updated_at")?, published_at: row.try_get("published_at")?, }) } fn map_operation_summary(row: &PgRow) -> Result { let target: Target = deserialize_json_value(row.try_get::, _>("target_json")?.0)?; let (target_url, target_action) = target_summary(&target); Ok(OperationSummary { id: OperationId::new(row.try_get::("id")?), workspace_id: WorkspaceId::new(row.try_get::("workspace_id")?), name: row.try_get("name")?, display_name: row.try_get("display_name")?, category: row.try_get("category")?, protocol: deserialize_enum_text(&row.try_get::("protocol")?, "protocol")?, target_url, target_action, status: deserialize_enum_text(&row.try_get::("status")?, "status")?, current_draft_version: from_db_version( row.try_get("current_draft_version")?, "current_draft_version", )?, latest_published_version: row .try_get::, _>("latest_published_version")? .map(|value| from_db_version(value, "latest_published_version")) .transpose()?, created_at: row.try_get("created_at")?, updated_at: row.try_get("updated_at")?, published_at: row.try_get("published_at")?, }) } fn map_operation_usage_summary(row: &PgRow) -> Result { Ok(OperationUsageSummary { operation_id: OperationId::new(row.try_get::("operation_id")?), calls_today: to_u64(row.try_get::("calls_today")?, "calls_today")?, error_rate_pct: row.try_get("error_rate_pct")?, avg_latency_ms: to_u64(row.try_get::("avg_latency_ms")?, "avg_latency_ms")?, }) } fn target_summary(target: &Target) -> (String, String) { match target { Target::Rest(rest) => ( join_target_url(&rest.base_url, &rest.path_template), match rest.method { HttpMethod::Get => "GET", HttpMethod::Post => "POST", HttpMethod::Put => "PUT", HttpMethod::Patch => "PATCH", HttpMethod::Delete => "DELETE", } .to_owned(), ), Target::Graphql(graphql) => ( graphql.endpoint.clone(), match graphql.operation_type { GraphqlOperationType::Query => "QUERY", GraphqlOperationType::Mutation => "MUTATION", } .to_owned(), ), Target::Grpc(grpc) => (grpc.server_addr.clone(), grpc.method.clone()), } } fn join_target_url(base: &str, path: &str) -> String { if path.is_empty() { return base.to_owned(); } if path.starts_with("http://") || path.starts_with("https://") || path.starts_with("grpc://") { return path.to_owned(); } let trimmed_base = base.trim_end_matches('/'); let trimmed_path = path.trim_start_matches('/'); if trimmed_base.is_empty() { format!("/{trimmed_path}") } else if trimmed_path.is_empty() { trimmed_base.to_owned() } else { format!("{trimmed_base}/{trimmed_path}") } } fn map_operation_agent_ref(row: &PgRow) -> Result { Ok(OperationAgentRef { operation_id: OperationId::new(row.try_get::("operation_id")?), agent_id: AgentId::new(row.try_get::("agent_id")?), agent_slug: row.try_get("agent_slug")?, display_name: row.try_get("display_name")?, }) } fn map_operation_version_record(row: &PgRow) -> Result { let operation_id = OperationId::new(row.try_get::("id")?); let version = from_db_version(row.try_get("version")?, "version")?; let status = deserialize_enum_text(&row.try_get::("status")?, "status")?; Ok(OperationVersionRecord { operation_id: operation_id.clone(), workspace_id: WorkspaceId::new(row.try_get::("workspace_id")?), version, status, change_note: row.try_get("change_note")?, created_at: row.try_get("created_at")?, created_by: row.try_get("created_by")?, snapshot: RegistryOperation { id: operation_id, name: row.try_get("name")?, display_name: row.try_get("display_name")?, category: row.try_get("category")?, protocol: deserialize_enum_text(&row.try_get::("protocol")?, "protocol")?, status, version, target: deserialize_json_value(row.try_get::, _>("target_json")?.0)?, input_schema: deserialize_json_value( row.try_get::, _>("input_schema_json")?.0, )?, output_schema: deserialize_json_value( row.try_get::, _>("output_schema_json")?.0, )?, input_mapping: deserialize_json_value( row.try_get::, _>("input_mapping_json")?.0, )?, output_mapping: deserialize_json_value( row.try_get::, _>("output_mapping_json")?.0, )?, execution_config: deserialize_json_value( row.try_get::, _>("execution_config_json")?.0, )?, tool_description: deserialize_json_value( row.try_get::, _>("tool_description_json")?.0, )?, samples: row .try_get::>, _>("samples_json")? .map(|value| deserialize_json_value(value.0)) .transpose()?, generated_draft: row .try_get::>, _>("generated_draft_json")? .map(|value| deserialize_json_value(value.0)) .transpose()?, config_export: row .try_get::>, _>("config_export_json")? .map(|value| deserialize_json_value(value.0)) .transpose()?, created_at: row.try_get("operation_created_at")?, updated_at: row.try_get("operation_updated_at")?, published_at: row.try_get("operation_published_at")?, }, }) } fn map_auth_profile(row: &PgRow) -> Result { Ok(AuthProfile { id: crank_core::AuthProfileId::new(row.try_get::("id")?), workspace_id: WorkspaceId::new(row.try_get::("workspace_id")?), name: row.try_get("name")?, kind: deserialize_enum_text(&row.try_get::("kind")?, "kind")?, config: deserialize_json_value(row.try_get::, _>("config_json")?.0)?, created_at: row.try_get("created_at")?, updated_at: row.try_get("updated_at")?, }) } fn map_agent_binding(row: &PgRow) -> Result { Ok(AgentOperationBinding { operation_id: OperationId::new(row.try_get::("operation_id")?), operation_version: from_db_version(row.try_get("operation_version")?, "operation_version")?, tool_name: row.try_get("tool_name")?, tool_title: row.try_get("tool_title")?, tool_description_override: row.try_get("tool_description_override")?, enabled: row.try_get("enabled")?, }) } fn map_agent_version_record( summary: &AgentSummary, row: &PgRow, bindings: Vec, ) -> Result { let version = from_db_version(row.try_get("version")?, "version")?; let status = deserialize_enum_text(&row.try_get::("status")?, "status")?; Ok(AgentVersionRecord { agent_id: summary.id.clone(), workspace_id: summary.workspace_id.clone(), version, status, created_at: row.try_get("created_at")?, bindings, snapshot: AgentVersion { agent_id: summary.id.clone(), version, status, instructions: row.try_get::, _>("instructions_json")?.0, tool_selection_policy: row .try_get::, _>("tool_selection_policy_json")? .0, created_at: row.try_get("created_at")?, }, }) } fn map_published_agent_tool(row: &PgRow) -> Result { let record = map_operation_version_record(row)?; Ok(PublishedAgentTool { workspace_id: WorkspaceId::new(row.try_get::("workspace_id")?), workspace_slug: row.try_get("workspace_slug")?, agent_id: AgentId::new(row.try_get::("agent_id")?), agent_slug: row.try_get("agent_slug")?, tool_name: row.try_get("tool_name")?, tool_title: row.try_get("tool_title")?, tool_description: row.try_get("tool_description")?, operation: record.snapshot, }) } fn map_sample_metadata(row: &PgRow) -> Result { Ok(OperationSampleMetadata { id: crank_core::SampleId::new(row.try_get::("id")?), operation_id: OperationId::new(row.try_get::("operation_id")?), version: from_db_version(row.try_get("version")?, "version")?, sample_kind: deserialize_enum_text( &row.try_get::("sample_kind")?, "sample_kind", )?, storage_ref: row.try_get("storage_ref")?, content_type: row.try_get("content_type")?, file_name: row.try_get("file_name")?, created_at: row.try_get("created_at")?, }) } fn map_descriptor_metadata(row: &PgRow) -> Result { Ok(DescriptorMetadata { id: crank_core::DescriptorId::new(row.try_get::("id")?), operation_id: row .try_get::, _>("operation_id")? .map(OperationId::new), version: row .try_get::, _>("version")? .map(|value| from_db_version(value, "version")) .transpose()?, descriptor_kind: deserialize_enum_text( &row.try_get::("descriptor_kind")?, "descriptor_kind", )?, storage_ref: row.try_get("storage_ref")?, source_name: row.try_get("source_name")?, package_index: row .try_get::>, _>("package_index_json")? .map(|value| value.0), created_at: row.try_get("created_at")?, }) } fn map_yaml_import_job(row: &PgRow) -> Result { Ok(YamlImportJob { id: YamlImportJobId::new(row.try_get::("id")?), source_sample_id: row .try_get::, _>("source_sample_id")? .map(crank_core::SampleId::new), status: deserialize_enum_text(&row.try_get::("status")?, "status")?, format_version: row.try_get("format_version")?, mode: deserialize_enum_text(&row.try_get::("mode")?, "mode")?, result_operation_id: row .try_get::, _>("result_operation_id")? .map(OperationId::new), result_version: row .try_get::, _>("result_version")? .map(|value| from_db_version(value, "result_version")) .transpose()?, error_text: row.try_get("error_text")?, created_at: row.try_get("created_at")?, finished_at: row.try_get("finished_at")?, }) } fn map_usage_operation_breakdown(row: &PgRow) -> Result { Ok(UsageOperationBreakdown { operation_id: OperationId::new(row.try_get::("operation_id")?), operation_name: row.try_get("operation_name")?, operation_display_name: row.try_get("operation_display_name")?, protocol: deserialize_enum_text(&row.try_get::("protocol")?, "protocol")?, calls_total: to_u64(row.try_get::("calls_total")?, "calls_total")?, calls_error: to_u64(row.try_get::("calls_error")?, "calls_error")?, p50_ms: to_u64(row.try_get::("p50_ms")?, "p50_ms")?, p95_ms: to_u64(row.try_get::("p95_ms")?, "p95_ms")?, p99_ms: to_u64(row.try_get::("p99_ms")?, "p99_ms")?, }) } fn map_usage_agent_breakdown(row: &PgRow) -> Result { Ok(UsageAgentBreakdown { agent_id: AgentId::new(row.try_get::("agent_id")?), agent_slug: row.try_get("agent_slug")?, agent_display_name: row.try_get("agent_display_name")?, calls_total: to_u64(row.try_get::("calls_total")?, "calls_total")?, calls_error: to_u64(row.try_get::("calls_error")?, "calls_error")?, p50_ms: to_u64(row.try_get::("p50_ms")?, "p50_ms")?, p95_ms: to_u64(row.try_get::("p95_ms")?, "p95_ms")?, p99_ms: to_u64(row.try_get::("p99_ms")?, "p99_ms")?, }) } fn serialize_json_value(value: &T) -> Result { Ok(serde_json::to_value(value)?) } fn serialize_option_json_value( value: &Option, ) -> Result, RegistryError> { value.as_ref().map(serialize_json_value).transpose() } fn deserialize_json_value(value: Value) -> Result { Ok(serde_json::from_value(value)?) } fn serialize_enum_text( value: &T, field: &'static str, ) -> Result { serde_json::to_value(value)? .as_str() .map(ToOwned::to_owned) .ok_or(RegistryError::InvalidEnumRepresentation { field }) } fn deserialize_enum_text( value: &str, field: &'static str, ) -> Result { serde_json::from_value(Value::String(value.to_owned())) .map_err(|_| RegistryError::InvalidEnumRepresentation { field }) } fn to_db_version(value: u32) -> i32 { value as i32 } fn from_db_version(value: i32, field: &'static str) -> Result { u32::try_from(value).map_err(|_| RegistryError::InvalidNumericValue { field, value: i64::from(value), }) } fn to_u64(value: i64, field: &'static str) -> Result { u64::try_from(value).map_err(|_| RegistryError::InvalidNumericValue { field, value }) } fn usage_bucket_sql(bucket: crate::model::UsageBucket) -> &'static str { match bucket { crate::model::UsageBucket::Hour => "hour", crate::model::UsageBucket::Day => "day", crate::model::UsageBucket::Week => "week", crate::model::UsageBucket::Month => "month", } } #[cfg(test)] mod tests { use std::{ collections::BTreeMap, env, time::{SystemTime, UNIX_EPOCH}, }; use crank_core::{ ApiKeyHeaderAuthConfig, AuthConfig, AuthKind, AuthProfile, ConfigExport, ExecutionConfig, ExportMode, GeneratedDraft, GeneratedDraftStatus, HttpMethod, OperationId, OperationStatus, Protocol, RestTarget, RetryPolicy, Samples, SecretRef, Target, ToolDescription, ToolExample, WorkspaceId, }; use crank_mapping::{MappingRule, MappingSet}; use crank_schema::{Schema, SchemaKind}; use serde_json::json; use sqlx::{Executor, PgPool, postgres::PgPoolOptions}; use crate::{ PostgresRegistry, RegistryError, model::{ CreateVersionRequest, CreateYamlImportJobRequest, DescriptorKind, DescriptorMetadata, OperationSampleMetadata, PublishRequest, RegistryOperation, SampleKind, SaveAuthProfileRequest, SaveDescriptorMetadataRequest, SaveSampleMetadataRequest, YamlImportJobCompletion, YamlImportJobId, YamlImportJobStatus, }, }; fn test_workspace_id() -> WorkspaceId { WorkspaceId::new("ws_default") } #[tokio::test] async fn stores_versions_and_published_operations() { let database = TestDatabase::new().await; let registry = database.registry().await; let operation_v1 = test_operation("op_rest_01", 1, OperationStatus::Draft); registry .create_operation(&test_workspace_id(), &operation_v1, Some("alice")) .await .unwrap(); let operation_v2 = test_operation("op_rest_01", 2, OperationStatus::Draft); registry .create_version(CreateVersionRequest { workspace_id: &test_workspace_id(), snapshot: &operation_v2, change_note: Some("add output mapping"), created_by: Some("alice"), }) .await .unwrap(); registry .publish_operation(PublishRequest { workspace_id: &test_workspace_id(), operation_id: &operation_v2.id, version: operation_v2.version, published_at: "2026-03-25T12:10:00Z", published_by: Some("alice"), }) .await .unwrap(); let summary = registry .get_operation_summary(&test_workspace_id(), &operation_v2.id) .await .unwrap() .unwrap(); let versions = registry .list_operation_versions(&test_workspace_id(), &operation_v2.id) .await .unwrap(); let published = registry .get_published_operation(&operation_v2.id) .await .unwrap() .unwrap(); assert_eq!(summary.current_draft_version, 2); assert_eq!(summary.latest_published_version, Some(2)); assert_eq!(summary.status, OperationStatus::Published); assert_eq!(versions.len(), 2); assert_eq!( versions[1].change_note.as_deref(), Some("add output mapping") ); assert_eq!(published.version, 2); assert!(published.is_published()); database.cleanup().await; } #[tokio::test] async fn rejects_out_of_order_versions() { let database = TestDatabase::new().await; let registry = database.registry().await; let operation = test_operation("op_rest_02", 1, OperationStatus::Draft); registry .create_operation(&test_workspace_id(), &operation, None) .await .unwrap(); let invalid = test_operation("op_rest_02", 3, OperationStatus::Draft); let error = registry .create_version(CreateVersionRequest { workspace_id: &test_workspace_id(), snapshot: &invalid, change_note: None, created_by: None, }) .await .unwrap_err(); assert!(matches!( error, RegistryError::InvalidVersionSequence { expected: 2, actual: 3, .. } )); database.cleanup().await; } #[tokio::test] async fn stores_auth_profiles_and_artifact_metadata() { let database = TestDatabase::new().await; let registry = database.registry().await; let operation = test_operation("op_rest_03", 1, OperationStatus::Draft); registry .create_operation(&test_workspace_id(), &operation, None) .await .unwrap(); let auth_profile = AuthProfile { id: "auth_crank".into(), workspace_id: test_workspace_id(), name: "Crank API key".to_owned(), kind: AuthKind::ApiKeyHeader, config: AuthConfig::ApiKeyHeader(ApiKeyHeaderAuthConfig { header_name: "X-Api-Key".to_owned(), secret_ref: SecretRef::new("vault://crank/api-key"), }), created_at: "2026-03-25T12:00:00Z".to_owned(), updated_at: "2026-03-25T12:00:00Z".to_owned(), }; registry .save_auth_profile(SaveAuthProfileRequest { workspace_id: &test_workspace_id(), profile: &auth_profile, }) .await .unwrap(); let input_sample = OperationSampleMetadata { id: "sample_input".into(), operation_id: operation.id.clone(), version: 1, sample_kind: SampleKind::InputJson, storage_ref: "file:///tmp/input.json".to_owned(), content_type: "application/json".to_owned(), file_name: Some("input.json".to_owned()), created_at: "2026-03-25T12:01:00Z".to_owned(), }; let descriptor = DescriptorMetadata { id: "descriptor_01".into(), operation_id: Some(operation.id.clone()), version: Some(1), descriptor_kind: DescriptorKind::DescriptorSet, storage_ref: "file:///tmp/schema.desc".to_owned(), source_name: Some("schema.desc".to_owned()), package_index: Some(json!({ "crm.v1": ["LeadService"] })), created_at: "2026-03-25T12:02:00Z".to_owned(), }; registry .save_sample_metadata(SaveSampleMetadataRequest { sample: &input_sample, }) .await .unwrap(); registry .save_descriptor_metadata(SaveDescriptorMetadataRequest { descriptor: &descriptor, }) .await .unwrap(); let auth_profiles = registry .list_auth_profiles(&test_workspace_id()) .await .unwrap(); let samples = registry .list_sample_metadata(&operation.id, 1) .await .unwrap(); let descriptors = registry .list_descriptor_metadata(&operation.id, 1) .await .unwrap(); assert_eq!(auth_profiles, vec![auth_profile]); assert_eq!(samples, vec![input_sample]); assert_eq!(descriptors, vec![descriptor]); database.cleanup().await; } #[tokio::test] async fn stores_and_finishes_yaml_import_jobs() { let database = TestDatabase::new().await; let registry = database.registry().await; let job_id = YamlImportJobId::new("job_yaml_01"); let operation = test_operation("op_rest_04", 1, OperationStatus::Draft); registry .create_operation(&test_workspace_id(), &operation, None) .await .unwrap(); registry .create_yaml_import_job(CreateYamlImportJobRequest { id: &job_id, source_sample_id: None, format_version: "v1", mode: ExportMode::Portable, created_at: "2026-03-25T12:00:00Z", }) .await .unwrap(); registry .finish_yaml_import_job( &job_id, &YamlImportJobCompletion { status: YamlImportJobStatus::Completed, result_operation_id: Some(operation.id.clone()), result_version: Some(2), error_text: None, finished_at: "2026-03-25T12:05:00Z".to_owned(), }, ) .await .unwrap(); let job = registry .get_yaml_import_job(&job_id) .await .unwrap() .unwrap(); assert_eq!(job.status, YamlImportJobStatus::Completed); assert_eq!(job.result_version, Some(2)); assert_eq!(job.mode, ExportMode::Portable); database.cleanup().await; } struct TestDatabase { admin_pool: PgPool, database_url: String, schema: String, } impl TestDatabase { async fn new() -> Self { let database_url = env::var("TEST_DATABASE_URL") .expect("TEST_DATABASE_URL must point to a reachable PostgreSQL database"); let admin_pool = PgPoolOptions::new() .max_connections(1) .connect(&database_url) .await .unwrap(); let schema = format!( "test_registry_{}_{}", std::process::id(), SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap() .as_nanos() ); admin_pool .execute(sqlx::query(&format!("create schema {schema}"))) .await .unwrap(); Self { admin_pool, database_url, schema, } } async fn registry(&self) -> PostgresRegistry { PostgresRegistry::connect_in_schema(&self.database_url, Some(&self.schema)) .await .unwrap() } async fn cleanup(&self) { self.admin_pool .execute(sqlx::query(&format!( "drop schema if exists {} cascade", self.schema ))) .await .unwrap(); } } fn test_operation(id: &str, version: u32, status: OperationStatus) -> RegistryOperation { RegistryOperation { id: OperationId::new(id), name: format!("{id}_tool"), display_name: format!("Display {id}"), category: "general".to_owned(), protocol: Protocol::Rest, status, version, target: Target::Rest(RestTarget { base_url: "https://api.example.com".to_owned(), method: HttpMethod::Post, path_template: "/v1/leads".to_owned(), static_headers: BTreeMap::from([("X-Static".to_owned(), "true".to_owned())]), }), input_schema: Schema { kind: SchemaKind::Object, description: Some("input".to_owned()), required: true, nullable: false, default_value: None, fields: BTreeMap::from([( "email".to_owned(), Schema { kind: SchemaKind::String, description: None, required: true, nullable: false, default_value: None, fields: BTreeMap::new(), items: None, enum_values: Vec::new(), variants: Vec::new(), }, )]), items: None, enum_values: Vec::new(), variants: Vec::new(), }, output_schema: Schema { kind: SchemaKind::Object, description: Some("output".to_owned()), required: true, nullable: false, default_value: None, fields: BTreeMap::from([( "id".to_owned(), Schema { kind: SchemaKind::String, description: None, required: true, nullable: false, default_value: None, fields: BTreeMap::new(), items: None, enum_values: Vec::new(), variants: Vec::new(), }, )]), items: None, enum_values: Vec::new(), variants: Vec::new(), }, input_mapping: MappingSet { rules: vec![MappingRule { source: "$.mcp.email".to_owned(), target: "$.request.body.email".to_owned(), required: true, default_value: None, transform: None, condition: None, notes: None, }], }, output_mapping: MappingSet { rules: vec![MappingRule { source: "$.response.body.id".to_owned(), target: "$.output.id".to_owned(), required: true, default_value: None, transform: None, condition: None, notes: None, }], }, execution_config: ExecutionConfig { timeout_ms: 10_000, retry_policy: Some(RetryPolicy { max_attempts: 3 }), auth_profile_ref: Some("auth_crank".into()), headers: BTreeMap::from([("X-Request-Id".to_owned(), "static".to_owned())]), protocol_options: None, }, tool_description: ToolDescription { title: "Create lead".to_owned(), description: "Creates CRM lead".to_owned(), tags: vec!["crm".to_owned()], examples: vec![ToolExample { input: json!({ "email": "a@example.com" }), }], }, samples: Some(Samples { input_json_sample_ref: Some("sample_input".into()), output_json_sample_ref: Some("sample_output".into()), proto_file_ref: None, descriptor_ref: None, }), generated_draft: Some(GeneratedDraft { status: GeneratedDraftStatus::Available, source_types: vec!["input_json".to_owned(), "output_json".to_owned()], generated_at: Some("2026-03-25T11:59:00Z".to_owned()), input_schema_generated: true, output_schema_generated: true, input_mapping_generated: true, output_mapping_generated: true, warnings: Vec::new(), }), config_export: Some(ConfigExport { format_version: "v1".to_owned(), export_mode: ExportMode::Portable, }), created_at: "2026-03-25T11:58:00Z".to_owned(), updated_at: format!("2026-03-25T12:{version:02}:00Z"), published_at: None, } } }