feat(mssql): restore dacpac and bacpac over an existing database

src/domain/mssql/connection.rs

@@ -4,10 +4,13 @@ use tiberius::{AuthMethod, Client, Config};
 use tokio::net::TcpStream;
 use tokio_util::compat::{Compat, TokioAsyncWriteCompatExt};
 
-pub async fn build_client(cfg: &DatabaseConfig) -> Result<Client<Compat<TcpStream>>> {
+pub async fn build_client(cfg: &DatabaseConfig, database: Option<&str>) -> Result<Client<Compat<TcpStream>>> {
     let mut config = Config::new();
     config.host(&cfg.host);
     config.port(cfg.port);
+    if let Some(database) = database {
+        config.database(database);
+    }
     config.authentication(AuthMethod::sql_server(&cfg.username, &cfg.password));
     config.trust_cert();
 

src/domain/mssql/ping.rs

@@ -5,7 +5,7 @@ use tracing::{error, info};
 pub async fn run(cfg: DatabaseConfig) -> Result<bool> {
     info!("Running ping for MSSQL database {}", cfg.name);
 
-    match super::connection::build_client(&cfg).await {
+    match super::connection::build_client(&cfg, None).await {
         Ok(mut client) => match client.simple_query("SELECT 1").await {
             Ok(_) => {
                 info!("MSSQL ping succeeded for {}", cfg.name);

src/domain/mssql/restore.rs

@@ -1,3 +1,4 @@
+use super::format::MssqlFormat;
 use crate::services::backup::logger::JobLogger;
 use crate::services::config::DatabaseConfig;
 use anyhow::{Context, Result};
@@ -6,10 +7,29 @@ use std::process::Command;
 use std::sync::Arc;
 use std::time::Instant;
 
-pub async fn run(cfg: DatabaseConfig, restore_file: PathBuf, logger: Arc<JobLogger>) -> Result<()> {
-    tokio::task::spawn_blocking(move || -> Result<()> {
-        logger.log("debug", format!("Starting MSSQL restore for database {}", cfg.name));
+const DROP_DATABASE_SQL: &str = "DECLARE @q nvarchar(258) = QUOTENAME(@P1);
+IF DB_ID(@P1) IS NOT NULL
+BEGIN
+    IF CAST(SERVERPROPERTY('EngineEdition') AS int) <> 5
+        EXEC(N'ALTER DATABASE ' + @q + N' SET SINGLE_USER WITH ROLLBACK IMMEDIATE');
+    EXEC(N'DROP DATABASE ' + @q);
+END";
 
+pub async fn run(cfg: DatabaseConfig, restore_file: PathBuf, logger: Arc<JobLogger>) -> Result<()> {
+    logger.log("debug", format!("Starting MSSQL restore for database {}", cfg.name));
+
+    let Some(format) = MssqlFormat::from_path(&restore_file) else {
+        let message = format!(
+            "Unsupported MSSQL backup file {}: expected .bacpac or .dacpac",
+            restore_file.display()
+        );
+        logger.log("error", message.clone());
+        anyhow::bail!(message);
+    };
+
+    drop_database(&cfg, &logger).await?;
+
+    tokio::task::spawn_blocking(move || -> Result<()> {
         let connection_string = format!(
             "Server=tcp:{},{};Database={};User Id={};Password={};TrustServerCertificate=True;Encrypt=False",
             cfg.host, cfg.port, cfg.database, cfg.username, cfg.password
@@ -24,8 +44,12 @@ pub async fn run(cfg: DatabaseConfig, restore_file: PathBuf, logger: Arc<JobLogg
         ));
 
         let start = Instant::now();
-        let output = Command::new("sqlpackage")
-            .arg("/a:Import")
+        let mut command = Command::new("sqlpackage");
+        match format {
+            MssqlFormat::Dacpac => command.args(["/a:Publish", "/p:AllowIncompatiblePlatform=True"]),
+            MssqlFormat::Bacpac => command.arg("/a:Import"),
+        };
+        let output = command
             .arg(format!("/tcs:{}", connection_string))
             .arg(format!("/sf:{}", restore_file.display()))
             .output()
@@ -52,3 +76,20 @@ pub async fn run(cfg: DatabaseConfig, restore_file: PathBuf, logger: Arc<JobLogg
     })
     .await?
 }
+
+async fn drop_database(cfg: &DatabaseConfig, logger: &JobLogger) -> Result<()> {
+    logger.log("info", format!("Dropping MSSQL database {} before restore", cfg.database));
+
+    let start = Instant::now();
+    let mut client = super::connection::build_client(cfg, Some("master"))
+        .await
+        .with_context(|| format!("Failed to connect to master for {}", cfg.name))?;
+    client
+        .execute(DROP_DATABASE_SQL, &[&cfg.database.as_str()])
+        .await
+        .with_context(|| format!("Failed to drop MSSQL database {}", cfg.database))?;
+
+    let duration_ms = start.elapsed().as_millis() as f64;
+    logger.log_command(format!("DROP DATABASE [{}]", cfg.database), None, Some(0), Some(duration_ms));
+    Ok(())
+}

src/tests/domain/mssql.rs

@@ -245,6 +245,93 @@ async fn mssql_backup_windows_principal_test() {
     assert_eq!(file_path.extension().and_then(|e| e.to_str()), Some("dacpac"));
 }
 
+async fn restore_over_existing(format: &str, generated_id: &str) {
+    init_tracing_for_test();
+    let (_container, host, port) = start().await;
+    create_user_database(&host, port, "appdb").await;
+    exec(&host, port, "master", "CREATE LOGIN app_login WITH PASSWORD = 'App!Str0ng1'").await;
+    exec(
+        &host,
+        port,
+        "appdb",
+        "CREATE TABLE dbo.users (id INT PRIMARY KEY, email NVARCHAR(100) NOT NULL); \
+         INSERT dbo.users VALUES (1, 'a@x'), (2, 'b@x'), (3, 'c@x'); \
+         CREATE USER app_user FOR LOGIN app_login; \
+         GRANT SELECT ON dbo.users TO app_user;",
+    )
+    .await;
+
+    let config = with_backup_format(make_config(host.clone(), port, "appdb", generated_id), format);
+    let temp_dir = TempDir::new().unwrap();
+    let file_path = DatabaseFactory::create_for_backup(config.clone())
+        .await
+        .backup(temp_dir.path(), Arc::new(JobLogger::new()))
+        .await
+        .unwrap();
+    assert_eq!(file_path.extension().and_then(|e| e.to_str()), Some(format));
+
+    exec(
+        &host,
+        port,
+        "appdb",
+        "DELETE FROM dbo.users WHERE id = 1; INSERT dbo.users VALUES (4, 'd@x'); CREATE TABLE dbo.extra (id INT);",
+    )
+    .await;
+
+    DatabaseFactory::create_for_restore(config, &file_path)
+        .await
+        .restore(&file_path, Arc::new(JobLogger::new()))
+        .await
+        .unwrap();
+
+    assert_eq!(scalar(&host, port, "appdb", "sa", SA_PASSWORD, "SELECT COUNT(*) FROM dbo.users").await, 3);
+    assert_eq!(
+        scalar(&host, port, "appdb", "sa", SA_PASSWORD, "SELECT COUNT(*) FROM dbo.users WHERE id IN (1, 2, 3)").await,
+        3
+    );
+    assert_eq!(
+        scalar(&host, port, "appdb", "sa", SA_PASSWORD, "SELECT CASE WHEN OBJECT_ID('dbo.extra') IS NULL THEN 1 ELSE 0 END").await,
+        1
+    );
+    assert_eq!(
+        scalar(&host, port, "appdb", "app_login", "App!Str0ng1", "SELECT COUNT(*) FROM dbo.users").await,
+        3
+    );
+}
+
+#[tokio::test]
+async fn mssql_restore_over_existing_dacpac_test() {
+    restore_over_existing("dacpac", "5a445eb4-c2c6-4bde-a423-ee1385dcf6d8").await;
+}
+
+#[tokio::test]
+async fn mssql_restore_over_existing_bacpac_test() {
+    restore_over_existing("bacpac", "5a445eb4-c2c6-4bde-a423-ee1385dcf6d9").await;
+}
+
+#[tokio::test]
+async fn mssql_restore_rejects_unknown_extension_test() {
+    init_tracing_for_test();
+    let (_container, host, port) = start().await;
+    create_user_database(&host, port, "keepdb").await;
+
+    let temp_dir = TempDir::new().unwrap();
+    let file_path = temp_dir.path().join("backup.dump");
+    std::fs::write(&file_path, b"not a package").unwrap();
+
+    let config = make_config(host.clone(), port, "keepdb", "5a445eb4-c2c6-4bde-a423-ee1385dcf6da");
+    let result = DatabaseFactory::create_for_restore(config, &file_path)
+        .await
+        .restore(&file_path, Arc::new(JobLogger::new()))
+        .await;
+
+    assert!(result.unwrap_err().to_string().contains("expected .bacpac or .dacpac"));
+    assert_eq!(
+        scalar(&host, port, "master", "sa", SA_PASSWORD, "SELECT CASE WHEN DB_ID('keepdb') IS NULL THEN 0 ELSE 1 END").await,
+        1
+    );
+}
+
 mod format_tests {
     use super::{make_config, with_backup_format};
     use crate::domain::mssql::format::MssqlFormat;