fix: improve flow version fix migration (#4050)

backend/.sqlx/query-a6af0760c20eb9a7db648ea5d643b4c5703b776dc6cb1c5315e26569a2daa240.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_flow_versioning_2')",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "exists",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "a6af0760c20eb9a7db648ea5d643b4c5703b776dc6cb1c5315e26569a2daa240"
+}

backend/.sqlx/query-c4d032639dbfb47c84a2fbcb4cdca75af3007499a9f6c8c492a32fac5274aeb2.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO windmill_migrations (name) VALUES ('fix_flow_versioning_2')",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": []
+  },
+  "hash": "c4d032639dbfb47c84a2fbcb4cdca75af3007499a9f6c8c492a32fac5274aeb2"
+}

backend/windmill-api/src/db.rs

@@ -166,6 +166,11 @@ impl Migrate for CustomMigrator {
 pub async fn migrate(db: &DB) -> Result<(), Error> {
     let migrator = db.acquire().await?;
     let mut custom_migrator = CustomMigrator { inner: migrator };
+
+    if let Err(err) = fix_flow_versioning_migration(&mut custom_migrator, db).await {
+        tracing::error!("Could not apply flow versioning fix migration: {err:#}");
+    }
+
     match sqlx::migrate!("../migrations")
         .run_direct(&mut custom_migrator)
         .await
@@ -181,6 +186,7 @@ pub async fn migrate(db: &DB) -> Result<(), Error> {
         Err(err) => Err(err),
     }?;
 
+    #[cfg(feature = "enterprise")]
     if let Err(e) = windmill_migrations(&mut custom_migrator, db).await {
         tracing::error!("Could not apply windmill custom migrations: {e:#}")
     }
@@ -188,8 +194,45 @@ pub async fn migrate(db: &DB) -> Result<(), Error> {
     Ok(())
 }
 
+async fn fix_flow_versioning_migration(
+    migrator: &mut CustomMigrator,
+    db: &DB,
+) -> Result<(), Error> {
+    migrator.lock().await?;
+
+    if migrator
+        .list_applied_migrations()
+        .await?
+        .iter()
+        .any(|x| x.version == 20240630102146)
+    {
+        let has_done_migration = sqlx::query_scalar!(
+                "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_flow_versioning_2')",
+            )
+            .fetch_one(db)
+            .await?
+            .unwrap_or(false);
+
+        if !has_done_migration {
+            let query = include_str!("../../custom_migrations/fix_flow_versioning_2.sql");
+            tracing::info!("Applying fix_flow_versioning_2.sql");
+            let mut tx: sqlx::Transaction<'_, Postgres> = db.begin().await?;
+            tx.execute(query).await?;
+            tracing::info!("Applied fix_flow_versioning_2.sql");
+            sqlx::query!("INSERT INTO windmill_migrations (name) VALUES ('fix_flow_versioning_2')")
+                .execute(&mut *tx)
+                .await?;
+            tx.commit().await?;
+        }
+    }
+
+    migrator.unlock().await?;
+
+    Ok(())
+}
+
+#[cfg(feature = "enterprise")]
 async fn windmill_migrations(migrator: &mut CustomMigrator, db: &DB) -> Result<(), Error> {
-    #[cfg(feature = "enterprise")]
     if std::env::var("MIGRATION_NO_BYPASSRLS").is_ok() {
         migrator.lock().await?;
         let has_done_migration = sqlx::query_scalar!(
@@ -212,16 +255,6 @@ async fn windmill_migrations(migrator: &mut CustomMigrator, db: &DB) -> Result<(
         }
         migrator.unlock().await?;
     }
-    migrator.lock().await?;
-
-    let query = include_str!("../../custom_migrations/fix_flow_versioning_2.sql");
-    tracing::info!("Applying fix_flow_versioning_2.sql");
-    let mut tx: sqlx::Transaction<'_, Postgres> = db.begin().await?;
-    tx.execute(query).await?;
-    tracing::info!("Applied fix_flow_versioning_2.sql");
-    tx.commit().await?;
-
-    migrator.unlock().await?;
     Ok(())
 }