Workspace CRUD API with MongoDB persistence

* Workspace CRUD API with MongoDB persistence

* Address Copilot review and CodeScene feedback

Author: makarski

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mstream"
-version = "0.46.0"
+version = "0.47.0"
 edition = "2024"
 authors = ["Simon Makarski"]
 
@@ -62,6 +62,7 @@ apache-avro = { version = "0.14", features = ["derive"] }
 tokio = { version = "1.0", features = ["full", "test-util"] }
 tempfile = "3.23.0"
 criterion = { version = "0.8", features = ["html_reports"] }
+tower = { version = "0.5.3", features = ["util"] }
 
 [[bench]]
 name = "json_schema_bench"

README.md

@@ -517,6 +517,50 @@ For MongoDB services, the `id` field is auto-assigned if omitted. For PubSub ser
 
 Test suites persist test cases, assertions, and optionally a reference to the transform script they were authored against. Used by the Transform Studio to save and reload testing sessions.
 
+### Workspaces
+
+Workspaces persist a named development context — script reference, sample input, schema, and test suite — so that work-in-progress can be saved, shared, and restored later.
+
+> **Workspaces vs Connectors (Pipelines)**
+>
+> A **connector** is a deployed, running pipeline: it binds a source → middlewares → sinks and processes live data. A **workspace** is a development-time concept: it captures what you're *working on* (which script, which sample input, which schema, which test suite) without running anything. When ready, a workspace can pre-fill a connector's configuration, but the two are independent records.
+
+| Method | Endpoint | Description |
+|--------|----------|-------------|
+| `GET` | `/workspaces` | List saved workspaces (id + name) |
+| `GET` | `/workspaces/{id}` | Get a workspace by ID |
+| `POST` | `/workspaces` | Create a workspace |
+| `PUT` | `/workspaces/{id}` | Update a workspace |
+| `DELETE` | `/workspaces/{id}` | Delete a workspace |
+
+#### Create Workspace
+
+```
+POST /workspaces
+```
+
+```json
+{
+  "name": "PII Masking Workspace",
+  "script": { "service_name": "udf-anonymizer", "resource": "mask_pii.rhai" },
+  "input": "{\"name\": \"John\", \"email\": \"john@acme.com\"}",
+  "schema": { "schema_id": "schema-1", "service_name": "etl-test-data" },
+  "test_suite_id": "suite-1"
+}
+```
+
+All fields except `name` are optional. The `id` is auto-assigned if omitted.
+
+#### System Configuration
+
+```toml
+[system.workspaces]
+service_name = "system-db"
+resource = "workspaces"
+```
+
+Requires a MongoDB service. Without this configuration, workspaces use a noop store and workspace data is not persisted.
+
 ## Checkpoints
 
 Checkpoints allow connectors to resume from their last processed position after a restart, preventing data loss or reprocessing.

mstream-config.toml.example

@@ -77,6 +77,11 @@ resource = "mstream-checkpoints"
 buffer_capacity = 10000       # Max entries in ring buffer (default: 10000)
 stream_preload_count = 100    # Entries sent on SSE connect (default: 100)
 
+# Optional: persist Transform Studio workspaces to MongoDB
+[system.workspaces]
+service_name = "mongodb-source"
+resource = "workspaces"
+
 # Connector Configurations
 [[connectors]]
 enabled = true

src/api/error.rs

@@ -8,7 +8,7 @@ use tracing::{error, warn};
 use crate::{
     api::types::Message, job_manager::error::JobManagerError,
     middleware::udf::rhai::RhaiMiddlewareError, schema::introspect::SchemaIntrospectError,
-    testing::store::TestSuiteStoreError,
+    testing::store::TestSuiteStoreError, workspace::WorkspaceStoreError,
 };
 
 #[derive(Debug, thiserror::Error)]
