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

merge: feat/workspace-foundation

Browse Source
This commit is contained in:
a.tolmachev
2026-03-29 22:12:17 +03:00
parent bf2d61f55b d757adb192
commit 873683cac6
19 changed files with 743 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TASKS.md
+5 -6
View File
@@ -2,23 +2,22 @@
## Current
### `feat/operations-workspace-contracts`
### `feat/workspace-foundation`
Status: completed
DoD:
- `workspace-scoped` контракты для `Operations` и `Wizard` зафиксированы
- определены точные DTO и lifecycle semantics для operations
- `PATCH/DELETE/ARCHIVE` и wizard DTO shape документированы
- добавлены `workspaces` и `workspace_id` в storage model
- `admin-api` переведен на `workspace-scoped` routes для operations и auth profiles
- тесты и registry migration проходят для workspace foundation
## Next
- `feat/workspace-foundation`
- `feat/agent-publishing`
## Backlog
- `feat/workspace-foundation`
- `feat/agent-publishing`
- `feat/platform-access`
- `feat/observability-api`
apps/admin-api/src/app.rs
+16 -3
View File
@@ -12,12 +12,13 @@ use crate::{
run_test, upload_descriptor_set, upload_input_json, upload_output_json,
upload_proto_descriptor,
},
workspaces::{create_workspace, get_workspace, list_workspaces, update_workspace},
},
state::AppState,
};
pub fn build_app(state: AppState) -> Router {
let admin_router = Router::new()
let workspace_router = Router::new()
.route("/operations", get(list_operations).post(create_operation))
.route("/operations/import", post(import_operation))
.route("/operations/{operation_id}", get(get_operation))
@@ -62,9 +63,16 @@ pub fn build_app(state: AppState) -> Router {
)
.route("/auth-profiles/{auth_profile_id}", get(get_auth_profile));
let admin_router = Router::new()
.route("/workspaces", get(list_workspaces).post(create_workspace))
.route(
"/workspaces/{workspace_id}",
get(get_workspace).patch(update_workspace),
)
.nest("/workspaces/{workspace_id}", workspace_router);
Router::new()
.route("/health", get(crate::routes::health))
.merge(admin_router.clone())
.nest("/api/admin", admin_router)
.with_state(state)
}
@@ -98,6 +106,8 @@ mod tests {
state::AppState,
};
const DEFAULT_WORKSPACE_ID: &str = "ws_default";
#[tokio::test]
async fn creates_publishes_and_tests_rest_operation() {
let registry = test_registry().await;
@@ -590,7 +600,10 @@ mod tests {
axum::serve(listener, app).await.unwrap();
});
format!("http://{}", address)
format!(
"http://{}/api/admin/workspaces/{}",
address, DEFAULT_WORKSPACE_ID
)
}
async fn create_lead(Json(payload): Json<Value>) -> Json<Value> {
apps/admin-api/src/error.rs
+6
View File
@@ -96,6 +96,9 @@ impl IntoResponse for ApiError {
impl From<RegistryError> for ApiError {
fn from(value: RegistryError) -> Self {
match value {
RegistryError::WorkspaceNotFound { workspace_id } => {
Self::not_found(format!("workspace {workspace_id} was not found"))
}
RegistryError::OperationNotFound { operation_id } => {
Self::not_found(format!("operation {operation_id} was not found"))
}
@@ -111,6 +114,9 @@ impl From<RegistryError> for ApiError {
RegistryError::OperationAlreadyExists { operation_id } => {
Self::conflict(format!("operation {operation_id} already exists"))
}
RegistryError::WorkspaceSlugAlreadyExists { slug } => {
Self::conflict(format!("workspace with slug {slug} already exists"))
}
RegistryError::InvalidInitialVersion { .. }
| RegistryError::InvalidVersionSequence { .. }
| RegistryError::ImmutableOperationFieldChanged { .. }
apps/admin-api/src/routes.rs
+1
View File
@@ -1,5 +1,6 @@
pub mod auth_profiles;
pub mod operations;
pub mod workspaces;
use axum::Json;
use serde_json::json;
apps/admin-api/src/routes/auth_profiles.rs
+30 -5
View File
@@ -2,30 +2,55 @@ use axum::{
Json,
extract::{Path, State},
};
use serde::Deserialize;
use serde_json::{Value, json};
use crate::{error::ApiError, service::AuthProfilePayload, state::AppState};
pub async fn list_auth_profiles(State(state): State<AppState>) -> Result<Json<Value>, ApiError> {
let items = state.service.list_auth_profiles().await?;
#[derive(Deserialize)]
pub struct WorkspacePath {
pub workspace_id: String,
}
#[derive(Deserialize)]
pub struct WorkspaceAuthProfilePath {
pub workspace_id: String,
pub auth_profile_id: String,
}
pub async fn list_auth_profiles(
Path(path): Path<WorkspacePath>,
State(state): State<AppState>,
) -> Result<Json<Value>, ApiError> {
let items = state
.service
.list_auth_profiles(&path.workspace_id.as_str().into())
.await?;
Ok(Json(json!({ "items": items })))
}
pub async fn create_auth_profile(
Path(path): Path<WorkspacePath>,
State(state): State<AppState>,
Json(payload): Json<AuthProfilePayload>,
) -> Result<Json<Value>, ApiError> {
let profile = state.service.create_auth_profile(payload).await?;
let profile = state
.service
.create_auth_profile(&path.workspace_id.as_str().into(), payload)
.await?;
Ok(Json(json!(profile)))
}
pub async fn get_auth_profile(
Path(auth_profile_id): Path<String>,
Path(path): Path<WorkspaceAuthProfilePath>,
State(state): State<AppState>,
) -> Result<Json<Value>, ApiError> {
let profile = state
.service
.get_auth_profile(&auth_profile_id.as_str().into())
.get_auth_profile(
&path.workspace_id.as_str().into(),
&path.auth_profile_id.as_str().into(),
)
.await?;
Ok(Json(json!(profile)))
}
apps/admin-api/src/routes/operations.rs
+96 -28
View File
@@ -5,6 +5,7 @@ use axum::{
http::{HeaderMap, StatusCode, header},
response::IntoResponse,
};
use serde::Deserialize;
use serde_json::{Value, json};
use crate::{
@@ -16,86 +17,134 @@ use crate::{
state::AppState,
};
pub async fn list_operations(State(state): State<AppState>) -> Result<Json<Value>, ApiError> {
let items = state.service.list_operations().await?;
#[derive(Deserialize)]
pub struct WorkspacePath {
pub workspace_id: String,
}
#[derive(Deserialize)]
pub struct WorkspaceOperationPath {
pub workspace_id: String,
pub operation_id: String,
}
#[derive(Deserialize)]
pub struct WorkspaceOperationVersionPath {
pub workspace_id: String,
pub operation_id: String,
pub version: u32,
}
pub async fn list_operations(
Path(path): Path<WorkspacePath>,
State(state): State<AppState>,
) -> Result<Json<Value>, ApiError> {
let items = state
.service
.list_operations(&path.workspace_id.as_str().into())
.await?;
Ok(Json(json!({ "items": items })))
}
pub async fn create_operation(
Path(path): Path<WorkspacePath>,
State(state): State<AppState>,
Json(payload): Json<OperationPayload>,
) -> Result<Json<Value>, ApiError> {
let created = state.service.create_operation(payload).await?;
let created = state
.service
.create_operation(&path.workspace_id.as_str().into(), payload)
.await?;
Ok(Json(json!(created)))
}
pub async fn get_operation(
Path(operation_id): Path<String>,
Path(path): Path<WorkspaceOperationPath>,
State(state): State<AppState>,
) -> Result<Json<Value>, ApiError> {
let operation = state
.service
.get_operation(&operation_id.as_str().into())
.get_operation(
&path.workspace_id.as_str().into(),
&path.operation_id.as_str().into(),
)
.await?;
Ok(Json(json!(operation)))
}
pub async fn get_operation_version(
Path((operation_id, version)): Path<(String, u32)>,
Path(path): Path<WorkspaceOperationVersionPath>,
State(state): State<AppState>,
) -> Result<Json<Value>, ApiError> {
let version = state
.service
.get_operation_version(&operation_id.as_str().into(), version)
.get_operation_version(
&path.workspace_id.as_str().into(),
&path.operation_id.as_str().into(),
path.version,
)
.await?;
Ok(Json(json!(version)))
}
pub async fn create_version(
Path(operation_id): Path<String>,
Path(path): Path<WorkspaceOperationPath>,
State(state): State<AppState>,
Json(payload): Json<NewVersionPayload>,
) -> Result<Json<Value>, ApiError> {
let created = state
.service
.create_version(&operation_id.as_str().into(), payload)
.create_version(
&path.workspace_id.as_str().into(),
&path.operation_id.as_str().into(),
payload,
)
.await?;
Ok(Json(json!(created)))
}
pub async fn publish_operation(
Path(operation_id): Path<String>,
Path(path): Path<WorkspaceOperationPath>,
State(state): State<AppState>,
Json(payload): Json<PublishPayload>,
) -> Result<Json<Value>, ApiError> {
let published = state
.service
.publish_operation(&operation_id.as_str().into(), payload.version)
.publish_operation(
&path.workspace_id.as_str().into(),
&path.operation_id.as_str().into(),
payload.version,
)
.await?;
Ok(Json(json!(published)))
}
pub async fn run_test(
Path(operation_id): Path<String>,
Path(path): Path<WorkspaceOperationPath>,
State(state): State<AppState>,
Json(payload): Json<TestRunPayload>,
) -> Result<Json<Value>, ApiError> {
let result = state
.service
.run_test(&operation_id.as_str().into(), payload)
.run_test(
&path.workspace_id.as_str().into(),
&path.operation_id.as_str().into(),
payload,
)
.await?;
Ok(Json(json!(result)))
}
pub async fn upload_input_json(
Path(operation_id): Path<String>,
Path(path): Path<WorkspaceOperationPath>,
State(state): State<AppState>,
Json(payload): Json<Value>,
) -> Result<Json<Value>, ApiError> {
let sample = state
.service
.save_json_sample(
&operation_id.as_str().into(),
&path.workspace_id.as_str().into(),
&path.operation_id.as_str().into(),
crank_registry::SampleKind::InputJson,
&payload,
)
@@ -109,14 +158,15 @@ pub async fn upload_input_json(
}
pub async fn upload_output_json(
Path(operation_id): Path<String>,
Path(path): Path<WorkspaceOperationPath>,
State(state): State<AppState>,
Json(payload): Json<Value>,
) -> Result<Json<Value>, ApiError> {
let sample = state
.service
.save_json_sample(
&operation_id.as_str().into(),
&path.workspace_id.as_str().into(),
&path.operation_id.as_str().into(),
crank_registry::SampleKind::OutputJson,
&payload,
)
@@ -130,7 +180,7 @@ pub async fn upload_output_json(
}
pub async fn upload_proto_descriptor(
Path(operation_id): Path<String>,
Path(path): Path<WorkspaceOperationPath>,
State(state): State<AppState>,
headers: HeaderMap,
body: Bytes,
@@ -139,7 +189,8 @@ pub async fn upload_proto_descriptor(
let descriptor = state
.service
.upload_proto_file(
&operation_id.as_str().into(),
&path.workspace_id.as_str().into(),
&path.operation_id.as_str().into(),
source_name.as_deref(),
body.as_ref(),
)
@@ -149,7 +200,7 @@ pub async fn upload_proto_descriptor(
}
pub async fn upload_descriptor_set(
Path(operation_id): Path<String>,
Path(path): Path<WorkspaceOperationPath>,
State(state): State<AppState>,
headers: HeaderMap,
body: Bytes,
@@ -158,7 +209,8 @@ pub async fn upload_descriptor_set(
let descriptor = state
.service
.upload_descriptor_set(
&operation_id.as_str().into(),
&path.workspace_id.as_str().into(),
&path.operation_id.as_str().into(),
source_name.as_deref(),
body.as_ref(),
)
@@ -168,37 +220,49 @@ pub async fn upload_descriptor_set(
}
pub async fn list_grpc_services(
Path(operation_id): Path<String>,
Path(path): Path<WorkspaceOperationPath>,
Query(query): Query<ExportQuery>,
State(state): State<AppState>,
) -> Result<Json<Value>, ApiError> {
let services = state
.service
.list_grpc_services(&operation_id.as_str().into(), query.version)
.list_grpc_services(
&path.workspace_id.as_str().into(),
&path.operation_id.as_str().into(),
query.version,
)
.await?;
Ok(Json(json!({ "services": services })))
}
pub async fn generate_draft(
Path(operation_id): Path<String>,
Path(path): Path<WorkspaceOperationPath>,
State(state): State<AppState>,
Json(payload): Json<GenerateDraftPayload>,
) -> Result<Json<Value>, ApiError> {
let draft = state
.service
.generate_draft(&operation_id.as_str().into(), payload)
.generate_draft(
&path.workspace_id.as_str().into(),
&path.operation_id.as_str().into(),
payload,
)
.await?;
Ok(Json(json!(draft)))
}
pub async fn export_operation(
Path(operation_id): Path<String>,
Path(path): Path<WorkspaceOperationPath>,
Query(query): Query<ExportQuery>,
State(state): State<AppState>,
) -> Result<impl IntoResponse, ApiError> {
let yaml = state
.service
.export_operation(&operation_id.as_str().into(), query)
.export_operation(
&path.workspace_id.as_str().into(),
&path.operation_id.as_str().into(),
query,
)
.await?;
let mut headers = HeaderMap::new();
headers.insert(
@@ -210,11 +274,15 @@ pub async fn export_operation(
}
pub async fn import_operation(
Path(path): Path<WorkspacePath>,
Query(query): Query<ImportQuery>,
State(state): State<AppState>,
body: String,
) -> Result<Json<Value>, ApiError> {
let imported = state.service.import_operation(query, &body).await?;
let imported = state
.service
.import_operation(&path.workspace_id.as_str().into(), query, &body)
.await?;
Ok(Json(json!(imported)))
}
apps/admin-api/src/routes/workspaces.rs
+47
View File
@@ -0,0 +1,47 @@
use axum::{
Json,
extract::{Path, State},
};
use serde_json::{Value, json};
use crate::{
error::ApiError,
service::{UpdateWorkspacePayload, WorkspacePayload},
state::AppState,
};
pub async fn list_workspaces(State(state): State<AppState>) -> Result<Json<Value>, ApiError> {
let items = state.service.list_workspaces().await?;
Ok(Json(json!({ "items": items })))
}
pub async fn create_workspace(
State(state): State<AppState>,
Json(payload): Json<WorkspacePayload>,
) -> Result<Json<Value>, ApiError> {
let workspace = state.service.create_workspace(payload).await?;
Ok(Json(json!(workspace)))
}
pub async fn get_workspace(
Path(workspace_id): Path<String>,
State(state): State<AppState>,
) -> Result<Json<Value>, ApiError> {
let workspace = state
.service
.get_workspace(&workspace_id.as_str().into())
.await?;
Ok(Json(json!(workspace)))
}
pub async fn update_workspace(
Path(workspace_id): Path<String>,
State(state): State<AppState>,
Json(payload): Json<UpdateWorkspacePayload>,
) -> Result<Json<Value>, ApiError> {
let workspace = state
.service
.update_workspace(&workspace_id.as_str().into(), payload)
.await?;
Ok(Json(json!(workspace)))
}
apps/admin-api/src/service.rs
+165 -26
View File
@@ -3,13 +3,15 @@ use std::path::PathBuf;
use crank_core::{
AuthConfig, AuthKind, AuthProfile, AuthProfileId, ConfigExport, ExportMode, GeneratedDraft,
GeneratedDraftStatus, OperationId, OperationStatus, Protocol, SampleId, Samples, Target,
Workspace, WorkspaceId, WorkspaceStatus,
};
use crank_mapping::{JsonPathRoot, MappingSet, infer_mapping_from_samples};
use crank_proto::{ProtoService, services_from_descriptor_set_bytes};
use crank_registry::{
CreateVersionRequest, OperationSampleMetadata, OperationSummary, OperationVersionRecord,
PostgresRegistry, PublishRequest, RegistryOperation, SampleKind, SaveAuthProfileRequest,
SaveDescriptorMetadataRequest, SaveSampleMetadataRequest,
CreateVersionRequest, CreateWorkspaceRequest, OperationSampleMetadata, OperationSummary,
OperationVersionRecord, PostgresRegistry, PublishRequest, RegistryOperation, SampleKind,
SaveAuthProfileRequest, SaveDescriptorMetadataRequest, SaveSampleMetadataRequest,
UpdateWorkspaceRequest, WorkspaceRecord,
};
use crank_runtime::{PreparedRequest, RuntimeError, RuntimeExecutor, RuntimeOperation};
use crank_schema::Schema;
@@ -76,6 +78,22 @@ pub struct AuthProfilePayload {
pub config: AuthConfig,
}
#[derive(Clone, Debug, Deserialize)]
pub struct WorkspacePayload {
pub slug: String,
pub display_name: String,
#[serde(default)]
pub settings: Value,
}
#[derive(Clone, Debug, Deserialize)]
pub struct UpdateWorkspacePayload {
pub slug: Option<String>,
pub display_name: Option<String>,
pub status: Option<WorkspaceStatus>,
pub settings: Option<Value>,
}
#[derive(Clone, Debug, Deserialize)]
pub struct GenerateDraftPayload {
#[serde(default)]
@@ -122,6 +140,7 @@ pub struct YamlOperationDocument {
#[derive(Clone, Debug, Serialize)]
pub struct CreatedOperationResponse {
pub operation_id: String,
pub workspace_id: String,
pub version: u32,
pub status: OperationStatus,
}
@@ -129,6 +148,7 @@ pub struct CreatedOperationResponse {
#[derive(Clone, Debug, Serialize)]
pub struct PublishResponse {
pub operation_id: String,
pub workspace_id: String,
pub published_version: u32,
pub published_at: String,
}
@@ -136,6 +156,7 @@ pub struct PublishResponse {
#[derive(Clone, Debug, Serialize)]
pub struct ImportResponse {
pub operation_id: String,
pub workspace_id: String,
pub version: u32,
pub import_mode: ImportMode,
pub warnings: Vec<String>,
@@ -172,17 +193,89 @@ impl AdminService {
}
#[instrument(skip(self))]
pub async fn list_operations(&self) -> Result<Vec<OperationSummary>, ApiError> {
Ok(self.registry.list_operations().await?)
pub async fn list_workspaces(&self) -> Result<Vec<WorkspaceRecord>, ApiError> {
Ok(self.registry.list_workspaces().await?)
}
#[instrument(skip(self, payload), fields(workspace_slug = %payload.slug))]
pub async fn create_workspace(
&self,
payload: WorkspacePayload,
) -> Result<WorkspaceRecord, ApiError> {
let now = now_string()?;
let workspace = Workspace {
id: WorkspaceId::new(new_prefixed_id("ws")),
slug: payload.slug,
display_name: payload.display_name,
status: WorkspaceStatus::Active,
settings: payload.settings,
created_at: now.clone(),
updated_at: now,
};
self.registry
.create_workspace(CreateWorkspaceRequest {
workspace: &workspace,
})
.await?;
Ok(WorkspaceRecord { workspace })
}
pub async fn get_workspace(
&self,
workspace_id: &WorkspaceId,
) -> Result<WorkspaceRecord, ApiError> {
self.registry
.get_workspace(workspace_id)
.await?
.ok_or_else(|| {
ApiError::not_found(format!("workspace {} was not found", workspace_id.as_str()))
})
}
#[instrument(skip(self, payload), fields(workspace_id = %workspace_id.as_str()))]
pub async fn update_workspace(
&self,
workspace_id: &WorkspaceId,
payload: UpdateWorkspacePayload,
) -> Result<WorkspaceRecord, ApiError> {
let existing = self.get_workspace(workspace_id).await?.workspace;
let workspace = Workspace {
id: existing.id,
slug: payload.slug.unwrap_or(existing.slug),
display_name: payload.display_name.unwrap_or(existing.display_name),
status: payload.status.unwrap_or(existing.status),
settings: payload.settings.unwrap_or(existing.settings),
created_at: existing.created_at,
updated_at: now_string()?,
};
self.registry
.update_workspace(UpdateWorkspaceRequest {
workspace: &workspace,
})
.await?;
Ok(WorkspaceRecord { workspace })
}
pub async fn list_operations(
&self,
workspace_id: &WorkspaceId,
) -> Result<Vec<OperationSummary>, ApiError> {
self.ensure_workspace_exists(workspace_id).await?;
Ok(self.registry.list_operations(workspace_id).await?)
}
#[instrument(skip(self))]
pub async fn get_operation(
&self,
workspace_id: &WorkspaceId,
operation_id: &OperationId,
) -> Result<OperationSummary, ApiError> {
self.registry
.get_operation_summary(operation_id)
.get_operation_summary(workspace_id, operation_id)
.await?
.ok_or_else(|| {
ApiError::not_found(format!("operation {} was not found", operation_id.as_str()))
@@ -192,11 +285,12 @@ impl AdminService {
#[instrument(skip(self))]
pub async fn get_operation_version(
&self,
workspace_id: &WorkspaceId,
operation_id: &OperationId,
version: u32,
) -> Result<OperationVersionRecord, ApiError> {
self.registry
.get_operation_version(operation_id, version)
.get_operation_version(workspace_id, operation_id, version)
.await?
.ok_or_else(|| {
ApiError::not_found(format!(
@@ -209,11 +303,17 @@ impl AdminService {
#[instrument(skip(self, payload), fields(protocol = ?payload.protocol, operation_name = %payload.name))]
pub async fn create_operation(
&self,
workspace_id: &WorkspaceId,
payload: OperationPayload,
) -> Result<CreatedOperationResponse, ApiError> {
self.validate_operation_payload(&payload)?;
self.ensure_workspace_exists(workspace_id).await?;
if self.find_operation_by_name(&payload.name).await?.is_some() {
if self
.find_operation_by_name(workspace_id, &payload.name)
.await?
.is_some()
{
return Err(ApiError::conflict(format!(
"operation with name {} already exists",
payload.name
@@ -247,11 +347,14 @@ impl AdminService {
published_at: None,
};
self.registry.create_operation(&snapshot, None).await?;
self.registry
.create_operation(workspace_id, &snapshot, None)
.await?;
info!(operation_id = %operation_id.as_str(), version = 1, "operation created");
Ok(CreatedOperationResponse {
operation_id: operation_id.as_str().to_owned(),
workspace_id: workspace_id.as_str().to_owned(),
version: 1,
status: OperationStatus::Draft,
})
@@ -260,12 +363,13 @@ impl AdminService {
#[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), protocol = ?payload.operation.protocol))]
pub async fn create_version(
&self,
workspace_id: &WorkspaceId,
operation_id: &OperationId,
payload: NewVersionPayload,
) -> Result<CreatedOperationResponse, ApiError> {
self.validate_operation_payload(&payload.operation)?;
let summary = self.get_operation(operation_id).await?;
let summary = self.get_operation(workspace_id, operation_id).await?;
let now = now_string()?;
let version = summary.current_draft_version + 1;
let snapshot = RegistryOperation {
@@ -295,6 +399,7 @@ impl AdminService {
self.registry
.create_version(CreateVersionRequest {
workspace_id,
snapshot: &snapshot,
change_note: payload.change_note.as_deref(),
created_by: None,
@@ -304,6 +409,7 @@ impl AdminService {
Ok(CreatedOperationResponse {
operation_id: operation_id.as_str().to_owned(),
workspace_id: workspace_id.as_str().to_owned(),
version,
status: OperationStatus::Draft,
})
@@ -312,12 +418,14 @@ impl AdminService {
#[instrument(skip(self), fields(operation_id = %operation_id.as_str(), version))]
pub async fn publish_operation(
&self,
workspace_id: &WorkspaceId,
operation_id: &OperationId,
version: u32,
) -> Result<PublishResponse, ApiError> {
let published_at = now_string()?;
self.registry
.publish_operation(PublishRequest {
workspace_id,
operation_id,
version,
published_at: &published_at,
@@ -328,6 +436,7 @@ impl AdminService {
Ok(PublishResponse {
operation_id: operation_id.as_str().to_owned(),
workspace_id: workspace_id.as_str().to_owned(),
published_version: version,
published_at,
})
@@ -336,11 +445,12 @@ impl AdminService {
#[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), version = payload.version))]
pub async fn run_test(
&self,
workspace_id: &WorkspaceId,
operation_id: &OperationId,
payload: TestRunPayload,
) -> Result<TestRunResult, ApiError> {
let record = self
.get_operation_version(operation_id, payload.version)
.get_operation_version(workspace_id, operation_id, payload.version)
.await?;
let runtime = RuntimeOperation::from(record.snapshot.clone());
let request_preview =
@@ -375,18 +485,23 @@ impl AdminService {
}
#[instrument(skip(self))]
pub async fn list_auth_profiles(&self) -> Result<Vec<AuthProfile>, ApiError> {
Ok(self.registry.list_auth_profiles().await?)
pub async fn list_auth_profiles(
&self,
workspace_id: &WorkspaceId,
) -> Result<Vec<AuthProfile>, ApiError> {
self.ensure_workspace_exists(workspace_id).await?;
Ok(self.registry.list_auth_profiles(workspace_id).await?)
}
#[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), source_name = source_name.unwrap_or("descriptor-set.bin")))]
pub async fn upload_descriptor_set(
&self,
workspace_id: &WorkspaceId,
operation_id: &OperationId,
source_name: Option<&str>,
payload: &[u8],
) -> Result<DescriptorUploadResponse, ApiError> {
let summary = self.get_operation(operation_id).await?;
let summary = self.get_operation(workspace_id, operation_id).await?;
if summary.protocol != Protocol::Grpc {
return Err(ApiError::validation(
"descriptor upload is only allowed for grpc operations",
@@ -441,11 +556,12 @@ impl AdminService {
#[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), source_name = source_name.unwrap_or("schema.proto")))]
pub async fn upload_proto_file(
&self,
workspace_id: &WorkspaceId,
operation_id: &OperationId,
source_name: Option<&str>,
payload: &[u8],
) -> Result<DescriptorUploadResponse, ApiError> {
let summary = self.get_operation(operation_id).await?;
let summary = self.get_operation(workspace_id, operation_id).await?;
if summary.protocol != Protocol::Grpc {
return Err(ApiError::validation(
"proto upload is only allowed for grpc operations",
@@ -495,10 +611,11 @@ impl AdminService {
#[instrument(skip(self), fields(operation_id = %operation_id.as_str(), version))]
pub async fn list_grpc_services(
&self,
workspace_id: &WorkspaceId,
operation_id: &OperationId,
version: Option<u32>,
) -> Result<Vec<GrpcServiceSummary>, ApiError> {
let summary = self.get_operation(operation_id).await?;
let summary = self.get_operation(workspace_id, operation_id).await?;
if summary.protocol != Protocol::Grpc {
return Err(ApiError::validation(
"grpc services are only available for grpc operations",
@@ -529,10 +646,11 @@ impl AdminService {
#[instrument(skip(self))]
pub async fn get_auth_profile(
&self,
workspace_id: &WorkspaceId,
auth_profile_id: &AuthProfileId,
) -> Result<AuthProfile, ApiError> {
self.registry
.get_auth_profile(auth_profile_id)
.get_auth_profile(workspace_id, auth_profile_id)
.await?
.ok_or_else(|| {
ApiError::not_found(format!(
@@ -545,13 +663,16 @@ impl AdminService {
#[instrument(skip(self, payload), fields(auth_profile_name = %payload.name, auth_kind = ?payload.kind))]
pub async fn create_auth_profile(
&self,
workspace_id: &WorkspaceId,
payload: AuthProfilePayload,
) -> Result<AuthProfile, ApiError> {
validate_auth_profile_kind(payload.kind, &payload.config)?;
self.ensure_workspace_exists(workspace_id).await?;
let now = now_string()?;
let profile = AuthProfile {
id: AuthProfileId::new(new_prefixed_id("auth")),
workspace_id: workspace_id.clone(),
name: payload.name,
kind: payload.kind,
config: payload.config,
@@ -560,7 +681,10 @@ impl AdminService {
};
self.registry
.save_auth_profile(SaveAuthProfileRequest { profile: &profile })
.save_auth_profile(SaveAuthProfileRequest {
workspace_id,
profile: &profile,
})
.await?;
info!(auth_profile_id = %profile.id.as_str(), "auth profile created");
@@ -570,18 +694,21 @@ impl AdminService {
#[instrument(skip(self), fields(operation_id = %operation_id.as_str(), version = query.version.unwrap_or_default(), mode = ?query.mode))]
pub async fn export_operation(
&self,
workspace_id: &WorkspaceId,
operation_id: &OperationId,
query: ExportQuery,
) -> Result<String, ApiError> {
let version = match query.version {
Some(version) => version,
None => {
self.get_operation(operation_id)
self.get_operation(workspace_id, operation_id)
.await?
.current_draft_version
}
};
let record = self.get_operation_version(operation_id, version).await?;
let record = self
.get_operation_version(workspace_id, operation_id, version)
.await?;
let document = YamlOperationDocument {
format_version: "1".to_owned(),
kind: "operation".to_owned(),
@@ -600,6 +727,7 @@ impl AdminService {
#[instrument(skip(self, yaml_document), fields(mode = ?query.mode))]
pub async fn import_operation(
&self,
workspace_id: &WorkspaceId,
query: ImportQuery,
yaml_document: &str,
) -> Result<ImportResponse, ApiError> {
@@ -624,9 +752,10 @@ impl AdminService {
match query.mode {
ImportMode::Create => {
let created = self.create_operation(payload).await?;
let created = self.create_operation(workspace_id, payload).await?;
Ok(ImportResponse {
operation_id: created.operation_id,
workspace_id: created.workspace_id,
version: created.version,
import_mode: ImportMode::Create,
warnings: Vec::new(),
@@ -634,11 +763,12 @@ impl AdminService {
}
ImportMode::Upsert => {
if let Some(existing) = self
.find_operation_by_name(&document.operation.name)
.find_operation_by_name(workspace_id, &document.operation.name)
.await?
{
let created = self
.create_version(
workspace_id,
&existing.id,
NewVersionPayload {
operation: payload,
@@ -649,6 +779,7 @@ impl AdminService {
let response = ImportResponse {
operation_id: created.operation_id,
workspace_id: created.workspace_id,
version: created.version,
import_mode: ImportMode::Upsert,
warnings: Vec::new(),
@@ -660,9 +791,10 @@ impl AdminService {
);
Ok(response)
} else {
let created = self.create_operation(payload).await?;
let created = self.create_operation(workspace_id, payload).await?;
let response = ImportResponse {
operation_id: created.operation_id,
workspace_id: created.workspace_id,
version: created.version,
import_mode: ImportMode::Upsert,
warnings: Vec::new(),
@@ -681,11 +813,12 @@ impl AdminService {
#[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), sample_kind = ?sample_kind))]
pub async fn save_json_sample(
&self,
workspace_id: &WorkspaceId,
operation_id: &OperationId,
sample_kind: SampleKind,
payload: &Value,
) -> Result<OperationSampleMetadata, ApiError> {
let summary = self.get_operation(operation_id).await?;
let summary = self.get_operation(workspace_id, operation_id).await?;
let version = summary.current_draft_version;
let sample_id = SampleId::new(new_prefixed_id("sample"));
let now = now_string()?;
@@ -725,10 +858,11 @@ impl AdminService {
#[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str()))]
pub async fn generate_draft(
&self,
workspace_id: &WorkspaceId,
operation_id: &OperationId,
payload: GenerateDraftPayload,
) -> Result<DraftGenerationResult, ApiError> {
let summary = self.get_operation(operation_id).await?;
let summary = self.get_operation(workspace_id, operation_id).await?;
let samples = self
.registry
.list_sample_metadata(operation_id, summary.current_draft_version)
@@ -795,15 +929,20 @@ impl AdminService {
async fn find_operation_by_name(
&self,
workspace_id: &WorkspaceId,
name: &str,
) -> Result<Option<OperationSummary>, ApiError> {
Ok(self
.registry
.list_operations()
.list_operations(workspace_id)
.await?
.into_iter()
.find(|operation| operation.name == name))
}
async fn ensure_workspace_exists(&self, workspace_id: &WorkspaceId) -> Result<(), ApiError> {
self.get_workspace(workspace_id).await.map(|_| ())
}
}
fn build_request_preview(
apps/mcp-server/src/main.rs
+13 -4
View File
@@ -53,6 +53,7 @@ mod tests {
use crank_core::{
DescriptorId, ExecutionConfig, GraphqlOperationType, GraphqlTarget, GrpcTarget, HttpMethod,
Operation, OperationId, OperationStatus, Protocol, RestTarget, Target, ToolDescription,
WorkspaceId,
};
use crank_mapping::{MappingRule, MappingSet};
use crank_registry::{PostgresRegistry, PublishRequest};
@@ -63,6 +64,10 @@ mod tests {
use crate::app::build_app;
fn test_workspace_id() -> WorkspaceId {
WorkspaceId::new("ws_default")
}
#[tokio::test]
async fn initializes_lists_and_calls_published_tool_via_mcp() {
let registry = test_registry().await;
@@ -70,11 +75,12 @@ mod tests {
let operation = test_operation(&upstream_base_url, "crm_create_lead");
registry
.create_operation(&operation, Some("alice"))
.create_operation(&test_workspace_id(), &operation, Some("alice"))
.await
.unwrap();
registry
.publish_operation(PublishRequest {
workspace_id: &test_workspace_id(),
operation_id: &operation.id,
version: 1,
published_at: "2026-03-26T10:00:00Z",
@@ -137,11 +143,12 @@ mod tests {
let operation = test_graphql_operation(&endpoint, "crm_create_lead_graphql");
registry
.create_operation(&operation, Some("alice"))
.create_operation(&test_workspace_id(), &operation, Some("alice"))
.await
.unwrap();
registry
.publish_operation(PublishRequest {
workspace_id: &test_workspace_id(),
operation_id: &operation.id,
version: 1,
published_at: "2026-03-26T10:00:00Z",
@@ -191,11 +198,12 @@ mod tests {
let operation = test_grpc_operation(&server_addr, "echo_unary_grpc");
registry
.create_operation(&operation, Some("alice"))
.create_operation(&test_workspace_id(), &operation, Some("alice"))
.await
.unwrap();
registry
.publish_operation(PublishRequest {
workspace_id: &test_workspace_id(),
operation_id: &operation.id,
version: 1,
published_at: "2026-03-26T10:00:00Z",
@@ -318,11 +326,12 @@ mod tests {
let operation = test_operation(&upstream_base_url, "crm_publish_later");
registry
.create_operation(&operation, Some("alice"))
.create_operation(&test_workspace_id(), &operation, Some("alice"))
.await
.unwrap();
registry
.publish_operation(PublishRequest {
workspace_id: &test_workspace_id(),
operation_id: &operation.id,
version: 1,
published_at: "2026-03-26T10:00:00Z",
crates/crank-core/src/auth.rs
+5 -1
View File
@@ -1,6 +1,9 @@
use serde::{Deserialize, Serialize};
use crate::{ids::AuthProfileId, protocol::AuthKind};
use crate::{
ids::{AuthProfileId, WorkspaceId},
protocol::AuthKind,
};
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct SecretRef(pub String);
@@ -51,6 +54,7 @@ pub enum AuthConfig {
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct AuthProfile {
pub id: AuthProfileId,
pub workspace_id: WorkspaceId,
pub name: String,
pub kind: AuthKind,
pub config: AuthConfig,
crates/crank-core/src/ids.rs
+2
View File
@@ -40,3 +40,5 @@ define_id!(DescriptorId);
define_id!(ToolId);
define_id!(SampleId);
define_id!(AuthProfileId);
define_id!(WorkspaceId);
define_id!(UserId);
crates/crank-core/src/lib.rs
+3 -1
View File
@@ -2,15 +2,17 @@ pub mod auth;
pub mod ids;
pub mod operation;
pub mod protocol;
pub mod workspace;
pub use auth::{
ApiKeyHeaderAuthConfig, ApiKeyQueryAuthConfig, AuthConfig, AuthProfile, BasicAuthConfig,
BearerAuthConfig, SecretRef,
};
pub use ids::{AuthProfileId, DescriptorId, OperationId, SampleId, ToolId};
pub use ids::{AuthProfileId, DescriptorId, OperationId, SampleId, ToolId, UserId, WorkspaceId};
pub use operation::{
ConfigExport, ExecutionConfig, GeneratedDraft, GeneratedDraftStatus, GraphqlTarget,
GrpcProtocolOptions, GrpcTarget, Operation, OperationStatus, ProtocolOptions, RestTarget,
RetryPolicy, Samples, Target, ToolDescription, ToolExample,
};
pub use protocol::{AuthKind, ExportMode, GraphqlOperationType, HttpMethod, Protocol};
pub use workspace::{Workspace, WorkspaceStatus};
crates/crank-core/src/operation.rs
+1
View File
@@ -289,6 +289,7 @@ mod tests {
fn auth_profile_serializes_secret_refs_without_secret_values() {
let profile = AuthProfile {
id: AuthProfileId::new("auth_01"),
workspace_id: crate::ids::WorkspaceId::new("ws_01"),
name: "crm-prod-bearer".to_owned(),
kind: AuthKind::Bearer,
config: AuthConfig::Bearer(BearerAuthConfig {
crates/crank-core/src/workspace.rs
+22
View File
@@ -0,0 +1,22 @@
use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::ids::WorkspaceId;
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum WorkspaceStatus {
Active,
Archived,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Workspace {
pub id: WorkspaceId,
pub slug: String,
pub display_name: String,
pub status: WorkspaceStatus,
pub settings: Value,
pub created_at: String,
pub updated_at: String,
}
crates/crank-registry/src/error.rs
+4
View File
@@ -6,6 +6,10 @@ pub enum RegistryError {
Storage(#[from] sqlx::Error),
#[error(transparent)]
Serialization(#[from] serde_json::Error),
#[error("workspace {workspace_id} was not found")]
WorkspaceNotFound { workspace_id: String },
#[error("workspace with slug {slug} already exists")]
WorkspaceSlugAlreadyExists { slug: String },
#[error("operation {operation_id} already exists")]
OperationAlreadyExists { operation_id: String },
#[error("operation {operation_id} was not found")]
crates/crank-registry/src/lib.rs
+5 -5
View File
@@ -5,10 +5,10 @@ mod postgres;
pub use error::RegistryError;
pub use model::{
CreateVersionRequest, CreateYamlImportJobRequest, DescriptorKind, DescriptorMetadata,
OperationSampleMetadata, OperationSummary, OperationVersionRecord, PublishRequest,
RegistryOperation, SampleKind, SaveAuthProfileRequest, SaveDescriptorMetadataRequest,
SaveSampleMetadataRequest, YamlImportJob, YamlImportJobCompletion, YamlImportJobId,
YamlImportJobStatus,
CreateVersionRequest, CreateWorkspaceRequest, CreateYamlImportJobRequest, DescriptorKind,
DescriptorMetadata, OperationSampleMetadata, OperationSummary, OperationVersionRecord,
PublishRequest, RegistryOperation, SampleKind, SaveAuthProfileRequest,
SaveDescriptorMetadataRequest, SaveSampleMetadataRequest, UpdateWorkspaceRequest,
WorkspaceRecord, YamlImportJob, YamlImportJobCompletion, YamlImportJobId, YamlImportJobStatus,
};
pub use postgres::PostgresRegistry;
crates/crank-registry/src/migrations.rs
+77 -2
View File
@@ -1,10 +1,48 @@
use sqlx::{PgPool, query};
pub async fn apply_postgres(pool: &PgPool) -> Result<(), sqlx::Error> {
query(
"create table if not exists workspaces (
id text primary key,
slug text not null unique,
display_name text not null,
status text not null,
settings_json jsonb not null default '{}'::jsonb,
created_at timestamptz not null,
updated_at timestamptz not null
)",
)
.execute(pool)
.await?;
query(
"insert into workspaces (
id,
slug,
display_name,
status,
settings_json,
created_at,
updated_at
) values (
'ws_default',
'default',
'Default Workspace',
'active',
'{}'::jsonb,
now(),
now()
)
on conflict (id) do nothing",
)
.execute(pool)
.await?;
query(
"create table if not exists operations (
id text primary key,
name text not null unique,
workspace_id text null references workspaces(id) on delete cascade,
name text not null,
display_name text not null,
protocol text not null,
status text not null,
@@ -18,6 +56,24 @@ pub async fn apply_postgres(pool: &PgPool) -> Result<(), sqlx::Error> {
.execute(pool)
.await?;
query("alter table operations add column if not exists workspace_id text null references workspaces(id) on delete cascade")
.execute(pool)
.await?;
query("update operations set workspace_id = 'ws_default' where workspace_id is null")
.execute(pool)
.await?;
query("alter table operations alter column workspace_id set not null")
.execute(pool)
.await?;
query("alter table operations drop constraint if exists operations_name_key")
.execute(pool)
.await?;
query(
"create unique index if not exists operations_workspace_name_idx on operations(workspace_id, name)",
)
.execute(pool)
.await?;
query(
"create table if not exists operation_versions (
operation_id text not null references operations(id) on delete cascade,
@@ -89,7 +145,8 @@ pub async fn apply_postgres(pool: &PgPool) -> Result<(), sqlx::Error> {
query(
"create table if not exists auth_profiles (
id text primary key,
name text not null unique,
workspace_id text null references workspaces(id) on delete cascade,
name text not null,
kind text not null,
config_json jsonb not null,
created_at timestamptz not null,
@@ -99,6 +156,24 @@ pub async fn apply_postgres(pool: &PgPool) -> Result<(), sqlx::Error> {
.execute(pool)
.await?;
query("alter table auth_profiles add column if not exists workspace_id text null references workspaces(id) on delete cascade")
.execute(pool)
.await?;
query("update auth_profiles set workspace_id = 'ws_default' where workspace_id is null")
.execute(pool)
.await?;
query("alter table auth_profiles alter column workspace_id set not null")
.execute(pool)
.await?;
query("alter table auth_profiles drop constraint if exists auth_profiles_name_key")
.execute(pool)
.await?;
query(
"create unique index if not exists auth_profiles_workspace_name_idx on auth_profiles(workspace_id, name)",
)
.execute(pool)
.await?;
query(
"create table if not exists yaml_import_jobs (
id text primary key,
crates/crank-registry/src/model.rs
+21 -1
View File
@@ -1,6 +1,6 @@
use crank_core::{
AuthProfile, DescriptorId, ExportMode, Operation, OperationId, OperationStatus, Protocol,
SampleId,
SampleId, Workspace, WorkspaceId,
};
use crank_mapping::MappingSet;
use crank_schema::Schema;
@@ -40,9 +40,15 @@ define_registry_id!(YamlImportJobId);
pub type RegistryOperation = Operation<Schema, MappingSet>;
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct WorkspaceRecord {
pub workspace: Workspace,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct OperationSummary {
pub id: OperationId,
pub workspace_id: WorkspaceId,
pub name: String,
pub display_name: String,
pub protocol: Protocol,
@@ -57,6 +63,7 @@ pub struct OperationSummary {
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct OperationVersionRecord {
pub operation_id: OperationId,
pub workspace_id: WorkspaceId,
pub version: u32,
pub status: OperationStatus,
pub change_note: Option<String>,
@@ -138,6 +145,7 @@ pub struct YamlImportJobCompletion {
#[derive(Clone, Debug, PartialEq)]
pub struct CreateVersionRequest<'a> {
pub workspace_id: &'a WorkspaceId,
pub snapshot: &'a RegistryOperation,
pub change_note: Option<&'a str>,
pub created_by: Option<&'a str>,
@@ -145,6 +153,7 @@ pub struct CreateVersionRequest<'a> {
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct PublishRequest<'a> {
pub workspace_id: &'a WorkspaceId,
pub operation_id: &'a OperationId,
pub version: u32,
pub published_at: &'a str,
@@ -162,9 +171,20 @@ pub struct CreateYamlImportJobRequest<'a> {
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SaveAuthProfileRequest<'a> {
pub workspace_id: &'a WorkspaceId,
pub profile: &'a AuthProfile,
}
#[derive(Clone, Debug, PartialEq)]
pub struct CreateWorkspaceRequest<'a> {
pub workspace: &'a Workspace,
}
#[derive(Clone, Debug, PartialEq)]
pub struct UpdateWorkspaceRequest<'a> {
pub workspace: &'a Workspace,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SaveSampleMetadataRequest<'a> {
pub sample: &'a OperationSampleMetadata,
crates/crank-registry/src/postgres.rs
+224 -27
View File
@@ -1,4 +1,4 @@
use crank_core::{AuthProfile, OperationId, OperationStatus};
use crank_core::{AuthProfile, OperationId, OperationStatus, Workspace, WorkspaceId};
use serde::{Serialize, de::DeserializeOwned};
use serde_json::Value;
use sqlx::{
@@ -11,11 +11,11 @@ use crate::{
error::RegistryError,
migrations,
model::{
CreateVersionRequest, CreateYamlImportJobRequest, DescriptorMetadata,
OperationSampleMetadata, OperationSummary, OperationVersionRecord, PublishRequest,
RegistryOperation, SaveAuthProfileRequest, SaveDescriptorMetadataRequest,
SaveSampleMetadataRequest, YamlImportJob, YamlImportJobCompletion, YamlImportJobId,
YamlImportJobStatus,
CreateVersionRequest, CreateWorkspaceRequest, CreateYamlImportJobRequest,
DescriptorMetadata, OperationSampleMetadata, OperationSummary, OperationVersionRecord,
PublishRequest, RegistryOperation, SaveAuthProfileRequest, SaveDescriptorMetadataRequest,
SaveSampleMetadataRequest, UpdateWorkspaceRequest, WorkspaceRecord, YamlImportJob,
YamlImportJobCompletion, YamlImportJobId, YamlImportJobStatus,
},
};
@@ -38,6 +38,126 @@ impl PostgresRegistry {
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 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)),
}
}
async fn connect_in_schema(
database_url: &str,
schema: Option<&str>,
@@ -60,6 +180,7 @@ impl PostgresRegistry {
pub async fn create_operation(
&self,
workspace_id: &WorkspaceId,
snapshot: &RegistryOperation,
created_by: Option<&str>,
) -> Result<(), RegistryError> {
@@ -70,7 +191,11 @@ impl PostgresRegistry {
});
}
if self.get_operation_summary(&snapshot.id).await?.is_some() {
if self
.get_operation_summary(workspace_id, &snapshot.id)
.await?
.is_some()
{
return Err(RegistryError::OperationAlreadyExists {
operation_id: snapshot.id.as_str().to_owned(),
});
@@ -81,6 +206,7 @@ impl PostgresRegistry {
sqlx::query(
"insert into operations (
id,
workspace_id,
name,
display_name,
protocol,
@@ -91,13 +217,14 @@ impl PostgresRegistry {
updated_at,
published_at
) values (
$1, $2, $3, $4, $5, $6, $7,
$8::timestamptz,
$1, $2, $3, $4, $5, $6, $7, $8,
$9::timestamptz,
$10::timestamptz
$10::timestamptz,
$11::timestamptz
)",
)
.bind(snapshot.id.as_str())
.bind(workspace_id.as_str())
.bind(&snapshot.name)
.bind(&snapshot.display_name)
.bind(serialize_enum_text(&snapshot.protocol, "protocol")?)
@@ -125,7 +252,10 @@ impl PostgresRegistry {
&self,
request: CreateVersionRequest<'_>,
) -> Result<(), RegistryError> {
let Some(summary) = self.get_operation_summary(&request.snapshot.id).await? else {
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(),
});
@@ -170,10 +300,14 @@ impl PostgresRegistry {
Ok(())
}
pub async fn list_operations(&self) -> Result<Vec<OperationSummary>, RegistryError> {
pub async fn list_operations(
&self,
workspace_id: &WorkspaceId,
) -> Result<Vec<OperationSummary>, RegistryError> {
let rows = sqlx::query(
"select
id,
workspace_id,
name,
display_name,
protocol,
@@ -184,8 +318,10 @@ impl PostgresRegistry {
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 operations
where workspace_id = $1
order by name asc",
)
.bind(workspace_id.as_str())
.fetch_all(&self.pool)
.await?;
@@ -194,11 +330,13 @@ impl PostgresRegistry {
pub async fn get_operation_summary(
&self,
workspace_id: &WorkspaceId,
operation_id: &OperationId,
) -> Result<Option<OperationSummary>, RegistryError> {
let row = sqlx::query(
"select
id,
workspace_id,
name,
display_name,
protocol,
@@ -209,8 +347,9 @@ impl PostgresRegistry {
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 operations
where id = $1",
where workspace_id = $1 and id = $2",
)
.bind(workspace_id.as_str())
.bind(operation_id.as_str())
.fetch_optional(&self.pool)
.await?;
@@ -220,12 +359,14 @@ impl PostgresRegistry {
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.protocol,
@@ -249,8 +390,9 @@ impl PostgresRegistry {
ov.created_by
from operation_versions ov
join operations o on o.id = ov.operation_id
where ov.operation_id = $1 and ov.version = $2",
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)
@@ -261,11 +403,13 @@ impl PostgresRegistry {
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.protocol,
@@ -289,9 +433,10 @@ impl PostgresRegistry {
ov.created_by
from operation_versions ov
join operations o on o.id = ov.operation_id
where ov.operation_id = $1
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?;
@@ -304,7 +449,7 @@ impl PostgresRegistry {
request: PublishRequest<'_>,
) -> Result<(), RegistryError> {
if self
.get_operation_version(request.operation_id, request.version)
.get_operation_version(request.workspace_id, request.operation_id, request.version)
.await?
.is_none()
{
@@ -373,6 +518,7 @@ impl PostgresRegistry {
let row = sqlx::query(
"select
o.id,
o.workspace_id,
o.name,
o.display_name,
o.protocol,
@@ -413,6 +559,7 @@ impl PostgresRegistry {
let rows = sqlx::query(
"select
o.id,
o.workspace_id,
o.name,
o.display_name,
o.protocol,
@@ -455,19 +602,22 @@ impl PostgresRegistry {
sqlx::query(
"insert into auth_profiles (
id,
workspace_id,
name,
kind,
config_json,
created_at,
updated_at
) values ($1, $2, $3, $4, $5::timestamptz, $6::timestamptz)
) 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)?))
@@ -481,19 +631,22 @@ impl PostgresRegistry {
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 id = $1",
where workspace_id = $1 and id = $2",
)
.bind(workspace_id.as_str())
.bind(auth_profile_id.as_str())
.fetch_optional(&self.pool)
.await?;
@@ -501,18 +654,24 @@ impl PostgresRegistry {
row.as_ref().map(map_auth_profile).transpose()
}
pub async fn list_auth_profiles(&self) -> Result<Vec<AuthProfile>, RegistryError> {
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?;
@@ -837,9 +996,24 @@ fn assert_immutable_fields(
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_operation_summary(row: &PgRow) -> Result<OperationSummary, RegistryError> {
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")?,
protocol: deserialize_enum_text(&row.try_get::<String, _>("protocol")?, "protocol")?,
@@ -865,6 +1039,7 @@ fn map_operation_version_record(row: &PgRow) -> Result<OperationVersionRecord, R
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")?,
@@ -918,6 +1093,7 @@ fn map_operation_version_record(row: &PgRow) -> Result<OperationVersionRecord, R
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)?,
@@ -1042,7 +1218,7 @@ mod tests {
ApiKeyHeaderAuthConfig, AuthConfig, AuthKind, AuthProfile, ConfigExport, ExecutionConfig,
ExportMode, GeneratedDraft, GeneratedDraftStatus, HttpMethod, OperationId, OperationStatus,
Protocol, RestTarget, RetryPolicy, Samples, SecretRef, Target, ToolDescription,
ToolExample,
ToolExample, WorkspaceId,
};
use crank_mapping::{MappingRule, MappingSet};
use crank_schema::{Schema, SchemaKind};
@@ -1059,6 +1235,10 @@ mod tests {
},
};
fn test_workspace_id() -> WorkspaceId {
WorkspaceId::new("ws_default")
}
#[tokio::test]
async fn stores_versions_and_published_operations() {
let database = TestDatabase::new().await;
@@ -1066,7 +1246,7 @@ mod tests {
let operation_v1 = test_operation("op_rest_01", 1, OperationStatus::Draft);
registry
.create_operation(&operation_v1, Some("alice"))
.create_operation(&test_workspace_id(), &operation_v1, Some("alice"))
.await
.unwrap();
@@ -1074,6 +1254,7 @@ mod tests {
registry
.create_version(CreateVersionRequest {
workspace_id: &test_workspace_id(),
snapshot: &operation_v2,
change_note: Some("add output mapping"),
created_by: Some("alice"),
@@ -1083,6 +1264,7 @@ mod tests {
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",
@@ -1092,12 +1274,12 @@ mod tests {
.unwrap();
let summary = registry
.get_operation_summary(&operation_v2.id)
.get_operation_summary(&test_workspace_id(), &operation_v2.id)
.await
.unwrap()
.unwrap();
let versions = registry
.list_operation_versions(&operation_v2.id)
.list_operation_versions(&test_workspace_id(), &operation_v2.id)
.await
.unwrap();
let published = registry
@@ -1126,11 +1308,15 @@ mod tests {
let registry = database.registry().await;
let operation = test_operation("op_rest_02", 1, OperationStatus::Draft);
registry.create_operation(&operation, None).await.unwrap();
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,
@@ -1156,10 +1342,14 @@ mod tests {
let registry = database.registry().await;
let operation = test_operation("op_rest_03", 1, OperationStatus::Draft);
registry.create_operation(&operation, None).await.unwrap();
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 {
@@ -1172,6 +1362,7 @@ mod tests {
registry
.save_auth_profile(SaveAuthProfileRequest {
workspace_id: &test_workspace_id(),
profile: &auth_profile,
})
.await
@@ -1211,7 +1402,10 @@ mod tests {
.await
.unwrap();
let auth_profiles = registry.list_auth_profiles().await.unwrap();
let auth_profiles = registry
.list_auth_profiles(&test_workspace_id())
.await
.unwrap();
let samples = registry
.list_sample_metadata(&operation.id, 1)
.await
@@ -1235,7 +1429,10 @@ mod tests {
let job_id = YamlImportJobId::new("job_yaml_01");
let operation = test_operation("op_rest_04", 1, OperationStatus::Draft);
registry.create_operation(&operation, None).await.unwrap();
registry
.create_operation(&test_workspace_id(), &operation, None)
.await
.unwrap();
registry
.create_yaml_import_job(CreateYamlImportJobRequest {