feat(export): support database setup and batch SQL scripts

crates/dbx-web/src/main.rs

@@ -586,7 +586,11 @@ async fn main() {
         )
         .route("/export/query-result/cancel", post(routes::query_result_export::cancel_query_result_export))
         // SQL file
-        .route("/sql-file/preview", post(routes::sql_file::preview_sql_file))
+        .route(
+            "/sql-file/preview",
+            post(routes::sql_file::preview_sql_file)
+                .layer(DefaultBodyLimit::max(routes::sql_file::SQL_FILE_UPLOAD_MAX_BYTES.saturating_add(1024 * 1024))),
+        )
         .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))

crates/dbx-web/src/routes/sql_file.rs

@@ -1,5 +1,6 @@
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
+use std::time::Duration;
 
 use axum::extract::{Multipart, Path as AxumPath, State};
 use axum::response::sse::{Event, Sse};
@@ -7,21 +8,28 @@ use axum::Json;
 use dbx_core::sql;
 use dbx_core::sql::{SqlFileProgress, SqlFileRequest, SqlFileStatus};
 use dbx_core::sql_file_import::{
-    execute_sql_file_content, sql_file_error_progress, sql_file_progress as build_sql_file_progress,
+    execute_sql_file_paths, sql_file_error_progress, sql_file_progress as build_sql_file_progress,
     SqlFileProgressEmitter,
 };
 use futures::stream::Stream;
 use serde::Deserialize;
 use tokio::sync::broadcast;
 use tokio_util::sync::CancellationToken;
+use uuid::Uuid;
 
 use crate::error::AppError;
 use crate::state::WebState;
 
+const PENDING_SQL_FILE_PROGRESS_CHANNEL_TTL: Duration = Duration::from_secs(30);
+const SQL_FILE_UPLOAD_MAX_AGE: Duration = Duration::from_secs(24 * 60 * 60);
+pub const SQL_FILE_UPLOAD_MAX_BYTES: usize = 200 * 1024 * 1024;
+
 #[derive(Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct SqlFileExecuteWrapper {
     pub request: SqlFileRequest,
+    #[serde(default)]
+    pub file_paths: Vec<String>,
 }
 
 #[derive(Deserialize)]