@@ -33,6 +33,9 @@ pub enum ApiError {
 
     #[error(transparent)]
     TestSuiteStore(#[from] TestSuiteStoreError),
+
+    #[error(transparent)]
+    WorkspaceStore(#[from] WorkspaceStoreError),
 }
 
 impl IntoResponse for ApiError {
@@ -46,6 +49,7 @@ impl IntoResponse for ApiError {
             ApiError::BadRequest(msg) => (StatusCode::BAD_REQUEST, msg),
             ApiError::Internal(msg) => (StatusCode::INTERNAL_SERVER_ERROR, msg),
             ApiError::TestSuiteStore(err) => err.response_data(),
+            ApiError::WorkspaceStore(err) => err.response_data(),
         };
 
         if status.is_server_error() {
@@ -136,6 +140,22 @@ impl ApiErrorDetails for TestSuiteStoreError {
     }
 }
 
+impl ApiErrorDetails for WorkspaceStoreError {
+    fn response_data(&self) -> (StatusCode, String) {
+        match self {
+            WorkspaceStoreError::NotFound(id) => (
+                StatusCode::NOT_FOUND,
+                format!("Workspace '{}' not found", id),
+            ),
+            WorkspaceStoreError::MongoDb(e) => (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                format!("Database error: {}", e),
+            ),
+            WorkspaceStoreError::Other(msg) => (StatusCode::INTERNAL_SERVER_ERROR, msg.clone()),
+        }
+    }
+}
+
 impl ApiErrorDetails for RhaiMiddlewareError {
     fn response_data(&self) -> (StatusCode, String) {
         match &self {
@@ -353,6 +373,29 @@ mod tests {
         }
     }
 
+    mod workspace_store_error_tests {
+        use super::*;
+
+        #[tokio::test]
+        async fn not_found_returns_404() {
+            let error: ApiError = WorkspaceStoreError::NotFound("ws-123".to_string()).into();
+            let (status, message) = extract_response(error.into_response()).await;
+
+            assert_eq!(status, StatusCode::NOT_FOUND);
+            assert!(message.contains("ws-123"));
+        }
+
+        #[tokio::test]
+        async fn other_error_returns_500() {
+            let error: ApiError =
+                WorkspaceStoreError::Other("unexpected failure".to_string()).into();
+            let (status, message) = extract_response(error.into_response()).await;
+
+            assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR);
+            assert!(message.contains("unexpected failure"));
+        }
+    }
+
     mod api_error_tests {
         use super::*;
 

src/api/handler/mod.rs

@@ -7,6 +7,7 @@ pub mod test_gen;
 pub mod test_suites;
 pub mod transform;
 pub mod validate;
+pub mod workspaces;
 
 pub use jobs::{create_start_job, list_checkpoints, list_jobs, restart_job, stop_job};
 pub use schema::{fill_schema, get_resource_schema, schema_convert};
@@ -20,3 +21,6 @@ pub use test_gen::{transform_test_generate, transform_test_run};
 pub use test_suites::{delete_test_suite, get_test_suite, list_test_suites, save_test_suite};
 pub use transform::transform_run;
 pub use validate::{transform_completions, transform_validate};
+pub use workspaces::{
+    delete_workspace, get_workspace, list_workspaces, save_workspace, update_workspace,
+};

src/api/handler/test_suites.rs

@@ -14,6 +14,7 @@ pub async fn list_test_suites(
     State(state): State<AppState>,
 ) -> Result<impl IntoResponse, ApiError> {
     let summaries = state
+        .stores
         .test_suite_store
         .list()
         .await
@@ -28,6 +29,7 @@ pub async fn get_test_suite(
     Path(id): Path<String>,
 ) -> Result<impl IntoResponse, ApiError> {
     let suite = state
+        .stores
         .test_suite_store
         .get(&id)
         .await
@@ -47,6 +49,7 @@ pub async fn save_test_suite(
     suite.updated_at = chrono::Utc::now();
 
     state
+        .stores
         .test_suite_store
         .save(&suite)
         .await
@@ -67,6 +70,7 @@ pub async fn delete_test_suite(
     Path(id): Path<String>,
 ) -> Result<impl IntoResponse, ApiError> {
     state
+        .stores
         .test_suite_store
         .delete(&id)
         .await