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() {