From 8932b0a308bf95ed702cf5be0058acea33793aa6 Mon Sep 17 00:00:00 2001 From: t8y2 <1156263951@qq.com> Date: Sun, 14 Jun 2026 17:04:56 +0800 Subject: [PATCH] fix(mongo): align collection loading state --- .../components/common/QueryLoadingState.vue | 45 +++++++++ .../components/document/DocumentBrowser.vue | 67 ++++++++++++- apps/desktop/src/components/grid/DataGrid.vue | 4 +- .../src/components/layout/ContentArea.vue | 38 +++----- apps/desktop/src/lib/http.ts | 15 +-- apps/desktop/src/lib/tauri.ts | 12 +-- apps/desktop/src/stores/queryStore.ts | 6 +- crates/dbx-web/src/routes/mongo.rs | 94 +++++++++++++------ src-tauri/src/commands/mongo_cmd.rs | 87 ++++++++++++----- 9 files changed, 266 insertions(+), 102 deletions(-) create mode 100644 apps/desktop/src/components/common/QueryLoadingState.vue diff --git a/apps/desktop/src/components/common/QueryLoadingState.vue b/apps/desktop/src/components/common/QueryLoadingState.vue new file mode 100644 index 0000000000..a842fe66bf --- /dev/null +++ b/apps/desktop/src/components/common/QueryLoadingState.vue @@ -0,0 +1,45 @@ + + + diff --git a/apps/desktop/src/components/document/DocumentBrowser.vue b/apps/desktop/src/components/document/DocumentBrowser.vue index 9028c97caa..5cbb861a62 100644 --- a/apps/desktop/src/components/document/DocumentBrowser.vue +++ b/apps/desktop/src/components/document/DocumentBrowser.vue @@ -11,6 +11,7 @@ import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@ import DangerConfirmDialog from "@/components/editor/DangerConfirmDialog.vue"; import ErrorBanner from "@/components/ui/ErrorBanner.vue"; import DataGrid from "@/components/grid/DataGrid.vue"; +import QueryLoadingState from "@/components/common/QueryLoadingState.vue"; import * as api from "@/lib/api"; import { clampSearchSplitWidth } from "@/lib/dataGridSearchSplit"; import { buildDocumentFilterCondition, combineDocumentFilterConditions, currentDocumentFilterJson, defaultDocumentFilterRule, documentFilterModeNeedsValue, documentFilterModeOptions, documentStoreProviderFor, type DocumentFilterMode, type DocumentFilterRule } from "@/lib/documentStoreProvider"; @@ -40,6 +41,9 @@ const documents = ref([]); const lastGridColumns = ref([]); const total = ref(0); const loading = ref(false); +const documentLoadExecutionId = ref(""); +const documentLoadCancelling = ref(false); +const documentLoadingElapsedSeconds = ref("0.0"); const page = ref(0); const pageSize = ref(normalizeResultPageSize(settingsStore.editorSettings.pageSize)); const selectedIdx = ref(null); @@ -145,6 +149,8 @@ const gridResult = computed(() => { }); const documentFilterFieldOptions = computed(() => gridResult.value.columns); const documentStructuredFilterCount = computed(() => (appliedDocumentFilter.value ? 1 : 0)); +const documentLoadingLabelKey = computed(() => (documentLoadCancelling.value ? "common.stopping" : "common.loading")); +let documentLoadingTimer: ReturnType | undefined; function createDocumentFilterRule(): DocumentFilterRule { return defaultDocumentFilterRule(uuid(), documentFilterFieldOptions.value[0] ?? ""); @@ -362,15 +368,35 @@ const customSaveHandler = computed(() => ({ preview: previewDocumentChanges, })); +function stopDocumentLoadingTimer() { + if (documentLoadingTimer) clearInterval(documentLoadingTimer); + documentLoadingTimer = undefined; +} + +function startDocumentLoadingTimer() { + stopDocumentLoadingTimer(); + const startedAt = Date.now(); + documentLoadingElapsedSeconds.value = "0.0"; + documentLoadingTimer = setInterval(() => { + documentLoadingElapsedSeconds.value = ((Date.now() - startedAt) / 1000).toFixed(1); + }, 100); +} + async function load() { + if (documentLoadExecutionId.value) void api.cancelQuery(documentLoadExecutionId.value); + const executionId = uuid(); loading.value = true; + documentLoadExecutionId.value = executionId; + documentLoadCancelling.value = false; + startDocumentLoadingTimer(); error.value = ""; const previousSelectedIdx = selectedIdx.value; const previousSelectedId = previousSelectedIdx === null ? null : documentIdentity(documents.value[previousSelectedIdx]); try { const filter = currentDocumentFilter(); const sort = sortInput.value.trim() || undefined; - const result = await api.documentFindDocuments(props.connectionId, props.database, props.collection, page.value * pageSize.value, pageSize.value, filter, sort); + const result = await api.documentFindDocuments(props.connectionId, props.database, props.collection, page.value * pageSize.value, pageSize.value, filter, sort, executionId); + if (documentLoadExecutionId.value !== executionId) return; const nextDocuments = result.documents.map(asRecord); documents.value = nextDocuments; if (nextDocuments.length > 0) { @@ -386,9 +412,30 @@ async function load() { total.value = result.total; syncSelectedDocumentAfterLoad(previousSelectedIdx, previousSelectedId); } catch (e: unknown) { - error.value = e instanceof Error ? e.message : String(e); + if (documentLoadExecutionId.value === executionId) error.value = e instanceof Error ? e.message : String(e); } finally { - loading.value = false; + if (documentLoadExecutionId.value === executionId) { + loading.value = false; + documentLoadExecutionId.value = ""; + documentLoadCancelling.value = false; + stopDocumentLoadingTimer(); + } + } +} + +async function cancelDocumentLoad() { + const executionId = documentLoadExecutionId.value; + if (!executionId || documentLoadCancelling.value) return; + documentLoadCancelling.value = true; + try { + await api.cancelQuery(executionId); + } finally { + if (documentLoadExecutionId.value === executionId) { + loading.value = false; + documentLoadExecutionId.value = ""; + documentLoadCancelling.value = false; + stopDocumentLoadingTimer(); + } } } @@ -675,6 +722,8 @@ function highlightedJson(json: string): string { onMounted(load); onBeforeUnmount(() => { + if (documentLoadExecutionId.value) void api.cancelQuery(documentLoadExecutionId.value); + stopDocumentLoadingTimer(); endTableSearchSplitResize(); }); @@ -817,8 +866,18 @@ function resetTableSearchSplitWidth() { + (() => { -
+ +
{{ t("grid.querySuccess") }}
diff --git a/apps/desktop/src/components/layout/ContentArea.vue b/apps/desktop/src/components/layout/ContentArea.vue index 307a227c91..46b0a7a687 100644 --- a/apps/desktop/src/components/layout/ContentArea.vue +++ b/apps/desktop/src/components/layout/ContentArea.vue @@ -2,13 +2,14 @@ import { computed, ref, defineAsyncComponent, watch, nextTick, onMounted, onUnmounted } from "vue"; import type { CSSProperties } from "vue"; import { useI18n } from "vue-i18n"; -import { Check, Columns3, Loader2, Search, Square, Bot, GitBranch, BarChart3, TableProperties, ChevronDown, ChevronUp, Inbox, RefreshCcw, Wrench, ListChecks, Download, X } from "@lucide/vue"; +import { Check, Columns3, Loader2, Search, Bot, GitBranch, BarChart3, TableProperties, ChevronDown, ChevronUp, Inbox, RefreshCcw, Wrench, ListChecks, Download, X } from "@lucide/vue"; import { Splitpanes, Pane } from "splitpanes"; import "splitpanes/dist/splitpanes.css"; import { Button } from "@/components/ui/button"; import { Popover, PopoverContent, PopoverTrigger } from "@/components/ui/popover"; import { Switch } from "@/components/ui/switch"; import LightTooltip from "@/components/ui/LightTooltip.vue"; +import QueryLoadingState from "@/components/common/QueryLoadingState.vue"; import QueryEditor from "@/components/editor/QueryEditor.vue"; import ColumnInfoPanel from "@/components/editor/ColumnInfoPanel.vue"; import type { ColumnInfo } from "@/components/editor/ColumnInfoPanel.vue"; @@ -700,18 +701,16 @@ defineExpose({ focusSearch, refreshData, handleModRTarget }); -
-
- - {{ t(queryExecutionLabelKey(activeTab)) }} - · {{ queryRunningElapsedSeconds }}s -
- -
+
{{ t("editor.pressToExecute", { mod: shortcutModifier }) }}
{{ t("editor.pressToSaveSql", { mod: shortcutModifier }) }}
@@ -847,18 +846,7 @@ defineExpose({ focusSearch, refreshData, handleModRTarget }); @paginate="(offset: number, limit: number, whereInput?: string, orderBy?: string) => emit('paginate', offset, limit, whereInput, orderBy)" @sort="(column: string, columnIndex: number, direction: 'asc' | 'desc' | null, whereInput?: string) => emit('sort', column, columnIndex, direction, whereInput)" /> -
-
- - {{ t(queryExecutionLabelKey(activeTab)) }} - · {{ queryRunningElapsedSeconds }}s -
- -
+
{{ t("grid.dataUnavailable") }}
diff --git a/apps/desktop/src/lib/http.ts b/apps/desktop/src/lib/http.ts index fbc06f5e4e..b755c382a9 100644 --- a/apps/desktop/src/lib/http.ts +++ b/apps/desktop/src/lib/http.ts @@ -545,7 +545,8 @@ export async function executeInTransaction(connectionId: string, database: strin } export async function cancelQuery(executionId: string): Promise { - return post("/api/query/cancel", { executionId }); + const result = await post("/api/query/cancel", { executionId }); + return typeof result === "boolean" ? result : result.cancelled === true; } export async function analyzeSqlReferences(sql: string, dialect?: string): Promise { @@ -1389,16 +1390,16 @@ export async function elasticsearchListIndices(connectionId: string): Promise { - return post("/api/mongo/find-documents", { connectionId, database, collection, skip, limit, filter, sort }); +export async function mongoFindDocuments(connectionId: string, database: string, collection: string, skip: number, limit: number, filter?: string, sort?: string, executionId?: string): Promise { + return post("/api/mongo/find-documents", { connectionId, database, collection, skip, limit, filter, sort, executionId }); } -export async function documentFindDocuments(connectionId: string, database: string, collection: string, skip: number, limit: number, filter?: string, sort?: string): Promise { - return post("/api/document-store/find-documents", { connectionId, database, collection, skip, limit, filter, sort }); +export async function documentFindDocuments(connectionId: string, database: string, collection: string, skip: number, limit: number, filter?: string, sort?: string, executionId?: string): Promise { + return post("/api/document-store/find-documents", { connectionId, database, collection, skip, limit, filter, sort, executionId }); } -export async function mongoAggregateDocuments(connectionId: string, database: string, collection: string, pipelineJson: string, maxRows?: number): Promise { - return post("/api/mongo/aggregate-documents", { connectionId, database, collection, pipelineJson, maxRows }); +export async function mongoAggregateDocuments(connectionId: string, database: string, collection: string, pipelineJson: string, maxRows?: number, executionId?: string): Promise { + return post("/api/mongo/aggregate-documents", { connectionId, database, collection, pipelineJson, maxRows, executionId }); } export async function mongoInsertDocument(connectionId: string, database: string, collection: string, docJson: string): Promise { diff --git a/apps/desktop/src/lib/tauri.ts b/apps/desktop/src/lib/tauri.ts index 8509163520..ee5c1008d7 100644 --- a/apps/desktop/src/lib/tauri.ts +++ b/apps/desktop/src/lib/tauri.ts @@ -1268,16 +1268,16 @@ export async function elasticsearchListIndices(connectionId: string): Promise { - return invoke("mongo_find_documents", { connectionId, database, collection, skip, limit, filter, sort }); +export async function mongoFindDocuments(connectionId: string, database: string, collection: string, skip: number, limit: number, filter?: string, sort?: string, executionId?: string): Promise { + return invoke("mongo_find_documents", { connectionId, database, collection, skip, limit, filter, sort, executionId }); } -export async function documentFindDocuments(connectionId: string, database: string, collection: string, skip: number, limit: number, filter?: string, sort?: string): Promise { - return invoke("document_find_documents", { connectionId, database, collection, skip, limit, filter, sort }); +export async function documentFindDocuments(connectionId: string, database: string, collection: string, skip: number, limit: number, filter?: string, sort?: string, executionId?: string): Promise { + return invoke("document_find_documents", { connectionId, database, collection, skip, limit, filter, sort, executionId }); } -export async function mongoAggregateDocuments(connectionId: string, database: string, collection: string, pipelineJson: string, maxRows?: number): Promise { - return invoke("mongo_aggregate_documents", { connectionId, database, collection, pipelineJson, maxRows }); +export async function mongoAggregateDocuments(connectionId: string, database: string, collection: string, pipelineJson: string, maxRows?: number, executionId?: string): Promise { + return invoke("mongo_aggregate_documents", { connectionId, database, collection, pipelineJson, maxRows, executionId }); } export async function mongoInsertDocument(connectionId: string, database: string, collection: string, docJson: string): Promise { diff --git a/apps/desktop/src/stores/queryStore.ts b/apps/desktop/src/stores/queryStore.ts index 4aa26df1f3..1c44266cb9 100644 --- a/apps/desktop/src/stores/queryStore.ts +++ b/apps/desktop/src/stores/queryStore.ts @@ -1225,7 +1225,7 @@ export const useQueryStore = defineStore("query", () => { if (mongoFind) { await connStore.ensureConnected(tab.connectionId); console.info("[DBX][executeTabSql:mongo-find:start]", { traceId, collection: mongoFind.collection }); - const result = await api.mongoFindDocuments(tab.connectionId, tab.database, mongoFind.collection, mongoFind.skip, mongoFind.limit, mongoFind.filter, mongoFind.sort); + const result = await api.mongoFindDocuments(tab.connectionId, tab.database, mongoFind.collection, mongoFind.skip, mongoFind.limit, mongoFind.filter, mongoFind.sort, executionId); console.info("[DBX][executeTabSql:mongo-find:done]", { traceId, rowCount: result.documents.length, @@ -1252,7 +1252,7 @@ export const useQueryStore = defineStore("query", () => { if (mongoCount) { await connStore.ensureConnected(tab.connectionId); console.info("[DBX][executeTabSql:mongo-count:start]", { traceId, collection: mongoCount.collection }); - const result = await api.mongoFindDocuments(tab.connectionId, tab.database, mongoCount.collection, 0, 1, mongoCount.filter); + const result = await api.mongoFindDocuments(tab.connectionId, tab.database, mongoCount.collection, 0, 1, mongoCount.filter, undefined, executionId); console.info("[DBX][executeTabSql:mongo-count:done]", { traceId, total: result.total, @@ -1283,7 +1283,7 @@ export const useQueryStore = defineStore("query", () => { } await connStore.ensureConnected(tab.connectionId); console.info("[DBX][executeTabSql:mongo-aggregate:start]", { traceId, collection: mongoAggregate.collection }); - const result = await api.mongoAggregateDocuments(tab.connectionId, tab.database, mongoAggregate.collection, mongoAggregate.pipeline, pageLimit); + const result = await api.mongoAggregateDocuments(tab.connectionId, tab.database, mongoAggregate.collection, mongoAggregate.pipeline, pageLimit, executionId); console.info("[DBX][executeTabSql:mongo-aggregate:done]", { traceId, rowCount: result.documents.length, diff --git a/crates/dbx-web/src/routes/mongo.rs b/crates/dbx-web/src/routes/mongo.rs index b09d26315a..0a7fede627 100644 --- a/crates/dbx-web/src/routes/mongo.rs +++ b/crates/dbx-web/src/routes/mongo.rs @@ -1,3 +1,4 @@ +use std::future::Future; use std::sync::Arc; use axum::extract::State; @@ -7,6 +8,26 @@ use serde::Deserialize; use crate::error::AppError; use crate::state::WebState; +async fn run_cancellable(state: &Arc, execution_id: Option, future: F) -> Result +where + F: Future>, +{ + let registered = execution_id + .as_ref() + .filter(|id| !id.trim().is_empty()) + .map(|id| state.app.running_queries.register(id.clone())); + if let Some(query) = registered.as_ref() { + let token = query.token(); + tokio::select! { + biased; + _ = token.cancelled() => Err(AppError(dbx_core::query::canceled_error())), + result = future => result.map_err(AppError), + } + } else { + future.await.map_err(AppError) + } +} + /// Check if a connection is read-only and return an error if so. async fn ensure_writable( app: &dbx_core::connection::AppState, @@ -45,6 +66,7 @@ pub struct MongoFindRequest { pub limit: Option, pub filter: Option, pub sort: Option, + pub execution_id: Option, } #[derive(Deserialize)] @@ -55,6 +77,7 @@ pub struct MongoAggregateRequest { pub collection: String, pub pipeline_json: String, pub max_rows: Option, + pub execution_id: Option, } #[derive(Deserialize)] @@ -138,18 +161,21 @@ pub async fn find_documents( State(state): State>, Json(req): Json, ) -> Result, AppError> { - let result = dbx_core::mongo_ops::mongo_find_documents_core( - &state.app, - &req.connection_id, - &req.database, - &req.collection, - req.skip.unwrap_or(0), - req.limit.unwrap_or(50), - req.filter.as_deref(), - req.sort.as_deref(), + let result = run_cancellable( + &state, + req.execution_id.clone(), + dbx_core::mongo_ops::mongo_find_documents_core( + &state.app, + &req.connection_id, + &req.database, + &req.collection, + req.skip.unwrap_or(0), + req.limit.unwrap_or(50), + req.filter.as_deref(), + req.sort.as_deref(), + ), ) - .await - .map_err(AppError)?; + .await?; Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?)) } @@ -157,18 +183,21 @@ pub async fn document_find_documents( State(state): State>, Json(req): Json, ) -> Result, AppError> { - let result = dbx_core::mongo_ops::document_find_documents_core( - &state.app, - &req.connection_id, - &req.database, - &req.collection, - req.skip.unwrap_or(0), - req.limit.unwrap_or(50), - req.filter.as_deref(), - req.sort.as_deref(), + let result = run_cancellable( + &state, + req.execution_id.clone(), + dbx_core::mongo_ops::document_find_documents_core( + &state.app, + &req.connection_id, + &req.database, + &req.collection, + req.skip.unwrap_or(0), + req.limit.unwrap_or(50), + req.filter.as_deref(), + req.sort.as_deref(), + ), ) - .await - .map_err(AppError)?; + .await?; Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?)) } @@ -176,16 +205,19 @@ pub async fn aggregate_documents( State(state): State>, Json(req): Json, ) -> Result, AppError> { - let result = dbx_core::mongo_ops::mongo_aggregate_documents_core( - &state.app, - &req.connection_id, - &req.database, - &req.collection, - &req.pipeline_json, - req.max_rows, + let result = run_cancellable( + &state, + req.execution_id.clone(), + dbx_core::mongo_ops::mongo_aggregate_documents_core( + &state.app, + &req.connection_id, + &req.database, + &req.collection, + &req.pipeline_json, + req.max_rows, + ), ) - .await - .map_err(AppError)?; + .await?; Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?)) } diff --git a/src-tauri/src/commands/mongo_cmd.rs b/src-tauri/src/commands/mongo_cmd.rs index 0fc4eeae2f..89c490ba2c 100644 --- a/src-tauri/src/commands/mongo_cmd.rs +++ b/src-tauri/src/commands/mongo_cmd.rs @@ -1,9 +1,28 @@ +use std::future::Future; use std::sync::Arc; use tauri::State; use crate::commands::connection::{ensure_connection_writable, AppState}; use dbx_core::db::mongo_driver::MongoDocumentResult; +async fn run_cancellable(state: &Arc, execution_id: Option, future: F) -> Result +where + F: Future>, +{ + let registered_query = + execution_id.as_ref().filter(|id| !id.trim().is_empty()).map(|id| state.running_queries.register(id.clone())); + if let Some(query) = registered_query.as_ref() { + let token = query.token(); + tokio::select! { + biased; + _ = token.cancelled() => Err(dbx_core::query::canceled_error()), + result = future => result, + } + } else { + future.await + } +} + #[tauri::command] pub async fn mongo_list_databases( state: State<'_, Arc>, @@ -32,16 +51,22 @@ pub async fn mongo_find_documents( limit: i64, filter: Option, sort: Option, + execution_id: Option, ) -> Result { - dbx_core::mongo_ops::mongo_find_documents_core( - &state, - &connection_id, - &database, - &collection, - skip, - limit, - filter.as_deref(), - sort.as_deref(), + let app = state.inner().clone(); + run_cancellable( + &app, + execution_id, + dbx_core::mongo_ops::mongo_find_documents_core( + &app, + &connection_id, + &database, + &collection, + skip, + limit, + filter.as_deref(), + sort.as_deref(), + ), ) .await } @@ -57,16 +82,22 @@ pub async fn document_find_documents( limit: i64, filter: Option, sort: Option, + execution_id: Option, ) -> Result { - dbx_core::mongo_ops::document_find_documents_core( - &state, - &connection_id, - &database, - &collection, - skip, - limit, - filter.as_deref(), - sort.as_deref(), + let app = state.inner().clone(); + run_cancellable( + &app, + execution_id, + dbx_core::mongo_ops::document_find_documents_core( + &app, + &connection_id, + &database, + &collection, + skip, + limit, + filter.as_deref(), + sort.as_deref(), + ), ) .await } @@ -79,14 +110,20 @@ pub async fn mongo_aggregate_documents( collection: String, pipeline_json: String, max_rows: Option, + execution_id: Option, ) -> Result { - dbx_core::mongo_ops::mongo_aggregate_documents_core( - &state, - &connection_id, - &database, - &collection, - &pipeline_json, - max_rows, + let app = state.inner().clone(); + run_cancellable( + &app, + execution_id, + dbx_core::mongo_ops::mongo_aggregate_documents_core( + &app, + &connection_id, + &database, + &collection, + &pipeline_json, + max_rows, + ), ) .await }