bsodfather/crank
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add operations mutation api for alpine ui

Browse Source
This commit is contained in:
a.tolmachev
2026-03-30 00:42:41 +03:00
parent 4721bc1948
commit 23ca2cda59
14 changed files with 907 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/crank-registry/src/postgres.rs
+289 -17
View File
@@ -18,13 +18,14 @@ use crate::{
AgentSummary, AgentVersionRecord, CreateAgentRequest, CreateInvitationRequest,
CreateInvocationLogRequest, CreatePlatformApiKeyRequest, CreateVersionRequest,
CreateWorkspaceRequest, CreateYamlImportJobRequest, DescriptorMetadata, InvitationRecord,
InvocationLogRecord, ListInvocationLogsQuery, MembershipRecord, OperationSampleMetadata,
OperationSummary, OperationVersionRecord, PlatformApiKeyRecord, PublishAgentRequest,
PublishRequest, PublishedAgentTool, RegistryOperation, SaveAgentBindingsRequest,
SaveAuthProfileRequest, SaveDescriptorMetadataRequest, SaveSampleMetadataRequest,
UpdateWorkspaceRequest, UsageAgentBreakdown, UsageOperationBreakdown, UsageQuery,
UsageRollupRecord, UsageSummary, UsageTimelinePoint, WorkspaceRecord, YamlImportJob,
YamlImportJobCompletion, YamlImportJobId, YamlImportJobStatus,
InvocationLogRecord, ListInvocationLogsQuery, MembershipRecord, OperationAgentRef,
OperationSampleMetadata, OperationSummary, OperationUsageSummary, OperationVersionRecord,
PlatformApiKeyRecord, PublishAgentRequest, PublishRequest, PublishedAgentTool,
RegistryOperation, SaveAgentBindingsRequest, SaveAuthProfileRequest,
SaveDescriptorMetadataRequest, SaveSampleMetadataRequest, UpdateWorkspaceRequest,
UsageAgentBreakdown, UsageOperationBreakdown, UsageQuery, UsageRollupRecord, UsageSummary,
UsageTimelinePoint, WorkspaceRecord, YamlImportJob, YamlImportJobCompletion,
YamlImportJobId, YamlImportJobStatus,
},
};
@@ -1084,6 +1085,7 @@ impl PostgresRegistry {
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,
@@ -1192,6 +1194,7 @@ impl PostgresRegistry {
workspace_id,
name,
display_name,
category,
protocol,
status,
current_draft_version,
@@ -1200,16 +1203,17 @@ impl PostgresRegistry {
updated_at,
published_at
) values (
$1, $2, $3, $4, $5, $6, $7, $8,
$9::timestamptz,
$1, $2, $3, $4, $5, $6, $7, $8, $9,
$10::timestamptz,
$11::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))
@@ -1283,6 +1287,203 @@ impl PostgresRegistry {
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,
@@ -1293,6 +1494,7 @@ impl PostgresRegistry {
workspace_id,
name,
display_name,
category,
protocol,
status,
current_draft_version,
@@ -1311,6 +1513,57 @@ impl PostgresRegistry {
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,
@@ -1322,6 +1575,7 @@ impl PostgresRegistry {
workspace_id,
name,
display_name,
category,
protocol,
status,
current_draft_version,
@@ -1352,6 +1606,7 @@ impl PostgresRegistry {
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,
@@ -1395,6 +1650,7 @@ impl PostgresRegistry {
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,
@@ -1504,6 +1760,7 @@ impl PostgresRegistry {
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,
@@ -1545,6 +1802,7 @@ impl PostgresRegistry {
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,
@@ -2058,13 +2316,6 @@ fn assert_immutable_fields(
});
}
if summary.display_name != snapshot.display_name {
return Err(RegistryError::ImmutableOperationFieldChanged {
operation_id: snapshot.id.as_str().to_owned(),
field: "display_name",
});
}
if summary.protocol != snapshot.protocol {
return Err(RegistryError::ImmutableOperationFieldChanged {
operation_id: snapshot.id.as_str().to_owned(),
@@ -2201,6 +2452,7 @@ fn map_operation_summary(row: &PgRow) -> Result<OperationSummary, RegistryError>
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")?,
status: deserialize_enum_text(&row.try_get::<String, _>("status")?, "status")?,
current_draft_version: from_db_version(
@@ -2217,6 +2469,24 @@ fn map_operation_summary(row: &PgRow) -> Result<OperationSummary, RegistryError>
})
}
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 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")?;
@@ -2234,6 +2504,7 @@ fn map_operation_version_record(row: &PgRow) -> Result<OperationVersionRecord, R
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,
@@ -2809,6 +3080,7 @@ mod tests {
id: OperationId::new(id),
name: format!("{id}_tool"),
display_name: format!("Display {id}"),
category: "general".to_owned(),
protocol: Protocol::Rest,
status,
version,