bsodfather/crank
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
86b61523bd7c1fbeaeadb3c52a38d172e45dc963
crank/crates/crank-registry/src/postgres.rs
T
a.tolmachev 86b61523bd feat: persist current workspace in user sessions
2026-03-31 16:09:10 +03:00

3796 lines
130 KiB
Rust
Raw Blame History

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, RegistryError> {
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<Vec<WorkspaceRecord>, 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<Vec<WorkspaceMembershipRecord>, 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<UserId, RegistryError> {
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::<String, _>("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<Option<AuthUserRecord>, RegistryError> {
let row = sqlx::query(
"select
id,
email,
display_name,
password_hash,
status,
to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at
from users
where email = $1
limit 1",
)
.bind(email)
.fetch_optional(&self.pool)
.await?;
row.as_ref().map(map_auth_user_record).transpose()
}
pub async fn get_auth_user_by_id(
&self,
user_id: &UserId,
) -> Result<Option<AuthUserRecord>, RegistryError> {
let row = sqlx::query(
"select
id,
email,
display_name,
password_hash,
status,
to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at
from users
where id = $1
limit 1",
)
.bind(user_id.as_str())
.fetch_optional(&self.pool)
.await?;
row.as_ref().map(map_auth_user_record).transpose()
}
pub async fn update_user_profile(
&self,
user_id: &UserId,
email: &str,
display_name: &str,
) -> Result<User, RegistryError> {
let row = sqlx::query(
"update users
set email = $2,
display_name = $3
where id = $1
returning
id,
email,
display_name,
status,
to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at",
)
.bind(user_id.as_str())
.bind(email)
.bind(display_name)
.fetch_one(&self.pool)
.await
.map_err(|error| map_user_update_error(error, user_id, email))?;
map_user(&row)
}
pub async fn update_user_password(
&self,
user_id: &UserId,
password_hash: &str,
) -> Result<(), RegistryError> {
let result = sqlx::query(
"update users
set password_hash = $2
where id = $1",
)
.bind(user_id.as_str())
.bind(password_hash)
.execute(&self.pool)
.await?;
if result.rows_affected() == 0 {
return Err(RegistryError::UserNotFound {
user_id: user_id.as_str().to_owned(),
});
}
Ok(())
}
pub async fn create_user_session(
&self,
session_id: &UserSessionId,
user_id: &UserId,
current_workspace_id: Option<&WorkspaceId>,
secret_hash: &str,
expires_at: &str,
) -> Result<(), RegistryError> {
sqlx::query(
"insert into user_sessions (
id,
user_id,
current_workspace_id,
secret_hash,
status,
expires_at,
last_seen_at,
created_at
) values (
$1,
$2,
$3,
$4,
'active',
$5::timestamptz,
now(),
now()
)",
)
.bind(session_id.as_str())
.bind(user_id.as_str())
.bind(current_workspace_id.map(|id| id.as_str()))
.bind(secret_hash)
.bind(expires_at)
.execute(&self.pool)
.await?;
Ok(())
}
pub async fn get_user_session(
&self,
session_id: &UserSessionId,
secret_hash: &str,
) -> Result<Option<SessionRecord>, RegistryError> {
let row = sqlx::query(
"select
s.id,
s.user_id,
s.current_workspace_id,
u.email,
u.display_name,
u.status,
to_char(u.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at
from user_sessions s
join users u on u.id = s.user_id
where s.id = $1
and s.secret_hash = $2
and s.status = 'active'
and s.expires_at > now()
limit 1",
)
.bind(session_id.as_str())
.bind(secret_hash)
.fetch_optional(&self.pool)
.await?;
let Some(row) = row else {
return Ok(None);
};
let user_id = UserId::new(row.try_get::<String, _>("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::<String, _>("status")?, "status")?,
created_at: row.try_get("created_at")?,
};
let memberships = self.list_workspaces_for_user(&user_id).await?;
let stored_workspace_id = row
.try_get::<Option<String>, _>("current_workspace_id")?
.map(WorkspaceId::new);
let current_workspace_id = stored_workspace_id
.filter(|workspace_id| {
memberships
.iter()
.any(|membership| membership.workspace.id == *workspace_id)
})
.or_else(|| {
memberships
.first()
.map(|membership| membership.workspace.id.clone())
});
Ok(Some(SessionRecord {
session_id: UserSessionId::new(row.try_get::<String, _>("id")?),
user,
memberships,
current_workspace_id,
}))
}
pub async fn touch_user_session(
&self,
session_id: &UserSessionId,
) -> Result<(), RegistryError> {
sqlx::query(
"update user_sessions
set last_seen_at = now()
where id = $1",
)
.bind(session_id.as_str())
.execute(&self.pool)
.await?;
Ok(())
}
pub async fn revoke_user_session(
&self,
session_id: &UserSessionId,
) -> Result<(), RegistryError> {
sqlx::query(
"update user_sessions
set status = 'revoked'
where id = $1",
)
.bind(session_id.as_str())
.execute(&self.pool)
.await?;
Ok(())
}
pub async fn set_user_session_current_workspace(
&self,
session_id: &UserSessionId,
workspace_id: &WorkspaceId,
) -> Result<(), RegistryError> {
sqlx::query(
"update user_sessions
set current_workspace_id = $2
where id = $1",
)
.bind(session_id.as_str())
.bind(workspace_id.as_str())
.execute(&self.pool)
.await?;
Ok(())
}
pub async fn user_has_workspace_access(
&self,
user_id: &UserId,
workspace_id: &WorkspaceId,
) -> Result<bool, RegistryError> {
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<Vec<MembershipRecord>, RegistryError> {
let rows = sqlx::query(
"select
m.workspace_id,
m.user_id,
m.role,
to_char(m.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
u.email,
u.display_name,
u.status,
to_char(u.created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as user_created_at
from memberships m
join users u on u.id = m.user_id
where m.workspace_id = $1
order by u.email asc",
)
.bind(workspace_id.as_str())
.fetch_all(&self.pool)
.await?;
rows.iter().map(map_membership_record).collect()
}
pub async fn update_membership_role(
&self,
workspace_id: &WorkspaceId,
user_id: &UserId,
role: MembershipRole,
) -> Result<(), RegistryError> {
let result = sqlx::query(
"update memberships
set role = $3
where workspace_id = $1 and user_id = $2",
)
.bind(workspace_id.as_str())
.bind(user_id.as_str())
.bind(serialize_enum_text(&role, "role")?)
.execute(&self.pool)
.await?;
if result.rows_affected() == 0 {
return Err(RegistryError::MembershipNotFound {
workspace_id: workspace_id.as_str().to_owned(),
user_id: user_id.as_str().to_owned(),
});
}
Ok(())
}
pub async fn delete_membership(
&self,
workspace_id: &WorkspaceId,
user_id: &UserId,
) -> Result<(), RegistryError> {
let result = sqlx::query(
"delete from memberships
where workspace_id = $1 and user_id = $2",
)
.bind(workspace_id.as_str())
.bind(user_id.as_str())
.execute(&self.pool)
.await?;
if result.rows_affected() == 0 {
return Err(RegistryError::MembershipNotFound {
workspace_id: workspace_id.as_str().to_owned(),
user_id: user_id.as_str().to_owned(),
});
}
Ok(())
}
pub async fn list_invitations(
&self,
workspace_id: &WorkspaceId,
) -> Result<Vec<InvitationRecord>, RegistryError> {
let rows = sqlx::query(
"select
id,
workspace_id,
email,
role,
status,
token_hash,
to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at,
to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at
from invitation_tokens
where workspace_id = $1
order by created_at desc",
)
.bind(workspace_id.as_str())
.fetch_all(&self.pool)
.await?;
rows.iter().map(map_invitation_record).collect()
}
pub async fn create_invitation(
&self,
request: CreateInvitationRequest<'_>,
) -> Result<(), RegistryError> {
sqlx::query(
"insert into invitation_tokens (
id,
workspace_id,
email,
role,
status,
token_hash,
expires_at,
created_at
) values (
$1, $2, $3, $4, $5, $6, $7::timestamptz, $8::timestamptz
)",
)
.bind(request.invitation.id.as_str())
.bind(request.invitation.workspace_id.as_str())
.bind(&request.invitation.email)
.bind(serialize_enum_text(&request.invitation.role, "role")?)
.bind(serialize_enum_text(&request.invitation.status, "status")?)
.bind(&request.invitation.token_hash)
.bind(&request.invitation.expires_at)
.bind(&request.invitation.created_at)
.execute(&self.pool)
.await?;
Ok(())
}
pub async fn delete_invitation(
&self,
workspace_id: &WorkspaceId,
invitation_id: &InvitationId,
) -> Result<(), RegistryError> {
let result =
sqlx::query("delete from invitation_tokens where workspace_id = $1 and id = $2")
.bind(workspace_id.as_str())
.bind(invitation_id.as_str())
.execute(&self.pool)
.await?;
if result.rows_affected() == 0 {
return Err(RegistryError::InvitationNotFound {
invitation_id: invitation_id.as_str().to_owned(),
});
}
Ok(())
}
pub async fn delete_workspace(&self, workspace_id: &WorkspaceId) -> Result<(), RegistryError> {
let result = sqlx::query("delete from workspaces where id = $1")
.bind(workspace_id.as_str())
.execute(&self.pool)
.await?;
if result.rows_affected() == 0 {
return Err(RegistryError::WorkspaceNotFound {
workspace_id: workspace_id.as_str().to_owned(),
});
}
Ok(())
}
pub async fn list_platform_api_keys(
&self,
workspace_id: &WorkspaceId,
) -> Result<Vec<PlatformApiKeyRecord>, 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<Vec<InvocationLogRecord>, 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<Option<InvocationLogRecord>, 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<UsageSummary, 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 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::<i64, _>("calls_total")?;
let calls_ok = row.try_get::<i64, _>("calls_ok")?;
let calls_error = row.try_get::<i64, _>("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::<i64, _>("p50_ms")?, "p50_ms")?,
p95_ms: to_u64(row.try_get::<i64, _>("p95_ms")?, "p95_ms")?,
p99_ms: to_u64(row.try_get::<i64, _>("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<Vec<UsageTimelinePoint>, 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::<i64, _>("calls_ok")?, "calls_ok")?,
calls_error: to_u64(row.try_get::<i64, _>("calls_error")?, "calls_error")?,
})
})
.collect()
}
pub async fn list_usage_by_operation(
&self,
query: UsageQuery<'_>,
) -> Result<Vec<UsageOperationBreakdown>, 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<Option<UsageRollupRecord>, 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::<i64, _>("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::<i64, _>("calls_ok")?, "calls_ok")?,
calls_error: to_u64(row.try_get::<i64, _>("calls_error")?, "calls_error")?,
p50_ms: to_u64(row.try_get::<i64, _>("p50_ms")?, "p50_ms")?,
p95_ms: to_u64(row.try_get::<i64, _>("p95_ms")?, "p95_ms")?,
p99_ms: to_u64(row.try_get::<i64, _>("p99_ms")?, "p99_ms")?,
},
}))
}
pub async fn list_usage_by_agent(
&self,
query: UsageQuery<'_>,
) -> Result<Vec<UsageAgentBreakdown>, 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<Option<UsageRollupRecord>, 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::<i64, _>("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::<i64, _>("calls_ok")?, "calls_ok")?,
calls_error: to_u64(row.try_get::<i64, _>("calls_error")?, "calls_error")?,
p50_ms: to_u64(row.try_get::<i64, _>("p50_ms")?, "p50_ms")?,
p95_ms: to_u64(row.try_get::<i64, _>("p95_ms")?, "p95_ms")?,
p99_ms: to_u64(row.try_get::<i64, _>("p99_ms")?, "p99_ms")?,
},
}))
}
pub async fn get_workspace(
&self,
workspace_id: &WorkspaceId,
) -> Result<Option<WorkspaceRecord>, 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<Vec<AgentSummary>, 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<Option<AgentSummary>, 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<Option<AgentVersionRecord>, 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<Vec<PublishedAgentTool>, 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<Self, RegistryError> {
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<bool, RegistryError> {
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<Vec<OperationSummary>, 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<Vec<OperationUsageSummary>, 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<Vec<OperationAgentRef>, 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<Option<OperationSummary>, 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<Option<OperationVersionRecord>, 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<Vec<OperationVersionRecord>, 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<Option<RegistryOperation>, 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<Vec<RegistryOperation>, 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<Option<AuthProfile>, 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<Vec<AuthProfile>, 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<Vec<OperationSampleMetadata>, 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<Vec<DescriptorMetadata>, 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<Option<YamlImportJob>, 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<Vec<AgentOperationBinding>, 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<WorkspaceRecord, RegistryError> {
Ok(WorkspaceRecord {
workspace: Workspace {
id: WorkspaceId::new(row.try_get::<String, _>("id")?),
slug: row.try_get("slug")?,
display_name: row.try_get("display_name")?,
status: deserialize_enum_text(&row.try_get::<String, _>("status")?, "status")?,
settings: row.try_get::<Json<Value>, _>("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<WorkspaceMembershipRecord, RegistryError> {
Ok(WorkspaceMembershipRecord {
workspace: Workspace {
id: WorkspaceId::new(row.try_get::<String, _>("id")?),
slug: row.try_get("slug")?,
display_name: row.try_get("display_name")?,
status: deserialize_enum_text(&row.try_get::<String, _>("status")?, "status")?,
settings: row.try_get::<Json<Value>, _>("settings_json")?.0,
created_at: row.try_get("created_at")?,
updated_at: row.try_get("updated_at")?,
},
role: deserialize_enum_text(&row.try_get::<String, _>("role")?, "role")?,
})
}
fn map_membership_record(row: &PgRow) -> Result<MembershipRecord, RegistryError> {
Ok(MembershipRecord {
workspace_id: WorkspaceId::new(row.try_get::<String, _>("workspace_id")?),
user: User {
id: UserId::new(row.try_get::<String, _>("user_id")?),
email: row.try_get("email")?,
display_name: row.try_get("display_name")?,
status: deserialize_enum_text(&row.try_get::<String, _>("status")?, "status")?,
created_at: row.try_get("user_created_at")?,
},
role: deserialize_enum_text(&row.try_get::<String, _>("role")?, "role")?,
created_at: row.try_get("created_at")?,
})
}
fn map_auth_user_record(row: &PgRow) -> Result<AuthUserRecord, RegistryError> {
Ok(AuthUserRecord {
user: map_user(row)?,
password_hash: row.try_get("password_hash")?,
})
}
fn map_user(row: &PgRow) -> Result<User, RegistryError> {
Ok(User {
id: UserId::new(row.try_get::<String, _>("id")?),
email: row.try_get("email")?,
display_name: row.try_get("display_name")?,
status: deserialize_enum_text(&row.try_get::<String, _>("status")?, "status")?,
created_at: row.try_get("created_at")?,
})
}
fn map_user_update_error(error: sqlx::Error, user_id: &UserId, email: &str) -> RegistryError {
match error {
sqlx::Error::RowNotFound => RegistryError::UserNotFound {
user_id: user_id.as_str().to_owned(),
},
sqlx::Error::Database(database_error)
if database_error.code().as_deref() == Some("23505") =>
{
RegistryError::UserEmailAlreadyExists {
email: email.to_owned(),
}
}
other => RegistryError::Storage(other),
}
}
fn map_invitation_record(row: &PgRow) -> Result<InvitationRecord, RegistryError> {
Ok(InvitationRecord {
invitation: InvitationToken {
id: InvitationId::new(row.try_get::<String, _>("id")?),
workspace_id: WorkspaceId::new(row.try_get::<String, _>("workspace_id")?),
email: row.try_get("email")?,
role: deserialize_enum_text(&row.try_get::<String, _>("role")?, "role")?,
status: deserialize_enum_text(&row.try_get::<String, _>("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<PlatformApiKeyRecord, RegistryError> {
Ok(PlatformApiKeyRecord {
api_key: PlatformApiKey {
id: PlatformApiKeyId::new(row.try_get::<String, _>("id")?),
workspace_id: WorkspaceId::new(row.try_get::<String, _>("workspace_id")?),
name: row.try_get("name")?,
prefix: row.try_get("prefix")?,
scopes: deserialize_json_value(row.try_get::<Json<Value>, _>("scopes_json")?.0)?,
status: deserialize_enum_text(&row.try_get::<String, _>("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<InvocationLogRecord, RegistryError> {
Ok(InvocationLogRecord {
log: InvocationLog {
id: InvocationLogId::new(row.try_get::<String, _>("id")?),
workspace_id: WorkspaceId::new(row.try_get::<String, _>("workspace_id")?),
agent_id: row
.try_get::<Option<String>, _>("agent_id")?
.map(AgentId::new),
operation_id: OperationId::new(row.try_get::<String, _>("operation_id")?),
source: deserialize_enum_text(&row.try_get::<String, _>("source")?, "source")?,
level: deserialize_enum_text(&row.try_get::<String, _>("level")?, "level")?,
status: deserialize_enum_text(&row.try_get::<String, _>("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::<Option<i32>, _>("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::<i64, _>("duration_ms")?, "duration_ms")?,
error_kind: row.try_get("error_kind")?,
request_preview: row.try_get::<Json<Value>, _>("request_preview_json")?.0,
response_preview: row.try_get::<Json<Value>, _>("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<AgentSummary, RegistryError> {
Ok(AgentSummary {
id: AgentId::new(row.try_get::<String, _>("id")?),
workspace_id: WorkspaceId::new(row.try_get::<String, _>("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::<String, _>("status")?, "status")?,
current_draft_version: from_db_version(
row.try_get("current_draft_version")?,
"current_draft_version",
)?,
latest_published_version: row
.try_get::<Option<i32>, _>("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<OperationSummary, RegistryError> {
let target: Target = deserialize_json_value(row.try_get::<Json<Value>, _>("target_json")?.0)?;
let (target_url, target_action) = target_summary(&target);
Ok(OperationSummary {
id: OperationId::new(row.try_get::<String, _>("id")?),
workspace_id: WorkspaceId::new(row.try_get::<String, _>("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::<String, _>("protocol")?, "protocol")?,
target_url,
target_action,
status: deserialize_enum_text(&row.try_get::<String, _>("status")?, "status")?,
current_draft_version: from_db_version(
row.try_get("current_draft_version")?,
"current_draft_version",
)?,
latest_published_version: row
.try_get::<Option<i32>, _>("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<OperationUsageSummary, RegistryError> {
Ok(OperationUsageSummary {
operation_id: OperationId::new(row.try_get::<String, _>("operation_id")?),
calls_today: to_u64(row.try_get::<i64, _>("calls_today")?, "calls_today")?,
error_rate_pct: row.try_get("error_rate_pct")?,
avg_latency_ms: to_u64(row.try_get::<i64, _>("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<OperationAgentRef, RegistryError> {
Ok(OperationAgentRef {
operation_id: OperationId::new(row.try_get::<String, _>("operation_id")?),
agent_id: AgentId::new(row.try_get::<String, _>("agent_id")?),
agent_slug: row.try_get("agent_slug")?,
display_name: row.try_get("display_name")?,
})
}
fn map_operation_version_record(row: &PgRow) -> Result<OperationVersionRecord, RegistryError> {
let operation_id = OperationId::new(row.try_get::<String, _>("id")?);
let version = from_db_version(row.try_get("version")?, "version")?;
let status = deserialize_enum_text(&row.try_get::<String, _>("status")?, "status")?;
Ok(OperationVersionRecord {
operation_id: operation_id.clone(),
workspace_id: WorkspaceId::new(row.try_get::<String, _>("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::<String, _>("protocol")?, "protocol")?,
status,
version,
target: deserialize_json_value(row.try_get::<Json<Value>, _>("target_json")?.0)?,
input_schema: deserialize_json_value(
row.try_get::<Json<Value>, _>("input_schema_json")?.0,
)?,
output_schema: deserialize_json_value(
row.try_get::<Json<Value>, _>("output_schema_json")?.0,
)?,
input_mapping: deserialize_json_value(
row.try_get::<Json<Value>, _>("input_mapping_json")?.0,
)?,
output_mapping: deserialize_json_value(
row.try_get::<Json<Value>, _>("output_mapping_json")?.0,
)?,
execution_config: deserialize_json_value(
row.try_get::<Json<Value>, _>("execution_config_json")?.0,
)?,
tool_description: deserialize_json_value(
row.try_get::<Json<Value>, _>("tool_description_json")?.0,
)?,
samples: row
.try_get::<Option<Json<Value>>, _>("samples_json")?
.map(|value| deserialize_json_value(value.0))
.transpose()?,
generated_draft: row
.try_get::<Option<Json<Value>>, _>("generated_draft_json")?
.map(|value| deserialize_json_value(value.0))
.transpose()?,
config_export: row
.try_get::<Option<Json<Value>>, _>("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<AuthProfile, RegistryError> {
Ok(AuthProfile {
id: crank_core::AuthProfileId::new(row.try_get::<String, _>("id")?),
workspace_id: WorkspaceId::new(row.try_get::<String, _>("workspace_id")?),
name: row.try_get("name")?,
kind: deserialize_enum_text(&row.try_get::<String, _>("kind")?, "kind")?,
config: deserialize_json_value(row.try_get::<Json<Value>, _>("config_json")?.0)?,
created_at: row.try_get("created_at")?,
updated_at: row.try_get("updated_at")?,
})
}
fn map_agent_binding(row: &PgRow) -> Result<AgentOperationBinding, RegistryError> {
Ok(AgentOperationBinding {
operation_id: OperationId::new(row.try_get::<String, _>("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<AgentOperationBinding>,
) -> Result<AgentVersionRecord, RegistryError> {
let version = from_db_version(row.try_get("version")?, "version")?;
let status = deserialize_enum_text(&row.try_get::<String, _>("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::<Json<Value>, _>("instructions_json")?.0,
tool_selection_policy: row
.try_get::<Json<Value>, _>("tool_selection_policy_json")?
.0,
created_at: row.try_get("created_at")?,
},
})
}
fn map_published_agent_tool(row: &PgRow) -> Result<PublishedAgentTool, RegistryError> {
let record = map_operation_version_record(row)?;
Ok(PublishedAgentTool {
workspace_id: WorkspaceId::new(row.try_get::<String, _>("workspace_id")?),
workspace_slug: row.try_get("workspace_slug")?,
agent_id: AgentId::new(row.try_get::<String, _>("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<OperationSampleMetadata, RegistryError> {
Ok(OperationSampleMetadata {
id: crank_core::SampleId::new(row.try_get::<String, _>("id")?),
operation_id: OperationId::new(row.try_get::<String, _>("operation_id")?),
version: from_db_version(row.try_get("version")?, "version")?,
sample_kind: deserialize_enum_text(
&row.try_get::<String, _>("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<DescriptorMetadata, RegistryError> {
Ok(DescriptorMetadata {
id: crank_core::DescriptorId::new(row.try_get::<String, _>("id")?),
operation_id: row
.try_get::<Option<String>, _>("operation_id")?
.map(OperationId::new),
version: row
.try_get::<Option<i32>, _>("version")?
.map(|value| from_db_version(value, "version"))
.transpose()?,
descriptor_kind: deserialize_enum_text(
&row.try_get::<String, _>("descriptor_kind")?,
"descriptor_kind",
)?,
storage_ref: row.try_get("storage_ref")?,
source_name: row.try_get("source_name")?,
package_index: row
.try_get::<Option<Json<Value>>, _>("package_index_json")?
.map(|value| value.0),
created_at: row.try_get("created_at")?,
})
}
fn map_yaml_import_job(row: &PgRow) -> Result<YamlImportJob, RegistryError> {
Ok(YamlImportJob {
id: YamlImportJobId::new(row.try_get::<String, _>("id")?),
source_sample_id: row
.try_get::<Option<String>, _>("source_sample_id")?
.map(crank_core::SampleId::new),
status: deserialize_enum_text(&row.try_get::<String, _>("status")?, "status")?,
format_version: row.try_get("format_version")?,
mode: deserialize_enum_text(&row.try_get::<String, _>("mode")?, "mode")?,
result_operation_id: row
.try_get::<Option<String>, _>("result_operation_id")?
.map(OperationId::new),
result_version: row
.try_get::<Option<i32>, _>("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<UsageOperationBreakdown, RegistryError> {
Ok(UsageOperationBreakdown {
operation_id: OperationId::new(row.try_get::<String, _>("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::<String, _>("protocol")?, "protocol")?,
calls_total: to_u64(row.try_get::<i64, _>("calls_total")?, "calls_total")?,
calls_error: to_u64(row.try_get::<i64, _>("calls_error")?, "calls_error")?,
p50_ms: to_u64(row.try_get::<i64, _>("p50_ms")?, "p50_ms")?,
p95_ms: to_u64(row.try_get::<i64, _>("p95_ms")?, "p95_ms")?,
p99_ms: to_u64(row.try_get::<i64, _>("p99_ms")?, "p99_ms")?,
})
}
fn map_usage_agent_breakdown(row: &PgRow) -> Result<UsageAgentBreakdown, RegistryError> {
Ok(UsageAgentBreakdown {
agent_id: AgentId::new(row.try_get::<String, _>("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::<i64, _>("calls_total")?, "calls_total")?,
calls_error: to_u64(row.try_get::<i64, _>("calls_error")?, "calls_error")?,
p50_ms: to_u64(row.try_get::<i64, _>("p50_ms")?, "p50_ms")?,
p95_ms: to_u64(row.try_get::<i64, _>("p95_ms")?, "p95_ms")?,
p99_ms: to_u64(row.try_get::<i64, _>("p99_ms")?, "p99_ms")?,
})
}
fn serialize_json_value<T: Serialize>(value: &T) -> Result<Value, RegistryError> {
Ok(serde_json::to_value(value)?)
}
fn serialize_option_json_value<T: Serialize>(
value: &Option<T>,
) -> Result<Option<Value>, RegistryError> {
value.as_ref().map(serialize_json_value).transpose()
}
fn deserialize_json_value<T: DeserializeOwned>(value: Value) -> Result<T, RegistryError> {
Ok(serde_json::from_value(value)?)
}
fn serialize_enum_text<T: Serialize>(
value: &T,
field: &'static str,
) -> Result<String, RegistryError> {
serde_json::to_value(value)?
.as_str()
.map(ToOwned::to_owned)
.ok_or(RegistryError::InvalidEnumRepresentation { field })
}
fn deserialize_enum_text<T: DeserializeOwned>(
value: &str,
field: &'static str,
) -> Result<T, RegistryError> {
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, RegistryError> {
u32::try_from(value).map_err(|_| RegistryError::InvalidNumericValue {
field,
value: i64::from(value),
})
}
fn to_u64(value: i64, field: &'static str) -> Result<u64, RegistryError> {
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,
}
}
}
Reference in New Issue View Git Blame Copy Permalink