@@ -34,12 +42,20 @@ pub async fn preview_sql_file(
     State(state): State<Arc<WebState>>,
     mut multipart: Multipart,
 ) -> Result<Json<serde_json::Value>, AppError> {
-    let tmp_dir = state.data_dir.join("tmp");
+    let tmp_dir = state.data_dir.join("tmp").join("sql_file");
     std::fs::create_dir_all(&tmp_dir).map_err(|e| AppError::from(e.to_string()))?;
+    cleanup_expired_sql_file_uploads(&tmp_dir);
 
     if let Some(field) = multipart.next_field().await.map_err(|e| AppError::from(e.to_string()))? {
         let file_name = field.file_name().unwrap_or("upload.sql").to_string();
         let data = field.bytes().await.map_err(|e| AppError::from(e.to_string()))?;
+        if data.len() > SQL_FILE_UPLOAD_MAX_BYTES {
+            return Err(AppError::from(format!(
+                "File too large: {} bytes (max {} bytes)",
+                data.len(),
+                SQL_FILE_UPLOAD_MAX_BYTES
+            )));
+        }
 
         let file_path = safe_uploaded_sql_path(&tmp_dir, &file_name)?;
         std::fs::write(&file_path, &data).map_err(|e| AppError::from(e.to_string()))?;
@@ -47,13 +63,15 @@ pub async fn preview_sql_file(
         let size_bytes = data.len() as u64;
         let content = sql::decode_sql_file_bytes(&data).map_err(AppError::from)?;
         let preview: String = content.chars().take(20_000).collect();
+        let bootstrap_analysis = dbx_core::sql_file_import::mysql_like_sql_file_bootstrap_analysis(&content);
 
         return Ok(Json(serde_json::json!({
             "fileName": file_name,
             "filePath": file_path.to_string_lossy(),
             "sizeBytes": size_bytes,
             "preview": preview,
-            "canExecuteWithoutSelectedDatabase": dbx_core::sql_file_import::mysql_like_sql_file_can_execute_without_selected_database(&content),
+            "canExecuteWithoutSelectedDatabase": bootstrap_analysis.can_execute_without_selected_database,
+            "establishesDatabaseContext": bootstrap_analysis.establishes_database_context,
         })));
     }
 
@@ -75,7 +93,14 @@ pub async fn execute_sql_file(
     }
 
     let execution_id = req.execution_id.clone();
-    let file_path = validated_uploaded_sql_path(&state.data_dir, &req.file_path)?;
+    let requested_paths = if body.file_paths.is_empty() { vec![req.file_path.clone()] } else { body.file_paths };
+    if requested_paths.is_empty() {
+        return Err(AppError::from("No SQL files selected".to_string()));
+    }
+    let file_paths = requested_paths
+        .iter()
+        .map(|file_path| validated_uploaded_sql_path(&state.data_dir, file_path))
+        .collect::<Result<Vec<_>, _>>()?;
     let token = CancellationToken::new();
 
     {
@@ -85,8 +110,10 @@ pub async fn execute_sql_file(
         }
         executions.insert(execution_id.clone(), token.clone());
     }
-    let (tx, _) = tokio::sync::broadcast::channel::<String>(256);
-    state.sse_channels.write().await.insert(execution_id.clone(), tx.clone());
+    let tx = {
+        let mut channels = state.sse_channels.write().await;
+        channels.entry(execution_id.clone()).or_insert_with(|| tokio::sync::broadcast::channel::<String>(256).0).clone()
+    };
 
     let app = state.app.clone();
     let state_clone = state.clone();
@@ -107,37 +134,30 @@ pub async fn execute_sql_file(
             "",
             None,
         ));
-        match std::fs::metadata(&file_path) {
-            Ok(meta) if meta.len() > 200 * 1024 * 1024 => {
-                progress_emitter.emit(sql_file_error_progress(
-                    &req.execution_id,
-                    started_at,
-                    format!("File too large: {} bytes (max {} bytes)", meta.len(), 200 * 1024 * 1024),
-                ));
-                cleanup_sql_file_execution(&state_clone, &req.execution_id).await;
-                return;
+        for file_path in &file_paths {
+            match std::fs::metadata(file_path) {
+                Ok(meta) if meta.len() > SQL_FILE_UPLOAD_MAX_BYTES as u64 => {
+                    progress_emitter.emit(sql_file_error_progress(
+                        &req.execution_id,
+                        started_at,
+                        format!("File too large: {} bytes (max {} bytes)", meta.len(), SQL_FILE_UPLOAD_MAX_BYTES),
+                    ));
+                    cleanup_sql_file_execution(&state_clone, &req.execution_id).await;
+                    return;
+                }
+                Err(e) => {
+                    progress_emitter.emit(sql_file_error_progress(&req.execution_id, started_at, e.to_string()));
+                    cleanup_sql_file_execution(&state_clone, &req.execution_id).await;
+                    return;
+                }
+                _ => {}
             }
-            Err(e) => {
-                progress_emitter.emit(sql_file_error_progress(&req.execution_id, started_at, e.to_string()));
-                cleanup_sql_file_execution(&state_clone, &req.execution_id).await;
-                return;
-            }
-            _ => {}
         }
 
-        let file_content = match std::fs::read(&file_path).and_then(|bytes| {
-            sql::decode_sql_file_bytes(&bytes)
-                .map_err(|message| std::io::Error::new(std::io::ErrorKind::InvalidData, message))
-        }) {
-            Ok(content) => content,
-            Err(e) => {
-                progress_emitter.emit(sql_file_error_progress(&req.execution_id, started_at, e.to_string()));
-                cleanup_sql_file_execution(&state_clone, &req.execution_id).await;
-                return;
-            }
-        };
-
-        let _ = execute_sql_file_content(&app, &req, &file_content, token, started_at, |progress| {
+        let file_path_refs: Vec<&Path> = file_paths.iter().map(PathBuf::as_path).collect();
+        // The core executor emits exactly one terminal Error with the latest
+        // cumulative counters before returning Err.
+        let _ = execute_sql_file_paths(&app, &req, &file_path_refs, token, started_at, |progress| {
             progress_emitter.emit(progress);
         })
         .await;
@@ -164,7 +184,38 @@ fn safe_uploaded_sql_path(tmp_dir: &Path, file_name: &str) -> Result<PathBuf, Ap
     if base_name.is_empty() || base_name == "." || base_name == ".." {
         return Err(AppError::from("Invalid SQL file name".to_string()));
     }
-    Ok(tmp_dir.join(base_name))
+    let file_name = Path::new(base_name);
+    let stem = file_name.file_stem().and_then(|stem| stem.to_str()).filter(|stem| !stem.is_empty()).unwrap_or("upload");
+    let extension =
+        file_name.extension().and_then(|extension| extension.to_str()).filter(|extension| !extension.is_empty());
+    let unique_name = match extension {
+        Some(extension) => format!("{stem}-{}.{}", Uuid::new_v4(), extension),
+        None => format!("{stem}-{}", Uuid::new_v4()),
+    };
+    Ok(tmp_dir.join(unique_name))
+}
+
+fn cleanup_expired_sql_file_uploads(tmp_dir: &Path) {
+    cleanup_sql_file_uploads_older_than(tmp_dir, SQL_FILE_UPLOAD_MAX_AGE);
+}
+
+fn cleanup_sql_file_uploads_older_than(tmp_dir: &Path, max_age: Duration) {
+    let Ok(entries) = std::fs::read_dir(tmp_dir) else {
+        return;
+    };
+    for entry in entries.flatten() {
+        let path = entry.path();
+        let expired = entry
+            .metadata()
+            .ok()
+            .filter(|metadata| metadata.is_file())
+            .and_then(|metadata| metadata.modified().ok())
+            .and_then(|modified| modified.elapsed().ok())
+            .is_some_and(|age| age >= max_age);
+        if expired {
+            let _ = std::fs::remove_file(path);
+        }
+    }
 }
 
 fn validated_uploaded_sql_path(data_dir: &Path, file_path: &str) -> Result<PathBuf, AppError> {
@@ -185,10 +236,28 @@ pub async fn sql_file_progress(
     State(state): State<Arc<WebState>>,
     AxumPath(execution_id): AxumPath<String>,
 ) -> Result<Sse<impl Stream<Item = Result<Event, std::convert::Infallible>>>, AppError> {
-    let channels = state.sse_channels.read().await;
-    let tx = channels.get(&execution_id).ok_or_else(|| AppError::from("Execution not found".to_string()))?;
+    // The client subscribes before it sends the execution request. Creating the
+    // channel on demand prevents that SSE connection from racing the POST and
+    // guarantees it is subscribed before the background task emits `Started`.
+    let (tx, created) = {
+        let mut channels = state.sse_channels.write().await;
+        let created = !channels.contains_key(&execution_id);
+        let tx = channels
+            .entry(execution_id.clone())
+            .or_insert_with(|| tokio::sync::broadcast::channel::<String>(256).0)
+            .clone();
+        (tx, created)
+    };
+    if created {
+        let state = state.clone();
+        tokio::spawn(async move {
+            tokio::time::sleep(PENDING_SQL_FILE_PROGRESS_CHANNEL_TTL).await;
+            if !state.sql_file_executions.read().await.contains_key(&execution_id) {
+                state.remove_sse_channel(&execution_id).await;
+            }
+        });
+    }
     let rx = tx.subscribe();
-    drop(channels);
     Ok(crate::sse::sse_from_lossy_channel(rx))
 }
 
@@ -207,19 +276,28 @@ pub async fn cancel_sql_file(
 
 #[cfg(test)]
 mod tests {
-    use super::{safe_uploaded_sql_path, validated_uploaded_sql_path};
+    use std::time::Duration;
+
+    use super::{cleanup_sql_file_uploads_older_than, safe_uploaded_sql_path, validated_uploaded_sql_path};
 
     #[test]
-    fn uploaded_sql_path_uses_only_the_file_name() {
+    fn uploaded_sql_paths_are_unique_and_keep_the_extension() {
         let data_dir = std::env::temp_dir().join(format!("dbx-web-sql-file-test-{}", uuid::Uuid::new_v4()));
         let tmp_dir = data_dir.join("tmp");
 
-        let path = match safe_uploaded_sql_path(&tmp_dir, "../outside.sql") {
+        let first = match safe_uploaded_sql_path(&tmp_dir, "../outside.sql") {
+            Ok(path) => path,
+            Err(error) => panic!("{}", error.message),
+        };
+        let second = match safe_uploaded_sql_path(&tmp_dir, "nested/outside.sql") {
             Ok(path) => path,
             Err(error) => panic!("{}", error.message),
         };
 
-        assert_eq!(path, tmp_dir.join("outside.sql"));
+        assert!(first.starts_with(&tmp_dir));
+        assert!(second.starts_with(&tmp_dir));
+        assert_ne!(first, second);
+        assert_eq!(first.extension().and_then(|extension| extension.to_str()), Some("sql"));
         let _ = std::fs::remove_dir_all(data_dir);
     }
 
@@ -236,4 +314,17 @@ mod tests {
         assert!(result.is_err());
         let _ = std::fs::remove_dir_all(data_dir);
     }
+
+    #[test]
+    fn expired_sql_file_uploads_are_removed() {
+        let tmp_dir = std::env::temp_dir().join(format!("dbx-web-sql-file-test-{}", uuid::Uuid::new_v4()));
+        std::fs::create_dir_all(&tmp_dir).unwrap();
+        let upload = tmp_dir.join("upload.sql");
+        std::fs::write(&upload, "select 1;").unwrap();
+
+        cleanup_sql_file_uploads_older_than(&tmp_dir, Duration::ZERO);
+
+        assert!(!upload.exists());
+        let _ = std::fs::remove_dir_all(tmp_dir);
+    }
 }