From 41bceffc0e98a4f15f538b7180e0d2d1bf8edcad Mon Sep 17 00:00:00 2001 From: t8y2 <1156263951@qq.com> Date: Tue, 5 May 2026 03:16:50 +0800 Subject: [PATCH] feat: add web deployment with axum server, Docker support, and browser UI Introduce src-web (axum HTTP server) exposing all 43 database operations as REST APIs with SSE streaming for AI, transfers, and SQL file execution. Frontend adapts automatically between Tauri and HTTP backends via api.ts. Includes password authentication, Docker packaging, and login page. --- .dockerignore | 12 + Dockerfile | 33 + docker-compose.yml | 13 + package.json | 3 + src-web/Cargo.toml | 25 + src-web/src/auth.rs | 121 ++++ src-web/src/error.rs | 22 + src-web/src/main.rs | 172 ++++++ src-web/src/routes/ai.rs | 187 ++++++ src-web/src/routes/connection.rs | 123 ++++ src-web/src/routes/history.rs | 56 ++ src-web/src/routes/layout.rs | 37 ++ src-web/src/routes/mod.rs | 12 + src-web/src/routes/mongo.rs | 149 +++++ src-web/src/routes/query.rs | 105 ++++ src-web/src/routes/redis.rs | 208 +++++++ src-web/src/routes/schema.rs | 145 +++++ src-web/src/routes/sql_file.rs | 238 ++++++++ src-web/src/routes/table_import.rs | 176 ++++++ src-web/src/routes/transfer.rs | 178 ++++++ src-web/src/routes/update.rs | 10 + src-web/src/sse.rs | 14 + src-web/src/state.rs | 13 + src/App.vue | 79 ++- src/components/auth/LoginPage.vue | 57 ++ .../connection/ConnectionDialog.vue | 19 +- .../diagram/SchemaDiagramDialog.vue | 34 +- src/components/diff/SchemaDiffDialog.vue | 2 +- src/components/editor/AiAssistant.vue | 12 +- src/components/grid/DataGrid.vue | 31 +- src/components/import/TableImportDialog.vue | 8 +- src/components/lineage/FieldLineageDialog.vue | 2 +- src/components/mongo/MongoDocBrowser.vue | 2 +- src/components/redis/RedisKeyBrowser.vue | 4 +- src/components/redis/RedisValueViewer.vue | 4 +- .../search/DatabaseSearchDialog.vue | 2 +- src/components/sidebar/TreeItem.vue | 41 +- .../sql-file/SqlFileExecutionDialog.vue | 8 +- .../structure/TableStructureEditorDialog.vue | 2 +- .../transfer/DataTransferDialog.vue | 4 +- src/lib/ai.ts | 2 +- src/lib/api.ts | 148 +++++ src/lib/http.ts | 566 ++++++++++++++++++ src/stores/connectionStore.ts | 55 +- src/stores/historyStore.ts | 4 +- src/stores/queryStore.ts | 2 +- src/stores/settingsStore.ts | 2 +- vite.config.ts | 11 +- 48 files changed, 3044 insertions(+), 109 deletions(-) create mode 100644 .dockerignore create mode 100644 Dockerfile create mode 100644 docker-compose.yml create mode 100644 src-web/Cargo.toml create mode 100644 src-web/src/auth.rs create mode 100644 src-web/src/error.rs create mode 100644 src-web/src/main.rs create mode 100644 src-web/src/routes/ai.rs create mode 100644 src-web/src/routes/connection.rs create mode 100644 src-web/src/routes/history.rs create mode 100644 src-web/src/routes/layout.rs create mode 100644 src-web/src/routes/mod.rs create mode 100644 src-web/src/routes/mongo.rs create mode 100644 src-web/src/routes/query.rs create mode 100644 src-web/src/routes/redis.rs create mode 100644 src-web/src/routes/schema.rs create mode 100644 src-web/src/routes/sql_file.rs create mode 100644 src-web/src/routes/table_import.rs create mode 100644 src-web/src/routes/transfer.rs create mode 100644 src-web/src/routes/update.rs create mode 100644 src-web/src/sse.rs create mode 100644 src-web/src/state.rs create mode 100644 src/components/auth/LoginPage.vue create mode 100644 src/lib/api.ts create mode 100644 src/lib/http.ts diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 000000000..5c3370a0f --- /dev/null +++ b/.dockerignore @@ -0,0 +1,12 @@ +target/ +node_modules/ +dist/ +src-tauri/target/ +.git/ +.github/ +.claude/ +*.log +.DS_Store +.env +.env.* +tmp/ diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 000000000..9be7687db --- /dev/null +++ b/Dockerfile @@ -0,0 +1,33 @@ +# Stage 1: Build frontend +FROM node:20-slim AS frontend +WORKDIR /app +RUN npm i -g pnpm +COPY package.json pnpm-lock.yaml ./ +RUN pnpm install --frozen-lockfile +COPY src/ src/ +COPY index.html vite.config.ts tsconfig.json tsconfig.app.json tsconfig.node.json ./ +COPY tailwind.config.* postcss.config.* ./ +RUN pnpm build + +# Stage 2: Build Rust backend +FROM rust:1-bookworm AS backend +WORKDIR /app +COPY Cargo.toml Cargo.lock ./ +COPY crates/ crates/ +COPY src-web/ src-web/ +# Create a dummy src-tauri to satisfy workspace (not built) +RUN mkdir -p src-tauri/src && echo 'fn main() {}' > src-tauri/src/main.rs && echo 'pub fn run() {}' > src-tauri/src/lib.rs +COPY src-tauri/Cargo.toml src-tauri/ +COPY src-tauri/build.rs src-tauri/ +COPY src-tauri/tauri.conf.json src-tauri/ +RUN cargo build --release -p dbx-web + +# Stage 3: Final image +FROM debian:bookworm-slim +RUN apt-get update && apt-get install -y ca-certificates && rm -rf /var/lib/apt/lists/* +COPY --from=backend /app/target/release/dbx-web /usr/local/bin/ +COPY --from=frontend /app/dist /app/static +ENV DBX_STATIC_DIR=/app/static +ENV DBX_DATA_DIR=/app/data +EXPOSE 4224 +CMD ["dbx-web"] diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 000000000..a56d9afb8 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,13 @@ +services: + dbx: + build: . + ports: + - "4224:4224" + environment: + - DBX_PASSWORD=changeme + volumes: + - dbx-data:/app/data + restart: unless-stopped + +volumes: + dbx-data: diff --git a/package.json b/package.json index 5b2d1968b..3b8c9484b 100644 --- a/package.json +++ b/package.json @@ -5,6 +5,9 @@ "type": "module", "scripts": { "dev": "vite", + "dev:tauri": "tauri dev", + "dev:web": "vite --port 5173 --mode web", + "dev:backend": "RUST_LOG=info cargo watch -x 'run -p dbx-web'", "build": "vue-tsc --noEmit && vite build", "preview": "vite preview", "tauri": "tauri" diff --git a/src-web/Cargo.toml b/src-web/Cargo.toml new file mode 100644 index 000000000..5de911b15 --- /dev/null +++ b/src-web/Cargo.toml @@ -0,0 +1,25 @@ +[package] +name = "dbx-web" +version = "0.4.0" +edition = "2021" + +[[bin]] +name = "dbx-web" +path = "src/main.rs" + +[dependencies] +dbx-core = { path = "../crates/dbx-core" } +axum = { version = "0.8", features = ["multipart"] } +tower-http = { version = "0.6", features = ["cors", "fs", "compression-gzip"] } +tokio = { version = "1", features = ["full"] } +serde = { version = "1.0", features = ["derive"] } +serde_json = "1.0" +uuid = { version = "1", features = ["v4"] } +sha2 = "0.10" +hex = "0.4" +log = "0.4" +env_logger = "0.11" +async-stream = "0.3" +futures = "0.3" +rustls = { version = "0.23", features = ["aws-lc-rs"] } +tokio-util = "0.7" diff --git a/src-web/src/auth.rs b/src-web/src/auth.rs new file mode 100644 index 000000000..c8c3ae97c --- /dev/null +++ b/src-web/src/auth.rs @@ -0,0 +1,121 @@ +use std::sync::Arc; + +use axum::extract::State; +use axum::http::{Request, StatusCode}; +use axum::middleware::Next; +use axum::response::{IntoResponse, Response}; +use axum::Json; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; + +use crate::state::WebState; + +#[derive(Deserialize)] +pub struct LoginRequest { + pub password: String, +} + +#[derive(Serialize)] +pub struct AuthCheckResponse { + pub authenticated: bool, + pub required: bool, +} + +pub async fn login( + State(state): State>, + Json(body): Json, +) -> Result { + let password_hash = match &state.password_hash { + Some(h) => h, + None => { + return Ok(( + StatusCode::OK, + Json(serde_json::json!({"ok": true})), + ) + .into_response()); + } + }; + + let mut hasher = Sha256::new(); + hasher.update(body.password.as_bytes()); + let hash = hex::encode(hasher.finalize()); + + if hash != *password_hash { + return Err(StatusCode::UNAUTHORIZED); + } + + let token = uuid::Uuid::new_v4().to_string(); + state.sessions.write().await.insert(token.clone()); + + let cookie = format!("dbx_session={token}; Path=/; HttpOnly; SameSite=Lax"); + Ok(( + StatusCode::OK, + [("set-cookie", cookie.as_str())], + Json(serde_json::json!({"ok": true})), + ) + .into_response()) +} + +pub async fn check(State(state): State>) -> Json { + Json(AuthCheckResponse { + authenticated: state.password_hash.is_none(), + required: state.password_hash.is_some(), + }) +} + +pub async fn logout(State(state): State>, req: Request) -> Response { + if let Some(token) = extract_session_token(&req) { + state.sessions.write().await.remove(&token); + } + let cookie = "dbx_session=; Path=/; HttpOnly; Max-Age=0"; + ( + StatusCode::OK, + [("set-cookie", cookie)], + Json(serde_json::json!({"ok": true})), + ) + .into_response() +} + +fn extract_session_token(req: &Request) -> Option { + let cookie_header = req.headers().get("cookie")?.to_str().ok()?; + for pair in cookie_header.split(';') { + let pair = pair.trim(); + if let Some(value) = pair.strip_prefix("dbx_session=") { + if !value.is_empty() { + return Some(value.to_string()); + } + } + } + None +} + +pub async fn auth_middleware( + State(state): State>, + req: Request, + next: Next, +) -> Response { + // No password set — allow everything + if state.password_hash.is_none() { + return next.run(req).await; + } + + // Auth endpoints are always accessible + let path = req.uri().path(); + if path.starts_with("/api/auth/") { + return next.run(req).await; + } + + // Non-API requests (static files) are always accessible + if !path.starts_with("/api/") { + return next.run(req).await; + } + + // Check session token + if let Some(token) = extract_session_token(&req) { + if state.sessions.read().await.contains(&token) { + return next.run(req).await; + } + } + + StatusCode::UNAUTHORIZED.into_response() +} diff --git a/src-web/src/error.rs b/src-web/src/error.rs new file mode 100644 index 000000000..7f18b8ecb --- /dev/null +++ b/src-web/src/error.rs @@ -0,0 +1,22 @@ +use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; + +pub struct AppError(pub String); + +impl IntoResponse for AppError { + fn into_response(self) -> Response { + (StatusCode::INTERNAL_SERVER_ERROR, self.0).into_response() + } +} + +impl From for AppError { + fn from(s: String) -> Self { + AppError(s) + } +} + +impl From<&str> for AppError { + fn from(s: &str) -> Self { + AppError(s.to_string()) + } +} diff --git a/src-web/src/main.rs b/src-web/src/main.rs new file mode 100644 index 000000000..6d77dadaa --- /dev/null +++ b/src-web/src/main.rs @@ -0,0 +1,172 @@ +mod auth; +mod error; +mod routes; +mod sse; +mod state; + +use std::collections::{HashMap, HashSet}; +use std::net::SocketAddr; +use std::sync::Arc; + +use axum::middleware; +use axum::routing::{delete, get, post}; +use axum::Router; +use dbx_core::connection::AppState; +use sha2::{Digest, Sha256}; +use tokio::sync::RwLock; +use tower_http::cors::{Any, CorsLayer}; + +use state::WebState; + +#[tokio::main] +async fn main() { + env_logger::init(); + + rustls::crypto::aws_lc_rs::default_provider() + .install_default() + .expect("Failed to install rustls crypto provider"); + + let app_state = Arc::new(AppState::new()); + + // Data directory + let data_dir = std::env::var("DBX_DATA_DIR") + .map(std::path::PathBuf::from) + .unwrap_or_else(|_| { + let home = std::env::var("HOME").unwrap_or_else(|_| ".".to_string()); + std::path::PathBuf::from(home).join(".dbx-web") + }); + std::fs::create_dir_all(&data_dir).expect("Failed to create data directory"); + + // Password hash + let password_hash = std::env::var("DBX_PASSWORD").ok().map(|pw| { + let mut hasher = Sha256::new(); + hasher.update(pw.as_bytes()); + hex::encode(hasher.finalize()) + }); + + let web_state = Arc::new(WebState { + app: app_state, + data_dir, + password_hash, + sessions: RwLock::new(HashSet::new()), + sse_channels: RwLock::new(HashMap::new()), + }); + + // CORS + let cors = CorsLayer::new() + .allow_origin(Any) + .allow_methods(Any) + .allow_headers(Any); + + // API routes + let api = Router::new() + // Auth + .route("/auth/login", post(auth::login)) + .route("/auth/check", get(auth::check)) + .route("/auth/logout", post(auth::logout)) + // Connection + .route("/connection/test", post(routes::connection::test_connection)) + .route("/connection/connect", post(routes::connection::connect_db)) + .route("/connection/disconnect", post(routes::connection::disconnect_db)) + .route("/connection/save", post(routes::connection::save_connections)) + .route("/connection/list", get(routes::connection::load_connections)) + // Schema + .route("/schema/databases", get(routes::schema::list_databases)) + .route("/schema/schemas", get(routes::schema::list_schemas)) + .route("/schema/tables", get(routes::schema::list_tables)) + .route("/schema/columns", get(routes::schema::list_columns)) + .route("/schema/indexes", get(routes::schema::list_indexes)) + .route("/schema/foreign-keys", get(routes::schema::list_foreign_keys)) + .route("/schema/triggers", get(routes::schema::list_triggers)) + .route("/schema/ddl", get(routes::schema::get_ddl)) + // Query + .route("/query/execute", post(routes::query::execute_query)) + .route("/query/execute-multi", post(routes::query::execute_multi)) + .route("/query/execute-batch", post(routes::query::execute_batch)) + .route("/query/cancel", post(routes::query::cancel_query)) + // Redis + .route("/redis/list-databases", post(routes::redis::list_databases)) + .route("/redis/scan-keys", post(routes::redis::scan_keys)) + .route("/redis/get-value", post(routes::redis::get_value)) + .route("/redis/set-string", post(routes::redis::set_string)) + .route("/redis/delete-key", post(routes::redis::delete_key)) + .route("/redis/hash-set", post(routes::redis::hash_set)) + .route("/redis/hash-del", post(routes::redis::hash_del)) + .route("/redis/list-push", post(routes::redis::list_push)) + .route("/redis/list-remove", post(routes::redis::list_remove)) + .route("/redis/set-add", post(routes::redis::set_add)) + .route("/redis/set-remove", post(routes::redis::set_remove)) + // MongoDB + .route("/mongo/list-databases", post(routes::mongo::list_databases)) + .route("/mongo/list-collections", post(routes::mongo::list_collections)) + .route("/mongo/find-documents", post(routes::mongo::find_documents)) + .route("/mongo/insert-document", post(routes::mongo::insert_document)) + .route("/mongo/update-document", post(routes::mongo::update_document)) + .route("/mongo/delete-document", post(routes::mongo::delete_document)) + // History + .route("/history", get(routes::history::load_history).delete(routes::history::clear_history)) + .route("/history/save", post(routes::history::save_history)) + .route("/history/{id}", delete(routes::history::delete_history_entry)) + // AI + .route("/ai/config", post(routes::ai::save_ai_config).get(routes::ai::load_ai_config)) + .route("/ai/conversation", post(routes::ai::save_ai_conversation)) + .route("/ai/conversations", get(routes::ai::load_ai_conversations)) + .route("/ai/conversation/{id}", delete(routes::ai::delete_ai_conversation)) + .route("/ai/complete", post(routes::ai::ai_complete)) + .route("/ai/stream", post(routes::ai::ai_stream)) + .route("/ai/cancel-stream", post(routes::ai::ai_cancel_stream)) + .route("/ai/test-connection", post(routes::ai::ai_test_connection)) + // Transfer + .route("/transfer/start", post(routes::transfer::start_transfer)) + .route("/transfer/progress/{transferId}", get(routes::transfer::transfer_progress)) + .route("/transfer/cancel", post(routes::transfer::cancel_transfer)) + // SQL file + .route("/sql-file/preview", post(routes::sql_file::preview_sql_file)) + .route("/sql-file/execute", post(routes::sql_file::execute_sql_file)) + .route("/sql-file/progress/{executionId}", get(routes::sql_file::sql_file_progress)) + .route("/sql-file/cancel", post(routes::sql_file::cancel_sql_file)) + // Table import + .route("/import/preview", post(routes::table_import::preview_import)) + .route("/import/execute", post(routes::table_import::execute_import)) + .route("/import/progress/{importId}", get(routes::table_import::import_progress)) + .route("/import/cancel", post(routes::table_import::cancel_import)) + // Update + .route("/update/check", get(routes::update::check_for_updates)) + // Layout + .route("/layout/sidebar", post(routes::layout::save_sidebar_layout).get(routes::layout::load_sidebar_layout)) + .with_state(web_state.clone()); + + // Build app with auth middleware + let mut app = Router::new() + .nest("/api", api) + .layer(middleware::from_fn_with_state(web_state.clone(), auth::auth_middleware)) + .layer(cors); + + // Static file serving + if let Ok(static_dir) = std::env::var("DBX_STATIC_DIR") { + use tower_http::services::{ServeDir, ServeFile}; + let index_path = format!("{}/index.html", static_dir); + let serve_dir = ServeDir::new(&static_dir) + .not_found_service(ServeFile::new(&index_path)); + app = app.fallback_service(serve_dir); + } + + // Bind address + let port: u16 = std::env::var("DBX_PORT") + .ok() + .and_then(|p| p.parse().ok()) + .unwrap_or(4224); + let addr = SocketAddr::from(([0, 0, 0, 0], port)); + + log::info!("DBX Web server starting on http://{}", addr); + if std::env::var("DBX_PASSWORD").is_ok() { + log::info!("Password protection is enabled"); + } + + let listener = tokio::net::TcpListener::bind(addr) + .await + .expect("Failed to bind address"); + axum::serve(listener, app) + .await + .expect("Server error"); +} diff --git a/src-web/src/routes/ai.rs b/src-web/src/routes/ai.rs new file mode 100644 index 000000000..2ce7635c6 --- /dev/null +++ b/src-web/src/routes/ai.rs @@ -0,0 +1,187 @@ +use std::sync::Arc; + +use axum::extract::{Path, State}; +use axum::response::sse::{Event, KeepAlive, Sse}; +use axum::Json; +use futures::stream::Stream; +use serde::Deserialize; + +use dbx_core::ai::{ + AiCompletionRequest, AiConfig, AiConversation, AiStreamChunk, +}; + +use crate::error::AppError; +use crate::state::WebState; + +// --------------------------------------------------------------------------- +// Request types +// --------------------------------------------------------------------------- + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SaveAiConfigRequest { + pub config: AiConfig, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SaveAiConversationRequest { + pub conversation: AiConversation, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct AiCompleteRequest { + pub request: AiCompletionRequest, +} + +#[derive(Deserialize)] +pub struct AiStreamRequest { + #[serde(alias = "sessionId")] + pub session_id: String, + pub request: AiCompletionRequest, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct AiTestConnectionRequest { + pub config: AiConfig, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct AiCancelStreamRequest { + pub session_id: String, +} + +// --------------------------------------------------------------------------- +// Config +// --------------------------------------------------------------------------- + +pub async fn save_ai_config( + State(state): State>, + Json(body): Json, +) -> Result, AppError> { + let path = state.data_dir.join("ai_config.json"); + dbx_core::ai::save_config(&path, &body.config).map_err(AppError)?; + Ok(Json(())) +} + +pub async fn load_ai_config( + State(state): State>, +) -> Result>, AppError> { + let path = state.data_dir.join("ai_config.json"); + let config = dbx_core::ai::load_config(&path).map_err(AppError)?; + Ok(Json(config)) +} + +// --------------------------------------------------------------------------- +// Conversations +// --------------------------------------------------------------------------- + +pub async fn save_ai_conversation( + State(state): State>, + Json(body): Json, +) -> Result, AppError> { + let path = state.data_dir.join("ai_conversations.json"); + dbx_core::ai::save_conversation(&path, body.conversation).map_err(AppError)?; + Ok(Json(())) +} + +pub async fn load_ai_conversations( + State(state): State>, +) -> Result>, AppError> { + let path = state.data_dir.join("ai_conversations.json"); + let conversations = dbx_core::ai::load_conversations(&path).map_err(AppError)?; + Ok(Json(conversations)) +} + +pub async fn delete_ai_conversation( + State(state): State>, + Path(id): Path, +) -> Result, AppError> { + let path = state.data_dir.join("ai_conversations.json"); + dbx_core::ai::delete_conversation(&path, &id).map_err(AppError)?; + Ok(Json(())) +} + +// --------------------------------------------------------------------------- +// AI complete (non-streaming) +// --------------------------------------------------------------------------- + +pub async fn ai_complete( + Json(body): Json, +) -> Result, AppError> { + let result = dbx_core::ai::complete(&body.request) + .await + .map_err(AppError)?; + Ok(Json(result)) +} + +// --------------------------------------------------------------------------- +// AI test connection +// --------------------------------------------------------------------------- + +pub async fn ai_test_connection( + Json(body): Json, +) -> Result, AppError> { + let result = dbx_core::ai::test_connection_core(&body.config) + .await + .map_err(AppError)?; + Ok(Json(result)) +} + +// --------------------------------------------------------------------------- +// AI cancel stream +// --------------------------------------------------------------------------- + +pub async fn ai_cancel_stream( + Json(body): Json, +) -> Result, AppError> { + let result = dbx_core::ai::cancel_stream(&body.session_id).await; + Ok(Json(result)) +} + +// --------------------------------------------------------------------------- +// AI stream (POST returns SSE directly) +// --------------------------------------------------------------------------- + +pub async fn ai_stream( + Json(body): Json, +) -> Result>>, AppError> { + let session_id = body.session_id; + let request = body.request; + + let cancelled = dbx_core::ai::register_stream(&session_id).await; + let (tx, rx) = tokio::sync::broadcast::channel::(256); + + let sid = session_id.clone(); + tokio::spawn(async move { + let result = dbx_core::ai::stream(&sid, &request, &cancelled, |chunk: AiStreamChunk| { + let json = serde_json::to_string(&chunk).unwrap_or_default(); + let _ = tx.send(json); + }) + .await; + + if let Err(_e) = result { + let error_chunk = AiStreamChunk { + session_id: sid.clone(), + delta: String::new(), + reasoning_delta: None, + done: true, + }; + let _ = tx.send(serde_json::to_string(&error_chunk).unwrap_or_default()); + } + + dbx_core::ai::unregister_stream(&sid).await; + }); + + let stream = async_stream::stream! { + let mut rx = rx; + while let Ok(data) = rx.recv().await { + yield Ok(Event::default().data(data)); + } + }; + + Ok(Sse::new(stream).keep_alive(KeepAlive::default())) +} diff --git a/src-web/src/routes/connection.rs b/src-web/src/routes/connection.rs new file mode 100644 index 000000000..72e2f87b7 --- /dev/null +++ b/src-web/src/routes/connection.rs @@ -0,0 +1,123 @@ +use std::sync::Arc; + +use axum::extract::State; +use axum::Json; +use dbx_core::connection_secrets::{ + load_connections_from_file, save_connections_to_file, FileSecretStore, +}; +use dbx_core::models::connection::ConnectionConfig; +use serde::Deserialize; + +use crate::error::AppError; +use crate::state::WebState; + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ConnectRequest { + pub config: ConnectionConfig, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct DisconnectRequest { + pub connection_id: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SaveConnectionsRequest { + pub configs: Vec, +} + +pub async fn test_connection( + State(state): State>, + Json(body): Json, +) -> Result, AppError> { + let config = body.config; + let app = &state.app; + + // Store config temporarily + let temp_id = format!("__test_{}", uuid::Uuid::new_v4()); + app.configs + .lock() + .await + .insert(temp_id.clone(), config.clone()); + + // Try to connect + let result = app.get_or_create_pool(&temp_id, config.database.as_deref()).await; + + // Clean up + app.connections.lock().await.remove(&temp_id); + app.configs.lock().await.remove(&temp_id); + + match result { + Ok(_) => Ok(Json("Connection successful".to_string())), + Err(e) => Err(AppError(e)), + } +} + +pub async fn connect_db( + State(state): State>, + Json(body): Json, +) -> Result, AppError> { + let config = body.config; + let app = &state.app; + let connection_id = config.id.clone(); + + app.configs + .lock() + .await + .insert(connection_id.clone(), config.clone()); + + let pool_key = app + .get_or_create_pool(&connection_id, config.database.as_deref()) + .await + .map_err(AppError)?; + + Ok(Json(pool_key)) +} + +pub async fn disconnect_db( + State(state): State>, + Json(body): Json, +) -> Result, AppError> { + let app = &state.app; + let mut connections = app.connections.lock().await; + + // Remove all pool keys that start with this connection_id + let keys_to_remove: Vec = connections + .keys() + .filter(|k| k.starts_with(&body.connection_id)) + .cloned() + .collect(); + for key in keys_to_remove { + connections.remove(&key); + } + drop(connections); + + app.configs.lock().await.remove(&body.connection_id); + app.tunnels.stop_tunnel(&body.connection_id).await; + + Ok(Json(())) +} + +pub async fn save_connections( + State(state): State>, + Json(body): Json, +) -> Result, AppError> { + let path = state.data_dir.join("connections.json"); + let secret_path = state.data_dir.join("secrets.json"); + let store = FileSecretStore::new(secret_path); + save_connections_to_file(&path, &body.configs, &store).map_err(AppError)?; + Ok(Json(())) +} + +pub async fn load_connections( + State(state): State>, +) -> Result>, AppError> { + let path = state.data_dir.join("connections.json"); + let secret_path = state.data_dir.join("secrets.json"); + let store = FileSecretStore::new(secret_path); + let configs = load_connections_from_file(&path, &store).map_err(AppError)?; + Ok(Json(configs)) +} diff --git a/src-web/src/routes/history.rs b/src-web/src/routes/history.rs new file mode 100644 index 000000000..bad30c865 --- /dev/null +++ b/src-web/src/routes/history.rs @@ -0,0 +1,56 @@ +use std::sync::Arc; + +use axum::extract::{Path, Query, State}; +use axum::Json; +use dbx_core::history::{self, HistoryEntry}; +use serde::Deserialize; + +use crate::error::AppError; +use crate::state::WebState; + +#[derive(Deserialize)] +pub struct HistoryQuery { + pub limit: Option, + pub offset: Option, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SaveHistoryRequest { + pub entry: HistoryEntry, +} + +pub async fn save_history( + State(state): State>, + Json(body): Json, +) -> Result, AppError> { + let path = state.data_dir.join("query_history.json"); + history::save_history_entry(&path, body.entry).map_err(AppError)?; + Ok(Json(())) +} + +pub async fn load_history( + State(state): State>, + Query(q): Query, +) -> Result>, AppError> { + let path = state.data_dir.join("query_history.json"); + let limit = q.limit.unwrap_or(100); + let offset = q.offset.unwrap_or(0); + let entries = history::load_history_entries(&path, limit, offset).map_err(AppError)?; + Ok(Json(entries)) +} + +pub async fn clear_history(State(state): State>) -> Result, AppError> { + let path = state.data_dir.join("query_history.json"); + history::clear_history_entries(&path).map_err(AppError)?; + Ok(Json(())) +} + +pub async fn delete_history_entry( + State(state): State>, + Path(id): Path, +) -> Result, AppError> { + let path = state.data_dir.join("query_history.json"); + history::delete_history_entry_by_id(&path, &id).map_err(AppError)?; + Ok(Json(())) +} diff --git a/src-web/src/routes/layout.rs b/src-web/src/routes/layout.rs new file mode 100644 index 000000000..00f7aedcd --- /dev/null +++ b/src-web/src/routes/layout.rs @@ -0,0 +1,37 @@ +use std::sync::Arc; + +use axum::extract::State; +use axum::Json; +use serde::Deserialize; + +use crate::error::AppError; +use crate::state::WebState; + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SaveLayoutRequest { + pub layout: serde_json::Value, +} + +pub async fn save_sidebar_layout( + State(state): State>, + Json(body): Json, +) -> Result, AppError> { + let path = state.data_dir.join("sidebar_layout.json"); + let json = serde_json::to_string_pretty(&body.layout).map_err(|e| AppError(e.to_string()))?; + std::fs::write(&path, json).map_err(|e| AppError(e.to_string()))?; + Ok(Json(())) +} + +pub async fn load_sidebar_layout( + State(state): State>, +) -> Result, AppError> { + let path = state.data_dir.join("sidebar_layout.json"); + if !path.exists() { + return Ok(Json(serde_json::json!(null))); + } + let json = std::fs::read_to_string(&path).map_err(|e| AppError(e.to_string()))?; + let layout: serde_json::Value = + serde_json::from_str(&json).map_err(|e| AppError(e.to_string()))?; + Ok(Json(layout)) +} diff --git a/src-web/src/routes/mod.rs b/src-web/src/routes/mod.rs new file mode 100644 index 000000000..835bef78e --- /dev/null +++ b/src-web/src/routes/mod.rs @@ -0,0 +1,12 @@ +pub mod ai; +pub mod connection; +pub mod history; +pub mod layout; +pub mod mongo; +pub mod query; +pub mod redis; +pub mod schema; +pub mod sql_file; +pub mod table_import; +pub mod transfer; +pub mod update; diff --git a/src-web/src/routes/mongo.rs b/src-web/src/routes/mongo.rs new file mode 100644 index 000000000..f9b14eb20 --- /dev/null +++ b/src-web/src/routes/mongo.rs @@ -0,0 +1,149 @@ +use std::sync::Arc; + +use axum::extract::State; +use axum::Json; +use serde::Deserialize; + +use crate::error::AppError; +use crate::state::WebState; + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct MongoConnectionRequest { + pub connection_id: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct MongoCollectionRequest { + pub connection_id: String, + pub database: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct MongoFindRequest { + pub connection_id: String, + pub database: String, + pub collection: String, + pub skip: Option, + pub limit: Option, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct MongoInsertRequest { + pub connection_id: String, + pub database: String, + pub collection: String, + pub doc_json: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct MongoUpdateRequest { + pub connection_id: String, + pub database: String, + pub collection: String, + pub id: String, + pub doc_json: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct MongoDeleteRequest { + pub connection_id: String, + pub database: String, + pub collection: String, + pub id: String, +} + +pub async fn list_databases( + State(state): State>, + Json(req): Json, +) -> Result>, AppError> { + let result = dbx_core::mongo_ops::mongo_list_databases_core(&state.app, &req.connection_id) + .await + .map_err(AppError)?; + Ok(Json(result)) +} + +pub async fn list_collections( + State(state): State>, + Json(req): Json, +) -> Result>, AppError> { + let result = dbx_core::mongo_ops::mongo_list_collections_core( + &state.app, + &req.connection_id, + &req.database, + ) + .await + .map_err(AppError)?; + Ok(Json(result)) +} + +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), + ) + .await + .map_err(AppError)?; + Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?)) +} + +pub async fn insert_document( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + let result = dbx_core::mongo_ops::mongo_insert_document_core( + &state.app, + &req.connection_id, + &req.database, + &req.collection, + &req.doc_json, + ) + .await + .map_err(AppError)?; + Ok(Json(result)) +} + +pub async fn update_document( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + let result = dbx_core::mongo_ops::mongo_update_document_core( + &state.app, + &req.connection_id, + &req.database, + &req.collection, + &req.id, + &req.doc_json, + ) + .await + .map_err(AppError)?; + Ok(Json(result)) +} + +pub async fn delete_document( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + let result = dbx_core::mongo_ops::mongo_delete_document_core( + &state.app, + &req.connection_id, + &req.database, + &req.collection, + &req.id, + ) + .await + .map_err(AppError)?; + Ok(Json(result)) +} diff --git a/src-web/src/routes/query.rs b/src-web/src/routes/query.rs new file mode 100644 index 000000000..a3354272b --- /dev/null +++ b/src-web/src/routes/query.rs @@ -0,0 +1,105 @@ +use std::sync::Arc; + +use axum::extract::State; +use axum::Json; +use serde::Deserialize; + +use crate::error::AppError; +use crate::state::WebState; + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ExecuteQueryRequest { + pub connection_id: String, + pub database: String, + pub sql: String, + pub execution_id: Option, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CancelRequest { + pub execution_id: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ExecuteBatchRequest { + pub connection_id: String, + pub database: String, + pub statements: Vec, +} + +pub async fn execute_query( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + let execution_id = req + .execution_id + .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()); + + let registered = state.app.running_queries.register(execution_id); + let cancel_token = registered.token(); + + let result = dbx_core::query::execute_sql_statement( + &state.app, + &req.connection_id, + &req.database, + &req.sql, + Some(cancel_token), + ) + .await + .map_err(AppError)?; + + drop(registered); + Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?)) +} + +pub async fn execute_multi( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + let execution_id = req + .execution_id + .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()); + + let registered = state.app.running_queries.register(execution_id); + let cancel_token = registered.token(); + + let result = dbx_core::query::execute_multi_core( + &state.app, + &req.connection_id, + &req.database, + &req.sql, + Some(cancel_token), + ) + .await + .map_err(AppError)?; + + drop(registered); + Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?)) +} + +pub async fn execute_batch( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + let result = dbx_core::query::execute_statements( + &state.app, + &req.connection_id, + &req.database, + &req.statements, + ) + .await + .map_err(AppError)?; + + Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?)) +} + +pub async fn cancel_query( + State(state): State>, + Json(req): Json, +) -> Json { + let cancelled = state.app.running_queries.cancel(&req.execution_id); + Json(serde_json::json!({ "cancelled": cancelled })) +} diff --git a/src-web/src/routes/redis.rs b/src-web/src/routes/redis.rs new file mode 100644 index 000000000..6bc944f4b --- /dev/null +++ b/src-web/src/routes/redis.rs @@ -0,0 +1,208 @@ +use std::sync::Arc; + +use axum::extract::State; +use axum::Json; +use serde::Deserialize; + +use crate::error::AppError; +use crate::state::WebState; + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct RedisConnectionRequest { + pub connection_id: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct RedisScanRequest { + pub connection_id: String, + pub db: u32, + pub cursor: u64, + pub pattern: String, + pub count: usize, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct RedisKeyRequest { + pub connection_id: String, + pub key: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct RedisSetStringRequest { + pub connection_id: String, + pub key: String, + pub value: String, + pub ttl: Option, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct RedisHashRequest { + pub connection_id: String, + pub key: String, + pub field: String, + pub value: Option, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct RedisListRequest { + pub connection_id: String, + pub key: String, + pub value: Option, + pub index: Option, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct RedisSetRequest { + pub connection_id: String, + pub key: String, + pub member: String, +} + +pub async fn list_databases( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + let result = dbx_core::redis_ops::redis_list_databases_core(&state.app, &req.connection_id) + .await + .map_err(AppError)?; + Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?)) +} + +pub async fn scan_keys( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + let result = dbx_core::redis_ops::redis_scan_keys_core( + &state.app, + &req.connection_id, + req.db, + req.cursor, + &req.pattern, + req.count, + ) + .await + .map_err(AppError)?; + Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?)) +} + +pub async fn get_value( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + let result = dbx_core::redis_ops::redis_get_value_core(&state.app, &req.connection_id, &req.key) + .await + .map_err(AppError)?; + Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?)) +} + +pub async fn set_string( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + dbx_core::redis_ops::redis_set_string_core( + &state.app, + &req.connection_id, + &req.key, + &req.value, + req.ttl, + ) + .await + .map_err(AppError)?; + Ok(Json(())) +} + +pub async fn delete_key( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + dbx_core::redis_ops::redis_delete_key_core(&state.app, &req.connection_id, &req.key) + .await + .map_err(AppError)?; + Ok(Json(())) +} + +pub async fn hash_set( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + let value = req.value.as_deref().unwrap_or(""); + dbx_core::redis_ops::redis_hash_set_core( + &state.app, + &req.connection_id, + &req.key, + &req.field, + value, + ) + .await + .map_err(AppError)?; + Ok(Json(())) +} + +pub async fn hash_del( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + dbx_core::redis_ops::redis_hash_del_core( + &state.app, + &req.connection_id, + &req.key, + &req.field, + ) + .await + .map_err(AppError)?; + Ok(Json(())) +} + +pub async fn list_push( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + let value = req.value.as_deref().unwrap_or(""); + dbx_core::redis_ops::redis_list_push_core(&state.app, &req.connection_id, &req.key, value) + .await + .map_err(AppError)?; + Ok(Json(())) +} + +pub async fn list_remove( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + let index = req.index.unwrap_or(0); + dbx_core::redis_ops::redis_list_remove_core(&state.app, &req.connection_id, &req.key, index) + .await + .map_err(AppError)?; + Ok(Json(())) +} + +pub async fn set_add( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + dbx_core::redis_ops::redis_set_add_core(&state.app, &req.connection_id, &req.key, &req.member) + .await + .map_err(AppError)?; + Ok(Json(())) +} + +pub async fn set_remove( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + dbx_core::redis_ops::redis_set_remove_core( + &state.app, + &req.connection_id, + &req.key, + &req.member, + ) + .await + .map_err(AppError)?; + Ok(Json(())) +} diff --git a/src-web/src/routes/schema.rs b/src-web/src/routes/schema.rs new file mode 100644 index 000000000..9139c530b --- /dev/null +++ b/src-web/src/routes/schema.rs @@ -0,0 +1,145 @@ +use std::sync::Arc; + +use axum::extract::{Query, State}; +use axum::Json; +use serde::Deserialize; + +use crate::error::AppError; +use crate::state::WebState; + +#[derive(Deserialize)] +pub struct SchemaQuery { + pub connection_id: String, + pub database: Option, + pub schema: Option, + pub table: Option, +} + +pub async fn list_databases( + State(state): State>, + Query(q): Query, +) -> Result, AppError> { + let result = dbx_core::schema::list_databases_core(&state.app, &q.connection_id) + .await + .map_err(AppError)?; + Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?)) +} + +pub async fn list_schemas( + State(state): State>, + Query(q): Query, +) -> Result>, AppError> { + let database = q.database.as_deref().unwrap_or(""); + let result = dbx_core::schema::list_schemas_core(&state.app, &q.connection_id, database) + .await + .map_err(AppError)?; + Ok(Json(result)) +} + +pub async fn list_tables( + State(state): State>, + Query(q): Query, +) -> Result, AppError> { + let database = q.database.as_deref().unwrap_or(""); + let schema = q.schema.as_deref().unwrap_or(""); + let result = + dbx_core::schema::list_tables_core(&state.app, &q.connection_id, database, schema) + .await + .map_err(AppError)?; + Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?)) +} + +pub async fn list_columns( + State(state): State>, + Query(q): Query, +) -> Result, AppError> { + let database = q.database.as_deref().unwrap_or(""); + let schema = q.schema.as_deref().unwrap_or(""); + let table = q.table.as_deref().unwrap_or(""); + let result = dbx_core::schema::get_columns_core( + &state.app, + &q.connection_id, + database, + schema, + table, + ) + .await + .map_err(AppError)?; + Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?)) +} + +pub async fn list_indexes( + State(state): State>, + Query(q): Query, +) -> Result, AppError> { + let database = q.database.as_deref().unwrap_or(""); + let schema = q.schema.as_deref().unwrap_or(""); + let table = q.table.as_deref().unwrap_or(""); + let result = dbx_core::schema::list_indexes_core( + &state.app, + &q.connection_id, + database, + schema, + table, + ) + .await + .map_err(AppError)?; + Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?)) +} + +pub async fn list_foreign_keys( + State(state): State>, + Query(q): Query, +) -> Result, AppError> { + let database = q.database.as_deref().unwrap_or(""); + let schema = q.schema.as_deref().unwrap_or(""); + let table = q.table.as_deref().unwrap_or(""); + let result = dbx_core::schema::list_foreign_keys_core( + &state.app, + &q.connection_id, + database, + schema, + table, + ) + .await + .map_err(AppError)?; + Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?)) +} + +pub async fn list_triggers( + State(state): State>, + Query(q): Query, +) -> Result, AppError> { + let database = q.database.as_deref().unwrap_or(""); + let schema = q.schema.as_deref().unwrap_or(""); + let table = q.table.as_deref().unwrap_or(""); + let result = dbx_core::schema::list_triggers_core( + &state.app, + &q.connection_id, + database, + schema, + table, + ) + .await + .map_err(AppError)?; + Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?)) +} + +pub async fn get_ddl( + State(state): State>, + Query(q): Query, +) -> Result, AppError> { + let database = q.database.as_deref().unwrap_or(""); + let schema = q.schema.as_deref().unwrap_or(""); + let table = q.table.as_deref().unwrap_or(""); + let result = dbx_core::schema::get_table_ddl_core( + &state.app, + &q.connection_id, + database, + schema, + table, + ) + .await + .map_err(AppError)?; + Ok(Json(result)) +} diff --git a/src-web/src/routes/sql_file.rs b/src-web/src/routes/sql_file.rs new file mode 100644 index 000000000..31ec8dbb8 --- /dev/null +++ b/src-web/src/routes/sql_file.rs @@ -0,0 +1,238 @@ +use std::sync::Arc; + +use axum::extract::{Multipart, Path, State}; +use axum::response::sse::{Event, Sse}; +use axum::Json; +use dbx_core::sql; +use dbx_core::query; +use futures::stream::Stream; +use serde::Deserialize; + +use crate::error::AppError; +use crate::state::WebState; + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SqlFileExecuteRequest { + pub execution_id: String, + pub connection_id: String, + pub database: String, + pub file_path: String, + pub continue_on_error: bool, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SqlFileExecuteWrapper { + pub request: SqlFileExecuteRequest, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CancelSqlFileRequest { + pub execution_id: String, +} + +pub async fn preview_sql_file( + State(state): State>, + mut multipart: Multipart, +) -> Result, AppError> { + let tmp_dir = state.data_dir.join("tmp"); + std::fs::create_dir_all(&tmp_dir).map_err(|e| AppError(e.to_string()))?; + + while let Some(field) = multipart + .next_field() + .await + .map_err(|e| AppError(e.to_string()))? + { + let file_name = field + .file_name() + .unwrap_or("upload.sql") + .to_string(); + let data = field.bytes().await.map_err(|e| AppError(e.to_string()))?; + + let file_path = tmp_dir.join(&file_name); + std::fs::write(&file_path, &data).map_err(|e| AppError(e.to_string()))?; + + let size_bytes = data.len() as u64; + let content = String::from_utf8_lossy(&data); + let preview: String = content.chars().take(5000).collect(); + + return Ok(Json(serde_json::json!({ + "fileName": file_name, + "filePath": file_path.to_string_lossy(), + "sizeBytes": size_bytes, + "preview": preview, + }))); + } + + Err(AppError("No file uploaded".to_string())) +} + +pub async fn execute_sql_file( + State(state): State>, + Json(body): Json, +) -> Result, AppError> { + let req = body.request; + let execution_id = req.execution_id.clone(); + + let (tx, _) = tokio::sync::broadcast::channel::(256); + state + .sse_channels + .write() + .await + .insert(execution_id.clone(), tx.clone()); + + let app = state.app.clone(); + let state_clone = state.clone(); + + tokio::spawn(async move { + let file_content = match std::fs::read_to_string(&req.file_path) { + Ok(c) => c, + Err(e) => { + let progress = dbx_core::sql::SqlFileProgress { + execution_id: req.execution_id.clone(), + status: dbx_core::sql::SqlFileStatus::Error, + statement_index: 0, + success_count: 0, + failure_count: 0, + affected_rows: 0, + elapsed_ms: 0, + statement_summary: String::new(), + error: Some(e.to_string()), + }; + if let Ok(json) = serde_json::to_string(&progress) { + let _ = tx.send(json); + } + return; + } + }; + + // Send started + let started = dbx_core::sql::SqlFileProgress { + execution_id: req.execution_id.clone(), + status: dbx_core::sql::SqlFileStatus::Started, + statement_index: 0, + success_count: 0, + failure_count: 0, + affected_rows: 0, + elapsed_ms: 0, + statement_summary: String::new(), + error: None, + }; + if let Ok(json) = serde_json::to_string(&started) { + let _ = tx.send(json); + } + + let statements = sql::split_sql_statements(&file_content); + let start = std::time::Instant::now(); + let mut success_count = 0usize; + let mut failure_count = 0usize; + let mut total_affected: u64 = 0; + + for (i, stmt) in statements.iter().enumerate() { + let summary = sql::statement_summary(stmt); + + // Send running + let running = dbx_core::sql::SqlFileProgress { + execution_id: req.execution_id.clone(), + status: dbx_core::sql::SqlFileStatus::Running, + statement_index: i, + success_count, + failure_count, + affected_rows: total_affected, + elapsed_ms: start.elapsed().as_millis(), + statement_summary: summary.clone(), + error: None, + }; + if let Ok(json) = serde_json::to_string(&running) { + let _ = tx.send(json); + } + + match query::execute_sql_statement(&app, &req.connection_id, &req.database, stmt, None) + .await + { + Ok(result) => { + success_count += 1; + total_affected += result.affected_rows; + let done = dbx_core::sql::SqlFileProgress { + execution_id: req.execution_id.clone(), + status: dbx_core::sql::SqlFileStatus::StatementDone, + statement_index: i, + success_count, + failure_count, + affected_rows: total_affected, + elapsed_ms: start.elapsed().as_millis(), + statement_summary: summary, + error: None, + }; + if let Ok(json) = serde_json::to_string(&done) { + let _ = tx.send(json); + } + } + Err(e) => { + failure_count += 1; + let failed = dbx_core::sql::SqlFileProgress { + execution_id: req.execution_id.clone(), + status: dbx_core::sql::SqlFileStatus::StatementFailed, + statement_index: i, + success_count, + failure_count, + affected_rows: total_affected, + elapsed_ms: start.elapsed().as_millis(), + statement_summary: summary, + error: Some(e), + }; + if let Ok(json) = serde_json::to_string(&failed) { + let _ = tx.send(json); + } + if !req.continue_on_error { + break; + } + } + } + } + + // Send final done + let final_done = dbx_core::sql::SqlFileProgress { + execution_id: req.execution_id.clone(), + status: dbx_core::sql::SqlFileStatus::Done, + statement_index: statements.len(), + success_count, + failure_count, + affected_rows: total_affected, + elapsed_ms: start.elapsed().as_millis(), + statement_summary: String::new(), + error: None, + }; + if let Ok(json) = serde_json::to_string(&final_done) { + let _ = tx.send(json); + } + + state_clone.sse_channels.write().await.remove(&req.execution_id); + }); + + Ok(Json(serde_json::json!({ "executionId": execution_id }))) +} + +pub async fn sql_file_progress( + State(state): State>, + Path(execution_id): Path, +) -> Result>>, AppError> { + let channels = state.sse_channels.read().await; + let tx = channels + .get(&execution_id) + .ok_or_else(|| AppError("Execution not found".to_string()))?; + let rx = tx.subscribe(); + drop(channels); + Ok(crate::sse::sse_from_channel(rx)) +} + +pub async fn cancel_sql_file( + State(state): State>, + Json(req): Json, +) -> Json { + // Remove the channel to stop the execution loop + state.sse_channels.write().await.remove(&req.execution_id); + Json(serde_json::json!({ "cancelled": true })) +} diff --git a/src-web/src/routes/table_import.rs b/src-web/src/routes/table_import.rs new file mode 100644 index 000000000..6f84cf3cf --- /dev/null +++ b/src-web/src/routes/table_import.rs @@ -0,0 +1,176 @@ +use std::sync::Arc; + +use axum::extract::{Multipart, Path, State}; +use axum::response::sse::{Event, Sse}; +use axum::Json; +use dbx_core::table_import::{ + self, TableImportRequest, +}; +use dbx_core::transfer; +use futures::stream::Stream; +use serde::Deserialize; + +use crate::error::AppError; +use crate::state::WebState; + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ExecuteImportWrapper { + pub request: TableImportRequest, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CancelImportRequest { + pub import_id: String, +} + +pub async fn preview_import( + State(state): State>, + mut multipart: Multipart, +) -> Result, AppError> { + let tmp_dir = state.data_dir.join("tmp"); + std::fs::create_dir_all(&tmp_dir).map_err(|e| AppError(e.to_string()))?; + + while let Some(field) = multipart + .next_field() + .await + .map_err(|e| AppError(e.to_string()))? + { + let file_name = field + .file_name() + .unwrap_or("upload.csv") + .to_string(); + let data = field.bytes().await.map_err(|e| AppError(e.to_string()))?; + + let file_path = tmp_dir.join(&file_name); + std::fs::write(&file_path, &data).map_err(|e| AppError(e.to_string()))?; + + let file_path_str = file_path.to_string_lossy().to_string(); + let preview = table_import::preview_table_import_file_core(&file_path_str) + .map_err(AppError)?; + + return Ok(Json( + serde_json::to_value(preview).map_err(|e| AppError(e.to_string()))?, + )); + } + + Err(AppError("No file uploaded".to_string())) +} + +pub async fn execute_import( + State(state): State>, + Json(body): Json, +) -> Result, AppError> { + let req = body.request; + let import_id = req.import_id.clone(); + + let (tx, _) = tokio::sync::broadcast::channel::(256); + state + .sse_channels + .write() + .await + .insert(import_id.clone(), tx.clone()); + + let app = state.app.clone(); + let state_clone = state.clone(); + + tokio::spawn(async move { + let db_type = match transfer::get_db_type(&app, &req.connection_id).await { + Ok(t) => t, + Err(e) => { + let _ = tx.send( + serde_json::json!({ + "importId": req.import_id, + "status": "error", + "error": e + }) + .to_string(), + ); + return; + } + }; + + let pool_key = match app + .get_or_create_pool(&req.connection_id, Some(&req.database)) + .await + { + Ok(k) => k, + Err(e) => { + let _ = tx.send( + serde_json::json!({ + "importId": req.import_id, + "status": "error", + "error": e + }) + .to_string(), + ); + return; + } + }; + + let tx_clone = tx.clone(); + let import_id_for_cancel = req.import_id.clone(); + let result = table_import::import_table_file_core( + &app, + &req, + &db_type, + &pool_key, + |id: &str| { + let id = id.to_string(); + Box::pin(async move { + transfer::is_cancelled(&id).await + }) + }, + |progress| { + if let Ok(json) = serde_json::to_string(&progress) { + let _ = tx_clone.send(json); + } + }, + ) + .await; + + match result { + Ok(summary) => { + if let Ok(json) = serde_json::to_string(&summary) { + let _ = tx.send(json); + } + } + Err(e) => { + let _ = tx.send( + serde_json::json!({ + "importId": import_id_for_cancel, + "status": "error", + "error": e + }) + .to_string(), + ); + } + } + + state_clone.sse_channels.write().await.remove(&req.import_id); + }); + + Ok(Json(serde_json::json!({ "importId": import_id }))) +} + +pub async fn import_progress( + State(state): State>, + Path(import_id): Path, +) -> Result>>, AppError> { + let channels = state.sse_channels.read().await; + let tx = channels + .get(&import_id) + .ok_or_else(|| AppError("Import not found".to_string()))?; + let rx = tx.subscribe(); + drop(channels); + Ok(crate::sse::sse_from_channel(rx)) +} + +pub async fn cancel_import( + State(_state): State>, + Json(req): Json, +) -> Json { + transfer::set_cancelled(&req.import_id).await; + Json(serde_json::json!({ "cancelled": true })) +} diff --git a/src-web/src/routes/transfer.rs b/src-web/src/routes/transfer.rs new file mode 100644 index 000000000..c3e48d083 --- /dev/null +++ b/src-web/src/routes/transfer.rs @@ -0,0 +1,178 @@ +use std::sync::Arc; + +use axum::extract::{Path, State}; +use axum::response::sse::{Event, Sse}; +use axum::Json; +use dbx_core::transfer::{self, TransferRequest, TransferStatus}; +use futures::stream::Stream; +use serde::Deserialize; + +use crate::error::AppError; +use crate::state::WebState; + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct StartTransferRequest { + pub request: TransferRequest, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CancelTransferRequest { + pub transfer_id: String, +} + +pub async fn start_transfer( + State(state): State>, + Json(body): Json, +) -> Result, AppError> { + let req = body.request; + let transfer_id = req.transfer_id.clone(); + + // Create a broadcast channel for progress + let (tx, _) = tokio::sync::broadcast::channel::(256); + state + .sse_channels + .write() + .await + .insert(transfer_id.clone(), tx.clone()); + + let app = state.app.clone(); + let state_clone = state.clone(); + + tokio::spawn(async move { + let source_db_type = match transfer::get_db_type(&app, &req.source_connection_id).await { + Ok(t) => t, + Err(e) => { + let _ = tx.send(serde_json::json!({"error": e}).to_string()); + return; + } + }; + let target_db_type = match transfer::get_db_type(&app, &req.target_connection_id).await { + Ok(t) => t, + Err(e) => { + let _ = tx.send(serde_json::json!({"error": e}).to_string()); + return; + } + }; + + let source_pool_key = match app + .get_or_create_pool(&req.source_connection_id, Some(&req.source_database)) + .await + { + Ok(k) => k, + Err(e) => { + let _ = tx.send(serde_json::json!({"error": e}).to_string()); + return; + } + }; + let target_pool_key = match app + .get_or_create_pool(&req.target_connection_id, Some(&req.target_database)) + .await + { + Ok(k) => k, + Err(e) => { + let _ = tx.send(serde_json::json!({"error": e}).to_string()); + return; + } + }; + + let tables = req.tables.clone(); + for (i, table) in tables.iter().enumerate() { + let tx_clone = tx.clone(); + let result = transfer::transfer_table( + &app, + &req, + table, + i, + &source_db_type, + &target_db_type, + &source_pool_key, + &target_pool_key, + |progress| { + if let Ok(json) = serde_json::to_string(&progress) { + let _ = tx_clone.send(json); + } + }, + ) + .await; + + match result { + Ok(_) => { + let progress = transfer::TransferProgress { + transfer_id: req.transfer_id.clone(), + table: table.clone(), + table_index: i, + total_tables: tables.len(), + rows_transferred: 0, + total_rows: None, + status: TransferStatus::TableDone, + error: None, + }; + if let Ok(json) = serde_json::to_string(&progress) { + let _ = tx.send(json); + } + } + Err(e) => { + let progress = transfer::TransferProgress { + transfer_id: req.transfer_id.clone(), + table: table.clone(), + table_index: i, + total_tables: tables.len(), + rows_transferred: 0, + total_rows: None, + status: TransferStatus::Error, + error: Some(e), + }; + if let Ok(json) = serde_json::to_string(&progress) { + let _ = tx.send(json); + } + break; + } + } + } + + // Send done + let done = transfer::TransferProgress { + transfer_id: req.transfer_id.clone(), + table: String::new(), + table_index: tables.len(), + total_tables: tables.len(), + rows_transferred: 0, + total_rows: None, + status: TransferStatus::Done, + error: None, + }; + if let Ok(json) = serde_json::to_string(&done) { + let _ = tx.send(json); + } + + // Clean up channel + state_clone.sse_channels.write().await.remove(&req.transfer_id); + }); + + Ok(Json( + serde_json::json!({ "transferId": transfer_id }), + )) +} + +pub async fn transfer_progress( + State(state): State>, + Path(transfer_id): Path, +) -> Result>>, AppError> { + let channels = state.sse_channels.read().await; + let tx = channels + .get(&transfer_id) + .ok_or_else(|| AppError("Transfer not found".to_string()))?; + let rx = tx.subscribe(); + drop(channels); + Ok(crate::sse::sse_from_channel(rx)) +} + +pub async fn cancel_transfer( + State(_state): State>, + Json(req): Json, +) -> Json { + transfer::set_cancelled(&req.transfer_id).await; + Json(serde_json::json!({ "cancelled": true })) +} diff --git a/src-web/src/routes/update.rs b/src-web/src/routes/update.rs new file mode 100644 index 000000000..7c0a2b8e3 --- /dev/null +++ b/src-web/src/routes/update.rs @@ -0,0 +1,10 @@ +use axum::Json; +use dbx_core::update; + +use crate::error::AppError; + +pub async fn check_for_updates() -> Result, AppError> { + let release = update::fetch_latest_release().await.map_err(AppError)?; + let info = update::build_update_info(release, env!("CARGO_PKG_VERSION")); + Ok(Json(serde_json::to_value(info).map_err(|e| AppError(e.to_string()))?)) +} diff --git a/src-web/src/sse.rs b/src-web/src/sse.rs new file mode 100644 index 000000000..ac46d5eca --- /dev/null +++ b/src-web/src/sse.rs @@ -0,0 +1,14 @@ +use axum::response::sse::{Event, KeepAlive, Sse}; +use futures::stream::Stream; +use tokio::sync::broadcast; + +pub fn sse_from_channel( + mut rx: broadcast::Receiver, +) -> Sse>> { + let stream = async_stream::stream! { + while let Ok(data) = rx.recv().await { + yield Ok(Event::default().data(data)); + } + }; + Sse::new(stream).keep_alive(KeepAlive::default()) +} diff --git a/src-web/src/state.rs b/src-web/src/state.rs new file mode 100644 index 000000000..94dbb30f9 --- /dev/null +++ b/src-web/src/state.rs @@ -0,0 +1,13 @@ +use dbx_core::connection::AppState; +use std::collections::{HashMap, HashSet}; +use std::path::PathBuf; +use std::sync::Arc; +use tokio::sync::{broadcast, RwLock}; + +pub struct WebState { + pub app: Arc, + pub data_dir: PathBuf, + pub password_hash: Option, + pub sessions: RwLock>, + pub sse_channels: RwLock>>, +} diff --git a/src/App.vue b/src/App.vue index 493c66bb8..26a002555 100644 --- a/src/App.vue +++ b/src/App.vue @@ -51,10 +51,8 @@ import { useHistoryStore } from "@/stores/historyStore"; import { useSettingsStore } from "@/stores/settingsStore"; import { useToast } from "@/composables/useToast"; import { setLocale, currentLocale, type Locale } from "@/i18n"; -import { getCurrentWindow, type Theme } from "@tauri-apps/api/window"; -import { getCurrentWebview } from "@tauri-apps/api/webview"; -import { getVersion } from "@tauri-apps/api/app"; -import * as api from "@/lib/tauri"; +import type { Theme } from "@tauri-apps/api/window"; +import * as api from "@/lib/api"; import { canCancelQueryExecution, queryExecutionLabelKey } from "@/lib/queryExecutionState"; import { connectionDriverLabel, connectionIconType, connectionOptionSubtitle } from "@/lib/connectionPresentation"; import { resolveExecutableSql } from "@/lib/sqlExecutionTarget"; @@ -62,6 +60,7 @@ import { buildTableSelectSql, quoteTableIdentifier } from "@/lib/tableSelectSql" import { isTauriRuntime } from "@/lib/tauriRuntime"; import type { SqlFormatDialect } from "@/lib/sqlFormatter"; import { isCloseTabShortcut, isExecuteSqlShortcut } from "@/lib/keyboardShortcuts"; +import LoginPage from "@/components/auth/LoginPage.vue"; const { t } = useI18n(); const connectionStore = useConnectionStore(); @@ -70,6 +69,10 @@ const historyStore = useHistoryStore(); const settingsStore = useSettingsStore(); const { message: toastMessage, visible: toastVisible, toast } = useToast(); +const isDesktop = isTauriRuntime(); +const needsAuth = ref(!isDesktop); +const authenticated = ref(isDesktop); + const showConnectionDialog = ref(false); const showSettingsDialog = ref(false); const showHistory = ref(false); @@ -869,9 +872,11 @@ const isDark = ref(localStorage.getItem("dbx-theme") === "dark"); function applyTheme() { document.documentElement.classList.toggle("dark", isDark.value); if (!isTauriRuntime()) return; - getCurrentWindow() - .setTheme(isDark.value ? "dark" as Theme : "light" as Theme) - .catch(() => {}); + import("@tauri-apps/api/window").then(({ getCurrentWindow }) => { + getCurrentWindow() + .setTheme(isDark.value ? "dark" as Theme : "light" as Theme) + .catch(() => {}); + }); } function toggleTheme() { @@ -880,14 +885,20 @@ function toggleTheme() { applyTheme(); } -import { open } from "@tauri-apps/plugin-shell"; +function openUrl(url: string) { + if (isTauriRuntime()) { + import("@tauri-apps/plugin-shell").then(({ open }) => open(url)); + } else { + window.open(url, "_blank"); + } +} function openGitHub() { - open("https://github.com/t8y2/dbx"); + openUrl("https://github.com/t8y2/dbx"); } function openMcpGuide() { - open("https://github.com/t8y2/dbx/blob/main/docs/mcp-guide.md"); + openUrl("https://github.com/t8y2/dbx/blob/main/docs/mcp-guide.md"); } async function checkUpdates(options: { silent?: boolean } = {}) { @@ -923,7 +934,7 @@ function formatUpdateError(message: string): string { function openLatestRelease() { const url = updateInfo.value?.release_url || latestReleaseUrl; - open(url); + openUrl(url); } const isDownloadingUpdate = ref(false); @@ -987,19 +998,44 @@ function handleKeydown(e: KeyboardEvent) { } } -onMounted(() => { - applyTheme(); +function initApp() { connectionStore.initFromDisk().catch((e: any) => { toast(t("connection.loadFailed", { message: e?.message || String(e) }), 5000); }); settingsStore.initAiConfig(); +} + +function onAuthenticated() { + authenticated.value = true; + initApp(); +} + +onMounted(async () => { + applyTheme(); window.addEventListener("keydown", handleKeydown, true); window.addEventListener("resize", updateScrollButtons); - if (isTauriRuntime()) { - setupFileDrop().catch(() => {}); - checkUpdates({ silent: true }); + + if (!isDesktop) { + try { + const res = await fetch("/api/auth/check"); + const data = await res.json(); + needsAuth.value = data.required; + authenticated.value = data.authenticated; + } catch { /* server unreachable */ } + if (!needsAuth.value || authenticated.value) { + initApp(); + } + api.checkForUpdates().then((info) => { appVersion.value = info.current_version; }).catch(() => {}); + return; + } + + initApp(); + setupFileDrop().catch(() => {}); + checkUpdates({ silent: true }); + import("@tauri-apps/api/app").then(({ getVersion }) => { getVersion().then((v) => { appVersion.value = v; }).catch(() => {}); - import("@tauri-apps/api/event").then(({ listen }) => { + }).catch(() => {}); + import("@tauri-apps/api/event").then(({ listen }) => { listen<{ connection_id: string; database: string; schema?: string; table: string }>("mcp-open-table", async (event) => { const { connection_id, database, schema, table } = event.payload; @@ -1019,7 +1055,7 @@ onMounted(() => { openLineageTarget({ connectionId: connection_id, database, schema, tableName: table }); } - getCurrentWindow().setFocus().catch(() => {}); + import("@tauri-apps/api/window").then(({ getCurrentWindow }) => getCurrentWindow().setFocus().catch(() => {})); }); listen<{ connection_id: string; database: string; sql: string }>("mcp-execute-query", async (event) => { const { connection_id, database, sql } = event.payload; @@ -1033,10 +1069,9 @@ onMounted(() => { const tabId = queryStore.createTab(connection_id, database, undefined, "query"); queryStore.updateSql(tabId, sql); await queryStore.executeTabSql(tabId, sql); - getCurrentWindow().setFocus().catch(() => {}); + import("@tauri-apps/api/window").then(({ getCurrentWindow }) => getCurrentWindow().setFocus().catch(() => {})); }); }).catch(() => {}); - } }); onUnmounted(() => { @@ -1064,6 +1099,7 @@ function getDataFileQuery(path: string): string | null { } async function setupFileDrop() { + const { getCurrentWebview } = await import("@tauri-apps/api/webview"); const webview = getCurrentWebview(); await webview.onDragDropEvent(async (event) => { if (event.payload.type !== "drop") return; @@ -1120,7 +1156,8 @@ async function setupFileDrop() {