diff --git a/crates/workspace-server/Cargo.toml b/crates/workspace-server/Cargo.toml index b2008cfb..73e775fb 100644 --- a/crates/workspace-server/Cargo.toml +++ b/crates/workspace-server/Cargo.toml @@ -35,6 +35,7 @@ ticket.workspace = true memory.workspace = true merge-request.workspace = true tokio = { workspace = true, features = ["fs", "macros", "net", "rt-multi-thread", "sync", "time"] } +tower.workspace = true tokio-tungstenite.workspace = true worker.workspace = true workdir = { workspace = true, features = ["http-client"] } diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index f02ac49f..1add0d14 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -2447,6 +2447,8 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { #[derive(Clone)] pub struct RemoteRuntimeConfig { pub runtime_id: String, + /// Explicit Workspace assignment granted by Server authority. + pub workspace_id: Option, pub display_name: String, pub base_url: String, pub bearer_token: Option, @@ -2489,6 +2491,7 @@ impl RemoteRuntimeConfig { ) -> Self { Self { runtime_id: runtime_id.into(), + workspace_id: None, display_name: display_name.into(), base_url: base_url.into(), bearer_token, @@ -2501,6 +2504,11 @@ impl RemoteRuntimeConfig { } } + pub fn with_workspace_id(mut self, workspace_id: impl Into) -> Self { + self.workspace_id = Some(workspace_id.into()); + self + } + pub fn with_cached_capabilities(mut self, capabilities: RuntimeCapabilitySummary) -> Self { self.cached_capabilities = capabilities; self diff --git a/crates/workspace-server/src/lib.rs b/crates/workspace-server/src/lib.rs index d504bb11..d8ca871d 100644 --- a/crates/workspace-server/src/lib.rs +++ b/crates/workspace-server/src/lib.rs @@ -27,6 +27,7 @@ pub mod server; pub mod skills; pub mod store; pub mod worker_source; +pub mod workspace_catalog; mod workspace_subscription; pub use authority::{ @@ -45,8 +46,15 @@ pub use repositories::{ ConfiguredRepository, GitCommitSummary, GitRemoteSummary, GitRepositorySummary, RepositoryLogRead, RepositoryRegistryReader, RepositorySummary, }; -pub use server::{AuthConfig, ServerConfig, WorkspaceApi, build_router, serve}; +pub use server::{ + AuthConfig, ServerConfig, WorkspaceApi, WorkspaceServerApi, build_router, + build_workspace_server_router, serve, serve_workspace_catalog, +}; pub use store::{ControlPlaneStore, SqliteWorkspaceStore, WorkspaceRecord}; +pub use workspace_catalog::{ + InitialRepositoryIntent, WorkspaceCatalogService, WorkspaceCreateRequest, + WorkspaceCreateResponse, +}; use worker_runtime::identity::RuntimeWorkerRef; diff --git a/crates/workspace-server/src/main.rs b/crates/workspace-server/src/main.rs index 77a59e4f..0ccee280 100644 --- a/crates/workspace-server/src/main.rs +++ b/crates/workspace-server/src/main.rs @@ -9,10 +9,11 @@ use serde::{Deserialize, Serialize}; use tokio::net::TcpListener; use worker_runtime::auth::{RuntimeIdentityMaterial, decode_public_key}; use yoi_workspace_server::hosts::{RemoteRuntimeAuthConfig, RemoteRuntimeConfig}; -use yoi_workspace_server::store::{RepositoryRecord, SqliteWorkspaceStore, TrustedRuntimeRecord}; +use yoi_workspace_server::store::{SqliteWorkspaceStore, TrustedRuntimeRecord}; use yoi_workspace_server::{ - BackendRuntimesConfigFile, ControlPlaneStore, ServerConfig, WORKSPACE_BACKEND_CONFIG_TEMPLATE, - WorkspaceBackendConfigFile, WorkspaceIdentity, WorkspaceRecord, serve, + BackendRuntimesConfigFile, ControlPlaneStore, InitialRepositoryIntent, ServerConfig, + WORKSPACE_BACKEND_CONFIG_TEMPLATE, WorkspaceBackendConfigFile, WorkspaceCatalogService, + WorkspaceCreateRequest, WorkspaceIdentity, WorkspaceRecord, serve_workspace_catalog, }; #[derive(Debug)] @@ -152,30 +153,21 @@ async fn run_init_with_database_path( if let Some(parent) = database_path.parent() { tokio::fs::create_dir_all(parent).await?; } - let store = SqliteWorkspaceStore::open(&database_path)?; - store - .upsert_workspace(&WorkspaceRecord { - workspace_id: identity.workspace_id.clone(), - owner_account_id: None, + let store = Arc::new(SqliteWorkspaceStore::open(&database_path)?); + let service = WorkspaceCatalogService::new(store); + service.create_with_workspace_id( + WorkspaceCreateRequest { + operation_key: format!("cli-init:{}", identity.workspace_id), display_name: identity.display_name.clone(), - state: "active".to_string(), - created_at: identity.created_at.clone(), - updated_at: identity.created_at.clone(), - }) - .await?; - store.upsert_repository(&RepositoryRecord { - workspace_id: identity.workspace_id.clone(), - repository_id: "main".to_string(), - name: "Main repository".to_string(), - kind: "git".to_string(), - provider: Some("git".to_string()), - uri: options.workspace.display().to_string(), - default_ref: Some("HEAD".to_string()), - auth_ref_kind: None, - auth_ref_key: None, - created_at: identity.created_at.clone(), - updated_at: identity.created_at.clone(), - })?; + repository: InitialRepositoryIntent { + uri: options.workspace.display().to_string(), + display_name: Some("Main repository".to_string()), + default_ref: Some("HEAD".to_string()), + }, + }, + None, + Some(identity.workspace_id.clone()), + )?; eprintln!( "yoi-server: initialized workspace `{}` ({}) in server DB `{}`", @@ -358,6 +350,7 @@ fn run_trust_runtime_command(args: Vec) -> Result<(), Box { let mut runtime_id = None; + let mut workspace_id = None; let mut base_url = None; let mut public_key = None; let mut display_name = None; @@ -368,6 +361,9 @@ fn run_trust_runtime_command(args: Vec) -> Result<(), Box { runtime_id = Some(take_value(&flag, inline_value, &mut args)?) } + "--workspace-id" => { + workspace_id = Some(take_value(&flag, inline_value, &mut args)?) + } "--base-url" | "--endpoint" => { base_url = Some(take_value(&flag, inline_value, &mut args)?) } @@ -390,15 +386,39 @@ fn run_trust_runtime_command(args: Vec) -> Result<(), Box) -> Result<(), Box Result<(), Box Result<(), Box Result<(), Box Result { - let workspaces = store - .list_workspaces() - .map_err(|error| CliError(format!("failed to list workspaces from server DB: {error}")))?; - match workspaces.as_slice() { - [] => Err(CliError( - "server DB has no workspace records; run `yoi-server init --workspace `" - .to_string(), - )), - [workspace] => Ok(workspace.clone()), - _ => Err(CliError(format!( - "server DB contains {} workspaces; serve workspace selection is not implemented yet", - workspaces.len() - ))), - } -} - fn infer_workspace_root_from_repositories( store: &SqliteWorkspaceStore, workspace: &WorkspaceRecord, @@ -914,7 +939,7 @@ fn parse_listen(value: &str) -> Result { fn print_help() { println!( - "yoi-server\n\nUsage:\n yoi-server init [OPTIONS]\n yoi-server config [OPTIONS]\n yoi-server identity init --server-id [--replace]\n yoi-server identity show [--json]\n yoi-server trust-runtime add --runtime-id --base-url --public-key [--display-name ] [--replace]\n yoi-server trust-runtime list [--json] [--include-revoked]\n yoi-server trust-runtime revoke --runtime-id \n yoi-server skills [OPTIONS]\n yoi-server migrate --dry-run [--database ] + "yoi-server\n\nUsage:\n yoi-server init [OPTIONS]\n yoi-server config [OPTIONS]\n yoi-server identity init --server-id [--replace]\n yoi-server identity show [--json]\n yoi-server trust-runtime add --runtime-id --workspace-id --base-url --public-key [--display-name ] [--replace]\n yoi-server trust-runtime list [--json] [--include-revoked]\n yoi-server trust-runtime revoke --runtime-id \n yoi-server skills [OPTIONS]\n yoi-server migrate --dry-run [--database ] yoi-server serve [OPTIONS]\n\nOptions:\n -h, --help Print help" ); } @@ -1038,6 +1063,7 @@ mod tests { store .upsert_trusted_runtime(&TrustedRuntimeRecord { runtime_id: "runtime-a".to_string(), + workspace_id: None, display_name: "Runtime A".to_string(), base_url: "http://127.0.0.1:18080".to_string(), public_key, @@ -1059,6 +1085,7 @@ mod tests { async fn init_creates_identity_local_config_and_server_records() { let temp = tempfile::tempdir().unwrap(); let database_path = temp.path().join("data").join("server").join("server.db"); + std::fs::create_dir(temp.path().join(".git")).unwrap(); run_init_with_database_path( InitOptions { workspace: temp.path().canonicalize().unwrap(), diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 578ad3d0..3642531f 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -32,9 +32,11 @@ use ticket::{ execute_ticket_backend_operation, }; use tokio::net::TcpListener; +use tokio::sync::Mutex as AsyncMutex; use tokio_tungstenite::connect_async; use tokio_tungstenite::tungstenite::Message as TungsteniteMessage; use tokio_tungstenite::tungstenite::client::IntoClientRequest; +use tower::ServiceExt; use url::Url; use uuid::Uuid; use webauthn_rs::prelude::{ @@ -111,6 +113,7 @@ use crate::store::{ TicketWorkerAssignmentRecord, UserRecord, WorkdirRegistryRecord, WorkerControlGrantRecord, WorkerRegistryRecord, WorkerWorkdirLinkRecord, WorkspaceRecord, WorkspaceResourceKind, }; +use crate::workspace_catalog::{WorkspaceCatalogService, WorkspaceCreateRequest}; use crate::{Error, Result}; use worker_runtime::catalog::{ ConfigBundleRef, ProfileSelector, RepositorySelector as RuntimeRepositorySelector, @@ -158,6 +161,9 @@ pub struct ServerConfig { pub remote_runtime_sources: Vec, pub runtime_config_path: Option, pub backend_base_url: Option, + /// Allows the first ownerless Workspace to be created without a session. + /// This must only be enabled for a loopback-bound local Server. + pub allow_local_workspace_bootstrap: bool, } impl ServerConfig { @@ -187,6 +193,7 @@ impl ServerConfig { remote_runtime_sources: Vec::new(), runtime_config_path: BackendRuntimesConfigFile::default_path(), backend_base_url: None, + allow_local_workspace_bootstrap: false, } } @@ -243,10 +250,69 @@ impl ServerConfig { Self::default_workspace_backend_data_root(workspace_id).join("embedded-runtime") } + pub fn with_local_workspace_bootstrap(mut self, enabled: bool) -> Self { + self.allow_local_workspace_bootstrap = enabled; + self + } + pub fn with_embedded_runtime_store_root(mut self, root: impl Into) -> Self { self.embedded_runtime_store_root = root.into(); self } + + fn for_catalog_workspace( + &self, + workspace: &WorkspaceRecord, + repositories: Vec, + ) -> Result { + let primary = repositories + .iter() + .find(|repository| repository.repository_id == "main") + .or_else(|| repositories.first()) + .ok_or_else(|| { + Error::Config(format!( + "Workspace {} has no registered repository", + workspace.workspace_id + )) + })?; + let workspace_root = PathBuf::from(&primary.uri); + if !workspace_root.is_absolute() { + return Err(Error::Config(format!( + "Workspace {} repository uri is not an absolute local path", + workspace.workspace_id + ))); + } + let repositories = repositories + .into_iter() + .map(|repository| ConfiguredRepository { + id: repository.repository_id, + provider: repository.provider.unwrap_or(repository.kind), + path: PathBuf::from(&repository.uri), + uri: repository.uri, + display_name: Some(repository.name), + default_selector: repository.default_ref, + }) + .collect(); + let mut scoped = self.clone(); + scoped.workspace_id.clone_from(&workspace.workspace_id); + scoped + .workspace_display_name + .clone_from(&workspace.display_name); + scoped + .workspace_created_at + .clone_from(&workspace.created_at); + scoped.workspace_root = workspace_root; + scoped.embedded_runtime_store_root = + Self::default_embedded_runtime_store_root(&workspace.workspace_id); + scoped.repositories = repositories; + // Runtime trust is server-global. Only explicitly assigned sources enter + // this Workspace's registry and receive Workspace-scoped capabilities. + scoped.remote_runtime_sources.retain(|runtime| { + runtime.workspace_id.as_deref() == Some(workspace.workspace_id.as_str()) + }); + scoped.runtime_event_sources.clear(); + Ok(scoped) + } } const ORCHESTRATOR_ATTENTION_TICKET_LIMIT: usize = 20; @@ -682,6 +748,216 @@ impl crate::worker_source::VerifiedWorkerRemoveExecutor for WorkspaceWorkerRemov } } +#[derive(Clone)] +pub struct WorkspaceServerApi { + template: Arc, + store: Arc, + catalog: WorkspaceCatalogService, + routers: Arc>>, +} + +impl WorkspaceServerApi { + pub fn new(template: ServerConfig, store: Arc) -> Self { + Self { + template: Arc::new(template), + catalog: WorkspaceCatalogService::new(store.clone()), + store, + routers: Arc::new(AsyncMutex::new(HashMap::new())), + } + } + + async fn router_for_workspace(&self, workspace_id: &str) -> Result> { + let mut routers = self.routers.lock().await; + if let Some(router) = routers.get(workspace_id) { + return Ok(Some(router.clone())); + } + let Some(workspace) = self.store.get_workspace(workspace_id).await? else { + return Ok(None); + }; + let repositories = self.store.list_repositories(workspace_id)?; + let config = self + .template + .for_catalog_workspace(&workspace, repositories)?; + let api = WorkspaceApi::new(config, self.store.clone()).await?; + tokio::spawn(run_orchestrator_turn_end_hook(api.clone())); + let router = build_router(api); + routers.insert(workspace_id.to_string(), router.clone()); + Ok(Some(router)) + } + + async fn preload(&self) -> Result<()> { + for workspace in self.store.list_workspaces()? { + let _ = self + .router_for_workspace(&workspace.workspace_id) + .await? + .ok_or_else(|| { + Error::Config(format!( + "Workspace {} disappeared while loading", + workspace.workspace_id + )) + })?; + } + Ok(()) + } +} + +fn server_error_response(error: Error) -> Response { + ApiError::from(error).into_response() +} + +fn forbidden_server_response(message: &str) -> Response { + ( + StatusCode::FORBIDDEN, + Json(serde_json::json!({ "error": message })), + ) + .into_response() +} + +#[derive(Debug, Deserialize)] +struct WorkspaceListQuery { + limit: Option, +} + +async fn list_server_workspaces( + State(api): State, + headers: HeaderMap, + Query(query): Query, +) -> Response { + let owner = match resolve_server_actor(&api, &headers).await { + Ok(Some(actor)) => Some(actor.account_id), + Ok(None) => None, + Err(error) => return server_error_response(error), + }; + match api + .catalog + .list(owner.as_deref(), query.limit.unwrap_or(100)) + { + Ok(workspaces) => Json(workspaces).into_response(), + Err(error) => server_error_response(error), + } +} + +async fn create_server_workspace( + State(api): State, + headers: HeaderMap, + Json(request): Json, +) -> Response { + let owner_account_id = match resolve_server_actor(&api, &headers).await { + Ok(Some(actor)) => Some(actor.account_id), + Ok(None) if api.template.allow_local_workspace_bootstrap => { + match api.store.list_workspaces() { + Ok(workspaces) if workspaces.is_empty() => None, + Ok(_) => { + return forbidden_server_response( + "Workspace creation requires an authenticated owner", + ); + } + Err(error) => return server_error_response(error), + } + } + Ok(None) => { + return forbidden_server_response("Workspace creation requires an authenticated owner"); + } + Err(error) => return server_error_response(error), + }; + let created = match api.catalog.create(request, owner_account_id) { + Ok(created) => created, + Err(error) => return server_error_response(error), + }; + if let Err(error) = api + .router_for_workspace(&created.workspace.workspace_id) + .await + { + return server_error_response(error); + } + let status = if created.replayed { + StatusCode::OK + } else { + StatusCode::CREATED + }; + (status, Json(created)).into_response() +} + +async fn resolve_server_actor( + api: &WorkspaceServerApi, + headers: &HeaderMap, +) -> std::result::Result, Error> { + let cookie_name = auth_public_config(api.template.as_ref()).cookie_name; + resolve_request_actor(api.store.as_ref(), headers, &cookie_name).await +} + +async fn dispatch_workspace_request( + State(api): State, + request: Request, +) -> Response { + let path = request.uri().path(); + let workspace_id = scoped_workspace_id(path); + let router = if let Some(workspace_id) = workspace_id { + match api.router_for_workspace(workspace_id).await { + Ok(Some(router)) => Some(router), + Ok(None) => None, + Err(error) => return server_error_response(error), + } + } else { + let workspaces = match api.store.list_workspaces() { + Ok(workspaces) => workspaces, + Err(error) => return server_error_response(error), + }; + if workspaces.len() == 1 || is_server_global_forward(path) { + match workspaces.first() { + Some(workspace) => match api.router_for_workspace(&workspace.workspace_id).await { + Ok(router) => router, + Err(error) => return server_error_response(error), + }, + None => None, + } + } else { + None + } + }; + let Some(router) = router else { + return StatusCode::NOT_FOUND.into_response(); + }; + match router.oneshot(request).await { + Ok(response) => response, + Err(error) => match error {}, + } +} + +fn is_server_global_forward(path: &str) -> bool { + path == "/api/auth" + || path.starts_with("/api/auth/") + || path == "/health" + || path == "/" + || path.starts_with("/assets/") +} + +fn scoped_workspace_id(path: &str) -> Option<&str> { + let mut segments = path.trim_start_matches('/').split('/'); + match (segments.next(), segments.next(), segments.next()) { + (Some("api"), Some("w"), Some(workspace_id)) if !workspace_id.is_empty() => { + Some(workspace_id) + } + (Some("w"), Some(workspace_id), _) if !workspace_id.is_empty() => Some(workspace_id), + _ => None, + } +} + +pub async fn build_workspace_server_router( + template: ServerConfig, + store: Arc, +) -> Result { + let api = WorkspaceServerApi::new(template, store); + api.preload().await?; + Ok(Router::new() + .route( + "/api/workspaces", + get(list_server_workspaces).post(create_server_workspace), + ) + .fallback(dispatch_workspace_request) + .with_state(api)) +} + impl WorkspaceApi { pub fn with_config_schema_provider( mut self, @@ -1872,6 +2148,16 @@ struct ApiFailureLogEvent<'a> { diagnostics: Option<&'a [RuntimeDiagnostic]>, } +pub async fn serve_workspace_catalog( + template: ServerConfig, + store: Arc, + listener: TcpListener, +) -> Result<()> { + let router = build_workspace_server_router(template, store).await?; + axum::serve(listener, router).await?; + Ok(()) +} + pub async fn serve( config: ServerConfig, store: Arc, @@ -14130,6 +14416,139 @@ mod tests { config } + #[tokio::test] + async fn server_router_dispatches_two_workspace_contexts_without_state_leakage() { + let dir = tempfile::tempdir().unwrap(); + let repository_a = dir.path().join("repository-a"); + let repository_b = dir.path().join("repository-b"); + std::fs::create_dir_all(repository_a.join(".git")).unwrap(); + std::fs::create_dir_all(repository_b.join(".git")).unwrap(); + let template = test_server_config(dir.path()); + let store = Arc::new(SqliteWorkspaceStore::open(&template.database_path).unwrap()); + let catalog = WorkspaceCatalogService::new(store.clone()); + let workspace_a = catalog + .create( + WorkspaceCreateRequest { + operation_key: "create-a".to_string(), + display_name: "Workspace A".to_string(), + repository: crate::workspace_catalog::InitialRepositoryIntent { + uri: repository_a.display().to_string(), + display_name: None, + default_ref: None, + }, + }, + None, + ) + .unwrap(); + let workspace_b = catalog + .create( + WorkspaceCreateRequest { + operation_key: "create-b".to_string(), + display_name: "Workspace B".to_string(), + repository: crate::workspace_catalog::InitialRepositoryIntent { + uri: repository_b.display().to_string(), + display_name: None, + default_ref: None, + }, + }, + None, + ) + .unwrap(); + let app = build_workspace_server_router(template, store) + .await + .unwrap(); + + let uri_a = format!("/api/w/{}/workspace", workspace_a.workspace.workspace_id); + let uri_b = format!("/api/w/{}/workspace", workspace_b.workspace.workspace_id); + let (a, b) = tokio::join!(get_json(app.clone(), &uri_a), get_json(app.clone(), &uri_b)); + assert_eq!(a["workspace_id"], workspace_a.workspace.workspace_id); + assert_eq!(a["display_name"], "Workspace A"); + assert_eq!(b["workspace_id"], workspace_b.workspace.workspace_id); + assert_eq!(b["display_name"], "Workspace B"); + + let missing = app + .oneshot( + Request::builder() + .uri("/api/w/00000000-0000-0000-0000-000000000001/workspace") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(missing.status(), StatusCode::NOT_FOUND); + } + + #[tokio::test] + async fn local_bootstrap_create_activates_workspace_without_server_restart() { + let dir = tempfile::tempdir().unwrap(); + let repository = dir.path().join("repository"); + std::fs::create_dir_all(repository.join(".git")).unwrap(); + let template = test_server_config(dir.path()).with_local_workspace_bootstrap(true); + let store = Arc::new(SqliteWorkspaceStore::open(&template.database_path).unwrap()); + let app = build_workspace_server_router(template, store) + .await + .unwrap(); + let payload = json!({ + "operation_key": "bootstrap-1", + "display_name": "Created Workspace", + "repository": { + "uri": repository, + "display_name": "Repository", + "default_ref": "HEAD" + } + }); + + let created = app + .clone() + .oneshot( + Request::builder() + .method(Method::POST) + .uri("/api/workspaces") + .header(axum::http::header::CONTENT_TYPE, "application/json") + .body(Body::from(payload.to_string())) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(created.status(), StatusCode::CREATED); + let body = to_bytes(created.into_body(), usize::MAX).await.unwrap(); + let body: Value = serde_json::from_slice(&body).unwrap(); + let workspace_id = body["workspace"]["workspace_id"].as_str().unwrap(); + + let workspace = get_json(app.clone(), &format!("/api/w/{workspace_id}/workspace")).await; + assert_eq!(workspace["display_name"], "Created Workspace"); + + let replayed = app + .oneshot( + Request::builder() + .method(Method::POST) + .uri("/api/workspaces") + .header(axum::http::header::CONTENT_TYPE, "application/json") + .body(Body::from(payload.to_string())) + .unwrap(), + ) + .await + .unwrap(); + // Local bootstrap authority is consumed after the first Workspace; even + // an exact HTTP retry must authenticate rather than creating another + // ownerless Workspace accidentally. + assert_eq!(replayed.status(), StatusCode::FORBIDDEN); + } + + #[test] + fn scoped_workspace_path_requires_an_explicit_workspace_segment() { + assert_eq!( + scoped_workspace_id("/api/w/workspace-a/tickets"), + Some("workspace-a") + ); + assert_eq!( + scoped_workspace_id("/w/workspace-b/workers"), + Some("workspace-b") + ); + assert_eq!(scoped_workspace_id("/api/workspaces"), None); + assert_eq!(scoped_workspace_id("/api/workspace"), None); + } + fn memory_staging_record_json(id: &str, claim: &str) -> String { json!({ "schema_version": 1, @@ -16022,6 +16441,7 @@ mod tests { let mut config = test_server_config(temp.path()); config.remote_runtime_sources.push(RemoteRuntimeConfig { runtime_id: "runtime-remote".to_string(), + workspace_id: Some(TEST_WORKSPACE_ID.to_string()), display_name: "Remote Runtime".to_string(), base_url: "https://runtime.invalid".to_string(), bearer_token: None, @@ -16049,8 +16469,20 @@ mod tests { timeout: std::time::Duration::from_secs(1), }); let store = SqliteWorkspaceStore::open(config.database_path.clone()).unwrap(); + store + .upsert_workspace(&WorkspaceRecord { + workspace_id: TEST_WORKSPACE_ID.to_string(), + owner_account_id: None, + display_name: "Test Workspace".to_string(), + state: "active".to_string(), + created_at: "2026-08-11T00:00:00Z".to_string(), + updated_at: "2026-08-11T00:00:00Z".to_string(), + }) + .await + .unwrap(); let trust = crate::store::TrustedRuntimeRecord { runtime_id: "runtime-remote".to_string(), + workspace_id: Some(TEST_WORKSPACE_ID.to_string()), display_name: "Remote Runtime".to_string(), base_url: "https://runtime.invalid".to_string(), public_key: identity.public_key.clone(), diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index a00fb91d..01a8f1b0 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -220,6 +220,11 @@ const MIGRATIONS: &[Migration] = &[ name: "enforce Workspace resource foreign keys", apply: enforce_workspace_resource_foreign_keys, }, + Migration { + version: 40, + name: "create atomic Workspace catalog operations", + apply: create_workspace_catalog_operations, + }, ]; struct Migration { @@ -266,9 +271,26 @@ pub struct RepositoryRecord { pub updated_at: String, } +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkspaceBootstrapRecord { + pub operation_key: String, + pub request_fingerprint: String, + pub workspace: WorkspaceRecord, + pub repository: RepositoryRecord, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkspaceBootstrapResult { + pub workspace: WorkspaceRecord, + pub repository: RepositoryRecord, + pub config_revision: u64, + pub replayed: bool, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct TrustedRuntimeRecord { pub runtime_id: String, + pub workspace_id: Option, pub display_name: String, pub base_url: String, pub public_key: String, @@ -584,6 +606,10 @@ pub trait ControlPlaneStore: Send + Sync { ) -> Result>; async fn upsert_workspace(&self, record: &WorkspaceRecord) -> Result<()>; async fn get_workspace(&self, workspace_id: &str) -> Result>; + fn create_workspace_bootstrap( + &self, + record: &WorkspaceBootstrapRecord, + ) -> Result; async fn get_trusted_runtime(&self, runtime_id: &str) -> Result>; async fn consume_worker_mutation_source_jti( &self, @@ -1217,8 +1243,8 @@ impl SqliteWorkspaceStore { self.with_conn(|conn| { conn.execute( r#"INSERT INTO trusted_runtime_records ( - runtime_id, display_name, base_url, public_key, created_at, updated_at, revoked_at - ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7) + runtime_id, workspace_id, display_name, base_url, public_key, created_at, updated_at, revoked_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) ON CONFLICT(runtime_id) DO UPDATE SET display_name = excluded.display_name, base_url = excluded.base_url, @@ -1227,6 +1253,7 @@ impl SqliteWorkspaceStore { revoked_at = excluded.revoked_at"#, params![ record.runtime_id, + record.workspace_id, record.display_name, record.base_url, record.public_key, @@ -1245,10 +1272,10 @@ impl SqliteWorkspaceStore { ) -> Result> { self.with_conn(|conn| { let sql = if include_revoked { - r#"SELECT runtime_id, display_name, base_url, public_key, created_at, updated_at, revoked_at + r#"SELECT runtime_id, workspace_id, display_name, base_url, public_key, created_at, updated_at, revoked_at FROM trusted_runtime_records ORDER BY runtime_id ASC"# } else { - r#"SELECT runtime_id, display_name, base_url, public_key, created_at, updated_at, revoked_at + r#"SELECT runtime_id, workspace_id, display_name, base_url, public_key, created_at, updated_at, revoked_at FROM trusted_runtime_records WHERE revoked_at IS NULL ORDER BY runtime_id ASC"# }; let mut stmt = conn.prepare(sql)?; @@ -1371,10 +1398,167 @@ impl ControlPlaneStore for SqliteWorkspaceStore { }) } + fn create_workspace_bootstrap( + &self, + record: &WorkspaceBootstrapRecord, + ) -> Result { + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + if let Some((fingerprint, workspace_id)) = tx + .query_row( + "SELECT request_fingerprint, workspace_id FROM workspace_create_operations WHERE operation_key = ?1", + params![record.operation_key], + |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)), + ) + .optional()? + { + if fingerprint != record.request_fingerprint { + return Err(Error::WorkspaceConfigConflict( + "Workspace create operation key was already used with different input" + .to_string(), + )); + } + let workspace = tx.query_row( + r#"SELECT workspace_id, owner_account_id, display_name, state, created_at, updated_at + FROM workspaces WHERE workspace_id = ?1"#, + params![workspace_id], + read_workspace_record, + )?; + let repository = tx.query_row( + r#"SELECT workspace_id, repository_id, name, kind, provider, uri, default_ref, + auth_ref_kind, auth_ref_key, created_at, updated_at + FROM repositories WHERE workspace_id = ?1 AND repository_id = ?2"#, + params![workspace.workspace_id, record.repository.repository_id], + read_repository_record, + )?; + let config_revision = crate::config_source::load_state(&tx, &workspace.workspace_id)? + .ok_or_else(|| Error::Store("Workspace config is missing".to_string()))? + .snapshot + .revision; + tx.commit()?; + return Ok(WorkspaceBootstrapResult { + workspace, + repository, + config_revision, + replayed: true, + }); + } + + if let Some(existing) = tx + .query_row( + r#"SELECT workspace_id, owner_account_id, display_name, state, created_at, updated_at + FROM workspaces WHERE workspace_id = ?1"#, + params![record.workspace.workspace_id], + read_workspace_record, + ) + .optional()? + { + if existing.owner_account_id != record.workspace.owner_account_id + || existing.display_name != record.workspace.display_name + || existing.state != record.workspace.state + { + return Err(Error::WorkspaceConfigConflict( + "Workspace identity already exists with different metadata".to_string(), + )); + } + let existing_repository = tx + .query_row( + r#"SELECT workspace_id, repository_id, name, kind, provider, uri, default_ref, + auth_ref_kind, auth_ref_key, created_at, updated_at + FROM repositories WHERE workspace_id = ?1 AND repository_id = ?2"#, + params![record.repository.workspace_id, record.repository.repository_id], + read_repository_record, + ) + .optional()?; + if existing_repository.as_ref() != Some(&record.repository) { + return Err(Error::WorkspaceConfigConflict( + "Workspace initial repository already exists with different metadata" + .to_string(), + )); + } + } else { + tx.execute( + r#"INSERT INTO workspaces ( + workspace_id, owner_account_id, display_name, state, created_at, updated_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6)"#, + params![ + record.workspace.workspace_id, + record.workspace.owner_account_id, + record.workspace.display_name, + record.workspace.state, + record.workspace.created_at, + record.workspace.updated_at, + ], + )?; + tx.execute( + r#"INSERT INTO repositories ( + workspace_id, repository_id, name, kind, provider, uri, default_ref, + auth_ref_kind, auth_ref_key, created_at, updated_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)"#, + params![ + record.repository.workspace_id, + record.repository.repository_id, + record.repository.name, + record.repository.kind, + record.repository.provider, + record.repository.uri, + record.repository.default_ref, + record.repository.auth_ref_kind, + record.repository.auth_ref_key, + record.repository.created_at, + record.repository.updated_at, + ], + )?; + } + if crate::config_source::load_state(&tx, &record.workspace.workspace_id)?.is_none() { + let state = crate::config_source::initial_state()?; + crate::config_source::insert_materialized_state( + &tx, + &record.workspace.workspace_id, + &state, + &record.workspace.created_at, + )?; + } + for resource_kind in ["ticket", "objective", "worker"] { + tx.execute( + r#"INSERT OR IGNORE INTO workspace_resource_human_key_counters ( + workspace_id, resource_kind, next_sequence + ) VALUES (?1, ?2, 1)"#, + params![record.workspace.workspace_id, resource_kind], + )?; + } + let config_revision = crate::config_source::load_state( + &tx, + &record.workspace.workspace_id, + )? + .ok_or_else(|| Error::Store("Workspace config is missing".to_string()))? + .snapshot + .revision; + tx.execute( + r#"INSERT INTO workspace_create_operations ( + operation_key, request_fingerprint, workspace_id, created_at + ) VALUES (?1, ?2, ?3, ?4)"#, + params![ + record.operation_key, + record.request_fingerprint, + record.workspace.workspace_id, + record.workspace.created_at, + ], + )?; + tx.commit()?; + Ok(WorkspaceBootstrapResult { + workspace: record.workspace.clone(), + repository: record.repository.clone(), + config_revision, + replayed: false, + }) + }) + } + async fn get_trusted_runtime(&self, runtime_id: &str) -> Result> { self.with_conn(|conn| { conn.query_row( - r#"SELECT runtime_id, display_name, base_url, public_key, created_at, updated_at, revoked_at + r#"SELECT runtime_id, workspace_id, display_name, base_url, public_key, created_at, updated_at, revoked_at FROM trusted_runtime_records WHERE runtime_id = ?1"#, params![runtime_id], read_trusted_runtime_record, @@ -3948,12 +4132,13 @@ fn account_select_sql(where_clause: &str) -> String { fn read_trusted_runtime_record(row: &rusqlite::Row<'_>) -> rusqlite::Result { Ok(TrustedRuntimeRecord { runtime_id: row.get(0)?, - display_name: row.get(1)?, - base_url: row.get(2)?, - public_key: row.get(3)?, - created_at: row.get(4)?, - updated_at: row.get(5)?, - revoked_at: row.get(6)?, + workspace_id: row.get(1)?, + display_name: row.get(2)?, + base_url: row.get(3)?, + public_key: row.get(4)?, + created_at: row.get(5)?, + updated_at: row.get(6)?, + revoked_at: row.get(7)?, }) } @@ -5129,6 +5314,26 @@ CREATE UNIQUE INDEX ux_worker_workdir_attachment_reservation_id Ok(()) } +fn create_workspace_catalog_operations(conn: &Connection) -> Result<()> { + conn.execute_batch( + r#" + ALTER TABLE trusted_runtime_records + ADD COLUMN workspace_id TEXT REFERENCES workspaces(workspace_id) ON DELETE RESTRICT; + CREATE INDEX idx_trusted_runtime_records_workspace + ON trusted_runtime_records(workspace_id, revoked_at, runtime_id); + + CREATE TABLE workspace_create_operations ( + operation_key TEXT PRIMARY KEY, + request_fingerprint TEXT NOT NULL, + workspace_id TEXT NOT NULL UNIQUE, + created_at TEXT NOT NULL, + FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE + ); + "#, + )?; + Ok(()) +} + fn verify_workspace_resource_constraints(conn: &Connection) -> Result<()> { if current_schema_version(conn)? < 39 { return Ok(()); @@ -7249,7 +7454,7 @@ mod tests { let before = std::fs::read(&path).unwrap(); let plan = SqliteWorkspaceStore::migration_plan(&path).unwrap(); assert_eq!(plan.current_schema_version, 36); - assert_eq!(plan.target_schema_version, 39); + assert_eq!(plan.target_schema_version, 40); assert!(plan.migration_required); assert_eq!(plan.worker_count, 1); assert_eq!(plan.mappings[0].legacy_worker_id, 7); @@ -7263,7 +7468,7 @@ mod tests { store .with_conn(|conn| { assert!(table_exists(conn, "worker_diagnostics_archives")?); - assert_eq!(current_schema_version(conn)?, 39); + assert_eq!(current_schema_version(conn)?, 40); Ok(()) }) .unwrap(); @@ -7342,7 +7547,7 @@ mod tests { ), ] ); - assert_eq!(current_schema_version(&conn).unwrap(), 39); + assert_eq!(current_schema_version(&conn).unwrap(), 40); let foreign_key_error: Option = conn .query_row("PRAGMA foreign_key_check", [], |row| row.get(0)) .optional() @@ -7471,7 +7676,7 @@ INSERT INTO worker_orphan_diagnostics ( apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 39); + assert_eq!(current_schema_version(&conn).unwrap(), 40); assert!(!table_exists(&conn, "worker_control_delegation_operations").unwrap()); let controller_worker_id: String = conn .query_row( @@ -7589,7 +7794,7 @@ INSERT INTO worker_orphan_diagnostics ( apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 39); + assert_eq!(current_schema_version(&conn).unwrap(), 40); assert!(table_exists(&conn, "worker_workdir_attachment_reservations").unwrap()); } @@ -7622,7 +7827,7 @@ CREATE TABLE flow_events (event_id TEXT PRIMARY KEY); apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 39); + assert_eq!(current_schema_version(&conn).unwrap(), 40); assert!(table_exists(&conn, "flow_sources").unwrap()); assert!(table_exists(&conn, "flow_source_revisions").unwrap()); assert!(!table_exists(&conn, "flow_instances").unwrap()); @@ -7689,7 +7894,7 @@ INSERT INTO worker_workdir_attachment_reservations ( apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 39); + assert_eq!(current_schema_version(&conn).unwrap(), 40); let repositories_sql: String = conn .query_row( "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'repositories'", @@ -7867,7 +8072,7 @@ INSERT INTO workdir_registry ( let db = dir.path().join("control-plane.sqlite"); let store = SqliteWorkspaceStore::open(&db).unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 39); + assert_eq!(store.schema_version().await.unwrap(), 40); assert!( !store .with_conn(|conn| table_exists(conn, "worker_workspace_credentials")) @@ -7884,7 +8089,7 @@ INSERT INTO workdir_registry ( store.upsert_workspace(&record).await.unwrap(); let reopened = SqliteWorkspaceStore::open(&db).unwrap(); - assert_eq!(reopened.schema_version().await.unwrap(), 39); + assert_eq!(reopened.schema_version().await.unwrap(), 40); assert_eq!( reopened.get_workspace("local-dev").await.unwrap(), Some(record) @@ -8383,13 +8588,13 @@ INSERT INTO worker_registry ( configure_sqlite(&conn).unwrap(); apply_migrations(&conn).unwrap(); conn.execute( - "INSERT INTO __yoi_schema_migrations (version, name) VALUES (40, 'future')", + "INSERT INTO __yoi_schema_migrations (version, name) VALUES (41, 'future')", [], ) .unwrap(); let error = apply_migrations(&conn).unwrap_err().to_string(); - assert!(error.contains("schema version 40 is newer"), "{error}"); + assert!(error.contains("schema version 41 is newer"), "{error}"); assert!(error.contains("refusing to serve"), "{error}"); } @@ -8587,6 +8792,41 @@ VALUES ('workspace-b', 'ticket-b', 'related', 'ticket-a', NULL, 'tester', '2026- assert_eq!(foreign_keys_enabled, 1); } + #[test] + fn schema_v40_adds_workspace_create_operations_and_fail_closed_runtime_assignment() { + let mut conn = Connection::open_in_memory().unwrap(); + configure_sqlite(&conn).unwrap(); + apply_migrations_through(&conn, 39).unwrap(); + let now = "2026-01-01T00:00:00Z"; + conn.execute( + r#"INSERT INTO workspaces ( + workspace_id, display_name, state, created_at, updated_at + ) VALUES ('workspace-a', 'Workspace A', 'active', ?1, ?1)"#, + params![now], + ) + .unwrap(); + conn.execute( + r#"INSERT INTO trusted_runtime_records ( + runtime_id, display_name, base_url, public_key, created_at, updated_at + ) VALUES ('runtime-a', 'Runtime A', 'http://runtime-a.test', 'key', ?1, ?1)"#, + params![now], + ) + .unwrap(); + + apply_migrations(&mut conn).unwrap(); + + assert_eq!(current_schema_version(&conn).unwrap(), 40); + let workspace_id: Option = conn + .query_row( + "SELECT workspace_id FROM trusted_runtime_records WHERE runtime_id = 'runtime-a'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(workspace_id, None); + assert!(table_exists(&conn, "workspace_create_operations").unwrap()); + } + #[test] fn workspace_resource_fk_migration_preflights_and_enforces_composite_identity() { let conn = Connection::open_in_memory().unwrap(); @@ -9076,7 +9316,7 @@ INSERT INTO ticket_worker_assignment_events ( .unwrap(); let store = SqliteWorkspaceStore::from_connection(conn).unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 39); + assert_eq!(store.schema_version().await.unwrap(), 40); store .with_conn(|conn| { @@ -9265,7 +9505,7 @@ CREATE TABLE ticket_assignment_operations ( #[tokio::test] async fn repository_records_round_trip() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 39); + assert_eq!(store.schema_version().await.unwrap(), 40); let workspace = WorkspaceRecord { workspace_id: "local-dev".to_string(), owner_account_id: None, @@ -9331,7 +9571,7 @@ CREATE TABLE ticket_assignment_operations ( #[tokio::test] async fn memory_authority_records_round_trip_and_close_staging() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 39); + assert_eq!(store.schema_version().await.unwrap(), 40); let workspace = WorkspaceRecord { workspace_id: "local-dev".to_string(), owner_account_id: None, @@ -9722,7 +9962,7 @@ CREATE TABLE ticket_assignment_operations ( #[tokio::test] async fn account_and_login_records_round_trip() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 39); + assert_eq!(store.schema_version().await.unwrap(), 40); let now = "2026-07-22T00:00:00Z".to_string(); let account = AccountRecord { account_id: "acct-user-alice".to_string(), @@ -9908,6 +10148,7 @@ CREATE TABLE ticket_assignment_operations ( store .upsert_trusted_runtime(&TrustedRuntimeRecord { runtime_id: "runtime-a".to_string(), + workspace_id: None, display_name: "Runtime A".to_string(), base_url: "https://runtime.invalid".to_string(), public_key: "public-key".to_string(), diff --git a/crates/workspace-server/src/workspace_catalog.rs b/crates/workspace-server/src/workspace_catalog.rs new file mode 100644 index 00000000..d0d3ed66 --- /dev/null +++ b/crates/workspace-server/src/workspace_catalog.rs @@ -0,0 +1,321 @@ +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use chrono::{SecondsFormat, Utc}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use uuid::Uuid; + +use crate::store::{ + ControlPlaneStore, RepositoryRecord, WorkspaceBootstrapRecord, WorkspaceRecord, +}; +use crate::{Error, Result}; + +const DEFAULT_REPOSITORY_ID: &str = "main"; +const MAX_DISPLAY_NAME_BYTES: usize = 200; +const MAX_OPERATION_KEY_BYTES: usize = 200; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct InitialRepositoryIntent { + pub uri: String, + #[serde(default)] + pub display_name: Option, + #[serde(default)] + pub default_ref: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct WorkspaceCreateRequest { + pub operation_key: String, + pub display_name: String, + pub repository: InitialRepositoryIntent, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkspaceCreateResponse { + pub workspace: WorkspaceRecord, + pub repository: RepositoryRecord, + pub config_revision: u64, + pub request_fingerprint: String, + pub replayed: bool, +} + +#[derive(Clone)] +pub struct WorkspaceCatalogService { + store: Arc, +} + +impl WorkspaceCatalogService { + pub fn new(store: Arc) -> Self { + Self { store } + } + + pub fn list( + &self, + owner_account_id: Option<&str>, + limit: usize, + ) -> Result> { + let limit = limit.clamp(1, 200); + Ok(self + .store + .list_workspaces()? + .into_iter() + .filter(|workspace| { + workspace.owner_account_id.is_none() + || owner_account_id + .is_some_and(|owner| workspace.owner_account_id.as_deref() == Some(owner)) + }) + .take(limit) + .collect()) + } + + pub fn create( + &self, + request: WorkspaceCreateRequest, + owner_account_id: Option, + ) -> Result { + self.create_with_workspace_id(request, owner_account_id, None) + } + + pub fn create_with_workspace_id( + &self, + request: WorkspaceCreateRequest, + owner_account_id: Option, + requested_workspace_id: Option, + ) -> Result { + let operation_key = normalize_required( + "operation_key", + request.operation_key, + MAX_OPERATION_KEY_BYTES, + )?; + let display_name = + normalize_required("display_name", request.display_name, MAX_DISPLAY_NAME_BYTES)?; + let repository_path = validate_repository_uri(&request.repository.uri)?; + let repository_uri = repository_path.to_string_lossy().into_owned(); + let repository_name = request + .repository + .display_name + .as_deref() + .map(str::trim) + .filter(|value| !value.is_empty()) + .unwrap_or("Main repository") + .to_string(); + let default_ref = request + .repository + .default_ref + .as_deref() + .map(str::trim) + .filter(|value| !value.is_empty()) + .unwrap_or("HEAD") + .to_string(); + let requested_workspace_id = requested_workspace_id + .map(|value| { + Uuid::parse_str(value.trim()) + .map(|id| id.to_string()) + .map_err(|_| Error::InvalidInput("workspace_id must be a UUID".to_string())) + }) + .transpose()?; + let workspace_id = requested_workspace_id + .clone() + .unwrap_or_else(|| Uuid::now_v7().to_string()); + let fingerprint = workspace_create_fingerprint( + requested_workspace_id.as_deref(), + &display_name, + owner_account_id.as_deref(), + &repository_uri, + &repository_name, + &default_ref, + ); + let now = Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true); + let result = self + .store + .create_workspace_bootstrap(&WorkspaceBootstrapRecord { + operation_key, + request_fingerprint: fingerprint.clone(), + workspace: WorkspaceRecord { + workspace_id: workspace_id.clone(), + owner_account_id, + display_name, + state: "active".to_string(), + created_at: now.clone(), + updated_at: now.clone(), + }, + repository: RepositoryRecord { + workspace_id, + repository_id: DEFAULT_REPOSITORY_ID.to_string(), + name: repository_name, + kind: "git".to_string(), + provider: Some("git".to_string()), + uri: repository_uri, + default_ref: Some(default_ref), + auth_ref_kind: None, + auth_ref_key: None, + created_at: now.clone(), + updated_at: now, + }, + })?; + Ok(WorkspaceCreateResponse { + workspace: result.workspace, + repository: result.repository, + config_revision: result.config_revision, + request_fingerprint: fingerprint, + replayed: result.replayed, + }) + } +} + +fn normalize_required(field: &str, value: String, max_bytes: usize) -> Result { + let value = value.trim(); + if value.is_empty() || value.len() > max_bytes { + return Err(Error::InvalidInput(format!( + "{field} must be between 1 and {max_bytes} bytes" + ))); + } + Ok(value.to_string()) +} + +fn validate_repository_uri(uri: &str) -> Result { + let uri = uri.trim(); + if uri.is_empty() || uri.contains("://") { + return Err(Error::InvalidInput( + "initial repository uri must be an absolute server-local path".to_string(), + )); + } + let path = Path::new(uri); + if !path.is_absolute() { + return Err(Error::InvalidInput( + "initial repository uri must be an absolute server-local path".to_string(), + )); + } + let path = path.canonicalize().map_err(|error| { + Error::InvalidInput(format!("initial repository path is unavailable: {error}")) + })?; + if !path.is_dir() { + return Err(Error::InvalidInput( + "initial repository path must be a directory".to_string(), + )); + } + let normal_git = path.join(".git").exists(); + let bare_git = path.join("HEAD").is_file() && path.join("objects").is_dir(); + if !normal_git && !bare_git { + return Err(Error::InvalidInput( + "initial repository path is not a Git repository".to_string(), + )); + } + Ok(path) +} + +fn workspace_create_fingerprint( + requested_workspace_id: Option<&str>, + display_name: &str, + owner_account_id: Option<&str>, + repository_uri: &str, + repository_name: &str, + default_ref: &str, +) -> String { + let payload = serde_json::json!({ + "requested_workspace_id": requested_workspace_id, + "display_name": display_name, + "owner_account_id": owner_account_id, + "repository": { + "repository_id": DEFAULT_REPOSITORY_ID, + "uri": repository_uri, + "display_name": repository_name, + "default_ref": default_ref, + "kind": "git", + } + }); + let mut hasher = Sha256::new(); + hasher.update(serde_json::to_vec(&payload).expect("workspace fingerprint serializes")); + let digest = hasher.finalize(); + let encoded = digest + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::(); + format!("sha256:{encoded}") +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::store::SqliteWorkspaceStore; + + fn git_repository() -> tempfile::TempDir { + let dir = tempfile::tempdir().unwrap(); + std::fs::create_dir(dir.path().join(".git")).unwrap(); + dir + } + + #[tokio::test] + async fn create_is_atomic_and_exact_retries_converge() { + let store = Arc::new(SqliteWorkspaceStore::in_memory().unwrap()); + let service = WorkspaceCatalogService::new(store.clone()); + let repository = git_repository(); + let request = WorkspaceCreateRequest { + operation_key: "request-1".to_string(), + display_name: "Workspace A".to_string(), + repository: InitialRepositoryIntent { + uri: repository.path().display().to_string(), + display_name: None, + default_ref: None, + }, + }; + + let created = service.create(request.clone(), None).unwrap(); + let replayed = service.create(request, None).unwrap(); + + assert!(!created.replayed); + assert!(replayed.replayed); + assert_eq!( + created.workspace.workspace_id, + replayed.workspace.workspace_id + ); + assert_eq!(store.list_workspaces().unwrap().len(), 1); + assert_eq!( + store + .list_repositories(&created.workspace.workspace_id) + .unwrap() + .len(), + 1 + ); + assert!( + store + .load_workspace_config(&created.workspace.workspace_id) + .unwrap() + .is_some() + ); + } + + #[tokio::test] + async fn idempotency_key_reuse_with_different_payload_is_rejected() { + let store = Arc::new(SqliteWorkspaceStore::in_memory().unwrap()); + let service = WorkspaceCatalogService::new(store); + let repository = git_repository(); + let mut request = WorkspaceCreateRequest { + operation_key: "request-1".to_string(), + display_name: "Workspace A".to_string(), + repository: InitialRepositoryIntent { + uri: repository.path().display().to_string(), + display_name: None, + default_ref: None, + }, + }; + service.create(request.clone(), None).unwrap(); + request.display_name = "Workspace B".to_string(); + + let error = service.create(request, None).unwrap_err().to_string(); + assert!(error.contains("different input"), "{error}"); + } + + #[test] + fn repository_intent_rejects_remote_and_non_git_paths() { + let remote = validate_repository_uri("https://example.test/repo.git").unwrap_err(); + assert!(remote.to_string().contains("server-local path")); + + let dir = tempfile::tempdir().unwrap(); + let non_git = validate_repository_uri(&dir.path().display().to_string()).unwrap_err(); + assert!(non_git.to_string().contains("not a Git repository")); + } +}