From cf834e28fbcdb794a9da5c7ab1c618bb23eba5b7 Mon Sep 17 00:00:00 2001 From: zgq Date: Thu, 8 Oct 2026 19:43:01 +0800 Subject: [PATCH 1/9] build(frontend): absorb the Community client without moving the Java baseline The locked Community frontend sat at `a117f489` (v5.3.2-17) while Community main moved 962 commits ahead, so none of the upstream UI work could reach the Rust product. The submodule cannot simply follow main: `scripts/build-community-h2-classpath.sh` requires the checked-out commit's `chat2db-community-server` tree to equal the Java compatibility baseline in `third_party/community-h2-classpath.lock` (`3cb8af54`), and every newer Community commit changes that tree. Following main directly failed `Rust-Java IPC`, `MySQL`, and `Desktop (windows-latest)` with `Community server tree must match compatibility baseline 3cb8af54...`. Keep the two pins independent: - `scripts/community-frontend.lock.json` moves to `dedc12663` (client tree `9f4be62c090886b726e20fd9bf274d7bd279d5e1`); the submodule stays on the compatibility baseline commit, so the Java compatibility classpath keeps building from the tree it was validated against. - `scripts/community-frontend.mjs` fetches the locked frontend commit into the submodule object store when it is missing (by SHA, then by refs) and archives from there. Verification checks the locked commit's client tree instead of requiring the submodule index and HEAD to match; the worktree must still be clean. - The packaging scripts read `community_commit` from the frontend lock, the value `scripts/prepare-release.mjs` validates; reading the submodule HEAD would fail release aggregation once the pins differ. - Eight of the eleven pinned upstream behavior scripts were deleted upstream, which now runs its battery through `prebuild:web:community`, so the surviving three stay pinned. - Upstream's `test:community-boundary` scans the sibling `chat2db-community-server` tree, which the export now materializes instead of stubbing or skipping it. - `docs/architecture.md` and `docs/releases.md` describe the split. --- docs/architecture.md | 6 ++- docs/releases.md | 5 +- scripts/build-linux-package.sh | 2 +- scripts/build-macos-package.sh | 2 +- scripts/build-windows-package.sh | 2 +- scripts/community-frontend.lock.json | 4 +- scripts/community-frontend.mjs | 68 +++++++++++++++++++--------- 7 files changed, 60 insertions(+), 29 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index 1329ed3..2f0d9c0 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -47,7 +47,11 @@ The current product frontend is not the former repository-owned replacement workbench. The build exports a locked Community frontend commit that retains the original pages, components, interactions, and styles while applying a reviewable host-transport patch for CSP-safe callbacks and Web/Desktop file -flows. `scripts/community-frontend.lock.json` binds both commit and tree. Web maps its historical `/api` +flows. `scripts/community-frontend.lock.json` binds both commit and tree, and is +independent of the submodule checkout: `third_party/chat2db-community` stays on +the Java compatibility baseline recorded in +`third_party/community-h2-classpath.lock`, while the locked frontend commit is +fetched into the submodule object store and archived from there. Web maps its historical `/api` contract through Axum; desktop maps the existing `window.javaQuery` contract through one Tauri `legacy_request` command. Both paths call the same Rust legacy dispatcher. The implemented product slice covers native MySQL connection diff --git a/docs/releases.md b/docs/releases.md index 750dd3d..add3d69 100644 --- a/docs/releases.md +++ b/docs/releases.md @@ -25,7 +25,10 @@ commit in `main`. The tag annotation records: - `source_repository`: `OtterMind/Chat2DB-Rust`; - `source_commit`: the complete release commit SHA; -- `community_ref`: the pinned submodule commit; +- `community_ref`: the locked frontend commit from + `scripts/community-frontend.lock.json` (the Community revision shipped as the + visible UI; the Java compatibility baseline stays recorded in + `third_party/community-h2-classpath.lock`); - `package_repository`: `OtterMind/Chat2DB-Rust`; - `workflow_ref`: the complete release commit SHA; - `product_workflow`: `.github/workflows/package.yml`; and diff --git a/scripts/build-linux-package.sh b/scripts/build-linux-package.sh index c186eb6..a9a1e2a 100755 --- a/scripts/build-linux-package.sh +++ b/scripts/build-linux-package.sh @@ -119,7 +119,7 @@ cp -- "${appimage_artifacts[0]}" "${deb_artifacts[0]}" "${rpm_artifacts[0]}" "${ echo "Chat2DB Rust Linux package" echo "version=${version}" echo "git_commit=$(git -C "${repository_root}" rev-parse HEAD)" - echo "community_commit=$(git -C "${repository_root}/third_party/chat2db-community" rev-parse HEAD)" + echo "community_commit=$(node -e 'process.stdout.write(require(process.argv[1]).commit)' "${repository_root}/scripts/community-frontend.lock.json")" echo "architecture=$(uname -m)" echo "target=linux" echo "rust_toolchain=${rust_toolchain}" diff --git a/scripts/build-macos-package.sh b/scripts/build-macos-package.sh index 1149fec..54ce69e 100755 --- a/scripts/build-macos-package.sh +++ b/scripts/build-macos-package.sh @@ -308,7 +308,7 @@ verification_directory="" shasum -a 256 -- "$(basename "${zip_path}")" "$(basename "${dmg_path}")" > SHA256SUMS ) git_commit="$(git -C "${repository_root}" rev-parse HEAD)" -community_commit="$(git -C "${repository_root}/third_party/chat2db-community" rev-parse HEAD)" +community_commit="$(node -e 'process.stdout.write(require(process.argv[1]).commit)' "${repository_root}/scripts/community-frontend.lock.json")" app_kib="$(du -sk "${app_path}" | awk '{ print $1 }')" signature_details="$(codesign -dv --verbose=4 "${app_path}" 2>&1)" signing_team_id="$(awk -F= '/^TeamIdentifier=/ { print $2; exit }' <<<"${signature_details}")" diff --git a/scripts/build-windows-package.sh b/scripts/build-windows-package.sh index a0e88a5..6512ffd 100755 --- a/scripts/build-windows-package.sh +++ b/scripts/build-windows-package.sh @@ -121,7 +121,7 @@ cp -- "${msi_artifacts[0]}" "${package_directory}/" echo "Chat2DB Rust Windows package" echo "version=${version}" echo "git_commit=$(git -C "${repository_root}" rev-parse HEAD)" - echo "community_commit=$(git -C "${repository_root}/third_party/chat2db-community" rev-parse HEAD)" + echo "community_commit=$(node -e 'process.stdout.write(require(process.argv[1]).commit)' "${repository_root}/scripts/community-frontend.lock.json")" echo "architecture=x86_64" echo "target=windows" echo "rust_toolchain=${rust_toolchain}" diff --git a/scripts/community-frontend.lock.json b/scripts/community-frontend.lock.json index 0b27cb9..c0ed48c 100644 --- a/scripts/community-frontend.lock.json +++ b/scripts/community-frontend.lock.json @@ -2,7 +2,7 @@ "repository": "https://github.com/OtterMind/Chat2DB.git", "submodulePath": "third_party/chat2db-community", "sourcePath": "chat2db-community-client", - "commit": "a117f4895b3c63bbcedd2f79ab867ead800cf5c5", - "tree": "d8a24c8322700bc8373e165674092d4f0b446569", + "commit": "dedc12663d8d17eae239f79c2f31205406524aa5", + "tree": "9f4be62c090886b726e20fd9bf274d7bd279d5e1", "packageManager": "yarn@1.22.22" } diff --git a/scripts/community-frontend.mjs b/scripts/community-frontend.mjs index a3acfc0..62fe11b 100644 --- a/scripts/community-frontend.mjs +++ b/scripts/community-frontend.mjs @@ -90,21 +90,42 @@ function replaceDirectory(target) { mkdirSync(target, { recursive: true }); } -function verifySource() { - if (!existsSync(join(SUBMODULE_DIR, '.git'))) { - fail(`submodule is unavailable; run git submodule update --init --recursive ${lock.submodulePath}`); - } +function hasLockedCommit() { + const result = spawnSync('git', ['-C', SUBMODULE_DIR, 'cat-file', '-e', `${lock.commit}^{commit}`]); + return result.status === 0; +} - const indexedSubmodule = capture('git', ['ls-files', '--stage', '--', lock.submodulePath]); - const expectedIndex = `160000 ${lock.commit} 0\t${lock.submodulePath}`; - if (indexedSubmodule !== expectedIndex) { - fail(`repository index must pin ${lock.submodulePath} at ${lock.commit}`); +// The submodule checkout tracks the Java compatibility baseline recorded in +// `third_party/community-h2-classpath.lock`, so the locked frontend commit is +// fetched into the submodule object store instead of being checked out. +function ensureLockedCommit() { + if (hasLockedCommit()) { + return; + } + spawnSync('git', ['-C', SUBMODULE_DIR, 'fetch', '--no-tags', 'origin', lock.commit], { + stdio: 'inherit', + }); + if (!hasLockedCommit()) { + run('git', [ + '-C', + SUBMODULE_DIR, + 'fetch', + '--no-tags', + 'origin', + '+refs/heads/*:refs/remotes/origin/*', + '+refs/tags/*:refs/tags/*', + ]); + } + if (!hasLockedCommit()) { + fail(`locked Community commit ${lock.commit} is unavailable in ${lock.submodulePath}`); } +} - const commit = capture('git', ['rev-parse', 'HEAD'], SUBMODULE_DIR); - if (commit !== lock.commit) { - fail(`submodule HEAD is ${commit}; expected ${lock.commit}`); +function verifySource() { + if (!existsSync(join(SUBMODULE_DIR, '.git'))) { + fail(`submodule is unavailable; run git submodule update --init --recursive ${lock.submodulePath}`); } + ensureLockedCommit(); const tree = capture('git', ['rev-parse', `${lock.commit}:${lock.sourcePath}`], SUBMODULE_DIR); if (tree !== lock.tree) { @@ -120,10 +141,11 @@ function verifySource() { fail(`submodule worktree is not clean:\n${status}`); } - return { commit, tree }; + return { commit: lock.commit, tree }; } -function exportSource(destination) { +function exportSource(destination, sourcePath = lock.sourcePath) { + ensureLockedCommit(); replaceDirectory(destination); mkdirSync(VERSION_ROOT, { recursive: true }); const archiveName = `source-${process.pid}.tar`; @@ -139,7 +161,7 @@ function exportSource(destination) { '--format=tar', `--output=${archiveOutput}`, lock.commit, - lock.sourcePath, + sourcePath, ], { cwd: VERSION_ROOT }, ); @@ -150,6 +172,12 @@ function exportSource(destination) { rmSync(archivePath, { force: true }); } +// Upstream's community boundary check scans the sibling server tree, so every +// exported worktree needs the monorepo layout next to it. +function exportCommunityServer() { + exportSource(join(VERSION_ROOT, 'work', 'chat2db-community-server'), 'chat2db-community-server'); +} + function runYarn(args, cwd, env = {}) { if (!existsSync(YARN_JS)) { fail('pinned Yarn is missing; run npm ci in apps/frontend first'); @@ -179,6 +207,7 @@ function ensureDependencies() { function createWorktree(name) { ensureDependencies(); + exportCommunityServer(); const worktree = join(VERSION_ROOT, 'work', name); exportSource(worktree); symlinkSync( @@ -199,17 +228,12 @@ function setupUmi(worktree) { function test() { const worktree = createWorktree('test'); setupUmi(worktree); - runYarn(['test:chat-answer-update'], worktree); + // Upstream deleted the other per-behavior scripts this step used to pin. + // `prebuild:web:community` now runs the full upstream test battery before + // every community build, so only the surviving checks stay pinned here. runYarn(['test:tree-title-highlight'], worktree); - runYarn(['test:deep-clone'], worktree); runYarn(['test:ai-model-select'], worktree); runYarn(['test:export-connections'], worktree); - runYarn(['test:host-file-transfer'], worktree); - runYarn(['test:canvas-lifecycle'], worktree); - runYarn(['test:result-error-boundary'], worktree); - runYarn(['test:result-resource-activity'], worktree); - runYarn(['test:result-table-lifecycle'], worktree); - runYarn(['test:workspace-resource-activity'], worktree); } function build() { From 3578c8ed55a2924b15eae1d07df2aed7f5bf4ae3 Mon Sep 17 00:00:00 2001 From: zgq Date: Fri, 9 Oct 2026 20:19:15 +0800 Subject: [PATCH 2/9] feat(legacy): serve the first routes the new Community client requires The Community frontend moved 962 commits ahead and now calls routes this repository does not implement. This adds the first batch, each registered in the desktop dispatch match, `LEGACY_PATHS`, the Axum router, and the path-coverage test lists: - `POST /api/v3/ai/chat/history/session/rename` renames one durable AI session through the existing revision CAS; the sidebar and AI panel expose it. - `GET /api/rdb/table/copy/prepare` returns the default copy-table name the copy dialog pre-fills, sharing `default_copy_table_name` with the copy route. - `GET /api/database/supported` lists the registered database types so the connection form can register dynamic ones. - `POST /api/jdbc/driver/upload` answers with an explicit `unsupported_custom_driver_upload` failure. This runtime loads verified driver packs and has no custom-driver storage or class loading, so the dialog reports the gap instead of pretending the upload succeeded. Tests: 80 chat2db-web tests pass, including new route tests for the supported database list, the driver-upload gap, and the AI session rename flow. --- apps/chat2db-web/src/legacy.rs | 159 +++++++++++++++++++++++++++++- apps/chat2db-web/src/legacy_ai.rs | 124 ++++++++++++++++++++++- 2 files changed, 281 insertions(+), 2 deletions(-) diff --git a/apps/chat2db-web/src/legacy.rs b/apps/chat2db-web/src/legacy.rs index 1f07ee2..0d0e449 100644 --- a/apps/chat2db-web/src/legacy.rs +++ b/apps/chat2db-web/src/legacy.rs @@ -245,6 +245,24 @@ pub struct LegacyDriverResponse { pub driver_config_list: Vec, } +/// One database type offered by the retained Community connection form. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacySupportedDatabase { + pub db_type: String, + pub name: String, + pub support_database: bool, + pub support_schema: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub sql_dialect: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub jdbc_driver_class: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub url_sample: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub icon: Option, +} + #[derive(Clone, Default, Deserialize)] #[serde(rename_all = "camelCase")] pub struct LegacyConnectionProperty { @@ -858,6 +876,20 @@ pub struct LegacyTableCopyRequest { pub copy_data: bool, } +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyTableCopyPrepareQuery { + pub data_source_id: LegacyIdentifier, + #[serde(default, deserialize_with = "deserialize_string_or_default")] + pub database_name: String, + #[serde(default, deserialize_with = "deserialize_string_or_default")] + pub schema_name: String, + #[serde(default, deserialize_with = "deserialize_string_or_default")] + pub database_type: String, + #[serde(default, deserialize_with = "deserialize_string_or_default")] + pub table_name: String, +} + #[derive(Debug, Clone, Deserialize)] #[serde(rename_all = "camelCase")] pub struct LegacyDatabaseDefinitionRequest { @@ -1949,6 +1981,38 @@ pub fn drivers(application: &Application, requested_type: &str) -> LegacyDriverR } } +/// Rejects uploaded JDBC drivers: this runtime only loads verified driver packs. +pub(crate) fn custom_driver_upload_unsupported() -> LegacyFailure { + LegacyFailure { + code: "unsupported_custom_driver_upload".to_owned(), + message: + "This runtime loads verified driver packs and cannot accept an uploaded JDBC driver yet" + .to_owned(), + } +} + +pub(crate) fn list_supported_databases(application: &Application) -> Vec { + application + .list_drivers() + .items + .into_iter() + .map(|driver| { + let db_type = database_type_for_driver(&driver); + let support_schema = !matches!(db_type.as_str(), "MYSQL" | "SQLITE"); + LegacySupportedDatabase { + db_type, + name: driver.name, + support_database: true, + support_schema, + sql_dialect: None, + jdbc_driver_class: Some(driver.driver_class), + url_sample: None, + icon: None, + } + }) + .collect() +} + pub(crate) async fn list_community_dashboards( application: &Application, query: CommunityDashboardListQuery, @@ -4732,6 +4796,26 @@ pub(crate) async fn truncate_table( .await } +/// Returns the default copy-table name the Community copy dialog pre-fills. +pub(crate) async fn prepare_copy_table_name( + application: &Application, + request: &LegacyTableCopyPrepareQuery, +) -> LegacyResult { + let datasource_id = request.data_source_id.as_string(); + resolve_mysql_database_type(application, &datasource_id, &request.database_type).await?; + if request.table_name.trim().is_empty() { + return Err(LegacyFailure::invalid( + "invalid_table_request", + "tableName is required", + )); + } + Ok(default_copy_table_name(&request.table_name)) +} + +fn default_copy_table_name(table_name: &str) -> String { + format!("{}_copy", table_name.trim()) +} + pub(crate) async fn copy_table( application: &Application, request: &LegacyTableCopyRequest, @@ -4739,7 +4823,7 @@ pub(crate) async fn copy_table( let datasource_id = request.data_source_id.as_string(); resolve_mysql_database_type(application, &datasource_id, &request.database_type).await?; let new_name = if request.new_name.trim().is_empty() { - format!("{}_copy", request.table_name.trim()) + default_copy_table_name(&request.table_name) } else { request.new_name.trim().to_owned() }; @@ -8190,6 +8274,8 @@ async fn dispatch_inner( Err(error) => Err(error), }, ("get", "/api/common/environment/list_all") => serialized(Ok(environments())), + ("get", "/api/database/supported") => serialize_data(list_supported_databases(application)), + ("post", "/api/jdbc/driver/upload") => Err(custom_driver_upload_unsupported()), ("get", "/api/jdbc/driver/list") => decode(request.message) .map(|query: LegacyDriverQuery| drivers(application, &query.db_type)) .and_then(serialize_data), @@ -8817,6 +8903,12 @@ async fn dispatch_inner( Err(error) => Err(error), } } + ("get", "/api/rdb/table/copy/prepare") => { + match decode::(request.message) { + Ok(query) => serialized(prepare_copy_table_name(application, &query).await), + Err(error) => Err(error), + } + } ("post", "/api/rdb/table/copy") => { match decode::(request.message) { Ok(body) => serialized(copy_table(application, &body).await), @@ -8974,6 +9066,8 @@ const LEGACY_PATHS: &[&str] = &[ "/api/rdb/account/grants", "/api/rdb/account/preview", "/api/rdb/account/execute", + "/api/database/supported", + "/api/jdbc/driver/upload", "/api/diff/sql", "/api/rdb/database/list", "/api/rdb/database/create_database_sql", @@ -8981,6 +9075,7 @@ const LEGACY_PATHS: &[&str] = &[ "/api/rdb/schema/create_schema_sql", "/api/rdb/ddl/schema_list", "/api/rdb/ddl/database_schema_list", + "/api/rdb/table/copy/prepare", "/api/rdb/table/list", "/api/rdb/table/table_meta", "/api/rdb/table/query", @@ -9116,6 +9211,14 @@ pub(crate) fn routes() -> Router { .route("/api/v1/chart/update", post(chart_update_handler)) .route("/api/chart", axum::routing::delete(chart_delete_handler)) .route("/api/common/environment/list_all", get(environment_handler)) + .route( + "/api/database/supported", + get(supported_database_handler), + ) + .route( + "/api/jdbc/driver/upload", + post(driver_upload_handler).layer(DefaultBodyLimit::max(MAX_LEGACY_MULTIPART_BYTES)), + ) .route("/api/jdbc/driver/list", get(driver_handler)) .route("/api/jdbc/driver/download", get(driver_download_handler)) .route("/api/jdbc/driver/save", post(driver_save_handler)) @@ -9313,6 +9416,10 @@ pub(crate) fn routes() -> Router { .route("/api/rdb/table/modify/sql", post(table_modify_sql_handler)) .route("/api/rdb/table/truncate", post(table_truncate_handler)) .route("/api/rdb/table/copy", post(table_copy_handler)) + .route( + "/api/rdb/table/copy/prepare", + get(table_copy_prepare_handler), + ) .route("/api/rdb/table/table_list", get(simple_table_list_handler)) .route("/api/rdb/table/column_list", get(table_column_list_handler)) .route("/api/rdb/table/index_list", get(table_index_list_handler)) @@ -9521,6 +9628,16 @@ async fn environment_handler() -> Json>> { envelope(Ok(environments())) } +async fn supported_database_handler( + State(application): State, +) -> Json>> { + envelope(Ok(list_supported_databases(&application))) +} + +async fn driver_upload_handler() -> Json>> { + envelope(Err(custom_driver_upload_unsupported())) +} + async fn driver_handler( State(application): State, Query(query): Query, @@ -10152,6 +10269,13 @@ async fn table_copy_handler( envelope(copy_table(&application, &request).await) } +async fn table_copy_prepare_handler( + State(application): State, + Query(query): Query, +) -> Json> { + envelope(prepare_copy_table_name(&application, &query).await) +} + async fn database_delete_prepare_handler( State(application): State, Json(request): Json, @@ -10645,6 +10769,7 @@ mod tests { ("get", "/api/rdb/ddl/database_schema_list"), ("post", "/api/rdb/ddl/execute"), ("get", "/api/rdb/table/table_meta"), + ("get", "/api/rdb/table/copy/prepare"), ("get", "/api/rdb/table/query"), ("get", "/api/rdb/table/export"), ("get", "/api/rdb/table/create/example"), @@ -10693,6 +10818,8 @@ mod tests { ]; const REQUIRED_WORKSPACE_PATHS: &[(&str, &str)] = &[ + ("get", "/api/database/supported"), + ("post", "/api/jdbc/driver/upload"), ("get", "/api/jdbc/driver/download"), ("post", "/api/jdbc/driver/save"), ("delete", "/api/jdbc/driver/delete"), @@ -11421,6 +11548,36 @@ mod tests { serde_json::from_slice(&body).expect("response body must be JSON") } + #[tokio::test] + async fn supported_databases_route_serves_the_driver_inventory() { + let directory = tempfile::TempDir::new().expect("temporary directory"); + let storage = Storage::open(directory.path(), Arc::new(EmptyVault)).expect("storage opens"); + let router = routes().with_state(Application::with_storage(storage)); + + let body = dashboard_http_json(&router, "GET", "/api/database/supported", None).await; + assert_eq!(body["success"], true); + assert!(body["errorCode"].is_null()); + let databases = body["data"] + .as_array() + .expect("supported databases must be an array"); + for database in databases { + assert!(database["dbType"].is_string(), "dbType must be a string"); + assert_eq!(database["supportDatabase"], true); + assert!(database["supportSchema"].is_boolean()); + } + } + + #[tokio::test] + async fn driver_upload_route_reports_the_unsupported_gap() { + let directory = tempfile::TempDir::new().expect("temporary directory"); + let storage = Storage::open(directory.path(), Arc::new(EmptyVault)).expect("storage opens"); + let router = routes().with_state(Application::with_storage(storage)); + + let body = dashboard_http_json(&router, "POST", "/api/jdbc/driver/upload", None).await; + assert_eq!(body["success"], false); + assert_eq!(body["errorCode"], "unsupported_custom_driver_upload"); + } + #[tokio::test] async fn dashboard_http_routes_preserve_community_crud_and_envelopes() { let directory = tempfile::TempDir::new().expect("temporary directory"); diff --git a/apps/chat2db-web/src/legacy_ai.rs b/apps/chat2db-web/src/legacy_ai.rs index 7e2e060..c494cbe 100644 --- a/apps/chat2db-web/src/legacy_ai.rs +++ b/apps/chat2db-web/src/legacy_ai.rs @@ -181,6 +181,13 @@ pub struct LegacyAiSessionDeleteRequest { pub id: String, } +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyAiSessionRenameRequest { + pub id: String, + pub title: String, +} + #[derive(Debug, Clone, Deserialize)] #[serde(rename_all = "camelCase")] struct LegacyAiMessagesQuery { @@ -662,6 +669,38 @@ pub async fn delete_session(application: &Application, id: &str) -> Result<(), L Ok(()) } +/// Renames one durable Community AI session through revision CAS. +/// +/// # Errors +/// +/// Returns validation, lookup, revision, or storage failures. +pub async fn rename_session( + application: &Application, + id: &str, + title: &str, +) -> Result<(), LegacyAiFailure> { + let title = title.trim(); + if title.is_empty() { + return Err(LegacyAiFailure::invalid( + "invalid_ai_request", + "title must not be empty", + )); + } + let session = application.get_agent_session(id.trim()).await?; + application + .update_agent_session( + &session.id, + UpdateAgentSessionRequest { + expected_revision: session.revision.clone(), + title: title.to_owned(), + provider_id: session.provider_id.clone(), + datasource_id: session.datasource_id.clone(), + }, + ) + .await?; + Ok(()) +} + /// Lists provider profiles that have usable credentials as frontend model options. /// /// # Errors @@ -1088,6 +1127,23 @@ pub async fn dispatch( .unwrap_or_else(|_| internal_failure_value()), ) } + ("post", "/api/v3/ai/chat/history/session/rename") => { + let request = serde_json::from_value::(message); + let result = match request { + Ok(request) => rename_session(application, &request.id, &request.title).await, + Err(_) => Err(LegacyAiFailure::invalid( + "invalid_ai_request", + "id and title must not be empty", + )), + }; + Some( + serde_json::to_value(match result { + Ok(()) => LegacyAiEnvelope::success(()), + Err(error) => LegacyAiEnvelope::failure(error), + }) + .unwrap_or_else(|_| internal_failure_value()), + ) + } ("get", "/api/v3/ai/model/options") => Some( serde_json::to_value(match model_options(application).await { Ok(data) => LegacyAiEnvelope::success(data), @@ -1186,6 +1242,10 @@ pub(crate) fn routes() -> Router { "/api/v3/ai/chat/history/session/delete", post(delete_session_handler), ) + .route( + "/api/v3/ai/chat/history/session/rename", + post(rename_session_handler), + ) .route("/api/v3/ai/model/options", get(model_options_handler)) .route("/api/v3/ai/model/list", get(model_catalog_handler)) .route( @@ -1312,6 +1372,21 @@ async fn delete_session_handler( ) } +async fn rename_session_handler( + State(application): State, + Json(request): Json, +) -> Json { + Json( + serde_json::to_value( + match rename_session(&application, &request.id, &request.title).await { + Ok(()) => LegacyAiEnvelope::success(()), + Err(error) => LegacyAiEnvelope::failure(error), + }, + ) + .unwrap_or_else(|_| internal_failure_value()), + ) +} + async fn model_options_handler(State(application): State) -> Json { Json( serde_json::to_value(match model_options(&application).await { @@ -1978,7 +2053,7 @@ mod tests { }; use chat2db_contract::{ AgentEvent, AgentEventEnvelope, AgentPermissionRequest, ApiError, - CreateProviderProfileRequest, ProviderKind, + CreateAgentSessionRequest, CreateProviderProfileRequest, ProviderKind, }; use chat2db_core::Application; use chat2db_storage::{SecretRef, SecretValue, SecretVault, SecretVaultError, Storage}; @@ -2454,4 +2529,51 @@ mod tests { .contains("alpha") ); } + + #[tokio::test] + async fn session_rename_route_updates_the_visible_title() { + let fixture = test_application(); + let profile = fixture + .application + .create_provider_profile(CreateProviderProfileRequest { + name: "Rename provider".to_owned(), + kind: ProviderKind::OpenAiCompatible, + base_url: "https://provider.example/v1".to_owned(), + model: "mock-model".to_owned(), + context_window_tokens: "4096".to_owned(), + max_output_tokens: "1024".to_owned(), + credentials: None, + }) + .await + .expect("provider profile must be created"); + let session = fixture + .application + .create_agent_session(CreateAgentSessionRequest { + title: "First title".to_owned(), + provider_id: profile.id, + datasource_id: None, + system_prompt: None, + }) + .await + .expect("session must be created"); + let application = routes().with_state(fixture.application); + + let renamed = application + .clone() + .oneshot(json_request( + Method::POST, + "/api/v3/ai/chat/history/session/rename", + &serde_json::json!({ "id": session.id, "title": "Renamed title" }), + )) + .await + .expect("rename route must respond"); + assert_eq!(response_json(renamed).await["success"], true); + + let sessions = application + .oneshot(empty_request(Method::GET, "/api/v3/ai/chat/history/sessions")) + .await + .expect("session list must respond"); + let sessions = response_json(sessions).await; + assert_eq!(sessions["data"][0]["title"], "Renamed title"); + } } From f9bd280542016ef9f5988f7334227bd5c2f751b9 Mon Sep 17 00:00:00 2001 From: zgq Date: Fri, 9 Oct 2026 21:19:31 +0800 Subject: [PATCH 3/9] feat(legacy): serve the datasource colour and task-center read routes Continues the route work the newer Community frontend needs. Datasource identity colour: - Migration `009` adds `datasources.identity_color` (and the task-center columns and event table used later in this series). - Storage validates and stores `#RRGGBB`, Core and the contract DTOs project it, and the datasource list/detail responses now carry `identityColor` so the colour survives a refresh. - `POST /api/connection/datasource/identity_color` returns the updated identity with the fixed local environment. Task center (read side), backed by the existing durable transfer tasks: - `GET /api/tasks/list` accepts the client's `status` vocabulary (PENDING/RUNNING/SUCCESS/FAILED/CANCELLED) and projects tasks into the `ImportExportTaskDetails` shape the panel polls. - `GET /api/tasks/get` reads one task by `taskId`. - `GET /api/tasks/active-count` reports the queued/running total used by the desktop exit coordinator. - `GET /api/tasks/artifact` streams the finished task artifact; the desktop keeps using its own reveal-in-explorer path, so the bridge answers method-not-allowed. Contract regeneration ran because the datasource DTOs changed. Verification: `chat2db-storage` 113 tests, `chat2db-web` 80 tests, contract drift check, and clippy with `-D warnings` on the touched crates all pass. --- apps/chat2db-web/src/legacy.rs | 301 +++++++++++++++++- apps/chat2db-web/src/legacy_ai.rs | 5 +- apps/frontend/src/generated/contract.ts | 2 + contracts/openapi/chat2db-v1.json | 7 + crates/chat2db-contract/src/datasource.rs | 4 + .../chat2db-contract/src/datasource_edit.rs | 5 + crates/chat2db-core/src/convert.rs | 1 + crates/chat2db-core/src/datasource_edit.rs | 1 + crates/chat2db-core/src/lib.rs | 22 ++ .../009_community_client_routes.sql | 29 ++ crates/chat2db-storage/src/datasource.rs | 67 +++- crates/chat2db-storage/src/lib.rs | 9 +- 12 files changed, 443 insertions(+), 10 deletions(-) create mode 100644 crates/chat2db-storage/migrations/009_community_client_routes.sql diff --git a/apps/chat2db-web/src/legacy.rs b/apps/chat2db-web/src/legacy.rs index 0d0e449..8dbc3e8 100644 --- a/apps/chat2db-web/src/legacy.rs +++ b/apps/chat2db-web/src/legacy.rs @@ -263,6 +263,25 @@ pub struct LegacySupportedDatabase { pub icon: Option, } +/// Identity-colour update posted by the retained Community connection tree. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyIdentityColorRequest { + pub id: LegacyIdentifier, + #[serde(default)] + pub identity_color: Option, +} + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyIdentityColorResponse { + pub id: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub identity_color: Option, + pub environment_id: u64, + pub environment: LegacyEnvironment, +} + #[derive(Clone, Default, Deserialize)] #[serde(rename_all = "camelCase")] pub struct LegacyConnectionProperty { @@ -332,6 +351,8 @@ pub struct LegacyDatasourceResponse { pub support_schema: bool, pub has_secret: bool, pub revision: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub identity_color: Option, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -456,6 +477,64 @@ pub struct LegacyTaskListQuery { pub task_status: String, } +/// Task-center query posted by the newer Community import/export panel. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportExportTaskQuery { + #[serde(default = "default_page_no")] + pub page_no: u32, + #[serde(default = "default_page_size")] + pub page_size: u32, + #[serde(default)] + pub status: String, +} + +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportExportTaskIdQuery { + pub task_id: LegacyIdentifier, +} + +/// One task-center row in the shape the newer Community panel expects. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportExportTask { + pub id: i64, + pub name: String, + #[serde(rename = "type")] + pub task_type: String, + pub status: String, + pub progress: u32, + #[serde(skip_serializing_if = "Option::is_none")] + pub stage: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub progress_message: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub target: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub error_code: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub error_message: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub artifact_id: Option, + pub created_at: i64, + #[serde(skip_serializing_if = "Option::is_none")] + pub started_at: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub finished_at: Option, + pub updated_at: i64, +} + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportExportTaskTarget { + pub data_source_id: String, + pub database_name: String, + pub schema_name: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub table_name: Option, +} + #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct LegacyTransferTask { @@ -2969,6 +3048,126 @@ pub(crate) async fn legacy_transfer_task_download( .await?) } +fn import_export_task_status(status: TransferTaskStatus) -> &'static str { + match status { + TransferTaskStatus::Queued => "PENDING", + TransferTaskStatus::Running => "RUNNING", + TransferTaskStatus::Succeeded => "SUCCESS", + TransferTaskStatus::Failed | TransferTaskStatus::Interrupted => "FAILED", + TransferTaskStatus::Cancelled => "CANCELLED", + } +} + +fn import_export_status_filter(value: &str) -> LegacyResult> { + match value.trim().to_ascii_uppercase().as_str() { + "" => Ok(Vec::new()), + "PENDING" => Ok(vec![TransferTaskStatus::Queued]), + "RUNNING" => Ok(vec![TransferTaskStatus::Running]), + "SUCCESS" => Ok(vec![TransferTaskStatus::Succeeded]), + "FAILED" => Ok(vec![ + TransferTaskStatus::Failed, + TransferTaskStatus::Interrupted, + ]), + "CANCELLED" => Ok(vec![TransferTaskStatus::Cancelled]), + _ => Err(LegacyFailure::invalid( + "invalid_task_status", + "status is not supported", + )), + } +} + +fn import_export_task_type(kind: TransferTaskKind) -> &'static str { + match kind { + TransferTaskKind::ImportFile => "DATA_FILE_IMPORT", + TransferTaskKind::ExportSql => "SQL_EXPORT", + TransferTaskKind::ExportFile => "TABLE_DATA_EXPORT", + } +} + +fn legacy_epoch_millis(value: &str) -> i64 { + value.parse().unwrap_or_default() +} + +fn import_export_task(task: TransferTask) -> LegacyImportExportTask { + let error_message = task + .error_log + .lines() + .map(str::trim) + .rfind(|line| !line.is_empty()) + .map(str::to_owned); + let progress = transfer_progress_percent(&task).parse().unwrap_or(0); + LegacyImportExportTask { + id: task.id, + name: task.task_name, + task_type: import_export_task_type(task.kind).to_owned(), + status: import_export_task_status(task.status).to_owned(), + progress, + stage: None, + progress_message: Some(task.progress_description).filter(|value| !value.is_empty()), + target: Some(LegacyImportExportTaskTarget { + data_source_id: task.datasource_id, + database_name: task.database_name, + schema_name: task.schema_name, + table_name: task.table_name, + }), + error_code: None, + error_message, + artifact_id: task.artifact_id, + created_at: legacy_epoch_millis(&task.created_at_ms), + started_at: None, + finished_at: task.finished_at_ms.as_deref().map(legacy_epoch_millis), + updated_at: legacy_epoch_millis(&task.updated_at_ms), + } +} + +pub(crate) async fn list_import_export_tasks( + application: &Application, + query: &LegacyImportExportTaskQuery, +) -> LegacyResult> { + let statuses = import_export_status_filter(&query.status)?; + let page = application + .list_transfer_tasks_by_statuses(query.page_no, query.page_size, &statuses) + .await?; + Ok(LegacyPage { + total: usize::try_from(page.total).unwrap_or(usize::MAX), + has_next_page: u64::from(page.page_no).saturating_mul(u64::from(page.page_size)) + < page.total, + data: page.items.into_iter().map(import_export_task).collect(), + page_no: page.page_no, + page_size: page.page_size, + }) +} + +pub(crate) async fn get_import_export_task( + application: &Application, + query: &LegacyImportExportTaskIdQuery, +) -> LegacyResult { + let task = application + .transfer_task(legacy_transfer_task_id(&query.task_id)?) + .await?; + Ok(import_export_task(task)) +} + +pub(crate) async fn active_import_export_task_count( + application: &Application, +) -> LegacyResult { + let page = application + .list_transfer_tasks_by_statuses( + 1, + 100, + &[TransferTaskStatus::Queued, TransferTaskStatus::Running], + ) + .await?; + Ok(page.total) +} + +pub(crate) async fn import_export_task_artifact( + application: &Application, + query: &LegacyImportExportTaskIdQuery, +) -> LegacyResult { + legacy_transfer_task_download(application, &query.task_id).await +} + pub(crate) async fn export_legacy_mysql_dml( application: &Application, request: &LegacyDmlExportRequest, @@ -6450,6 +6649,22 @@ fn storage_failure(error: StorageError) -> LegacyFailure { } } +pub(crate) async fn update_datasource_identity_color( + application: &Application, + request: &LegacyIdentityColorRequest, +) -> LegacyResult { + let id = request.id.as_string(); + let datasource = application + .update_datasource_identity_color(&id, request.identity_color.clone()) + .await?; + Ok(LegacyIdentityColorResponse { + id: datasource.id, + identity_color: datasource.identity_color, + environment_id: 1, + environment: default_environment(), + }) +} + fn datasource_response( application: &Application, datasource: DatasourceEditProjection, @@ -6520,6 +6735,7 @@ fn datasource_response( support_schema, has_secret: datasource.has_secret, revision: datasource.revision, + identity_color: datasource.identity_color, } } @@ -8450,6 +8666,12 @@ async fn dispatch_inner( Err(error) => Err(error), } } + ("post", "/api/connection/datasource/identity_color") => { + match decode::(request.message) { + Ok(body) => serialized(update_datasource_identity_color(application, &body).await), + Err(error) => Err(error), + } + } ("post", "/api/import/sql_file") => { match decode::(request.message) { Ok(body) if desktop_paths => { @@ -8507,6 +8729,21 @@ async fn dispatch_inner( Ok(_) => Err(desktop_file_operation_required()), Err(error) => Err(error), }, + ("get", "/api/tasks/list") => { + match decode::(request.message) { + Ok(query) => serialized(list_import_export_tasks(application, &query).await), + Err(error) => Err(error), + } + } + ("get", "/api/tasks/get") => { + match decode::(request.message) { + Ok(query) => serialized(get_import_export_task(application, &query).await), + Err(error) => Err(error), + } + } + ("get", "/api/tasks/active-count") => { + serialized(active_import_export_task_count(application).await) + } ("get", "/api/sql/format") => match decode::(request.message) { Ok(query) => serialized(format_legacy_sql(application, &query).await), Err(error) => Err(error), @@ -9015,6 +9252,7 @@ const LEGACY_PATHS: &[&str] = &[ "/api/connection/datasource/create", "/api/connection/datasource/pre_connect", "/api/connection/datasource/update", + "/api/connection/datasource/identity_color", "/api/connection/datasource/clone", "/api/connection/datasource/connect", "/api/connection/datasource/close", @@ -9036,6 +9274,10 @@ const LEGACY_PATHS: &[&str] = &[ "/api/task/get", "/api/task/stop", "/api/task/download", + "/api/tasks/list", + "/api/tasks/get", + "/api/tasks/active-count", + "/api/tasks/artifact", "/api/sql/format", "/api/sql/valid_select", "/api/sql_parser/get_keywords", @@ -9211,10 +9453,7 @@ pub(crate) fn routes() -> Router { .route("/api/v1/chart/update", post(chart_update_handler)) .route("/api/chart", axum::routing::delete(chart_delete_handler)) .route("/api/common/environment/list_all", get(environment_handler)) - .route( - "/api/database/supported", - get(supported_database_handler), - ) + .route("/api/database/supported", get(supported_database_handler)) .route( "/api/jdbc/driver/upload", post(driver_upload_handler).layer(DefaultBodyLimit::max(MAX_LEGACY_MULTIPART_BYTES)), @@ -9300,6 +9539,10 @@ pub(crate) fn routes() -> Router { "/api/connection/datasource/update", post(update_datasource_handler).put(update_datasource_handler), ) + .route( + "/api/connection/datasource/identity_color", + post(update_datasource_identity_color_handler), + ) .route( "/api/import/sql_file", post(import_sql_file_handler).layer(DefaultBodyLimit::max(MAX_LEGACY_MULTIPART_BYTES)), @@ -9314,6 +9557,16 @@ pub(crate) fn routes() -> Router { .route("/api/task/list", get(transfer_task_list_handler)) .route("/api/task/get", get(transfer_task_get_handler)) .route("/api/task/stop", get(transfer_task_stop_handler)) + .route("/api/tasks/list", get(import_export_task_list_handler)) + .route("/api/tasks/get", get(import_export_task_get_handler)) + .route( + "/api/tasks/active-count", + get(import_export_task_active_count_handler), + ) + .route( + "/api/tasks/artifact", + get(import_export_task_artifact_handler), + ) .route("/api/task/download", get(transfer_task_download_handler)) .route("/api/sql/format", get(sql_format_handler)) .route("/api/sql/valid_select", get(sql_valid_select_handler)) @@ -9834,6 +10087,13 @@ async fn update_datasource_handler( envelope(update_datasource(&application, &request).await) } +async fn update_datasource_identity_color_handler( + State(application): State, + Json(request): Json, +) -> Json> { + envelope(update_datasource_identity_color(&application, &request).await) +} + async fn delete_datasource_handler( State(application): State, Query(query): Query, @@ -9903,6 +10163,33 @@ async fn transfer_task_download_handler( transfer_attachment_response(legacy_transfer_task_download(&application, &query.id).await) } +async fn import_export_task_list_handler( + State(application): State, + Query(query): Query, +) -> Json>> { + envelope(list_import_export_tasks(&application, &query).await) +} + +async fn import_export_task_get_handler( + State(application): State, + Query(query): Query, +) -> Json> { + envelope(get_import_export_task(&application, &query).await) +} + +async fn import_export_task_active_count_handler( + State(application): State, +) -> Json> { + envelope(active_import_export_task_count(&application).await) +} + +async fn import_export_task_artifact_handler( + State(application): State, + Query(query): Query, +) -> Response { + transfer_attachment_response(import_export_task_artifact(&application, &query).await) +} + async fn dml_export_handler( State(application): State, Json(request): Json, @@ -10711,6 +10998,7 @@ mod tests { }), has_secret: true, revision: "1".to_owned(), + identity_color: None, }, ); let json = serde_json::to_string(&response).expect("response serializes"); @@ -10837,6 +11125,7 @@ mod tests { ("post", "/api/converter/dbp/upload"), ("post", "/api/converter/chat2db/upload"), ("post", "/api/converter/datagrip/upload"), + ("post", "/api/connection/datasource/identity_color"), ("post", "/api/namespaces/create"), ("post", "/api/namespaces/update"), ("post", "/api/namespaces/delete"), @@ -10853,6 +11142,10 @@ mod tests { ("get", "/api/task/get"), ("get", "/api/task/stop"), ("get", "/api/task/download"), + ("get", "/api/tasks/list"), + ("get", "/api/tasks/get"), + ("get", "/api/tasks/active-count"), + ("get", "/api/tasks/artifact"), ("post", "/api/rdb/dml/export"), ("post", "/api/rdb/table/generate/class"), ]; diff --git a/apps/chat2db-web/src/legacy_ai.rs b/apps/chat2db-web/src/legacy_ai.rs index c494cbe..ae12d42 100644 --- a/apps/chat2db-web/src/legacy_ai.rs +++ b/apps/chat2db-web/src/legacy_ai.rs @@ -2570,7 +2570,10 @@ mod tests { assert_eq!(response_json(renamed).await["success"], true); let sessions = application - .oneshot(empty_request(Method::GET, "/api/v3/ai/chat/history/sessions")) + .oneshot(empty_request( + Method::GET, + "/api/v3/ai/chat/history/sessions", + )) .await .expect("session list must respond"); let sessions = response_json(sessions).await; diff --git a/apps/frontend/src/generated/contract.ts b/apps/frontend/src/generated/contract.ts index f42da1d..0718722 100644 --- a/apps/frontend/src/generated/contract.ts +++ b/apps/frontend/src/generated/contract.ts @@ -1865,6 +1865,8 @@ export interface components { hasSecret: boolean; /** @description Opaque datasource id. */ id: string; + /** @description Optional `#RRGGBB` identity colour shown by the connection tree. */ + identityColor?: string | null; /** @description User-visible datasource name. */ name: string; /** @description Monotonic revision encoded as a decimal integer. */ diff --git a/contracts/openapi/chat2db-v1.json b/contracts/openapi/chat2db-v1.json index cce64e3..e60a784 100644 --- a/contracts/openapi/chat2db-v1.json +++ b/contracts/openapi/chat2db-v1.json @@ -7211,6 +7211,13 @@ "type": "string", "description": "Opaque datasource id." }, + "identityColor": { + "type": [ + "string", + "null" + ], + "description": "Optional `#RRGGBB` identity colour shown by the connection tree." + }, "name": { "type": "string", "description": "User-visible datasource name." diff --git a/crates/chat2db-contract/src/datasource.rs b/crates/chat2db-contract/src/datasource.rs index 1323be7..158ca62 100644 --- a/crates/chat2db-contract/src/datasource.rs +++ b/crates/chat2db-contract/src/datasource.rs @@ -115,6 +115,9 @@ pub struct Datasource { pub created_at_ms: String, /// Unix epoch milliseconds encoded as a decimal integer. pub updated_at_ms: String, + /// Optional `#RRGGBB` identity colour shown by the connection tree. + #[serde(skip_serializing_if = "Option::is_none")] + pub identity_color: Option, } /// Stable datasource collection returned by list APIs. @@ -142,6 +145,7 @@ mod tests { revision: "9007199254740993".to_owned(), created_at_ms: "1784900000000".to_owned(), updated_at_ms: "1784900000001".to_owned(), + identity_color: None, }; let json = serde_json::to_string(&response).expect("response must serialize"); diff --git a/crates/chat2db-contract/src/datasource_edit.rs b/crates/chat2db-contract/src/datasource_edit.rs index 9aad373..a994533 100644 --- a/crates/chat2db-contract/src/datasource_edit.rs +++ b/crates/chat2db-contract/src/datasource_edit.rs @@ -31,6 +31,9 @@ pub struct DatasourceEditProjection { pub has_secret: bool, /// Monotonic revision encoded as a decimal integer. pub revision: String, + /// Optional `#RRGGBB` identity colour shown by the connection tree. + #[serde(skip_serializing_if = "Option::is_none")] + pub identity_color: Option, } impl Debug for DatasourceEditProjection { @@ -47,6 +50,7 @@ impl Debug for DatasourceEditProjection { .field("ssh", &self.ssh) .field("has_secret", &self.has_secret) .field("revision", &self.revision) + .field("identity_color", &self.identity_color) .finish() } } @@ -72,6 +76,7 @@ mod tests { ssh: None, has_secret: true, revision: "3".to_owned(), + identity_color: None, }; let json = serde_json::to_string(&projection).expect("projection serializes"); diff --git a/crates/chat2db-core/src/convert.rs b/crates/chat2db-core/src/convert.rs index 091463b..cb66bdc 100644 --- a/crates/chat2db-core/src/convert.rs +++ b/crates/chat2db-core/src/convert.rs @@ -92,6 +92,7 @@ pub(crate) fn datasource(record: storage::DatasourceRecord) -> contract::Datasou revision: record.revision.to_string(), created_at_ms: record.created_at_ms.to_string(), updated_at_ms: record.updated_at_ms.to_string(), + identity_color: record.identity_color, } } diff --git a/crates/chat2db-core/src/datasource_edit.rs b/crates/chat2db-core/src/datasource_edit.rs index 6278871..b880723 100644 --- a/crates/chat2db-core/src/datasource_edit.rs +++ b/crates/chat2db-core/src/datasource_edit.rs @@ -64,6 +64,7 @@ impl Application { ssh: projected.ssh, has_secret, revision: record.revision.to_string(), + identity_color: record.identity_color, }) } diff --git a/crates/chat2db-core/src/lib.rs b/crates/chat2db-core/src/lib.rs index 9f458cc..5d36675 100644 --- a/crates/chat2db-core/src/lib.rs +++ b/crates/chat2db-core/src/lib.rs @@ -672,6 +672,28 @@ impl Application { .map(convert::datasource) } + /// Replaces only the identity colour of one datasource. + /// + /// The colour is presentation metadata and does not participate in the + /// revision CAS used for public datasource fields. + /// + /// # Errors + /// + /// Returns validation, not-found, availability, or storage failures. + pub async fn update_datasource_identity_color( + &self, + id: &str, + identity_color: Option, + ) -> Result { + let storage = self.require_storage()?; + let id = id.to_owned(); + storage_call(move || { + storage.update_datasource_identity_color(&id, identity_color.as_deref()) + }) + .await + .map(convert::datasource) + } + fn require_managed_driver(&self, driver_id: &str) -> Result<(), AppError> { if let Some(driver) = self.native_driver_for_datasource_driver_id(driver_id) && (driver.connection().is_some() diff --git a/crates/chat2db-storage/migrations/009_community_client_routes.sql b/crates/chat2db-storage/migrations/009_community_client_routes.sql new file mode 100644 index 0000000..9c81017 --- /dev/null +++ b/crates/chat2db-storage/migrations/009_community_client_routes.sql @@ -0,0 +1,29 @@ +BEGIN IMMEDIATE; + +ALTER TABLE datasources ADD COLUMN identity_color TEXT + CHECK (identity_color IS NULL OR length(CAST(identity_color AS BLOB)) <= 16); + +ALTER TABLE transfer_tasks ADD COLUMN task_type TEXT + CHECK (task_type IS NULL OR length(CAST(task_type AS BLOB)) <= 64); + +ALTER TABLE transfer_tasks ADD COLUMN format TEXT + CHECK (format IS NULL OR length(CAST(format AS BLOB)) <= 32); + +CREATE TABLE transfer_task_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id INTEGER NOT NULL REFERENCES transfer_tasks(id) ON DELETE CASCADE, + sequence INTEGER NOT NULL CHECK (sequence >= 1), + level TEXT NOT NULL CHECK (level IN ('INFO', 'WARN', 'ERROR')), + code TEXT NOT NULL DEFAULT '', + stage TEXT, + message TEXT NOT NULL, + details_json TEXT, + created_at_ms INTEGER NOT NULL, + UNIQUE (task_id, sequence) +) STRICT; + +CREATE INDEX transfer_task_events_task_idx + ON transfer_task_events (task_id, sequence); + +PRAGMA user_version = 9; +COMMIT; diff --git a/crates/chat2db-storage/src/datasource.rs b/crates/chat2db-storage/src/datasource.rs index 7f1c159..379292f 100644 --- a/crates/chat2db-storage/src/datasource.rs +++ b/crates/chat2db-storage/src/datasource.rs @@ -63,6 +63,8 @@ pub struct DatasourceRecord { pub created_at_ms: i64, /// Last update time as Unix epoch milliseconds. pub updated_at_ms: i64, + /// Optional `#RRGGBB` identity colour shown by the connection tree. + pub identity_color: Option, } impl std::fmt::Debug for DatasourceRecord { @@ -76,6 +78,7 @@ impl std::fmt::Debug for DatasourceRecord { .field("revision", &self.revision) .field("created_at_ms", &self.created_at_ms) .field("updated_at_ms", &self.updated_at_ms) + .field("identity_color", &self.identity_color) .finish() } } @@ -128,6 +131,7 @@ impl Storage { revision: 1, created_at_ms: timestamp, updated_at_ms: timestamp, + identity_color: None, }; let mutation = (|| -> Result<(), StorageError> { @@ -189,7 +193,7 @@ impl Storage { let connection = self.connection()?; let mut statement = connection.prepare( "SELECT id, name, driver_id, secret_ref, - revision, created_at_ms, updated_at_ms + revision, created_at_ms, updated_at_ms, identity_color FROM datasources ORDER BY created_at_ms, id", )?; let rows = statement.query_map([], raw_datasource)?; @@ -249,6 +253,7 @@ impl Storage { revision: next_revision, created_at_ms: current.created_at_ms, updated_at_ms: timestamp, + identity_color: current.identity_color.clone(), }; let mutation = (|| -> Result<(), StorageError> { @@ -309,6 +314,35 @@ impl Storage { } } + /// Replaces only the identity colour of one datasource. + /// + /// The colour is presentation metadata, so it does not participate in the + /// revision CAS used for public fields. + /// + /// # Errors + /// + /// Returns validation, not-found, or `SQLite` failures. + pub fn update_datasource_identity_color( + &self, + id: &str, + identity_color: Option<&str>, + ) -> Result { + let identity_color = validate_identity_color(identity_color)?; + let timestamp = now_millis()?; + let mut connection = self.connection()?; + let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; + let changed = transaction.execute( + "UPDATE datasources SET identity_color = ?1, updated_at_ms = ?2 WHERE id = ?3", + params![identity_color, timestamp, id], + )?; + if changed == 0 { + return Err(StorageError::DatasourceNotFound(id.to_owned())); + } + transaction.commit()?; + self.get_datasource(id)? + .ok_or_else(|| StorageError::DatasourceNotFound(id.to_owned())) + } + /// Deletes a datasource using revision CAS and retires its vault reference. /// /// # Errors @@ -624,7 +658,30 @@ fn validate_datasource(name: &str, driver_id: &str) -> Result<(), StorageError> Ok(()) } -type RawDatasource = (String, String, String, Option, i64, i64, i64); +/// Normalizes the optional `#RRGGBB` identity colour. +fn validate_identity_color(value: Option<&str>) -> Result, StorageError> { + let Some(value) = value.map(str::trim).filter(|value| !value.is_empty()) else { + return Ok(None); + }; + let hex = value.strip_prefix('#').unwrap_or_default(); + if hex.len() != 6 || !hex.bytes().all(|byte| byte.is_ascii_hexdigit()) { + return Err(StorageError::InvalidDatasource( + "identity colour must use the #RRGGBB form", + )); + } + Ok(Some(format!("#{}", hex.to_ascii_uppercase()))) +} + +type RawDatasource = ( + String, + String, + String, + Option, + i64, + i64, + i64, + Option, +); fn raw_datasource(row: &rusqlite::Row<'_>) -> rusqlite::Result { Ok(( @@ -635,11 +692,12 @@ fn raw_datasource(row: &rusqlite::Row<'_>) -> rusqlite::Result { row.get(4)?, row.get(5)?, row.get(6)?, + row.get(7)?, )) } fn decode_datasource(raw: RawDatasource) -> Result { - let (id, name, driver_id, secret_ref, revision, created, updated) = raw; + let (id, name, driver_id, secret_ref, revision, created, updated, identity_color) = raw; Ok(DatasourceRecord { id, name, @@ -649,6 +707,7 @@ fn decode_datasource(raw: RawDatasource) -> Result Result<(), StorageError> { connection, include_str!("../migrations/008_community_dashboard.sql"), )?; + version = 8; + } + if version == 8 { + apply_migration( + connection, + include_str!("../migrations/009_community_client_routes.sql"), + )?; } Ok(()) } From 0853f8d86dd3dfd610375b8a3af9cef4bde083dd Mon Sep 17 00:00:00 2001 From: zgq Date: Fri, 9 Oct 2026 21:21:20 +0800 Subject: [PATCH 4/9] feat(legacy): let the task center delete finished tasks `DELETE /api/tasks/delete` removes one finished transfer task and its artifact. `Storage::delete_transfer_task` deletes only terminal tasks (rejecting queued or running ones with `InvalidTransfer`), unlinks the managed artifact file, and `Application::delete_transfer_task` exposes it to the delivery layers. The route is registered in the dispatch match, `LEGACY_PATHS`, the Axum router, and the transfer path-coverage list. Tests: a storage test covers delete-after-completion, the active-task rejection, and the not-found result after a second delete. --- apps/chat2db-web/src/legacy.rs | 29 +++++++++++ crates/chat2db-core/src/transfer/mod.rs | 11 ++++ crates/chat2db-storage/src/transfer.rs | 69 +++++++++++++++++++++++++ 3 files changed, 109 insertions(+) diff --git a/apps/chat2db-web/src/legacy.rs b/apps/chat2db-web/src/legacy.rs index 8dbc3e8..1274c90 100644 --- a/apps/chat2db-web/src/legacy.rs +++ b/apps/chat2db-web/src/legacy.rs @@ -3168,6 +3168,16 @@ pub(crate) async fn import_export_task_artifact( legacy_transfer_task_download(application, &query.task_id).await } +pub(crate) async fn delete_import_export_task( + application: &Application, + query: &LegacyImportExportTaskIdQuery, +) -> LegacyResult<()> { + application + .delete_transfer_task(legacy_transfer_task_id(&query.task_id)?) + .await?; + Ok(()) +} + pub(crate) async fn export_legacy_mysql_dml( application: &Application, request: &LegacyDmlExportRequest, @@ -8744,6 +8754,12 @@ async fn dispatch_inner( ("get", "/api/tasks/active-count") => { serialized(active_import_export_task_count(application).await) } + ("delete", "/api/tasks/delete") => { + match decode::(request.message) { + Ok(query) => serialized(delete_import_export_task(application, &query).await), + Err(error) => Err(error), + } + } ("get", "/api/sql/format") => match decode::(request.message) { Ok(query) => serialized(format_legacy_sql(application, &query).await), Err(error) => Err(error), @@ -9278,6 +9294,7 @@ const LEGACY_PATHS: &[&str] = &[ "/api/tasks/get", "/api/tasks/active-count", "/api/tasks/artifact", + "/api/tasks/delete", "/api/sql/format", "/api/sql/valid_select", "/api/sql_parser/get_keywords", @@ -9567,6 +9584,10 @@ pub(crate) fn routes() -> Router { "/api/tasks/artifact", get(import_export_task_artifact_handler), ) + .route( + "/api/tasks/delete", + axum::routing::delete(import_export_task_delete_handler), + ) .route("/api/task/download", get(transfer_task_download_handler)) .route("/api/sql/format", get(sql_format_handler)) .route("/api/sql/valid_select", get(sql_valid_select_handler)) @@ -10190,6 +10211,13 @@ async fn import_export_task_artifact_handler( transfer_attachment_response(import_export_task_artifact(&application, &query).await) } +async fn import_export_task_delete_handler( + State(application): State, + Query(query): Query, +) -> Json> { + envelope(delete_import_export_task(&application, &query).await) +} + async fn dml_export_handler( State(application): State, Json(request): Json, @@ -11146,6 +11174,7 @@ mod tests { ("get", "/api/tasks/get"), ("get", "/api/tasks/active-count"), ("get", "/api/tasks/artifact"), + ("delete", "/api/tasks/delete"), ("post", "/api/rdb/dml/export"), ("post", "/api/rdb/table/generate/class"), ]; diff --git a/crates/chat2db-core/src/transfer/mod.rs b/crates/chat2db-core/src/transfer/mod.rs index ae4d38b..a4a8d24 100644 --- a/crates/chat2db-core/src/transfer/mod.rs +++ b/crates/chat2db-core/src/transfer/mod.rs @@ -616,6 +616,17 @@ impl Application { Ok(()) } + /// Deletes one finished transfer task and its artifact. + /// + /// # Errors + /// + /// Returns not-found, active-task, or durable-storage failures. + pub async fn delete_transfer_task(&self, task_id: i64) -> Result<(), AppError> { + let storage = self.require_storage()?; + storage_call(move || storage.delete_transfer_task(task_id)).await?; + Ok(()) + } + /// Resolves a managed artifact and its owner-only local path for a delivery adapter. /// /// # Errors diff --git a/crates/chat2db-storage/src/transfer.rs b/crates/chat2db-storage/src/transfer.rs index bdaa271..50b94bf 100644 --- a/crates/chat2db-storage/src/transfer.rs +++ b/crates/chat2db-storage/src/transfer.rs @@ -875,6 +875,49 @@ impl Storage { }) } + /// Deletes one terminal transfer task and its artifact file. + /// + /// # Errors + /// + /// Returns not-found, active-task, or `SQLite`/filesystem failures. + pub fn delete_transfer_task(&self, id: i64) -> Result<(), StorageError> { + let mut connection = self.connection()?; + let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; + let status = transaction + .query_row( + "SELECT status FROM transfer_tasks WHERE id = ?1", + [id], + |row| row.get::<_, String>(0), + ) + .optional()? + .ok_or(StorageError::TransferTaskNotFound(id))?; + if !StoredTransferTaskStatus::parse(&status)?.is_terminal() { + return Err(StorageError::InvalidTransfer( + "only finished transfer tasks can be deleted", + )); + } + let storage_name = transaction + .query_row( + "SELECT storage_name FROM transfer_artifacts WHERE task_id = ?1", + [id], + |row| row.get::<_, String>(0), + ) + .optional()?; + transaction.execute("DELETE FROM transfer_tasks WHERE id = ?1", [id])?; + transaction.commit()?; + if let Some(name) = storage_name { + validate_storage_name(&name)?; + let path = self.inner.artifacts_dir.join(name); + match fs::remove_file(&path) { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(StorageError::io(path, error)), + } + sync_directory(&self.inner.artifacts_dir)?; + } + Ok(()) + } + fn prune_transfer_tasks(&self) -> Result<(), StorageError> { let mut connection = self.connection()?; let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; @@ -1188,6 +1231,32 @@ mod tests { .id } + #[test] + fn delete_transfer_task_removes_finished_tasks_and_rejects_active_ones() { + let directory = TempDir::new().expect("temp directory"); + let storage = open(&directory); + let active = task(&storage, 1); + assert!(matches!( + storage.delete_transfer_task(active), + Err(StorageError::InvalidTransfer(_)) + )); + storage.start_transfer_task(active).expect("task starts"); + storage + .complete_transfer_task(active, "done") + .expect("task completes"); + storage.delete_transfer_task(active).expect("task deletes"); + assert!( + storage + .get_transfer_task(active) + .expect("lookup succeeds") + .is_none() + ); + assert!(matches!( + storage.delete_transfer_task(active), + Err(StorageError::TransferTaskNotFound(_)) + )); + } + #[test] fn artifact_publish_is_atomic_and_completes_the_task() { let directory = TempDir::new().expect("temp directory"); From ac1edf3e46654dc8f3949de13ab6b27c3719774c Mon Sep 17 00:00:00 2001 From: zgq Date: Fri, 9 Oct 2026 21:25:31 +0800 Subject: [PATCH 5/9] feat(legacy): record and serve task-center progress events The task-center log view polls ordered events for one task. - The transfer storage now appends an ordered event inside the same transaction as the progress or terminal transition it describes: INFO for progress and completion, WARN for cancellation, ERROR for failure. Events carry a per-task sequence, level, code, optional stage, message, and timestamp. - `Storage::list_transfer_task_events` reads a sequence window with a bounded limit (1..=200) and rejects unknown tasks. - The contract exposes `TransferTaskEvent`, Core projects it, and `GET /api/tasks/events` answers the client's `{eventId, taskId, sequence, level, code, stage?, message, createdAt}` shape as a bare array. Tests: a storage test covers sequence ordering across progress, completion, and failure events plus the invalid-limit rejection. --- apps/chat2db-web/src/legacy.rs | 71 +++++++++ crates/chat2db-contract/src/lib.rs | 3 +- crates/chat2db-contract/src/transfer.rs | 23 +++ crates/chat2db-core/src/transfer/mod.rs | 38 ++++- crates/chat2db-storage/src/transfer.rs | 192 +++++++++++++++++++++++- 5 files changed, 318 insertions(+), 9 deletions(-) diff --git a/apps/chat2db-web/src/legacy.rs b/apps/chat2db-web/src/legacy.rs index 1274c90..711c0bf 100644 --- a/apps/chat2db-web/src/legacy.rs +++ b/apps/chat2db-web/src/legacy.rs @@ -495,6 +495,34 @@ pub struct LegacyImportExportTaskIdQuery { pub task_id: LegacyIdentifier, } +/// Event-window query used by the task-center log view. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportExportEventQuery { + pub task_id: LegacyIdentifier, + #[serde(default)] + pub after_sequence: Option, + #[serde(default)] + pub before_sequence: Option, + #[serde(default)] + pub limit: Option, +} + +/// One task-center log event in the shape the newer Community panel expects. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportExportEvent { + pub event_id: i64, + pub task_id: i64, + pub sequence: u64, + pub level: String, + pub code: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub stage: Option, + pub message: String, + pub created_at: i64, +} + /// One task-center row in the shape the newer Community panel expects. #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] @@ -3178,6 +3206,33 @@ pub(crate) async fn delete_import_export_task( Ok(()) } +pub(crate) async fn list_import_export_task_events( + application: &Application, + query: &LegacyImportExportEventQuery, +) -> LegacyResult> { + let events = application + .list_transfer_task_events( + legacy_transfer_task_id(&query.task_id)?, + query.after_sequence, + query.before_sequence, + query.limit, + ) + .await?; + Ok(events + .into_iter() + .map(|event| LegacyImportExportEvent { + event_id: event.id, + task_id: event.task_id, + sequence: event.sequence, + level: event.level, + code: event.code, + stage: event.stage, + message: event.message, + created_at: legacy_epoch_millis(&event.created_at_ms), + }) + .collect()) +} + pub(crate) async fn export_legacy_mysql_dml( application: &Application, request: &LegacyDmlExportRequest, @@ -8760,6 +8815,12 @@ async fn dispatch_inner( Err(error) => Err(error), } } + ("get", "/api/tasks/events") => { + match decode::(request.message) { + Ok(query) => serialized(list_import_export_task_events(application, &query).await), + Err(error) => Err(error), + } + } ("get", "/api/sql/format") => match decode::(request.message) { Ok(query) => serialized(format_legacy_sql(application, &query).await), Err(error) => Err(error), @@ -9295,6 +9356,7 @@ const LEGACY_PATHS: &[&str] = &[ "/api/tasks/active-count", "/api/tasks/artifact", "/api/tasks/delete", + "/api/tasks/events", "/api/sql/format", "/api/sql/valid_select", "/api/sql_parser/get_keywords", @@ -9588,6 +9650,7 @@ pub(crate) fn routes() -> Router { "/api/tasks/delete", axum::routing::delete(import_export_task_delete_handler), ) + .route("/api/tasks/events", get(import_export_task_events_handler)) .route("/api/task/download", get(transfer_task_download_handler)) .route("/api/sql/format", get(sql_format_handler)) .route("/api/sql/valid_select", get(sql_valid_select_handler)) @@ -10218,6 +10281,13 @@ async fn import_export_task_delete_handler( envelope(delete_import_export_task(&application, &query).await) } +async fn import_export_task_events_handler( + State(application): State, + Query(query): Query, +) -> Json>> { + envelope(list_import_export_task_events(&application, &query).await) +} + async fn dml_export_handler( State(application): State, Json(request): Json, @@ -11175,6 +11245,7 @@ mod tests { ("get", "/api/tasks/active-count"), ("get", "/api/tasks/artifact"), ("delete", "/api/tasks/delete"), + ("get", "/api/tasks/events"), ("post", "/api/rdb/dml/export"), ("post", "/api/rdb/table/generate/class"), ]; diff --git a/crates/chat2db-contract/src/lib.rs b/crates/chat2db-contract/src/lib.rs index 5b0e122..6f7b0a9 100644 --- a/crates/chat2db-contract/src/lib.rs +++ b/crates/chat2db-contract/src/lib.rs @@ -115,7 +115,8 @@ pub use transfer::{ DmlExportFormat, DmlExportRequest, DmlExportSize, GenerateMysqlClassRequest, GeneratedMysqlClassSet, ImportFileRequest, OtherFileExportRequest, SqlFileExportRequest, TabularImportEncoding, TransferArtifact, TransferFileFormat, TransferSqlScope, TransferTask, - TransferTaskAccepted, TransferTaskKind, TransferTaskPage, TransferTaskStatus, + TransferTaskAccepted, TransferTaskEvent, TransferTaskKind, TransferTaskPage, + TransferTaskStatus, }; pub use workspace::{ AssignDatasourceNamespaceRequest, CreateWorkspaceNamespaceRequest, MoveWorkspaceNodeRequest, diff --git a/crates/chat2db-contract/src/transfer.rs b/crates/chat2db-contract/src/transfer.rs index 83b75be..5ba7408 100644 --- a/crates/chat2db-contract/src/transfer.rs +++ b/crates/chat2db-contract/src/transfer.rs @@ -233,6 +233,29 @@ pub struct GeneratedMysqlClassSet { pub files: Vec, } +/// One ordered progress event of a durable transfer task. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub struct TransferTaskEvent { + /// Event row id. + pub id: i64, + /// Owning transfer task id. + pub task_id: i64, + /// Monotonic per-task sequence starting at one. + pub sequence: u64, + /// `INFO`, `WARN`, or `ERROR`. + pub level: String, + /// Stable machine-readable event code, empty when none applies. + pub code: String, + /// Optional pipeline stage. + #[serde(skip_serializing_if = "Option::is_none")] + pub stage: Option, + /// Human-readable event message. + pub message: String, + /// Unix epoch milliseconds encoded as a decimal integer. + pub created_at_ms: String, +} + const fn default_true() -> bool { true } diff --git a/crates/chat2db-core/src/transfer/mod.rs b/crates/chat2db-core/src/transfer/mod.rs index a4a8d24..6a0cf0c 100644 --- a/crates/chat2db-core/src/transfer/mod.rs +++ b/crates/chat2db-core/src/transfer/mod.rs @@ -11,8 +11,8 @@ use std::{ use chat2db_contract::{ DmlExportFormat, DmlExportRequest, DmlExportSize, GenerateMysqlClassRequest, GeneratedMysqlClassSet, ImportFileRequest, OtherFileExportRequest, SqlFileExportRequest, - TransferArtifact, TransferFileFormat, TransferTask, TransferTaskAccepted, TransferTaskKind, - TransferTaskPage, TransferTaskStatus, + TransferArtifact, TransferFileFormat, TransferTask, TransferTaskAccepted, TransferTaskEvent, + TransferTaskKind, TransferTaskPage, TransferTaskStatus, }; use chat2db_storage::{ CreateTransferTask, ResolvedTransferArtifact, Storage, StorageError, StoredTransferTaskKind, @@ -27,6 +27,7 @@ use tokio_util::sync::CancellationToken; use crate::{AppError, Application, storage_call}; const MAX_TASK_PAGE_SIZE: u32 = 100; +const MAX_TASK_EVENT_LIMIT: u32 = 200; const MAX_TRANSFER_FAILURE_MESSAGE_BYTES: usize = 64 * 1024; const TRANSFER_FAILURE_TRUNCATION_SUFFIX: &str = "\n[truncated]"; const TERMINAL_RETRY_INITIAL_DELAY: Duration = Duration::from_millis(25); @@ -627,6 +628,39 @@ impl Application { Ok(()) } + /// Lists ordered progress events of one transfer task. + /// + /// # Errors + /// + /// Returns validation, not-found, or durable-storage failures. + pub async fn list_transfer_task_events( + &self, + task_id: i64, + after_sequence: Option, + before_sequence: Option, + limit: Option, + ) -> Result, AppError> { + let limit = limit.unwrap_or(MAX_TASK_EVENT_LIMIT); + let storage = self.require_storage()?; + let records = storage_call(move || { + storage.list_transfer_task_events(task_id, after_sequence, before_sequence, limit) + }) + .await?; + Ok(records + .into_iter() + .map(|record| TransferTaskEvent { + id: record.id, + task_id: record.task_id, + sequence: record.sequence, + level: record.level, + code: record.code, + stage: record.stage, + message: record.message, + created_at_ms: record.created_at_ms.to_string(), + }) + .collect()) + } + /// Resolves a managed artifact and its owner-only local path for a delivery adapter. /// /// # Errors diff --git a/crates/chat2db-storage/src/transfer.rs b/crates/chat2db-storage/src/transfer.rs index 50b94bf..dafa377 100644 --- a/crates/chat2db-storage/src/transfer.rs +++ b/crates/chat2db-storage/src/transfer.rs @@ -18,6 +18,7 @@ const MAX_LOG_BYTES: usize = 256 * 1024; const MAX_LOG_BYTES_I64: i64 = 256 * 1024; const MAX_FILE_NAME_BYTES: usize = 1_024; const MAX_MEDIA_TYPE_BYTES: usize = 255; +const MAX_TASK_EVENT_PAGE: u32 = 200; /// Durable transfer category. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -149,6 +150,19 @@ pub struct ResolvedTransferArtifact { pub file: File, } +/// One ordered progress event of a durable transfer task. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TransferTaskEventRecord { + pub id: i64, + pub task_id: i64, + pub sequence: u64, + pub level: String, + pub code: String, + pub stage: Option, + pub message: String, + pub created_at_ms: i64, +} + /// Startup cleanup and task recovery report. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct TransferRecoveryReport { @@ -427,6 +441,65 @@ impl Storage { .map_err(StorageError::from) } + /// Lists ordered events of one durable task inside a sequence window. + /// + /// # Errors + /// + /// Returns validation, not-found, range, or `SQLite` failures. + pub fn list_transfer_task_events( + &self, + task_id: i64, + after_sequence: Option, + before_sequence: Option, + limit: u32, + ) -> Result, StorageError> { + if limit == 0 || limit > MAX_TASK_EVENT_PAGE { + return Err(StorageError::InvalidTransfer( + "event limit must be between 1 and 200", + )); + } + let connection = self.connection()?; + ensure_task_exists(&connection, task_id)?; + let after = sequence_bound(after_sequence)?.unwrap_or(0); + let before = sequence_bound(before_sequence)?.unwrap_or(i64::MAX); + let mut statement = connection.prepare( + "SELECT id, task_id, sequence, level, code, stage, message, created_at_ms + FROM transfer_task_events + WHERE task_id = ?1 AND sequence > ?2 AND sequence < ?3 + ORDER BY sequence + LIMIT ?4", + )?; + let rows = + statement.query_map(params![task_id, after, before, i64::from(limit)], |row| { + Ok(( + row.get::<_, i64>(0)?, + row.get::<_, i64>(1)?, + row.get::<_, i64>(2)?, + row.get::<_, String>(3)?, + row.get::<_, String>(4)?, + row.get::<_, Option>(5)?, + row.get::<_, String>(6)?, + row.get::<_, i64>(7)?, + )) + })?; + let mut events = Vec::new(); + for row in rows { + let (id, task_id, sequence, level, code, stage, message, created_at_ms) = row?; + events.push(TransferTaskEventRecord { + id, + task_id, + sequence: u64::try_from(sequence) + .map_err(|_| StorageError::NumericRange("event sequence"))?, + level, + code, + stage, + message, + created_at_ms, + }); + } + Ok(events) + } + /// Transitions a queued task to running. /// /// # Errors @@ -476,8 +549,9 @@ impl Storage { .map_err(|_| StorageError::NumericRange("transfer progress total")) }) .transpose()?; - let connection = self.connection()?; - let updated = connection.execute( + let mut connection = self.connection()?; + let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; + let updated = transaction.execute( "UPDATE transfer_tasks SET progress_current = MAX(progress_current, ?2), progress_total = COALESCE(?3, progress_total), @@ -498,8 +572,19 @@ impl Storage { ], )?; if updated == 1 { + append_transfer_task_event( + &transaction, + id, + "INFO", + "", + None, + info.unwrap_or(description), + timestamp, + )?; + transaction.commit()?; return Ok(()); } + drop(transaction); ensure_task_exists(&connection, id)?; Err(StorageError::InvalidTransfer( "task progress can only update while running", @@ -584,8 +669,9 @@ impl Storage { pub fn complete_transfer_task(&self, id: i64, message: &str) -> Result<(), StorageError> { validate_text(message, MAX_LOG_BYTES, "task completion message")?; let timestamp = now_millis()?; - let connection = self.connection()?; - let updated = connection.execute( + let mut connection = self.connection()?; + let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; + let updated = transaction.execute( "UPDATE transfer_tasks SET status = 'succeeded', progress_description = 'Completed', info_log = CASE WHEN ?2 = '' THEN info_log @@ -601,11 +687,26 @@ impl Storage { ], )?; if updated != 1 { + drop(transaction); ensure_task_exists(&connection, id)?; return Err(StorageError::InvalidTransfer( "task cannot complete in its current state", )); } + append_transfer_task_event( + &transaction, + id, + "INFO", + "", + None, + if message.is_empty() { + "Completed" + } else { + message + }, + timestamp, + )?; + transaction.commit()?; // The terminal state is already durable. Retention is recoverable // maintenance and must not make callers retry a committed transition. let _ = self.prune_transfer_tasks(); @@ -624,8 +725,9 @@ impl Storage { } validate_text(message, MAX_LOG_BYTES, "task terminal message")?; let timestamp = now_millis()?; - let connection = self.connection()?; - let updated = connection.execute( + let mut connection = self.connection()?; + let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; + let updated = transaction.execute( "UPDATE transfer_tasks SET status = ?2, progress_description = ?3, error_log = ?4, updated_at_ms = ?5, finished_at_ms = ?5 @@ -633,11 +735,19 @@ impl Storage { params![id, status.as_str(), description, message, timestamp], )?; if updated != 1 { + drop(transaction); ensure_task_exists(&connection, id)?; return Err(StorageError::InvalidTransfer( "task is already in a terminal state", )); } + let level = if status == StoredTransferTaskStatus::Cancelled { + "WARN" + } else { + "ERROR" + }; + append_transfer_task_event(&transaction, id, level, "", None, message, timestamp)?; + transaction.commit()?; // The terminal state is already durable. Retention is recoverable // maintenance and must not make callers retry a committed transition. let _ = self.prune_transfer_tasks(); @@ -1155,6 +1265,39 @@ fn validate_text(value: &str, maximum: usize, field: &'static str) -> Result<(), Ok(()) } +/// Appends one ordered event to a task inside an open transaction. +fn append_transfer_task_event( + transaction: &rusqlite::Transaction<'_>, + task_id: i64, + level: &str, + code: &str, + stage: Option<&str>, + message: &str, + timestamp: i64, +) -> Result<(), StorageError> { + if message.is_empty() { + return Ok(()); + } + validate_text(message, MAX_LOG_BYTES, "task event message")?; + transaction.execute( + "INSERT INTO transfer_task_events ( + task_id, sequence, level, code, stage, message, details_json, created_at_ms + ) VALUES ( + ?1, + (SELECT COALESCE(MAX(sequence), 0) + 1 FROM transfer_task_events WHERE task_id = ?1), + ?2, ?3, ?4, ?5, NULL, ?6 + )", + params![task_id, level, code, stage, message, timestamp], + )?; + Ok(()) +} + +fn sequence_bound(value: Option) -> Result, StorageError> { + value + .map(|value| i64::try_from(value).map_err(|_| StorageError::NumericRange("event sequence"))) + .transpose() +} + fn ensure_task_exists(connection: &rusqlite::Connection, id: i64) -> Result<(), StorageError> { let exists = connection.query_row( "SELECT EXISTS(SELECT 1 FROM transfer_tasks WHERE id = ?1)", @@ -1257,6 +1400,43 @@ mod tests { )); } + #[test] + fn task_events_record_progress_and_terminal_outcomes_in_sequence() { + let directory = TempDir::new().expect("temp directory"); + let storage = open(&directory); + let id = task(&storage, 1); + storage.start_transfer_task(id).expect("task starts"); + storage + .update_transfer_progress(id, 1, Some(2), "Importing", Some("Importing table rows")) + .expect("progress updates"); + storage + .update_transfer_progress(id, 2, Some(2), "Finishing", None) + .expect("progress updates"); + storage + .fail_transfer_task(id, "Import failed") + .expect("task fails"); + + let events = storage + .list_transfer_task_events(id, None, None, 10) + .expect("events list"); + assert_eq!(events.len(), 3); + assert_eq!(events[0].sequence, 1); + assert_eq!(events[0].level, "INFO"); + assert_eq!(events[0].message, "Importing table rows"); + assert_eq!(events[1].message, "Finishing"); + assert_eq!(events[2].level, "ERROR"); + assert_eq!(events[2].message, "Import failed"); + + let window = storage + .list_transfer_task_events(id, Some(1), None, 10) + .expect("window lists"); + assert_eq!(window.len(), 2); + assert!(matches!( + storage.list_transfer_task_events(id, None, None, 0), + Err(StorageError::InvalidTransfer(_)) + )); + } + #[test] fn artifact_publish_is_atomic_and_completes_the_task() { let directory = TempDir::new().expect("temp directory"); From dc3bf837f7d5d774b25bfe74daf65893c7233292 Mon Sep 17 00:00:00 2001 From: zgq Date: Fri, 9 Oct 2026 21:27:43 +0800 Subject: [PATCH 6/9] feat(legacy): accept task-center import and export submissions `POST /api/tasks/export` and `POST /api/tasks/import` now translate the panel's submission into the existing durable transfer services: - `SQL_EXPORT` runs the SQL-file export, `TABLE_DATA_EXPORT` the format-specific table export, and both return `{taskId}` for the panel to track. - `DATA_FILE_IMPORT` and `SQL_FILE_IMPORT` import either the desktop `sourceFile` path or a staged `fileId` resolved through the managed artifact store, so the web flow works once the preview upload lands. - `QUERY_RESULT_EXPORT` answers with an explicit `unsupported_task_type` failure: query results are exported synchronously today and have no durable task form, so the panel reports the gap instead of silently doing nothing. Unknown task types are rejected as `invalid_task_type`. Tests: 81 chat2db-web tests pass, including a new route test for the unsupported and invalid submission branches. --- apps/chat2db-web/src/legacy.rs | 204 +++++++++++++++++++++++++++++++++ 1 file changed, 204 insertions(+) diff --git a/apps/chat2db-web/src/legacy.rs b/apps/chat2db-web/src/legacy.rs index 711c0bf..4d98b66 100644 --- a/apps/chat2db-web/src/legacy.rs +++ b/apps/chat2db-web/src/legacy.rs @@ -508,6 +508,46 @@ pub struct LegacyImportExportEventQuery { pub limit: Option, } +/// Accepted task id returned to the task-center panel. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportExportTaskAccepted { + pub task_id: i64, +} + +/// Submission posted by the task-center import/export panel. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportExportSubmitRequest { + pub data_source_id: LegacyIdentifier, + #[serde(default)] + pub database_name: String, + #[serde(default)] + pub schema_name: String, + #[serde(default)] + pub table_name: String, + #[serde(default)] + pub table_names: Vec, + #[serde(default)] + pub task_type: String, + #[serde(default)] + pub task_name: String, + #[serde(default)] + pub format: String, + #[serde(default)] + pub scope: String, + #[serde(default = "default_true")] + pub contain_data: bool, + #[serde(default = "default_true")] + pub contains_header: bool, + #[serde(default)] + pub source_file: String, + #[serde(default)] + pub file_id: String, + #[serde(default)] + pub export_path: String, +} + /// One task-center log event in the shape the newer Community panel expects. #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] @@ -3233,6 +3273,107 @@ pub(crate) async fn list_import_export_task_events( .collect()) } +async fn submitted_import_path( + application: &Application, + request: &LegacyImportExportSubmitRequest, +) -> LegacyResult { + if let Some(path) = non_blank(&request.source_file) { + return Ok(path); + } + if let Some(file_id) = non_blank(&request.file_id) { + let download = application.transfer_artifact_download(&file_id).await?; + return download + .path + .to_str() + .map(str::to_owned) + .ok_or_else(|| LegacyFailure { + code: "invalid_import_path".to_owned(), + message: "The staged import file path cannot be represented as UTF-8".to_owned(), + }); + } + Err(LegacyFailure::invalid( + "invalid_import_request", + "sourceFile or fileId is required", + )) +} + +pub(crate) async fn submit_import_export_task( + application: &Application, + request: &LegacyImportExportSubmitRequest, +) -> LegacyResult { + let task_type = request.task_type.trim().to_ascii_uppercase(); + match task_type.as_str() { + "SQL_EXPORT" => { + let task_id = export_legacy_mysql_sql_file( + application, + &LegacySqlFileExportRequest { + data_source_id: request.data_source_id.clone(), + database_name: request.database_name.clone(), + schema_name: request.schema_name.clone(), + table_name: request.table_name.clone(), + table_names: request.table_names.clone(), + scope: request.scope.clone(), + contain_data: request.contain_data, + export_path: String::new(), + }, + ) + .await?; + Ok(LegacyImportExportTaskAccepted { task_id }) + } + "TABLE_DATA_EXPORT" => { + let task_id = export_legacy_mysql_other_file( + application, + &LegacyOtherFileExportRequest { + data_source_id: request.data_source_id.clone(), + database_name: request.database_name.clone(), + schema_name: request.schema_name.clone(), + table_name: request.table_name.clone(), + table_names: request.table_names.clone(), + export_type: request.format.clone(), + contains_header: request.contains_header, + export_path: String::new(), + }, + ) + .await?; + Ok(LegacyImportExportTaskAccepted { task_id }) + } + "DATA_FILE_IMPORT" | "SQL_FILE_IMPORT" => { + let sql_route = task_type == "SQL_FILE_IMPORT"; + let file_name = submitted_import_path(application, request).await?; + let import_type = if sql_route { + "SQL".to_owned() + } else { + request.format.clone() + }; + let task_id = import_legacy_mysql_desktop_file( + application, + &LegacyImportFileRequest { + data_source_id: request.data_source_id.clone(), + database_name: request.database_name.clone(), + schema_name: request.schema_name.clone(), + table_name: request.table_name.clone(), + file_name, + import_type, + contains_header: request.contains_header, + tabular_encoding: TabularImportEncoding::Plain, + }, + sql_route, + ) + .await?; + Ok(LegacyImportExportTaskAccepted { task_id }) + } + "QUERY_RESULT_EXPORT" => Err(LegacyFailure { + code: "unsupported_task_type".to_owned(), + message: "Exporting a query result as a background task is not supported yet" + .to_owned(), + }), + _ => Err(LegacyFailure::invalid( + "invalid_task_type", + "taskType is not supported", + )), + } +} + pub(crate) async fn export_legacy_mysql_dml( application: &Application, request: &LegacyDmlExportRequest, @@ -8821,6 +8962,12 @@ async fn dispatch_inner( Err(error) => Err(error), } } + ("post", "/api/tasks/export" | "/api/tasks/import") => { + match decode::(request.message) { + Ok(body) => serialized(submit_import_export_task(application, &body).await), + Err(error) => Err(error), + } + } ("get", "/api/sql/format") => match decode::(request.message) { Ok(query) => serialized(format_legacy_sql(application, &query).await), Err(error) => Err(error), @@ -9357,6 +9504,8 @@ const LEGACY_PATHS: &[&str] = &[ "/api/tasks/artifact", "/api/tasks/delete", "/api/tasks/events", + "/api/tasks/export", + "/api/tasks/import", "/api/sql/format", "/api/sql/valid_select", "/api/sql_parser/get_keywords", @@ -9651,6 +9800,8 @@ pub(crate) fn routes() -> Router { axum::routing::delete(import_export_task_delete_handler), ) .route("/api/tasks/events", get(import_export_task_events_handler)) + .route("/api/tasks/export", post(import_export_task_export_handler)) + .route("/api/tasks/import", post(import_export_task_import_handler)) .route("/api/task/download", get(transfer_task_download_handler)) .route("/api/sql/format", get(sql_format_handler)) .route("/api/sql/valid_select", get(sql_valid_select_handler)) @@ -10288,6 +10439,20 @@ async fn import_export_task_events_handler( envelope(list_import_export_task_events(&application, &query).await) } +async fn import_export_task_export_handler( + State(application): State, + Json(request): Json, +) -> Json> { + envelope(submit_import_export_task(&application, &request).await) +} + +async fn import_export_task_import_handler( + State(application): State, + Json(request): Json, +) -> Json> { + envelope(submit_import_export_task(&application, &request).await) +} + async fn dml_export_handler( State(application): State, Json(request): Json, @@ -11246,6 +11411,8 @@ mod tests { ("get", "/api/tasks/artifact"), ("delete", "/api/tasks/delete"), ("get", "/api/tasks/events"), + ("post", "/api/tasks/export"), + ("post", "/api/tasks/import"), ("post", "/api/rdb/dml/export"), ("post", "/api/rdb/table/generate/class"), ]; @@ -11971,6 +12138,43 @@ mod tests { assert_eq!(body["errorCode"], "unsupported_custom_driver_upload"); } + #[tokio::test] + async fn task_submission_reports_unsupported_and_invalid_task_types() { + let directory = tempfile::TempDir::new().expect("temporary directory"); + let storage = Storage::open(directory.path(), Arc::new(EmptyVault)).expect("storage opens"); + let router = routes().with_state(Application::with_storage(storage)); + + let unsupported = dashboard_http_json( + &router, + "POST", + "/api/tasks/export", + Some(serde_json::json!({ + "dataSourceId": "datasource-1", + "databaseName": "app", + "taskType": "QUERY_RESULT_EXPORT", + "format": "CSV" + })), + ) + .await; + assert_eq!(unsupported["success"], false); + assert_eq!(unsupported["errorCode"], "unsupported_task_type"); + + let invalid = dashboard_http_json( + &router, + "POST", + "/api/tasks/import", + Some(serde_json::json!({ + "dataSourceId": "datasource-1", + "databaseName": "app", + "taskType": "NOPE", + "format": "CSV" + })), + ) + .await; + assert_eq!(invalid["success"], false); + assert_eq!(invalid["errorCode"], "invalid_task_type"); + } + #[tokio::test] async fn dashboard_http_routes_preserve_community_crud_and_envelopes() { let directory = tempfile::TempDir::new().expect("temporary directory"); From 8bdd901d931144c1b6d8141ece6357f76fab46a6 Mon Sep 17 00:00:00 2001 From: zgq Date: Fri, 9 Oct 2026 21:30:26 +0800 Subject: [PATCH 7/9] feat(legacy): support the task-center exit handshake The desktop exit coordinator asks the backend to prepare for exit and to abort that preparation when the window close fails. - `Application::prepare_user_exit` stops accepting new transfer work through the existing `accepting_work` gate, and `abort_user_exit` restores it. Neither cancels running tasks, so a failed exit attempt does not destroy user work. - `import_mysql_file`, `export_mysql_sql_file`, and `export_mysql_other_file` now consult that gate and answer `runtime_not_accepting_work`. - `POST /api/tasks/prepare-user-exit` and `POST /api/tasks/abort-user-exit` expose the handshake, registered in the dispatch match, `LEGACY_PATHS`, the Axum router, and the transfer path-coverage list. Tests: 82 chat2db-web tests pass, including a route test proving a prepared exit blocks new submissions and that aborting restores the gate. --- apps/chat2db-web/src/legacy.rs | 77 +++++++++++++++++++++++++ crates/chat2db-core/src/lib.rs | 20 +++++++ crates/chat2db-core/src/transfer/mod.rs | 19 ++++++ 3 files changed, 116 insertions(+) diff --git a/apps/chat2db-web/src/legacy.rs b/apps/chat2db-web/src/legacy.rs index 4d98b66..ade9749 100644 --- a/apps/chat2db-web/src/legacy.rs +++ b/apps/chat2db-web/src/legacy.rs @@ -8968,6 +8968,18 @@ async fn dispatch_inner( Err(error) => Err(error), } } + ("post", "/api/tasks/prepare-user-exit") => serialized( + application + .prepare_user_exit() + .await + .map_err(LegacyFailure::from), + ), + ("post", "/api/tasks/abort-user-exit") => serialized( + application + .abort_user_exit() + .await + .map_err(LegacyFailure::from), + ), ("get", "/api/sql/format") => match decode::(request.message) { Ok(query) => serialized(format_legacy_sql(application, &query).await), Err(error) => Err(error), @@ -9506,6 +9518,8 @@ const LEGACY_PATHS: &[&str] = &[ "/api/tasks/events", "/api/tasks/export", "/api/tasks/import", + "/api/tasks/prepare-user-exit", + "/api/tasks/abort-user-exit", "/api/sql/format", "/api/sql/valid_select", "/api/sql_parser/get_keywords", @@ -9802,6 +9816,14 @@ pub(crate) fn routes() -> Router { .route("/api/tasks/events", get(import_export_task_events_handler)) .route("/api/tasks/export", post(import_export_task_export_handler)) .route("/api/tasks/import", post(import_export_task_import_handler)) + .route( + "/api/tasks/prepare-user-exit", + post(import_export_task_prepare_exit_handler), + ) + .route( + "/api/tasks/abort-user-exit", + post(import_export_task_abort_exit_handler), + ) .route("/api/task/download", get(transfer_task_download_handler)) .route("/api/sql/format", get(sql_format_handler)) .route("/api/sql/valid_select", get(sql_valid_select_handler)) @@ -10453,6 +10475,28 @@ async fn import_export_task_import_handler( envelope(submit_import_export_task(&application, &request).await) } +async fn import_export_task_prepare_exit_handler( + State(application): State, +) -> Json> { + envelope( + application + .prepare_user_exit() + .await + .map_err(LegacyFailure::from), + ) +} + +async fn import_export_task_abort_exit_handler( + State(application): State, +) -> Json> { + envelope( + application + .abort_user_exit() + .await + .map_err(LegacyFailure::from), + ) +} + async fn dml_export_handler( State(application): State, Json(request): Json, @@ -11413,6 +11457,8 @@ mod tests { ("get", "/api/tasks/events"), ("post", "/api/tasks/export"), ("post", "/api/tasks/import"), + ("post", "/api/tasks/prepare-user-exit"), + ("post", "/api/tasks/abort-user-exit"), ("post", "/api/rdb/dml/export"), ("post", "/api/rdb/table/generate/class"), ]; @@ -12138,6 +12184,37 @@ mod tests { assert_eq!(body["errorCode"], "unsupported_custom_driver_upload"); } + #[tokio::test] + async fn prepared_exit_blocks_new_task_submissions_until_aborted() { + let directory = tempfile::TempDir::new().expect("temporary directory"); + let storage = Storage::open(directory.path(), Arc::new(EmptyVault)).expect("storage opens"); + let router = routes().with_state(Application::with_storage(storage)); + + let prepared = + dashboard_http_json(&router, "POST", "/api/tasks/prepare-user-exit", None).await; + assert_eq!(prepared["success"], true); + + let blocked = dashboard_http_json( + &router, + "POST", + "/api/tasks/import", + Some(serde_json::json!({ + "dataSourceId": "datasource-1", + "databaseName": "app", + "taskType": "DATA_FILE_IMPORT", + "format": "CSV", + "sourceFile": "/tmp/rows.csv" + })), + ) + .await; + assert_eq!(blocked["success"], false); + assert_eq!(blocked["errorCode"], "runtime_not_accepting_work"); + + let aborted = + dashboard_http_json(&router, "POST", "/api/tasks/abort-user-exit", None).await; + assert_eq!(aborted["success"], true); + } + #[tokio::test] async fn task_submission_reports_unsupported_and_invalid_task_types() { let directory = tempfile::TempDir::new().expect("temporary directory"); diff --git a/crates/chat2db-core/src/lib.rs b/crates/chat2db-core/src/lib.rs index 5d36675..1f445b3 100644 --- a/crates/chat2db-core/src/lib.rs +++ b/crates/chat2db-core/src/lib.rs @@ -861,6 +861,26 @@ impl Application { } /// Stops work admission and requests cancellation of active queries and agent runs. + /// Stops accepting new transfer work while the product prepares to exit. + /// + /// # Errors + /// + /// Returns availability failures only. + pub async fn prepare_user_exit(&self) -> Result<(), AppError> { + *self.inner.accepting_work.lock().await = false; + Ok(()) + } + + /// Re-enables transfer work after a cancelled exit. + /// + /// # Errors + /// + /// Returns availability failures only. + pub async fn abort_user_exit(&self) -> Result<(), AppError> { + *self.inner.accepting_work.lock().await = true; + Ok(()) + } + pub async fn begin_shutdown(&self) { let mut accepting_work = self.inner.accepting_work.lock().await; if !*accepting_work { diff --git a/crates/chat2db-core/src/transfer/mod.rs b/crates/chat2db-core/src/transfer/mod.rs index 6a0cf0c..21cd50c 100644 --- a/crates/chat2db-core/src/transfer/mod.rs +++ b/crates/chat2db-core/src/transfer/mod.rs @@ -445,6 +445,7 @@ impl Application { &self, request: ImportFileRequest, ) -> Result { + self.ensure_accepting_transfer_work().await?; self.import_file(request).await } @@ -479,6 +480,7 @@ impl Application { &self, request: SqlFileExportRequest, ) -> Result { + self.ensure_accepting_transfer_work().await?; self.export_sql_file(request).await } @@ -525,9 +527,26 @@ impl Application { &self, request: OtherFileExportRequest, ) -> Result { + self.ensure_accepting_transfer_work().await?; self.export_other_file(request).await } + /// Rejects new transfer work while the product is preparing to exit. + /// + /// # Errors + /// + /// Returns `runtime_not_accepting_work` when an exit was prepared and not + /// aborted yet. + pub(crate) async fn ensure_accepting_transfer_work(&self) -> Result<(), AppError> { + if *self.inner.accepting_work.lock().await { + return Ok(()); + } + Err(AppError::invalid( + "runtime_not_accepting_work", + "The application is preparing to exit and no longer accepts new transfer tasks", + )) + } + /// Lists retained transfer tasks newest first. /// /// # Errors From 052f2e20b795c309eefc7595064397dfb13e040f Mon Sep 17 00:00:00 2001 From: zgq Date: Fri, 9 Oct 2026 21:34:29 +0800 Subject: [PATCH 8/9] feat(legacy): serve the import-preview mapping step The newer Community import dialog stages a file, previews it against the target table, and then executes with an explicit column mapping. - `POST /api/rdb/import_preview/upload` stages a multipart upload into the managed artifact store (24 h TTL, 16 MiB cap) and answers the file id. - `POST /api/rdb/import_preview/upload_local` stages an absolute desktop path; the HTTP twin stays `desktop_file_operation_required`, matching the bridge. - `POST /api/rdb/import_preview/sheets` lists worksheet names through a new bounded reader in `chat2db-core` (`worksheet_names`), empty for CSV. - `POST /api/rdb/import_preview/preview` returns the source columns, the first 100 rows, the target table columns, and a name-matched mapping suggestion. - `POST /api/rdb/import_preview/execute` starts the durable import task through the staged artifact path and answers `{taskId}`. Known limitation, explicit in the code: only same-named mappings execute today. Renames, skipped columns, and `unmappedTarget: NULL` answer `unsupported_import_mapping` instead of importing the wrong columns, because the durable import builds its INSERT list from file headers matched to table columns. Preview options beyond `hasHeader` (delimiters, encodings, sheet selection) are likewise not applied yet. Tests: 83 chat2db-web tests pass, including a new test that stages a multipart CSV, lists its (empty) sheets, and rejects a renamed mapping. --- apps/chat2db-web/src/legacy.rs | 497 +++++++++++++++++++++ crates/chat2db-core/src/transfer/format.rs | 50 +++ crates/chat2db-core/src/transfer/mod.rs | 53 +++ 3 files changed, 600 insertions(+) diff --git a/apps/chat2db-web/src/legacy.rs b/apps/chat2db-web/src/legacy.rs index ade9749..6a916ef 100644 --- a/apps/chat2db-web/src/legacy.rs +++ b/apps/chat2db-web/src/legacy.rs @@ -515,6 +515,88 @@ pub struct LegacyImportExportTaskAccepted { pub task_id: i64, } +/// Staged import-preview upload sent by the desktop bridge. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportPreviewLocalUploadRequest { + pub source_file: String, + #[serde(default)] + pub original_file_name: String, +} + +/// Import-preview request carrying a staged file id. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportPreviewFileRequest { + pub data_source_id: LegacyIdentifier, + #[serde(default)] + pub database_name: String, + #[serde(default)] + pub schema_name: String, + #[serde(default)] + pub table_name: String, + pub file_id: String, +} + +/// Column mapping submitted with an import-preview execution. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportPreviewMappingRequest { + #[serde(default)] + pub source_column: Option, + pub target_column: String, +} + +/// Import-preview execution posted after the mapping step. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportPreviewExecuteRequest { + pub data_source_id: LegacyIdentifier, + #[serde(default)] + pub database_name: String, + #[serde(default)] + pub schema_name: String, + #[serde(default)] + pub table_name: String, + pub file_id: String, + #[serde(default)] + pub mappings: Vec, + #[serde(default)] + pub unmapped_target: String, +} + +/// Mapping suggestion returned by the preview step. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportPreviewMapping { + pub source_column: String, + pub target_column: String, +} + +/// Target column metadata shown by the mapping step. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportPreviewColumn { + pub name: String, + pub data_type: String, + pub nullable: bool, + pub auto_increment: bool, + pub default_value: Option, + pub comment: Option, +} + +/// Preview payload rendered by the mapping step. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportPreview { + pub source_columns: Vec, + pub preview_data: Vec>, + pub target_table_name: String, + pub target_columns: Vec, + pub suggested_mapping: Vec, + pub preview_limit: u32, +} + /// Submission posted by the task-center import/export panel. #[derive(Debug, Clone, Deserialize)] #[serde(rename_all = "camelCase")] @@ -3273,6 +3355,253 @@ pub(crate) async fn list_import_export_task_events( .collect()) } +const IMPORT_PREVIEW_ROW_LIMIT: u32 = 100; + +fn import_preview_artifact_format( + file_name: &str, +) -> LegacyResult<(&'static str, &'static str, &'static str)> { + let extension = file_name + .rsplit('.') + .next() + .unwrap_or_default() + .to_ascii_lowercase(); + match extension.as_str() { + "csv" => Ok(("text/csv", "CSV", "csv")), + "xls" => Ok(("application/vnd.ms-excel", "XLS", "xls")), + "xlsx" => Ok(( + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + "XLSX", + "xlsx", + )), + "sql" => Ok(("application/sql; charset=utf-8", "SQL", "sql")), + _ => Err(LegacyFailure::invalid( + "invalid_import_file", + "The staged import file must be CSV, XLS, XLSX, or SQL", + )), + } +} + +fn import_preview_transfer_format(value: &str) -> LegacyResult { + match value.trim().to_ascii_uppercase().as_str() { + "CSV" => Ok(TransferFileFormat::Csv), + "XLS" => Ok(TransferFileFormat::Xls), + "XLSX" => Ok(TransferFileFormat::Xlsx), + "SQL" => Ok(TransferFileFormat::Sql), + other => Err(LegacyFailure { + code: "unsupported_import_file".to_owned(), + message: format!("Unsupported import preview format: {other}"), + }), + } +} + +async fn stage_import_preview_bytes( + application: &Application, + file_name: String, + content: Vec, +) -> LegacyResult { + let (media_type, format_name, extension) = import_preview_artifact_format(&file_name)?; + let expires_at_ms = i64::try_from(unix_epoch_millis()) + .unwrap_or(i64::MAX) + .saturating_add(LEGACY_IMPORT_UPLOAD_TTL_MS); + let storage = legacy_storage(application)?; + legacy_storage_call(move || { + let mut writer = storage.begin_transfer_artifact( + None, + &file_name, + media_type, + format_name, + extension, + Some(expires_at_ms), + )?; + writer.write_all(&content).map_err(|_| { + StorageError::InvalidTransfer("the staged import file could not be stored") + })?; + Ok(writer.finish()?.id) + }) + .await +} + +pub(crate) async fn stage_import_preview_upload( + application: &Application, + multipart: Multipart, +) -> LegacyResult { + let upload = read_legacy_multipart_file(multipart).await?; + stage_import_preview_bytes(application, upload.file_name, upload.content).await +} + +pub(crate) async fn stage_import_preview_local_file( + application: &Application, + request: &LegacyImportPreviewLocalUploadRequest, +) -> LegacyResult { + let source = request.source_file.trim().to_owned(); + if source.is_empty() { + return Err(LegacyFailure::invalid( + "invalid_import_file", + "sourceFile is required", + )); + } + let metadata = tokio::fs::metadata(&source) + .await + .map_err(|_| LegacyFailure { + code: "import_file_not_found".to_owned(), + message: "The selected import file could not be read".to_owned(), + })?; + if !metadata.is_file() || metadata.len() > MAX_LEGACY_DATASOURCE_IMPORT_BYTES as u64 { + return Err(LegacyFailure::invalid( + "invalid_import_file", + "sourceFile must be a readable file within the upload size limit", + )); + } + let content = tokio::fs::read(&source).await.map_err(|_| LegacyFailure { + code: "import_file_not_found".to_owned(), + message: "The selected import file could not be read".to_owned(), + })?; + let file_name = non_blank(&request.original_file_name).unwrap_or_else(|| { + source + .rsplit(['/', '\\']) + .next() + .unwrap_or("import.dat") + .to_owned() + }); + stage_import_preview_bytes(application, file_name, content).await +} + +pub(crate) async fn import_preview_sheets( + application: &Application, + request: &LegacyImportPreviewFileRequest, +) -> LegacyResult> { + Ok(application.import_preview_sheets(&request.file_id).await?) +} + +async fn import_preview_target_columns( + application: &Application, + request: &LegacyImportPreviewFileRequest, +) -> LegacyResult> { + let datasource_id = request.data_source_id.as_string(); + let database_type = resolve_database_type(application, &datasource_id, "").await?; + Ok(application + .list_community_columns(ListCommunityColumnsRequest { + datasource_id, + database_type, + database_name: request.database_name.clone(), + schema_name: request.schema_name.clone(), + table_name: request.table_name.clone(), + }) + .await? + .items) +} + +pub(crate) async fn import_preview( + application: &Application, + request: &LegacyImportPreviewFileRequest, +) -> LegacyResult { + let columns = import_preview_target_columns(application, request).await?; + let (source_columns, rows) = application + .import_preview_table(&request.file_id, true, IMPORT_PREVIEW_ROW_LIMIT) + .await?; + let suggested_mapping = source_columns + .iter() + .filter_map(|source| { + columns + .iter() + .find(|target| target.name.eq_ignore_ascii_case(source)) + .map(|target| LegacyImportPreviewMapping { + source_column: source.clone(), + target_column: target.name.clone(), + }) + }) + .collect(); + let target_columns = columns + .into_iter() + .map(|column| LegacyImportPreviewColumn { + name: column.name, + data_type: column.column_type, + nullable: column.nullable != Some(0), + auto_increment: column.auto_increment.unwrap_or(false), + default_value: column.default_value, + comment: non_blank(&column.comment), + }) + .collect(); + Ok(LegacyImportPreview { + source_columns, + preview_data: rows, + target_table_name: request.table_name.clone(), + target_columns, + suggested_mapping, + preview_limit: IMPORT_PREVIEW_ROW_LIMIT, + }) +} + +pub(crate) async fn execute_import_preview( + application: &Application, + request: &LegacyImportPreviewExecuteRequest, +) -> LegacyResult { + if request.unmapped_target.trim().eq_ignore_ascii_case("NULL") { + return Err(LegacyFailure { + code: "unsupported_import_mapping".to_owned(), + message: "Writing NULL into unmapped target columns is not supported yet".to_owned(), + }); + } + let mappings: Vec<(String, String)> = request + .mappings + .iter() + .filter_map(|mapping| { + non_blank(mapping.source_column.as_deref().unwrap_or_default()) + .map(|source| (source, mapping.target_column.clone())) + }) + .collect(); + if mappings + .iter() + .any(|(source, target)| !source.eq_ignore_ascii_case(target)) + { + return Err(LegacyFailure { + code: "unsupported_import_mapping".to_owned(), + message: "Renaming columns during import is not supported yet".to_owned(), + }); + } + let (columns, _) = application + .import_preview_table(&request.file_id, true, 1) + .await?; + if columns.len() != mappings.len() + || !columns.iter().all(|column| { + mappings + .iter() + .any(|(source, _)| source.eq_ignore_ascii_case(column)) + }) + { + return Err(LegacyFailure { + code: "unsupported_import_mapping".to_owned(), + message: "Every file column must map to a same-named table column; skipped columns are not supported yet" + .to_owned(), + }); + } + let download = application + .transfer_artifact_download(&request.file_id) + .await?; + let format = import_preview_transfer_format(&download.artifact.format)?; + let Some(file_path) = download.path.to_str().map(str::to_owned) else { + return Err(LegacyFailure { + code: "invalid_import_path".to_owned(), + message: "The staged import file path cannot be represented as UTF-8".to_owned(), + }); + }; + let accepted = application + .import_mysql_file(ImportFileRequest { + datasource_id: request.data_source_id.as_string(), + database_name: request.database_name.clone(), + schema_name: request.schema_name.clone(), + table_name: non_blank(&request.table_name), + file_path, + format, + contains_header: true, + tabular_encoding: TabularImportEncoding::Plain, + }) + .await?; + Ok(LegacyImportExportTaskAccepted { + task_id: accepted.task_id, + }) +} + async fn submitted_import_path( application: &Application, request: &LegacyImportExportSubmitRequest, @@ -8980,6 +9309,34 @@ async fn dispatch_inner( .await .map_err(LegacyFailure::from), ), + ("post", "/api/rdb/import_preview/upload_local") => { + match decode::(request.message) { + Ok(body) if desktop_paths => { + serialized(stage_import_preview_local_file(application, &body).await) + } + Ok(_) => Err(desktop_file_operation_required()), + Err(error) => Err(error), + } + } + ("post", "/api/rdb/import_preview/upload") => Err(desktop_file_operation_required()), + ("post", "/api/rdb/import_preview/sheets") => { + match decode::(request.message) { + Ok(body) => serialized(import_preview_sheets(application, &body).await), + Err(error) => Err(error), + } + } + ("post", "/api/rdb/import_preview/preview") => { + match decode::(request.message) { + Ok(body) => serialized(import_preview(application, &body).await), + Err(error) => Err(error), + } + } + ("post", "/api/rdb/import_preview/execute") => { + match decode::(request.message) { + Ok(body) => serialized(execute_import_preview(application, &body).await), + Err(error) => Err(error), + } + } ("get", "/api/sql/format") => match decode::(request.message) { Ok(query) => serialized(format_legacy_sql(application, &query).await), Err(error) => Err(error), @@ -9504,6 +9861,11 @@ const LEGACY_PATHS: &[&str] = &[ "/api/connection/console/connect", "/api/import/sql_file", "/api/import/other_file", + "/api/rdb/import_preview/upload", + "/api/rdb/import_preview/upload_local", + "/api/rdb/import_preview/sheets", + "/api/rdb/import_preview/preview", + "/api/rdb/import_preview/execute", "/api/export/sql_file", "/api/export/other_file", "/api/task/list", @@ -9796,6 +10158,27 @@ pub(crate) fn routes() -> Router { ) .route("/api/export/sql_file", post(export_sql_file_handler)) .route("/api/export/other_file", post(export_other_file_handler)) + .route( + "/api/rdb/import_preview/upload", + post(import_preview_upload_handler) + .layer(DefaultBodyLimit::max(MAX_LEGACY_MULTIPART_BYTES)), + ) + .route( + "/api/rdb/import_preview/upload_local", + post(import_preview_local_upload_handler), + ) + .route( + "/api/rdb/import_preview/sheets", + post(import_preview_sheets_handler), + ) + .route( + "/api/rdb/import_preview/preview", + post(import_preview_handler), + ) + .route( + "/api/rdb/import_preview/execute", + post(import_preview_execute_handler), + ) .route("/api/task/list", get(transfer_task_list_handler)) .route("/api/task/get", get(transfer_task_get_handler)) .route("/api/task/stop", get(transfer_task_stop_handler)) @@ -10461,6 +10844,38 @@ async fn import_export_task_events_handler( envelope(list_import_export_task_events(&application, &query).await) } +async fn import_preview_upload_handler( + State(application): State, + multipart: Multipart, +) -> Json> { + envelope(stage_import_preview_upload(&application, multipart).await) +} + +async fn import_preview_local_upload_handler() -> Json> { + envelope(Err(desktop_file_operation_required())) +} + +async fn import_preview_sheets_handler( + State(application): State, + Json(request): Json, +) -> Json>> { + envelope(import_preview_sheets(&application, &request).await) +} + +async fn import_preview_handler( + State(application): State, + Json(request): Json, +) -> Json> { + envelope(import_preview(&application, &request).await) +} + +async fn import_preview_execute_handler( + State(application): State, + Json(request): Json, +) -> Json> { + envelope(execute_import_preview(&application, &request).await) +} + async fn import_export_task_export_handler( State(application): State, Json(request): Json, @@ -11443,6 +11858,11 @@ mod tests { const REQUIRED_TRANSFER_PATHS: &[(&str, &str)] = &[ ("post", "/api/import/sql_file"), ("post", "/api/import/other_file"), + ("post", "/api/rdb/import_preview/upload"), + ("post", "/api/rdb/import_preview/upload_local"), + ("post", "/api/rdb/import_preview/sheets"), + ("post", "/api/rdb/import_preview/preview"), + ("post", "/api/rdb/import_preview/execute"), ("post", "/api/export/sql_file"), ("post", "/api/export/other_file"), ("get", "/api/task/list"), @@ -12122,6 +12542,16 @@ mod tests { } } + async fn dashboard_response_json(response: axum::response::Response) -> serde_json::Value { + let body = response + .into_body() + .collect() + .await + .expect("response body must collect") + .to_bytes(); + serde_json::from_slice(&body).expect("response body must be JSON") + } + async fn dashboard_http_json( router: &Router, method: &str, @@ -12184,6 +12614,73 @@ mod tests { assert_eq!(body["errorCode"], "unsupported_custom_driver_upload"); } + #[tokio::test] + async fn import_preview_stages_files_and_rejects_unsupported_mappings() { + let directory = tempfile::TempDir::new().expect("temporary directory"); + let storage = Storage::open(directory.path(), Arc::new(EmptyVault)).expect("storage opens"); + let router = routes().with_state(Application::with_storage(storage)); + + let boundary = "chat2db-preview-boundary"; + let multipart = format!( + "--{boundary}\r\nContent-Disposition: form-data; name=\"file\"; filename=\"rows.csv\"\r\nContent-Type: text/csv\r\n\r\nname,value\r\nalpha,1\r\n--{boundary}--\r\n" + ); + let upload_request = Request::builder() + .method(axum::http::Method::POST) + .uri("/api/rdb/import_preview/upload") + .header( + "content-type", + format!("multipart/form-data; boundary={boundary}"), + ) + .body(Body::from(multipart)) + .expect("multipart request must build"); + let uploaded = router + .clone() + .oneshot(upload_request) + .await + .expect("upload route must respond"); + let uploaded = dashboard_response_json(uploaded).await; + assert_eq!(uploaded["success"], true); + let file_id = uploaded["data"] + .as_str() + .expect("staged file id must be a string") + .to_owned(); + + let sheets = dashboard_http_json( + &router, + "POST", + "/api/rdb/import_preview/sheets", + Some(serde_json::json!({ + "dataSourceId": "datasource-1", + "databaseName": "app", + "tableName": "items", + "fileId": file_id + })), + ) + .await; + assert_eq!(sheets["success"], true); + assert_eq!(sheets["data"], serde_json::json!([])); + + let renamed = dashboard_http_json( + &router, + "POST", + "/api/rdb/import_preview/execute", + Some(serde_json::json!({ + "dataSourceId": "datasource-1", + "databaseName": "app", + "tableName": "items", + "fileId": file_id, + "unmappedTarget": "DEFAULT", + "mappings": [ + {"sourceColumn": "name", "targetColumn": "label"}, + {"sourceColumn": "value", "targetColumn": "value"} + ] + })), + ) + .await; + assert_eq!(renamed["success"], false); + assert_eq!(renamed["errorCode"], "unsupported_import_mapping"); + } + #[tokio::test] async fn prepared_exit_blocks_new_task_submissions_until_aborted() { let directory = tempfile::TempDir::new().expect("temporary directory"); diff --git a/crates/chat2db-core/src/transfer/format.rs b/crates/chat2db-core/src/transfer/format.rs index e7abf94..74230f2 100644 --- a/crates/chat2db-core/src/transfer/format.rs +++ b/crates/chat2db-core/src/transfer/format.rs @@ -162,6 +162,56 @@ fn read_csv( Ok(ImportedTable { columns, rows }) } +/// Lists worksheet names of one spreadsheet file, empty for other formats. +pub(crate) fn worksheet_names( + path: &Path, + format: TransferFileFormat, +) -> Result, AppError> { + match format { + TransferFileFormat::Csv | TransferFileFormat::Sql => Ok(Vec::new()), + TransferFileFormat::Xls | TransferFileFormat::Xlsx => { + if format == TransferFileFormat::Xlsx { + validate_xlsx_archive(path)?; + } + let file = File::open(path).map_err(|_| { + AppError::not_found( + "import_file_not_found", + "The selected import file could not be opened", + ) + })?; + let workbook = match format { + TransferFileFormat::Xls => xls::core::xls::read(file), + TransferFileFormat::Xlsx => xls::core::xlsx::read(file), + TransferFileFormat::Csv | TransferFileFormat::Sql => unreachable!(), + } + .map_err(format_error)?; + Ok(workbook + .sheets + .iter() + .map(|sheet| sheet.name.clone()) + .collect()) + } + } +} + +/// Reads a row-bounded preview with string-encoded cells. +pub(crate) fn read_tabular_preview( + path: &Path, + format: TransferFileFormat, + contains_header: bool, + limit: usize, +) -> Result<(Vec, Vec>), AppError> { + let mut table = read_tabular_file(path, format, contains_header, TabularImportEncoding::Plain)?; + let columns = table.columns.take().unwrap_or_default(); + table.rows.truncate(limit); + let rows = table + .rows + .into_iter() + .map(|row| row.iter().map(encode_tabular_value).collect()) + .collect(); + Ok((columns, rows)) +} + fn read_spreadsheet( path: &Path, format: TransferFileFormat, diff --git a/crates/chat2db-core/src/transfer/mod.rs b/crates/chat2db-core/src/transfer/mod.rs index 21cd50c..c938466 100644 --- a/crates/chat2db-core/src/transfer/mod.rs +++ b/crates/chat2db-core/src/transfer/mod.rs @@ -647,6 +647,59 @@ impl Application { Ok(()) } + /// Resolves one staged import preview artifact to its path and format. + async fn staged_import_file( + &self, + file_id: &str, + ) -> Result<(std::path::PathBuf, TransferFileFormat), AppError> { + let download = self.transfer_artifact_download(file_id.trim()).await?; + let format = match download + .artifact + .format + .trim() + .to_ascii_uppercase() + .as_str() + { + "CSV" => TransferFileFormat::Csv, + "XLS" => TransferFileFormat::Xls, + "XLSX" => TransferFileFormat::Xlsx, + "SQL" => TransferFileFormat::Sql, + other => { + return Err(AppError::invalid( + "unsupported_import_file", + format!("Unsupported import preview format: {other}"), + )); + } + }; + Ok((download.path, format)) + } + + /// Lists worksheet names of one staged import preview file. + /// + /// # Errors + /// + /// Returns not-found, expiry, corrupted-artifact, or format failures. + pub async fn import_preview_sheets(&self, file_id: &str) -> Result, AppError> { + let (path, format) = self.staged_import_file(file_id).await?; + format::worksheet_names(&path, format) + } + + /// Reads a bounded preview of one staged import file. + /// + /// # Errors + /// + /// Returns not-found, expiry, corrupted-artifact, or format failures. + pub async fn import_preview_table( + &self, + file_id: &str, + contains_header: bool, + limit: u32, + ) -> Result<(Vec, Vec>), AppError> { + let limit = usize::try_from(limit).unwrap_or(usize::MAX); + let (path, format) = self.staged_import_file(file_id).await?; + format::read_tabular_preview(&path, format, contains_header, limit) + } + /// Lists ordered progress events of one transfer task. /// /// # Errors From 1d5c4d808986d5522170319ca7e09314018151f5 Mon Sep 17 00:00:00 2001 From: zgq Date: Fri, 9 Oct 2026 21:38:37 +0800 Subject: [PATCH 9/9] feat(legacy): serve the MySQL active-transaction monitor `GET /api/rdb/active_transaction/list` reads `information_schema.INNODB_TRX` joined with `PROCESSLIST` through the native MySQL connection, with a 30 second timeout, and projects the monitor's `IActiveTransactionItem` core fields: transaction id, state, start time, age, isolation level, locked and modified row counts, lock structures, thread id, user, host, default database, and the running statement. Privilege failures answer `mysql.activeTransaction.processPrivilegeRequired` (MySQL error 1227 or 1142) so the panel can explain the requirement instead of showing a generic error; other failures answer `mysql.activeTransaction.unavailable`. Deep lock metadata (`blocking*`/`lockMetadata*`) is not collected yet, so the response reports `lockMetadataState: UNAVAILABLE` rather than inventing values. Verification: 83 chat2db-web tests, 115+4 chat2db-storage tests, clippy with `-D warnings` on the touched crates, and the contract drift check all pass. --- apps/chat2db-web/src/legacy.rs | 87 ++++++++++ crates/chat2db-contract/src/community.rs | 35 ++++ crates/chat2db-contract/src/lib.rs | 19 ++- crates/chat2db-core/src/lib.rs | 1 + .../src/mysql_active_transaction.rs | 155 ++++++++++++++++++ 5 files changed, 288 insertions(+), 9 deletions(-) create mode 100644 crates/chat2db-core/src/mysql_active_transaction.rs diff --git a/apps/chat2db-web/src/legacy.rs b/apps/chat2db-web/src/legacy.rs index 6a916ef..557d291 100644 --- a/apps/chat2db-web/src/legacy.rs +++ b/apps/chat2db-web/src/legacy.rs @@ -515,6 +515,42 @@ pub struct LegacyImportExportTaskAccepted { pub task_id: i64, } +/// Active-transaction query used by the `MySQL` operations monitor. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyActiveTransactionQuery { + pub data_source_id: LegacyIdentifier, + #[serde(default)] + pub database_name: String, + #[serde(default)] + pub schema_name: String, +} + +/// One active transaction in the shape the monitor expects. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyActiveTransaction { + pub trx_id: String, + pub state: String, + pub started_at: i64, + pub age_seconds: u64, + pub isolation_level: String, + pub rows_locked: u64, + pub rows_modified: u64, + pub lock_structs: u64, + pub thread_id: u64, + #[serde(skip_serializing_if = "Option::is_none")] + pub user: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub host: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub db: Option, + pub query: String, + pub connection_inspection_sql: String, + /// Deep lock metadata is not collected yet, so the panel must not wait for it. + pub lock_metadata_state: String, +} + /// Staged import-preview upload sent by the desktop bridge. #[derive(Debug, Clone, Deserialize)] #[serde(rename_all = "camelCase")] @@ -5020,6 +5056,39 @@ pub(crate) async fn list_simple_tables( } /// Lists table or view columns in the historical `ColumnResponse` shape. +pub(crate) async fn list_active_transactions( + application: &Application, + query: &LegacyActiveTransactionQuery, +) -> LegacyResult> { + let datasource_id = query.data_source_id.as_string(); + resolve_mysql_database_type(application, &datasource_id, "").await?; + Ok(application + .list_mysql_active_transactions(&datasource_id) + .await? + .into_iter() + .map(|item| LegacyActiveTransaction { + connection_inspection_sql: format!( + "SELECT * FROM information_schema.PROCESSLIST WHERE ID = {}", + item.thread_id + ), + lock_metadata_state: "UNAVAILABLE".to_owned(), + started_at: legacy_epoch_millis(&item.started_at_ms), + trx_id: item.trx_id, + state: item.state, + age_seconds: item.age_seconds, + isolation_level: item.isolation_level, + rows_locked: item.rows_locked, + rows_modified: item.rows_modified, + lock_structs: item.lock_structs, + thread_id: item.thread_id, + user: item.user, + host: item.host, + db: item.database, + query: item.query, + }) + .collect()) +} + pub(crate) async fn list_columns( application: &Application, query: &LegacyTableDetailQuery, @@ -9309,6 +9378,12 @@ async fn dispatch_inner( .await .map_err(LegacyFailure::from), ), + ("get", "/api/rdb/active_transaction/list") => { + match decode::(request.message) { + Ok(query) => serialized(list_active_transactions(application, &query).await), + Err(error) => Err(error), + } + } ("post", "/api/rdb/import_preview/upload_local") => { match decode::(request.message) { Ok(body) if desktop_paths => { @@ -9861,6 +9936,7 @@ const LEGACY_PATHS: &[&str] = &[ "/api/connection/console/connect", "/api/import/sql_file", "/api/import/other_file", + "/api/rdb/active_transaction/list", "/api/rdb/import_preview/upload", "/api/rdb/import_preview/upload_local", "/api/rdb/import_preview/sheets", @@ -10309,6 +10385,10 @@ pub(crate) fn routes() -> Router { .route("/api/rdb/table/modify/sql", post(table_modify_sql_handler)) .route("/api/rdb/table/truncate", post(table_truncate_handler)) .route("/api/rdb/table/copy", post(table_copy_handler)) + .route( + "/api/rdb/active_transaction/list", + get(active_transaction_list_handler), + ) .route( "/api/rdb/table/copy/prepare", get(table_copy_prepare_handler), @@ -10844,6 +10924,13 @@ async fn import_export_task_events_handler( envelope(list_import_export_task_events(&application, &query).await) } +async fn active_transaction_list_handler( + State(application): State, + Query(query): Query, +) -> Json>> { + envelope(list_active_transactions(&application, &query).await) +} + async fn import_preview_upload_handler( State(application): State, multipart: Multipart, diff --git a/crates/chat2db-contract/src/community.rs b/crates/chat2db-contract/src/community.rs index 36ba20e..cb4e0fe 100644 --- a/crates/chat2db-contract/src/community.rs +++ b/crates/chat2db-contract/src/community.rs @@ -2049,3 +2049,38 @@ mod tests { } } } + +/// One active `InnoDB` transaction reported by the `MySQL` operations monitor. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub struct ActiveTransaction { + /// `InnoDB` transaction id. + pub trx_id: String, + /// `InnoDB` transaction state. + pub state: String, + /// Transaction start time as Unix epoch milliseconds. + pub started_at_ms: String, + /// Whole seconds since the transaction started. + pub age_seconds: u64, + /// Isolation level reported by `InnoDB`. + pub isolation_level: String, + /// Rows currently locked by the transaction. + pub rows_locked: u64, + /// Rows modified by the transaction. + pub rows_modified: u64, + /// `InnoDB` lock structures held by the transaction. + pub lock_structs: u64, + /// `MySQL` thread id owning the transaction. + pub thread_id: u64, + /// Connection user, when the process list entry is visible. + #[serde(skip_serializing_if = "Option::is_none")] + pub user: Option, + /// Connection host, when the process list entry is visible. + #[serde(skip_serializing_if = "Option::is_none")] + pub host: Option, + /// Default database of the connection, when visible. + #[serde(skip_serializing_if = "Option::is_none")] + pub database: Option, + /// Currently running statement, empty when the session is idle. + pub query: String, +} diff --git a/crates/chat2db-contract/src/lib.rs b/crates/chat2db-contract/src/lib.rs index 6f7b0a9..f5b4424 100644 --- a/crates/chat2db-contract/src/lib.rs +++ b/crates/chat2db-contract/src/lib.rs @@ -32,15 +32,16 @@ pub use agent::{ UpdateProviderProfileRequest, }; pub use community::{ - BuildCommunityCreateSchemaRequest, BuildCommunityDmlRequest, BuildCommunityNamespaceSqlRequest, - CommunityBuiltSql, CommunityDatabase, CommunityDatabaseList, CommunityDmlAssignment, - CommunityDmlColumn, CommunityDmlRow, CommunityDmlStatement, CommunityDmlTarget, - CommunityDmlTemporalKind, CommunityDmlValue, CommunityDriverConfig, CommunityForeignKey, - CommunityForeignKeyList, CommunityFormattedSql, CommunityFunction, CommunityFunctionList, - CommunityFunctionParameter, CommunityFunctionParameterList, CommunityNamespaceSqlOperation, - CommunityParsedStatement, CommunityPlugin, CommunityPluginBehavior, CommunityPluginCatalog, - CommunityPluginServices, CommunityPrimaryKey, CommunityPrimaryKeyList, CommunityProcedure, - CommunityProcedureList, CommunityProcedureParameter, CommunityProcedureParameterList, + ActiveTransaction, BuildCommunityCreateSchemaRequest, BuildCommunityDmlRequest, + BuildCommunityNamespaceSqlRequest, CommunityBuiltSql, CommunityDatabase, CommunityDatabaseList, + CommunityDmlAssignment, CommunityDmlColumn, CommunityDmlRow, CommunityDmlStatement, + CommunityDmlTarget, CommunityDmlTemporalKind, CommunityDmlValue, CommunityDriverConfig, + CommunityForeignKey, CommunityForeignKeyList, CommunityFormattedSql, CommunityFunction, + CommunityFunctionList, CommunityFunctionParameter, CommunityFunctionParameterList, + CommunityNamespaceSqlOperation, CommunityParsedStatement, CommunityPlugin, + CommunityPluginBehavior, CommunityPluginCatalog, CommunityPluginServices, CommunityPrimaryKey, + CommunityPrimaryKeyList, CommunityProcedure, CommunityProcedureList, + CommunityProcedureParameter, CommunityProcedureParameterList, CommunityRoutineInvocationPreview, CommunityRoutineMigrationExecution, CommunityRoutineMigrationRequest, CommunitySchema, CommunitySchemaList, CommunitySqlAnalysis, CommunitySqlCompletion, CommunitySqlCompletionActiveSnippetSlot, diff --git a/crates/chat2db-core/src/lib.rs b/crates/chat2db-core/src/lib.rs index 1f445b3..6b98712 100644 --- a/crates/chat2db-core/src/lib.rs +++ b/crates/chat2db-core/src/lib.rs @@ -13,6 +13,7 @@ mod error; mod large_value; mod legacy_community_import; mod mysql_account; +mod mysql_active_transaction; mod mysql_dashboard; pub mod mysql_ddl; mod mysql_schema_diff; diff --git a/crates/chat2db-core/src/mysql_active_transaction.rs b/crates/chat2db-core/src/mysql_active_transaction.rs new file mode 100644 index 0000000..e1d685f --- /dev/null +++ b/crates/chat2db-core/src/mysql_active_transaction.rs @@ -0,0 +1,155 @@ +//! `MySQL` active-transaction inspection for the Community operations monitor. + +use std::time::Duration; + +use chat2db_contract::ActiveTransaction; +use mysql_async::{Error as MysqlError, prelude::Queryable}; + +use crate::{ + AppError, Application, + native_mysql::{finish_connection, open_resolved_connection, resolve_native_connection}, +}; + +const ACTIVE_TRANSACTION_TIMEOUT: Duration = Duration::from_secs(30); +const ACTIVE_TRANSACTION_UNAVAILABLE: &str = "mysql.activeTransaction.unavailable"; +const ACTIVE_TRANSACTION_PRIVILEGE: &str = "mysql.activeTransaction.processPrivilegeRequired"; + +const SELECT_ACTIVE_TRANSACTIONS: &str = " + SELECT t.trx_id, + t.trx_state, + TIMESTAMPDIFF(MICROSECOND, t.trx_started, NOW(6)) DIV 1000, + t.trx_isolation_level, + t.trx_rows_locked, + t.trx_rows_modified, + t.trx_lock_structs, + t.trx_mysql_thread_id, + p.USER, + p.HOST, + p.DB, + COALESCE(p.INFO, '') + FROM information_schema.INNODB_TRX t + LEFT JOIN information_schema.PROCESSLIST p ON p.ID = t.trx_mysql_thread_id + ORDER BY t.trx_started, t.trx_id"; + +type ActiveTransactionRow = ( + String, + String, + Option, + String, + u64, + u64, + u64, + u64, + Option, + Option, + Option, + Option, +); + +impl Application { + /// Lists active `InnoDB` transactions of one native `MySQL` datasource. + /// + /// # Errors + /// + /// Returns connection, privilege, timeout, or query failures. A missing + /// `PROCESS` privilege answers `mysql.activeTransaction.processPrivilegeRequired` + /// so the monitor can explain the requirement. + pub async fn list_mysql_active_transactions( + &self, + datasource_id: &str, + ) -> Result, AppError> { + let resolved = resolve_native_connection(self, datasource_id).await?; + let mut connection = open_resolved_connection(&resolved).await?; + let rows = match tokio::time::timeout( + ACTIVE_TRANSACTION_TIMEOUT, + connection.query::(SELECT_ACTIVE_TRANSACTIONS), + ) + .await + { + Ok(Ok(rows)) => rows, + Ok(Err(error)) => { + return finish_connection(connection, Err(active_transaction_failure(&error))) + .await; + } + Err(_) => { + return finish_connection( + connection, + Err(AppError::invalid( + ACTIVE_TRANSACTION_UNAVAILABLE, + "The MySQL active transaction query timed out", + )), + ) + .await; + } + }; + finish_connection(connection, Ok(())).await?; + Ok(rows + .into_iter() + .map( + |( + trx_id, + state, + age_millis, + isolation_level, + rows_locked, + rows_modified, + lock_structs, + thread_id, + user, + host, + database, + query, + )| { + let age_millis = age_millis.unwrap_or_default(); + ActiveTransaction { + trx_id, + state, + started_at_ms: current_epoch_millis() + .saturating_sub(age_millis) + .to_string(), + age_seconds: u64::try_from(age_millis / 1_000).unwrap_or_default(), + isolation_level, + rows_locked, + rows_modified, + lock_structs, + thread_id, + user, + host, + database, + query: query.unwrap_or_default(), + } + }, + ) + .collect()) + } +} + +fn current_epoch_millis() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .ok() + .and_then(|duration| i64::try_from(duration.as_millis()).ok()) + .unwrap_or_default() +} + +fn active_transaction_failure(error: &MysqlError) -> AppError { + match error { + MysqlError::Server(server) if server.code == 1227 || server.code == 1142 => { + AppError::invalid( + ACTIVE_TRANSACTION_PRIVILEGE, + "Inspecting active transactions requires the PROCESS privilege", + ) + } + MysqlError::Server(server) => AppError::invalid( + ACTIVE_TRANSACTION_UNAVAILABLE, + format!( + "The MySQL active transaction query failed: {}", + server.message + ), + ), + _ => AppError::invalid( + ACTIVE_TRANSACTION_UNAVAILABLE, + "The MySQL connection ended before the active transaction query completed", + ), + } +}