windmill/backend/windmill-api-workspaces/src/datatable_migrations.rs

/*
 * Author: Ruben Fiszel
 * Copyright: Windmill Labs, Inc 2022
 * This file and its contents are licensed under the AGPLv3 License.
 * Please see the included NOTICE for copyright information and
 * LICENSE-AGPL for a copy of the license.
 */

//! Data table SQL migrations: CRUD endpoints, run/rollback execution, opt-in
//! management, and the workspace-merge diff helper. Split out of `workspaces.rs`
//! to keep that file focused on core workspace configuration.

use crate::workspaces::{
    is_instance_datatable, pg_dump_database, strip_unreplayable_dump_lines, ItemComparison,
    PgDumpOptions,
};

use axum::{
    extract::{Extension, Path, Query},
    routing::{delete, get, post},
    Json, Router,
};
use chrono::Utc;
use serde::{Deserialize, Serialize};
use sqlx::{Postgres, Transaction};
use std::collections::{HashMap, HashSet};
use tokio_postgres::error::SqlState;

use windmill_api_auth::{require_super_admin, ApiAuthed};
use windmill_api_jobs::run_wait_result_internal;
use windmill_audit::audit_oss::audit_log;
use windmill_audit::ActionKind;
use windmill_common::datatable_roles::ADMIN_DATATABLE_ROLE;
use windmill_common::db::UserDB;
use windmill_common::error::{pg_error_message, Error, JsonResult, Result};
use windmill_common::jobs::{JobPayload, RawCode};
use windmill_common::query_builders::{render_db_quoted_identifier, DbType};
use windmill_common::runnable_settings::{ConcurrencySettingsWithCustom, DebouncingSettings};
use windmill_common::scripts::ScriptLang;
use windmill_common::users::username_to_permissioned_as;
use windmill_common::worker::to_raw_value;
use windmill_common::worker::SqlAnnotations;
use windmill_common::workspaces::{
    ensure_can_use_datatable_role, ensure_datatable_admin_access,
    get_datatable_resource_from_db_unchecked, resolve_governing_datatable, DatatableAccess,
};
use windmill_common::{PgDatabase, DB};
use windmill_git_sync::{
    handle_deployment_metadata, handle_deployment_metadata_batch, DeployedObject,
};
use windmill_queue::{push, PushArgs, PushIsolationLevel};

pub(crate) fn routes() -> Router {
    Router::new()
        .route(
            "/run_datatable_migrations/{datatable_name}",
            post(run_datatable_migrations),
        )
        .route(
            "/rollback_datatable_migrations/{datatable_name}",
            post(rollback_datatable_migrations),
        )
        .route("/list_datatable_migrations", get(list_datatable_migrations))
        .route(
            "/datatable_migrations_status/{datatable_name}",
            get(datatable_migrations_status),
        )
        .route(
            "/enable_datatable_migrations/{datatable_name}",
            post(enable_datatable_migrations),
        )
        .route(
            "/disable_datatable_migrations/{datatable_name}",
            post(disable_datatable_migrations),
        )
        .route(
            "/create_datatable_migration/{datatable_name}",
            post(create_datatable_migration),
        )
        .route(
            "/delete_datatable_migration/{datatable_name}/{timestamp}",
            delete(delete_datatable_migration),
        )
        .route(
            "/upsert_datatable_migration/{datatable_name}",
            post(upsert_datatable_migration),
        )
        .route(
            "/generate_initial_datatable_migration/{datatable_name}",
            post(generate_initial_datatable_migration),
        )
}

/// Refuse a migration whose role this caller may not use, before a job is pushed or a version
/// recorded.
///
/// A migration that declares `-- role <name>` runs as that role, so the caller has to be one of its
/// tenants. One that declares none runs as `admin` and reaches every object in the database
/// whatever the roles grant, so it is for the admins of the workspace that governs the data table
/// — a fork can run a migration under a role it holds, never a migration under `admin`.
///
/// The executor re-checks the role when it resolves the connection, so this is not the boundary. It
/// is what makes the refusal legible: which migration, and which role.
async fn ensure_migration_role_allowed(
    db: &DB,
    w_id: &str,
    datatable_name: &str,
    authed: &ApiAuthed,
    sql: &str,
    timestamp: i64,
    name: &str,
) -> Result<()> {
    let context = format!("Migration {timestamp} ({name})");
    let access = DatatableAccess::Authed(authed.to_authed_ref());
    match SqlAnnotations::datatable_role(sql)? {
        Some(role) => {
            ensure_can_use_datatable_role(db, w_id, datatable_name, Some(&role), &access, &context)
                .await
        }
        None => ensure_datatable_admin_access(db, w_id, datatable_name, &access)
            .await
            .map_err(|e| {
                Error::NotAuthorized(format!(
                    "{context} declares no role, so it would run as admin. {e}"
                ))
            }),
    }
}

#[derive(Serialize)]
struct AppliedMigration {
    version: i64,
    name: String,
}

#[derive(Serialize)]
struct RunDatatableMigrationsResult {
    applied: Vec<AppliedMigration>,
}

#[derive(Deserialize)]
struct RunDatatableMigrationsQuery {
    /// When set, only apply pending migrations up to and including this version.
    up_to: Option<i64>,
    /// When set, apply only this specific migration version (if not already
    /// applied), ignoring any other pending migrations. Takes precedence over
    /// `up_to`.
    only: Option<i64>,
}

/// Build the `database` argument for a migration job. Both resource-backed and
/// instance data tables pass a `datatable://<name>` reference; the pg executor
/// resolves it to real credentials server-side at run time. It must never be
/// resolved here: the resolved instance credentials include a single
/// instance-wide Postgres password, and the job's `args` are readable by the —
/// possibly non-admin — user who ran the migration.
async fn datatable_database_arg(
    db: &DB,
    w_id: &str,
    datatable_name: &str,
) -> Result<Box<serde_json::value::RawValue>> {
    // Fail fast with a clear error if the data table doesn't exist.
    sqlx::query_scalar!(
        "SELECT ws.datatable->'datatables'->$2 FROM workspace_settings ws WHERE ws.workspace_id = $1",
        w_id,
        datatable_name,
    )
    .fetch_one(db)
    .await?
    .ok_or_else(|| Error::internal_err(format!("datatable {datatable_name} not found")))?;

    // `?role=admin` rather than a bare reference, so a migration that declares no `-- role` runs
    // as the connection that owns the schema instead of falling through to the data table's
    // default role — which is what `ensure_migration_role_allowed` gated it as, and which is the
    // only role a DDL statement can be expected to succeed under. A migration that does declare a
    // role overrides this: the annotation wins over the reference.
    //
    // A legacy name containing `?` cannot be migrated through this reference: the appended query
    // makes it neither an exact name nor a parseable one. Accepted on purpose, since such names can
    // no longer be created and none are expected to carry migrations.
    Ok(to_raw_value(&format!(
        "datatable://{datatable_name}?role={ADMIN_DATATABLE_ROLE}"
    )))
}

/// Run a migration's SQL as a normal Windmill `postgresql` job, permissioned as
/// the requesting user and labelled `datatable_migration` for traceability, then
/// wait for it. Errors if the job fails.
async fn run_datatable_migration_job(
    db: &DB,
    user_db: &UserDB,
    authed: &ApiAuthed,
    w_id: &str,
    database_arg: &Box<serde_json::value::RawValue>,
    sql: &str,
) -> Result<()> {
    let mut args = HashMap::new();
    args.insert("database".to_string(), database_arg.clone());
    let push_args = PushArgs { extra: None, args: &args };

    let (uuid, mut tx) = push(
        db,
        PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into()),
        w_id,
        JobPayload::Code(RawCode {
            content: sql.to_string(),
            path: Some("datatable_migration".to_string()),
            hash: None,
            language: ScriptLang::Postgresql,
            lock: None,
            cache_ttl: None,
            cache_ignore_s3_path: None,
            dedicated_worker: None,
            tag: None,
            concurrency_settings: ConcurrencySettingsWithCustom::default(),
            debouncing_settings: DebouncingSettings::default(),
            modules: None,
        }),
        push_args,
        authed.display_username(),
        &authed.email,
        username_to_permissioned_as(&authed.username),
        authed.token_prefix.as_deref(),
        authed.username_override.as_deref(),
        None,
        None,
        None,
        None,
        None,
        None,
        false,
        false,
        None,
        true,
        None,
        None,
        None,
        None,
        Some(&authed.clone().into()),
        false,
        None,
        None,
        None,
    )
    .await?;

    // Tag the job so migration runs are easy to find in the run history.
    sqlx::query!(
        "UPDATE v2_job SET labels = (
                SELECT array_agg(DISTINCT l)
                FROM unnest(coalesce(labels, ARRAY[]::TEXT[]) || $2) l
            ) WHERE id = $1",
        uuid,
        &vec!["datatable_migration".to_string()],
    )
    .execute(&mut *tx)
    .await?;
    tx.commit().await?;

    let (result, success) =
        run_wait_result_internal(db, uuid, w_id, None, false, &authed.username).await?;
    if !success {
        // On failure the job result is `{"error": {"name", "message", ...}}`;
        // surface the executor's message (the Postgres error, e.g. `relation
        // "foo" does not exist`) instead of the raw JSON envelope.
        let detail = serde_json::from_str::<serde_json::Value>(result.get())
            .ok()
            .as_ref()
            .and_then(|v| v.get("error"))
            .and_then(|e| e.get("message"))
            .and_then(|m| m.as_str())
            .map(str::to_string)
            .unwrap_or_else(|| result.get().to_string());
        return Err(Error::internal_err(detail));
    }
    Ok(())
}

/// Ensure the `_wm_migrations` bookkeeping table exists. Migration versions are
/// only unique per data table, but several data-table configs can point at one
/// physical database, so it is keyed by `(datatable, version)` — a version-only
/// key would let one data table's migration mark another's same-version
/// migration as already applied (and rollback could touch the wrong row).
async fn ensure_wm_migrations_schema(client: &tokio_postgres::Client) -> Result<()> {
    // `CREATE TABLE IF NOT EXISTS` checks CREATE on the schema before it checks
    // existence, so probing first is what lets a data table whose role only holds
    // DML grants keep migrating against an already-created bookkeeping table.
    // `to_regclass` resolves through search_path, like the unqualified statements
    // the rest of this module runs against it. Takes no parameters, so it goes
    // through the simple protocol: a named prepared statement is what stalls
    // behind a transaction-pooling proxy (see `pg_get_full_schema`).
    let rows = client
        .simple_query("SELECT to_regclass('_wm_migrations') IS NOT NULL AS present")
        .await
        .map_err(|e| {
            Error::internal_err(format!(
                "Failed to look up _wm_migrations table: {}",
                pg_error_message(&e)
            ))
        })?;
    let exists = rows.iter().any(|msg| match msg {
        tokio_postgres::SimpleQueryMessage::Row(row) => row.get("present") == Some("t"),
        _ => false,
    });
    if exists {
        return Ok(());
    }

    let Err(e) = client
        .batch_execute(
            "CREATE TABLE IF NOT EXISTS _wm_migrations (\
                datatable TEXT NOT NULL, \
                version BIGINT NOT NULL, \
                installed_at TIMESTAMPTZ NOT NULL DEFAULT now(), \
                PRIMARY KEY (datatable, version))",
        )
        .await
    else {
        return Ok(());
    };

    let mut msg = format!(
        "Failed to ensure _wm_migrations table: {}",
        pg_error_message(&e)
    );
    // A role with only table-level grants cannot create it: since Postgres 15 the
    // `public` schema no longer grants CREATE to PUBLIC, so this is the usual
    // failure on a bring-your-own database.
    if e.code() == Some(&SqlState::INSUFFICIENT_PRIVILEGE) {
        // Windmill connects as the role that lacks the privilege, so it cannot
        // grant it: hand over the statement a schema owner has to run instead.
        // Keep it ahead of the explanation below — the UI collapses everything
        // past the first couple of lines behind a "Show more".
        if let Some((user, schema)) = connection_identity(client).await {
            // Both come back unquoted, so a mixed-case or hyphenated name would
            // otherwise render a statement that targets a different schema.
            msg.push_str(&format!(
                ". Run: GRANT CREATE ON SCHEMA {} TO {}",
                render_db_quoted_identifier(&schema, DbType::Postgresql),
                render_db_quoted_identifier(&user, DbType::Postgresql),
            ));
        }
        msg.push_str(
            ". Applied migrations are recorded in a `_wm_migrations` table in the data \
             table's own database, so its user needs to be able to create it",
        );
    }
    Err(Error::internal_err(msg))
}

/// The role and default schema of a data table connection, for grant hints.
/// Both come from the server so the statement we suggest names what the
/// connection actually resolves to, not what the resource happens to say.
async fn connection_identity(client: &tokio_postgres::Client) -> Option<(String, String)> {
    let rows = client
        .simple_query("SELECT current_user AS usr, current_schema() AS sch")
        .await
        .ok()?;
    rows.iter().find_map(|msg| match msg {
        tokio_postgres::SimpleQueryMessage::Row(row) => {
            Some((row.get("usr")?.to_string(), row.get("sch")?.to_string()))
        }
        _ => None,
    })
}

/// Open a connection to a data table's own database and hold the session-level
/// advisory lock that serializes migration runs/rollbacks. The lock is released
/// when the returned client is dropped, so callers must keep it in scope for the
/// whole critical section.
///
/// Runs, rollbacks *and* definition rewrites/deletes all take this lock: a run
/// snapshots a migration's `code_up` from `datatable_migrations` and only records
/// its version in `_wm_migrations` after the job succeeds, so an unserialized edit
/// could rewrite the definition in that window and leave `_wm_migrations` pointing
/// at SQL that was never applied. `_wm_migrations` is per-database, so a single
/// key is sufficient.
async fn lock_datatable_migration_runs(
    db: &DB,
    w_id: &str,
    datatable_name: &str,
) -> Result<tokio_postgres::Client> {
    let db_resource = get_datatable_resource_from_db_unchecked(db, w_id, datatable_name).await?;
    let pg_db: PgDatabase = serde_json::from_value(db_resource)
        .map_err(|e| Error::internal_err(format!("Failed to parse database credentials: {}", e)))?;
    let (client, connection) = pg_db.connect(Some(db)).await?;
    tokio::spawn(async move {
        if let Err(e) = connection.await {
            tracing::error!("Datatable connection error: {}", e);
        }
    });
    client
        .batch_execute("SELECT pg_advisory_lock(hashtext('windmill_datatable_migrations')::int8)")
        .await
        .map_err(|e| {
            Error::internal_err(format!(
                "Failed to acquire migration lock: {}",
                pg_error_message(&e)
            ))
        })?;
    Ok(client)
}

/// Read the versions recorded as applied in a data table's `_wm_migrations`,
/// scoped to that data table, using an existing connection. An absent table
/// (`42P01`) means nothing has been migrated yet.
async fn read_applied_versions_on_client(
    client: &tokio_postgres::Client,
    datatable_name: &str,
) -> Result<HashSet<i64>> {
    match client
        .query(
            "SELECT version FROM _wm_migrations WHERE datatable = $1",
            &[&datatable_name],
        )
        .await
    {
        Ok(rows) => Ok(rows.iter().map(|row| row.get::<_, i64>(0)).collect()),
        Err(e) if e.as_db_error().map(|d| d.code().code()) == Some("42P01") => Ok(HashSet::new()),
        Err(e) => Err(Error::internal_err(format!(
            "Failed to read _wm_migrations: {}",
            pg_error_message(&e)
        ))),
    }
}

/// Apply the workspace's pending data table migrations to a given data table.
/// Each migration runs as a normal Windmill `postgresql` job (permissioned as
/// the requester, labelled `datatable_migration`); applied versions are then
/// recorded in the data table's own `_wm_migrations` table, so only migrations
/// not recorded there are run, in ascending `timestamp` order.
async fn run_datatable_migrations(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Extension(user_db): Extension<UserDB>,
    Path((w_id, datatable_name)): Path<(String, String)>,
    Query(query): Query<RunDatatableMigrationsQuery>,
) -> JsonResult<RunDatatableMigrationsResult> {
    // Before the admin connection is opened at all: the bookkeeping below is created and read
    // through it, so a caller no role covers must be refused here rather than after the fact.
    crate::datatable_permissions::ensure_reaches_datatable(&db, &w_id, &datatable_name, &authed)
        .await?;

    audit_log(
        &db,
        &authed,
        "workspaces.run_datatable_migrations",
        ActionKind::Update,
        &w_id,
        Some(datatable_name.as_str()),
        None,
    )
    .await?;

    let database_arg = datatable_database_arg(&db, &w_id, &datatable_name).await?;

    // Take the run-serialization lock before snapshotting the migration
    // definitions: a concurrent definition rewrite/delete takes the same lock, so
    // the `code_up` we read here can't change between now and when we record its
    // version below. The lock is held until `client` drops at return.
    let client = lock_datatable_migration_runs(&db, &w_id, &datatable_name).await?;

    let migrations = sqlx::query!(
        "SELECT timestamp, name, code_up FROM datatable_migrations \
         WHERE workspace_id = $1 AND datatable = $2 ORDER BY timestamp ASC",
        &w_id,
        &datatable_name,
    )
    .fetch_all(&db)
    .await?;

    ensure_wm_migrations_schema(&client).await?;

    let applied_versions = read_applied_versions_on_client(&client, &datatable_name).await?;

    // How the user scoped the run, for the counter emitted on the first migration
    // that lands below.
    let scope = if query.only.is_some() {
        "only"
    } else if query.up_to.is_some() {
        "up_to"
    } else {
        "all"
    };

    let mut applied = Vec::new();
    for m in migrations {
        if let Some(only) = query.only {
            // Run a single specific migration, skipping every other one.
            if m.timestamp != only {
                continue;
            }
        } else if query.up_to.is_some_and(|up_to| m.timestamp > up_to) {
            // Migrations are ordered ascending, so once we pass `up_to` we're done.
            break;
        }
        if applied_versions.contains(&m.timestamp) {
            continue;
        }
        ensure_migration_role_allowed(
            &db,
            &w_id,
            &datatable_name,
            &authed,
            &m.code_up,
            m.timestamp,
            &m.name,
        )
        .await?;
        run_datatable_migration_job(&db, &user_db, &authed, &w_id, &database_arg, &m.code_up)
            .await
            .map_err(|e| {
                Error::internal_err(format!(
                    "Failed to apply migration {} ({}): {}",
                    m.timestamp, m.name, e
                ))
            })?;
        // Record the migration as installed once its job has succeeded.
        client
            .execute(
                "INSERT INTO _wm_migrations (datatable, version) VALUES ($1, $2) \
                 ON CONFLICT (datatable, version) DO NOTHING",
                &[&datatable_name, &m.timestamp],
            )
            .await
            .map_err(|e| {
                Error::internal_err(format!(
                    "Failed to record migration: {}",
                    pg_error_message(&e)
                ))
            })?;
        applied.push(AppliedMigration { version: m.timestamp, name: m.name });
        // One event per run that moved the data table forward, emitted on the
        // first migration that lands rather than after the loop: a later one
        // failing returns early, and that run still advanced the data table. A
        // run with nothing pending stays uncounted — it is the common outcome of
        // opening the list and would drown out the runs that did something.
        if applied.len() == 1 {
            windmill_common::feature_usage::log_feature_usage("datatable", "migration_run", scope);
        }
    }

    Ok(Json(RunDatatableMigrationsResult { applied }))
}

#[derive(Serialize)]
struct RolledBackMigration {
    version: i64,
    name: String,
}

#[derive(Serialize)]
struct RollbackDatatableMigrationsResult {
    rolled_back: Vec<RolledBackMigration>,
}

#[derive(Deserialize)]
struct RollbackDatatableMigrationsQuery {
    /// When set, roll back this specific applied migration version instead of
    /// the most recently applied one.
    only: Option<i64>,
}

/// Roll back a migration on a given data table: run its `code_down` as a normal
/// Windmill `postgresql` job (permissioned as the requester, labelled
/// `datatable_migration`) then drop its `_wm_migrations` row. Without `only` this
/// targets the most recently applied migration (one step); with `only` it
/// targets that specific applied version.
async fn rollback_datatable_migrations(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Extension(user_db): Extension<UserDB>,
    Path((w_id, datatable_name)): Path<(String, String)>,
    Query(query): Query<RollbackDatatableMigrationsQuery>,
) -> JsonResult<RollbackDatatableMigrationsResult> {
    // Before the admin connection is opened at all: the bookkeeping below is created and read
    // through it, so a caller no role covers must be refused here rather than after the fact.
    crate::datatable_permissions::ensure_reaches_datatable(&db, &w_id, &datatable_name, &authed)
        .await?;

    audit_log(
        &db,
        &authed,
        "workspaces.rollback_datatable_migrations",
        ActionKind::Update,
        &w_id,
        Some(datatable_name.as_str()),
        None,
    )
    .await?;

    // The data table's `_wm_migrations` bookkeeping is read here and the version
    // dropped after the job succeeds; the down SQL itself runs in the job. The
    // lock is held (until `client` drops at return) so a concurrent run or
    // definition rewrite can't interleave with this rollback.
    let client = lock_datatable_migration_runs(&db, &w_id, &datatable_name).await?;

    ensure_wm_migrations_schema(&client).await?;

    // Resolve which applied version to roll back: a specific one when `only` is
    // given (and actually applied), otherwise the most recently applied. Scoped
    // to this data table so a shared physical database can't surface another
    // data table's version.
    let target = match query.only {
        Some(only) => client
            .query_opt(
                "SELECT version FROM _wm_migrations WHERE datatable = $1 AND version = $2",
                &[&datatable_name, &only],
            )
            .await
            .map_err(|e| {
                Error::internal_err(format!(
                    "Failed to read _wm_migrations: {}",
                    pg_error_message(&e)
                ))
            })?,
        None => client
            .query_opt(
                "SELECT version FROM _wm_migrations WHERE datatable = $1 \
                 ORDER BY version DESC LIMIT 1",
                &[&datatable_name],
            )
            .await
            .map_err(|e| {
                Error::internal_err(format!(
                    "Failed to read _wm_migrations: {}",
                    pg_error_message(&e)
                ))
            })?,
    };

    let version: i64 = match target {
        Some(row) => row.get::<_, i64>(0),
        None => {
            return Ok(Json(RollbackDatatableMigrationsResult {
                rolled_back: vec![],
            }))
        }
    };

    let definition = sqlx::query!(
        "SELECT name, code_down FROM datatable_migrations \
         WHERE workspace_id = $1 AND datatable = $2 AND timestamp = $3",
        &w_id,
        &datatable_name,
        version
    )
    .fetch_optional(&db)
    .await?
    .ok_or_else(|| {
        Error::BadRequest(format!(
            "Cannot roll back migration {version}: its definition no longer exists"
        ))
    })?;

    let code_down = definition.code_down.ok_or_else(|| {
        Error::BadRequest(format!(
            "Cannot roll back migration {} ({}): it has no down migration",
            version, definition.name
        ))
    })?;

    ensure_migration_role_allowed(
        &db,
        &w_id,
        &datatable_name,
        &authed,
        &code_down,
        version,
        &definition.name,
    )
    .await?;

    let database_arg = datatable_database_arg(&db, &w_id, &datatable_name).await?;
    run_datatable_migration_job(&db, &user_db, &authed, &w_id, &database_arg, &code_down)
        .await
        .map_err(|e| {
            Error::internal_err(format!(
                "Failed to roll back migration {} ({}): {}",
                version, definition.name, e
            ))
        })?;

    // Forget the version once its down job has succeeded.
    client
        .execute(
            "DELETE FROM _wm_migrations WHERE datatable = $1 AND version = $2",
            &[&datatable_name, &version],
        )
        .await
        .map_err(|e| {
            Error::internal_err(format!(
                "Failed to drop migration record: {}",
                pg_error_message(&e)
            ))
        })?;

    windmill_common::feature_usage::log_feature_usage(
        "datatable",
        "migration_rollback",
        if query.only.is_some() { "only" } else { "last" },
    );

    Ok(Json(RollbackDatatableMigrationsResult {
        rolled_back: vec![RolledBackMigration { version, name: definition.name }],
    }))
}

#[derive(Serialize, Deserialize)]
pub struct DatatableMigration {
    pub datatable: String,
    pub timestamp: i64,
    pub name: String,
    pub code_up: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub code_down: Option<String>,
}

async fn list_datatable_migrations(
    _authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Path(w_id): Path<String>,
) -> JsonResult<Vec<DatatableMigration>> {
    let migrations = sqlx::query_as!(
        DatatableMigration,
        "SELECT datatable, timestamp, name, code_up, code_down FROM datatable_migrations \
         WHERE workspace_id = $1 ORDER BY datatable, timestamp ASC",
        &w_id
    )
    .fetch_all(&db)
    .await?;

    Ok(Json(migrations))
}

#[derive(Serialize)]
#[serde(rename_all = "snake_case")]
enum DatatableMigrationRunStatus {
    /// Recorded in the data table's `_wm_migrations` table.
    Ran,
    /// Defined but not yet applied.
    NotRun,
    /// Applied status could not be determined (connection failure).
    Unknown,
}

#[derive(Serialize)]
struct DatatableMigrationWithStatus {
    timestamp: i64,
    name: String,
    code_up: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    code_down: Option<String>,
    status: DatatableMigrationRunStatus,
}

#[derive(Serialize)]
struct DatatableMigrationsStatusResult {
    /// Whether the migrations feature is opted in for this data table.
    enabled: bool,
    migrations: Vec<DatatableMigrationWithStatus>,
    /// Set when the applied status couldn't be read from the data table.
    #[serde(skip_serializing_if = "Option::is_none")]
    error: Option<String>,
}

/// Whether the SQL-migrations feature is enabled for a data table. Honors the
/// explicit `migrations_enabled` flag; when unset (data tables predating the
/// feature) it is considered enabled only if migrations already exist.
async fn datatable_migrations_enabled(db: &DB, w_id: &str, datatable_name: &str) -> Result<bool> {
    let flag: Option<bool> = sqlx::query_scalar!(
        "SELECT (ws.datatable->'datatables'->$2->>'migrations_enabled')::boolean \
         FROM workspace_settings ws WHERE ws.workspace_id = $1",
        w_id,
        datatable_name,
    )
    .fetch_optional(db)
    .await?
    .flatten();

    match flag {
        Some(v) => Ok(v),
        None => Ok(sqlx::query_scalar!(
            "SELECT EXISTS(SELECT 1 FROM datatable_migrations \
             WHERE workspace_id = $1 AND datatable = $2)",
            w_id,
            datatable_name,
        )
        .fetch_one(db)
        .await?
        .unwrap_or(false)),
    }
}

/// Reject the request when migrations are not enabled for the data table.
async fn ensure_datatable_migrations_enabled(
    db: &DB,
    w_id: &str,
    datatable_name: &str,
) -> Result<()> {
    if !datatable_migrations_enabled(db, w_id, datatable_name).await? {
        return Err(Error::BadRequest(format!(
            "Migrations are not enabled for data table '{}'. Enable them first.",
            datatable_name
        )));
    }
    Ok(())
}

/// Read the versions recorded in a data table's `_wm_migrations` table. A
/// missing table means nothing has been applied yet (empty set, not an error).
async fn read_applied_datatable_versions(
    db: &DB,
    w_id: &str,
    datatable_name: &str,
) -> Result<HashSet<i64>> {
    let db_resource = get_datatable_resource_from_db_unchecked(db, w_id, datatable_name).await?;
    let pg_db: PgDatabase = serde_json::from_value(db_resource)
        .map_err(|e| Error::internal_err(format!("Failed to parse database credentials: {}", e)))?;
    let (client, connection) = pg_db.connect(Some(db)).await?;
    tokio::spawn(async move {
        if let Err(e) = connection.await {
            tracing::error!("Datatable connection error: {}", e);
        }
    });

    // Read-only status path: don't create the table here, and don't take the run
    // lock — a stale-by-a-moment applied set is fine for display.
    read_applied_versions_on_client(&client, datatable_name).await
}

/// List a data table's migrations annotated with whether each has been applied.
async fn datatable_migrations_status(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Path((w_id, datatable_name)): Path<(String, String)>,
) -> JsonResult<DatatableMigrationsStatusResult> {
    // Reads `_wm_migrations` through the data table's admin connection, so it answers to the same
    // question as running one: may you reach this data table at all.
    crate::datatable_permissions::ensure_reaches_datatable(&db, &w_id, &datatable_name, &authed)
        .await?;

    let enabled = datatable_migrations_enabled(&db, &w_id, &datatable_name).await?;
    if !enabled {
        return Ok(Json(DatatableMigrationsStatusResult {
            enabled: false,
            migrations: vec![],
            error: None,
        }));
    }

    let defs = sqlx::query!(
        "SELECT timestamp, name, code_up, code_down FROM datatable_migrations \
         WHERE workspace_id = $1 AND datatable = $2 ORDER BY timestamp ASC",
        &w_id,
        &datatable_name,
    )
    .fetch_all(&db)
    .await?;

    let (applied, error) = match read_applied_datatable_versions(&db, &w_id, &datatable_name).await
    {
        Ok(set) => (Some(set), None),
        Err(e) => (None, Some(e.to_string())),
    };

    let migrations = defs
        .into_iter()
        .map(|m| {
            let status = match &applied {
                Some(set) if set.contains(&m.timestamp) => DatatableMigrationRunStatus::Ran,
                Some(_) => DatatableMigrationRunStatus::NotRun,
                None => DatatableMigrationRunStatus::Unknown,
            };
            DatatableMigrationWithStatus {
                timestamp: m.timestamp,
                name: m.name,
                code_up: m.code_up,
                code_down: m.code_down,
                status,
            }
        })
        .collect();

    Ok(Json(DatatableMigrationsStatusResult {
        enabled: true,
        migrations,
        error,
    }))
}

/// Only workspace admins and super admins may opt a data table in or out of
/// migrations.
async fn require_datatable_migrations_manager(db: &DB, authed: &ApiAuthed) -> Result<()> {
    if authed.is_admin || require_super_admin(db, &authed).await.is_ok() {
        Ok(())
    } else {
        Err(Error::BadRequest(
            "Only workspace admins and super admins can enable or disable data table migrations"
                .to_string(),
        ))
    }
}

async fn enable_datatable_migrations(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Path((w_id, datatable_name)): Path<(String, String)>,
) -> Result<String> {
    require_datatable_migrations_manager(&db, &authed).await?;

    let updated = sqlx::query_scalar!(
        "UPDATE workspace_settings \
         SET datatable = jsonb_set(datatable, ARRAY['datatables', $2::text, 'migrations_enabled'], 'true'::jsonb) \
         WHERE workspace_id = $1 AND jsonb_exists(datatable->'datatables', $2) \
         RETURNING 1",
        &w_id,
        &datatable_name,
    )
    .fetch_optional(&db)
    .await?;
    if updated.is_none() {
        return Err(Error::NotFound(format!(
            "data table {datatable_name} not found"
        )));
    }

    audit_log(
        &db,
        &authed,
        "workspaces.enable_datatable_migrations",
        ActionKind::Update,
        &w_id,
        Some(datatable_name.as_str()),
        None,
    )
    .await?;

    windmill_common::feature_usage::log_feature_usage("datatable", "migrations_toggled", "on");

    Ok(format!(
        "Enabled migrations for data table {datatable_name}"
    ))
}

/// Opt a data table out of migrations. Deletes ALL of its migration definitions.
async fn disable_datatable_migrations(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Path((w_id, datatable_name)): Path<(String, String)>,
) -> Result<String> {
    require_datatable_migrations_manager(&db, &authed).await?;

    let mut tx = db.begin().await?;

    let updated = sqlx::query_scalar!(
        "UPDATE workspace_settings \
         SET datatable = jsonb_set(datatable, ARRAY['datatables', $2::text, 'migrations_enabled'], 'false'::jsonb) \
         WHERE workspace_id = $1 AND jsonb_exists(datatable->'datatables', $2) \
         RETURNING 1",
        &w_id,
        &datatable_name,
    )
    .fetch_optional(&mut *tx)
    .await?;
    if updated.is_none() {
        return Err(Error::NotFound(format!(
            "data table {datatable_name} not found"
        )));
    }

    // Capture the deleted definitions so each removal is tallied as a deployed
    // object (like single-migration deletion), keeping workspace comparison and
    // git-sync callbacks in sync when a fork opts back out of migrations.
    let deleted = sqlx::query!(
        "DELETE FROM datatable_migrations WHERE workspace_id = $1 AND datatable = $2 \
         RETURNING timestamp, name",
        &w_id,
        &datatable_name,
    )
    .fetch_all(&mut *tx)
    .await?;

    audit_log(
        &mut *tx,
        &authed,
        "workspaces.disable_datatable_migrations",
        ActionKind::Delete,
        &w_id,
        Some(datatable_name.as_str()),
        None,
    )
    .await?;

    tx.commit().await?;

    for m in deleted {
        record_datatable_migration_deployment(
            &authed,
            &db,
            &w_id,
            &datatable_name,
            m.timestamp,
            &m.name,
        )
        .await?;
    }

    windmill_common::feature_usage::log_feature_usage("datatable", "migrations_toggled", "off");

    Ok(format!(
        "Disabled migrations for data table {datatable_name} and deleted its migrations"
    ))
}

#[derive(Deserialize)]
pub struct CreateDatatableMigration {
    pub name: String,
    pub code_up: String,
    #[serde(default)]
    pub code_down: Option<String>,
}

/// Migration names map onto on-disk file names and the `_wm_migrations` record,
/// so keep them to a safe path-segment charset (matches the CLI scaffold).
fn validate_migration_name(name: &str) -> Result<()> {
    if name.is_empty()
        || !name
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
    {
        return Err(Error::BadRequest(format!(
            "Invalid migration name '{name}': use only letters, digits, '_' and '-'"
        )));
    }
    Ok(())
}

/// The data table name becomes a directory segment in the sync export
/// (`migrations/datatable/<datatable>/...`); reject anything that could escape it.
pub(crate) fn validate_datatable_path_segment(datatable: &str) -> Result<()> {
    if datatable.is_empty()
        || datatable.contains('/')
        || datatable.contains('\\')
        || datatable.contains("..")
    {
        return Err(Error::BadRequest(format!(
            "Invalid data table name '{datatable}': must not contain '/', '\\' or '..'"
        )));
    }
    Ok(())
}

/// A data table's name is a path segment of every migration it owns, and git sync
/// carries that path through three matchers that all speak the same restricted
/// charset: `transform_regexp` expands a repo's `**` filter to `[a-zA-Z0-9_\-./]*`,
/// the CLI reuses the path as a minimatch pattern (where a leading `.` also needs
/// `dot: true`, which the CLI does not set), and the deploy stages it as a
/// `git add` pathspec. A name outside the charset matches nothing in all three, so
/// its migrations would silently never reach the repo.
///
/// Only names being introduced are held to this — an existing data table keeps
/// saving (and can be renamed to a syncable name) instead of locking its workspace
/// out of the settings form.
pub(crate) fn validate_new_datatable_name(datatable: &str) -> Result<()> {
    validate_datatable_path_segment(datatable)?;
    let starts_alphanumeric = datatable
        .chars()
        .next()
        .is_some_and(|c| c.is_ascii_alphanumeric());
    if !starts_alphanumeric
        || !datatable
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || matches!(c, '_' | '-' | '.'))
    {
        return Err(Error::BadRequest(format!(
            "Invalid data table name '{datatable}': start with a letter or digit and use only \
             letters, digits, '_', '-' and '.' so its SQL migrations can be synced to a git \
             repository"
        )));
    }
    Ok(())
}

/// Record a data table migration change as a deployed object so it is tallied
/// into `workspace_diff` and shows up as a `datatable_migration` item in the
/// workspace-merge diff. The diff path is `<datatable>/<timestamp>_<name>`,
/// matching `parse_datatable_migration_diff_path`.
async fn record_datatable_migration_deployment(
    authed: &ApiAuthed,
    db: &DB,
    w_id: &str,
    datatable: &str,
    timestamp: i64,
    name: &str,
) -> Result<()> {
    // A data table predating `validate_new_datatable_name` can carry a name no
    // repo filter can match; say so once per change instead of letting the deploy
    // vanish inside the path filter.
    if validate_new_datatable_name(datatable).is_err() {
        tracing::warn!(
            "Data table '{datatable}' has a name git sync cannot match, so its SQL migrations \
             will not reach a linked repository. Rename it to letters, digits, '_', '-' and '.'."
        );
    }
    handle_deployment_metadata(
        &authed.email,
        &authed.username,
        db,
        w_id,
        DeployedObject::DatatableMigration { path: format!("{datatable}/{timestamp}_{name}") },
        Some(format!(
            "Data table migration {name} ({timestamp}) on {datatable}"
        )),
        false,
        None,
    )
    .await
}

/// Allocate the next version for a data table and insert the migration
/// definition, in one transaction. A per-(workspace, data table) advisory lock
/// serializes concurrent version allocation so two creates can't read the same
/// `MAX(timestamp)` and collide on the `(workspace_id, datatable, timestamp)`
/// primary key. The version is the current UTC `YYYYMMDDHHMMSS`, bumped past any
/// existing version to stay unique and monotonically increasing.
async fn insert_datatable_migration_def(
    tx: &mut Transaction<'_, Postgres>,
    w_id: &str,
    datatable: &str,
    name: &str,
    code_up: &str,
    code_down: Option<&str>,
) -> Result<i64> {
    sqlx::query!(
        "SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))",
        w_id,
        datatable,
    )
    .execute(&mut **tx)
    .await?;

    let now_ts: i64 = Utc::now()
        .format("%Y%m%d%H%M%S")
        .to_string()
        .parse()
        .map_err(|e| Error::internal_err(format!("Failed to build migration version: {}", e)))?;
    let max_existing: Option<i64> = sqlx::query_scalar!(
        "SELECT MAX(timestamp) FROM datatable_migrations WHERE workspace_id = $1 AND datatable = $2",
        w_id,
        datatable,
    )
    .fetch_one(&mut **tx)
    .await?;
    let timestamp = match max_existing {
        Some(m) if m >= now_ts => m + 1,
        _ => now_ts,
    };

    sqlx::query!(
        "INSERT INTO datatable_migrations (workspace_id, datatable, timestamp, name, code_up, code_down) \
         VALUES ($1, $2, $3, $4, $5, $6)",
        w_id,
        datatable,
        timestamp,
        name,
        code_up,
        code_down,
    )
    .execute(&mut **tx)
    .await?;

    Ok(timestamp)
}

/// Mark a version as already installed in a data table's `_wm_migrations` table
/// (ensuring the table exists first).
async fn mark_datatable_version_installed(
    db: &DB,
    pg_db: &PgDatabase,
    datatable: &str,
    version: i64,
) -> Result<()> {
    let (client, connection) = pg_db.connect(Some(db)).await?;
    tokio::spawn(async move {
        if let Err(e) = connection.await {
            tracing::error!("Datatable connection error: {}", e);
        }
    });
    ensure_wm_migrations_schema(&client).await?;
    client
        .execute(
            "INSERT INTO _wm_migrations (datatable, version) VALUES ($1, $2) \
             ON CONFLICT (datatable, version) DO NOTHING",
            &[&datatable, &version],
        )
        .await
        .map_err(|e| {
            Error::internal_err(format!(
                "Failed to mark initial migration installed: {}",
                pg_error_message(&e)
            ))
        })?;
    Ok(())
}

/// Create a single migration for a data table. The version is generated
/// server-side (current UTC `YYYYMMDDHHMMSS`), bumped past any existing version
/// so it stays unique and monotonically increasing.
async fn create_datatable_migration(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Path((w_id, datatable_name)): Path<(String, String)>,
    Json(payload): Json<CreateDatatableMigration>,
) -> JsonResult<DatatableMigration> {
    validate_datatable_path_segment(&datatable_name)?;
    validate_migration_name(&payload.name)?;
    ensure_datatable_migrations_enabled(&db, &w_id, &datatable_name).await?;

    let mut tx = db.begin().await?;
    let timestamp = insert_datatable_migration_def(
        &mut tx,
        &w_id,
        &datatable_name,
        &payload.name,
        &payload.code_up,
        payload.code_down.as_deref(),
    )
    .await?;
    tx.commit().await?;

    audit_log(
        &db,
        &authed,
        "workspaces.create_datatable_migration",
        ActionKind::Create,
        &w_id,
        Some(datatable_name.as_str()),
        None,
    )
    .await?;

    record_datatable_migration_deployment(
        &authed,
        &db,
        &w_id,
        &datatable_name,
        timestamp,
        &payload.name,
    )
    .await?;

    windmill_common::feature_usage::log_feature_usage("datatable", "migration_created", "manual");

    Ok(Json(DatatableMigration {
        datatable: datatable_name,
        timestamp,
        name: payload.name,
        code_up: payload.code_up,
        code_down: payload.code_down,
    }))
}

/// Delete a single migration definition from a data table.
async fn delete_datatable_migration(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Path((w_id, datatable_name, timestamp)): Path<(String, String, i64)>,
) -> Result<String> {
    // Hold the run-serialization lock across the applied-check and the delete: a
    // run snapshots a migration's SQL before recording its version, so an
    // unserialized delete could race it and leave `_wm_migrations` pointing at a
    // definition that no longer exists (breaking rollback and hiding the applied
    // version). Held until the handler returns. Fail closed if we can't verify.
    let unreachable = |e| {
        Error::internal_err(format!(
            "Cannot verify whether migration {} on data table '{}' has already been applied \
             (its database is unreachable: {}). Refusing to delete it; retry once the database \
             is reachable.",
            timestamp, datatable_name, e
        ))
    };
    let lock_client = lock_datatable_migration_runs(&db, &w_id, &datatable_name)
        .await
        .map_err(unreachable)?;
    let applied = read_applied_versions_on_client(&lock_client, &datatable_name)
        .await
        .map_err(unreachable)?;
    if applied.contains(&timestamp) {
        return Err(Error::BadRequest(format!(
            "Migration {} on data table '{}' has already been applied and cannot be deleted. \
             Revert it first.",
            timestamp, datatable_name
        )));
    }

    let deleted_name = sqlx::query_scalar!(
        "DELETE FROM datatable_migrations \
         WHERE workspace_id = $1 AND datatable = $2 AND timestamp = $3 \
         RETURNING name",
        &w_id,
        &datatable_name,
        timestamp,
    )
    .fetch_optional(&db)
    .await?;
    // The definition is removed; runs may resume (audit/deploy metadata below
    // don't need the lock).
    drop(lock_client);

    audit_log(
        &db,
        &authed,
        "workspaces.delete_datatable_migration",
        ActionKind::Delete,
        &w_id,
        Some(datatable_name.as_str()),
        None,
    )
    .await?;

    // Only tally a change if a migration was actually deleted.
    if let Some(name) = deleted_name {
        record_datatable_migration_deployment(
            &authed,
            &db,
            &w_id,
            &datatable_name,
            timestamp,
            &name,
        )
        .await?;
    }

    Ok(format!(
        "Deleted migration {} from {}",
        timestamp, datatable_name
    ))
}

#[derive(Deserialize)]
pub struct UpsertDatatableMigration {
    pub timestamp: i64,
    pub name: String,
    pub code_up: String,
    #[serde(default)]
    pub code_down: Option<String>,
}

/// Insert or update a single migration at an explicit version. Used by
/// `wmill sync` to push a `migrations/datatable/<dt>/<version>_<name>.up.sql`
/// (and `.down.sql`) file as the source of truth.
async fn upsert_datatable_migration(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Path((w_id, datatable_name)): Path<(String, String)>,
    Json(payload): Json<UpsertDatatableMigration>,
) -> Result<String> {
    validate_datatable_path_segment(&datatable_name)?;
    validate_migration_name(&payload.name)?;
    ensure_datatable_migrations_enabled(&db, &w_id, &datatable_name).await?;

    // Guard against silently rewriting a migration that has already run in the
    // data table's database: its `_wm_migrations` record would no longer match
    // its SQL, so a later `migrate up` would skip it and a rollback would run a
    // `down` that doesn't correspond to what was applied. Only an actual change
    // to an existing migration is guarded; unchanged re-pushes (e.g.
    // `wmill sync push`) always proceed, and so does filling in a down migration
    // that was missing — the up that ran is untouched, and that is the only way
    // to make an already-applied migration revertable.
    let existing = sqlx::query!(
        "SELECT name, code_up, code_down FROM datatable_migrations \
         WHERE workspace_id = $1 AND datatable = $2 AND timestamp = $3",
        &w_id,
        &datatable_name,
        payload.timestamp,
    )
    .fetch_optional(&db)
    .await?;
    // When overwriting the SQL of an existing definition, hold the
    // run-serialization lock across the applied-check and the write below so an
    // in-flight run can't record a version for the SQL we're about to overwrite.
    // Held until the end of the handler (well past the write). The exempt
    // upserts need no lock: a new or unchanged one overwrites nothing, and one
    // that only adds a down leaves the `code_up` a concurrent run is recording
    // a version for untouched.
    let only_adds_down = existing.as_ref().is_some_and(|existing| {
        existing.name == payload.name
            && existing.code_up == payload.code_up
            && existing.code_down.is_none()
            && payload.code_down.is_some()
    });
    let unchanged = existing.as_ref().is_some_and(|existing| {
        existing.name == payload.name
            && existing.code_up == payload.code_up
            && existing.code_down == payload.code_down
    });
    let _run_lock = match existing.as_ref() {
        Some(_) if !only_adds_down && !unchanged => {
            // Fail closed: if we can't lock/read the applied set (e.g. the
            // data-table database is temporarily unreachable), refuse the change
            // rather than risk overwriting a migration that has already run.
            let unreachable = |e| {
                Error::internal_err(format!(
                    "Cannot verify whether migration {} on data table '{}' has already been \
                     applied (its database is unreachable: {}). Refusing to modify it; retry \
                     once the database is reachable.",
                    payload.timestamp, datatable_name, e
                ))
            };
            let client = lock_datatable_migration_runs(&db, &w_id, &datatable_name)
                .await
                .map_err(unreachable)?;
            let applied = read_applied_versions_on_client(&client, &datatable_name)
                .await
                .map_err(unreachable)?;
            if applied.contains(&payload.timestamp) {
                return Err(Error::BadRequest(format!(
                    "Migration {} on data table '{}' has already been applied and cannot be modified. \
                     Revert it first, or add a new migration instead.",
                    payload.timestamp, datatable_name
                )));
            }
            Some(client)
        }
        _ => None,
    };

    // An exempt upsert judged the row from an unlocked read and then writes
    // without the lock, so that whole read is re-tested here, where `ON CONFLICT
    // DO UPDATE` re-reads the row under a row lock. Otherwise a request working
    // from a stale definition silently reverts whatever changed in between — a
    // second addition's down, or a locked rewrite of the up whose new SQL a run
    // may already have recorded a version for. Reading no row at all is part of
    // the premise: the equality is NULL when `$8` is, so a version created in the
    // meantime is refused rather than overwritten.
    let recheck_observed = _run_lock.is_none();
    let written = sqlx::query!(
        "INSERT INTO datatable_migrations (workspace_id, datatable, timestamp, name, code_up, code_down) \
         VALUES ($1, $2, $3, $4, $5, $6) \
         ON CONFLICT (workspace_id, datatable, timestamp) DO UPDATE \
         SET name = EXCLUDED.name, code_up = EXCLUDED.code_up, code_down = EXCLUDED.code_down \
         WHERE NOT $7 \
            OR (datatable_migrations.name = $8::text \
                AND datatable_migrations.code_up = $9::text \
                AND datatable_migrations.code_down IS NOT DISTINCT FROM $10::text)",
        &w_id,
        &datatable_name,
        payload.timestamp,
        &payload.name,
        &payload.code_up,
        payload.code_down.as_deref(),
        recheck_observed,
        existing.as_ref().map(|e| e.name.as_str()),
        existing.as_ref().map(|e| e.code_up.as_str()),
        existing.as_ref().and_then(|e| e.code_down.as_deref()),
    )
    .execute(&db)
    .await?;
    if written.rows_affected() == 0 {
        return Err(Error::BadRequest(format!(
            "Migration {} on data table '{}' changed while this change was being saved. \
             Reload it before editing.",
            payload.timestamp, datatable_name
        )));
    }
    // The definition is written; runs may resume (audit/deploy metadata below
    // don't need the lock).
    drop(_run_lock);

    audit_log(
        &db,
        &authed,
        "workspaces.upsert_datatable_migration",
        ActionKind::Update,
        &w_id,
        Some(datatable_name.as_str()),
        None,
    )
    .await?;

    record_datatable_migration_deployment(
        &authed,
        &db,
        &w_id,
        &datatable_name,
        payload.timestamp,
        &payload.name,
    )
    .await?;

    // An unchanged re-push is not counted: `wmill sync push` sends every migration
    // on every sync, so counting those would swamp the definitions people write.
    if !unchanged {
        windmill_common::feature_usage::log_feature_usage(
            "datatable",
            "migration_created",
            if existing.is_none() {
                "synced"
            } else {
                "edited"
            },
        );
    }

    Ok(format!(
        "Upserted migration {} in {}",
        payload.timestamp, datatable_name
    ))
}

/// Generate the first migration for a data table by snapshotting its current
/// schema with `pg_dump`. The migration is recorded as already installed (the
/// definition is written first, then its version is marked in the data table's
/// `_wm_migrations`, so it ends up considered applied and is never re-run) and
/// has no down migration.
async fn generate_initial_datatable_migration(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Path((w_id, datatable_name)): Path<(String, String)>,
) -> JsonResult<DatatableMigration> {
    // Returns a `pg_dump` of the whole schema and writes into the data table's own bookkeeping, so
    // it answers to the workspace that governs it rather than to whoever is asking.
    ensure_datatable_admin_access(
        &db,
        &w_id,
        &datatable_name,
        &DatatableAccess::Authed(authed.to_authed_ref()),
    )
    .await?;
    validate_datatable_path_segment(&datatable_name)?;
    ensure_datatable_migrations_enabled(&db, &w_id, &datatable_name).await?;

    // The initial snapshot only makes sense on a data table with no migrations
    // yet; reject otherwise so repeated calls don't pile up duplicate "initial"
    // definitions (each at a distinct timestamp, each marked installed).
    let has_migrations: bool = sqlx::query_scalar!(
        "SELECT EXISTS(SELECT 1 FROM datatable_migrations WHERE workspace_id = $1 AND datatable = $2)",
        &w_id,
        &datatable_name,
    )
    .fetch_one(&db)
    .await?
    .unwrap_or(false);
    if has_migrations {
        return Err(Error::BadRequest(format!(
            "Data table '{datatable_name}' already has migrations; the initial migration can only be generated when there are none."
        )));
    }

    let db_resource = get_datatable_resource_from_db_unchecked(&db, &w_id, &datatable_name).await?;
    let pg_db: PgDatabase = serde_json::from_value(db_resource)
        .map_err(|e| Error::internal_err(format!("Failed to parse database credentials: {}", e)))?;

    // Snapshot the schema without `_wm_migrations`, Windmill's own bookkeeping table, and
    // without what a replay elsewhere cannot run: the replaying user owns none of this
    // database's objects, and the grants Windmill plants in an instance database (`ALTER
    // DEFAULT PRIVILEGES FOR ROLE ...`) fail even replaying onto the same server.
    let no_acl = is_instance_datatable(&db, &w_id, &datatable_name).await?;
    let dump_file = pg_dump_database(
        &pg_db,
        PgDumpOptions {
            schema_only: true,
            exclude_tables: &["_wm_migrations"],
            no_owner: true,
            no_acl,
        },
    )
    .await?;
    let raw_dump = tokio::fs::read_to_string(&dump_file.path)
        .await
        .map_err(|e| Error::internal_err(format!("Failed to read schema dump: {}", e)))?;
    let code_up = strip_unreplayable_dump_lines(&raw_dump);

    // Record the definition first, then mark it installed. If marking fails we
    // delete the definition, so a failure leaves no phantom "initial" (rather
    // than a `_wm_migrations` version with no definition that the UI can't
    // clear). The narrow window where it briefly shows "not run" is benign:
    // running it would just no-op/fail harmlessly against the existing schema.
    let mut tx = db.begin().await?;
    let timestamp =
        insert_datatable_migration_def(&mut tx, &w_id, &datatable_name, "initial", &code_up, None)
            .await?;
    tx.commit().await?;

    if let Err(e) = mark_datatable_version_installed(&db, &pg_db, &datatable_name, timestamp).await
    {
        let _ = sqlx::query!(
            "DELETE FROM datatable_migrations \
             WHERE workspace_id = $1 AND datatable = $2 AND timestamp = $3",
            &w_id,
            &datatable_name,
            timestamp,
        )
        .execute(&db)
        .await;
        return Err(e);
    }

    audit_log(
        &db,
        &authed,
        "workspaces.generate_initial_datatable_migration",
        ActionKind::Create,
        &w_id,
        Some(datatable_name.as_str()),
        None,
    )
    .await?;

    record_datatable_migration_deployment(
        &authed,
        &db,
        &w_id,
        &datatable_name,
        timestamp,
        "initial",
    )
    .await?;

    windmill_common::feature_usage::log_feature_usage(
        "datatable",
        "migration_created",
        "initial_snapshot",
    );

    Ok(Json(DatatableMigration {
        datatable: datatable_name,
        timestamp,
        name: "initial".to_string(),
        code_up,
        code_down: None,
    }))
}

/// A datatable migration diff item has path `<datatable>/<timestamp>_<name>`.
/// Parse out the (datatable, timestamp) needed to look it up.
pub(crate) fn parse_datatable_migration_diff_path(path: &str) -> Option<(String, i64)> {
    let (datatable, file) = path.split_once('/')?;
    let ts_str: String = file.chars().take_while(|c| c.is_ascii_digit()).collect();
    let timestamp = ts_str.parse::<i64>().ok()?;
    Some((datatable.to_string(), timestamp))
}

pub(crate) async fn compare_two_datatable_migration(
    db: &DB,
    source_workspace_id: &str,
    fork_workspace_id: &str,
    path: &str,
) -> Result<ItemComparison> {
    let (datatable, timestamp) = match parse_datatable_migration_diff_path(path) {
        Some(v) => v,
        None => {
            return Ok(ItemComparison {
                has_changes: false,
                exists_in_source: false,
                exists_in_fork: false,
            })
        }
    };

    let source = sqlx::query!(
        "SELECT name, code_up, code_down FROM datatable_migrations \
         WHERE workspace_id = $1 AND datatable = $2 AND timestamp = $3",
        source_workspace_id,
        datatable,
        timestamp,
    )
    .fetch_optional(db)
    .await?;
    let target = sqlx::query!(
        "SELECT name, code_up, code_down FROM datatable_migrations \
         WHERE workspace_id = $1 AND datatable = $2 AND timestamp = $3",
        fork_workspace_id,
        datatable,
        timestamp,
    )
    .fetch_optional(db)
    .await?;

    let has_changes = match (&source, &target) {
        (Some(s), Some(t)) => {
            s.name != t.name || s.code_up != t.code_up || s.code_down != t.code_down
        }
        _ => source.is_some() || target.is_some(),
    };

    Ok(ItemComparison {
        has_changes,
        exists_in_source: source.is_some(),
        exists_in_fork: target.is_some(),
    })
}

#[derive(Deserialize, Debug)]
pub(crate) struct DatatableRename {
    pub(crate) from: String,
    pub(crate) to: String,
}

/// The database whose `_wm_migrations` a rename or delete of `datatable` in `w_id` should touch —
/// `None` when that is somebody else's.
///
/// A fork's entry points at the workspace that governs the data table, so renaming or removing it
/// changes what the fork calls the data table and nothing more. Following the pointer here would
/// let a fork admin relabel or wipe the *governing* workspace's migration bookkeeping through
/// their own settings form, and the parent would then re-run every migration from zero.
async fn resolve_datatable_pg(db: &DB, w_id: &str, datatable: &str) -> Result<Option<PgDatabase>> {
    let governing = resolve_governing_datatable(db, w_id, datatable).await?;
    if governing.workspace_id != w_id {
        return Ok(None);
    }
    let db_resource = get_datatable_resource_from_db_unchecked(db, w_id, datatable).await?;
    serde_json::from_value(db_resource)
        .map(Some)
        .map_err(|e| Error::internal_err(format!("Failed to parse database credentials: {}", e)))
}

/// Tolerate a data table whose database has no `_wm_migrations` yet (42P01 =
/// undefined_table): it has never run a migration, so nothing to rename or forget.
fn ignore_missing_wm_migrations(e: tokio_postgres::Error) -> Result<()> {
    match e.as_db_error().map(|d| d.code().code()) {
        Some("42P01") => Ok(()),
        _ => Err(Error::internal_err(format!(
            "Failed to update _wm_migrations: {}",
            pg_error_message(&e)
        ))),
    }
}

/// Drop a data table's rows from its own database's `_wm_migrations`.
async fn remote_forget_datatable_migrations(db: &DB, w_id: &str, datatable: &str) -> Result<()> {
    let Some(pg_db) = resolve_datatable_pg(db, w_id, datatable).await? else {
        return Ok(());
    };
    let (client, connection) = pg_db.connect(Some(db)).await?;
    tokio::spawn(async move {
        let _ = connection.await;
    });
    client
        .execute(
            "DELETE FROM _wm_migrations WHERE datatable = $1",
            &[&datatable],
        )
        .await
        .map(|_| ())
        .or_else(ignore_missing_wm_migrations)
}

/// Relabel a data table's rows in its own database's `_wm_migrations`. `resolve_by`
/// names the config entry used to find the database (the old name, still present
/// pre-commit); `from`/`to` are the `datatable` column values to move between.
async fn remote_rename_datatable_migrations(
    db: &DB,
    w_id: &str,
    resolve_by: &str,
    from: &str,
    to: &str,
) -> Result<()> {
    let Some(pg_db) = resolve_datatable_pg(db, w_id, resolve_by).await? else {
        return Ok(());
    };
    let (client, connection) = pg_db.connect(Some(db)).await?;
    tokio::spawn(async move {
        let _ = connection.await;
    });
    client
        .execute(
            "UPDATE _wm_migrations SET datatable = $2 WHERE datatable = $1",
            &[&from, &to],
        )
        .await
        .map(|_| ())
        .or_else(ignore_missing_wm_migrations)
}

/// Keep migration bookkeeping in sync when data tables are renamed or deleted in
/// the workspace config: the control table `datatable_migrations` (this database,
/// in `tx`) and each data table's own `_wm_migrations` (its own database, keyed
/// by data table name — see [`ensure_wm_migrations_schema`]).
///
/// Renames are applied in two phases through a temporary key so that a rename
/// chain or a swap (A->B, B->A) can't transiently collide on the
/// (datatable, ...) uniqueness mid-update.
///
/// The `_wm_migrations` updates are best-effort: they run just before `tx`
/// commits (resolved via the pool, which still exposes the old names), and a
/// temporarily unreachable data-table database is logged rather than failing the
/// whole config edit. If one is missed, the next run re-applies its migrations
/// against the existing schema.
///
/// Returns every migration path the cascade moved or removed, so the caller can
/// deploy them once `tx` commits — otherwise a linked git repo keeps the files of
/// a deleted or pre-rename data table forever.
pub(crate) async fn cascade_datatable_migration_renames_and_deletes(
    db: &DB,
    tx: &mut Transaction<'_, Postgres>,
    w_id: &str,
    renames: &[DatatableRename],
    deleted_datatables: &[String],
) -> Result<Vec<String>> {
    let mut changed_paths: Vec<String> = vec![];

    if !deleted_datatables.is_empty() {
        let deleted = sqlx::query!(
            "DELETE FROM datatable_migrations WHERE workspace_id = $1 AND datatable = ANY($2::text[]) \
             RETURNING datatable, timestamp, name",
            w_id,
            deleted_datatables
        )
        .fetch_all(&mut **tx)
        .await?;
        changed_paths.extend(
            deleted
                .into_iter()
                .map(|m| format!("{}/{}_{}", m.datatable, m.timestamp, m.name)),
        );
    }

    for (i, r) in renames.iter().enumerate() {
        let tmp = format!("__wm_rename_tmp/{i}");
        sqlx::query!(
            "UPDATE datatable_migrations SET datatable = $3 WHERE workspace_id = $1 AND datatable = $2",
            w_id,
            &r.from,
            &tmp
        )
        .execute(&mut **tx)
        .await?;
    }
    for (i, r) in renames.iter().enumerate() {
        let tmp = format!("__wm_rename_tmp/{i}");
        let moved = sqlx::query!(
            "UPDATE datatable_migrations SET datatable = $3 WHERE workspace_id = $1 AND datatable = $2 \
             RETURNING timestamp, name",
            w_id,
            &tmp,
            &r.to
        )
        .fetch_all(&mut **tx)
        .await?;
        for m in moved {
            // Both ends: the old path so its files are dropped from the repo, the
            // new one so they reappear under the renamed data table.
            changed_paths.push(format!("{}/{}_{}", r.from, m.timestamp, m.name));
            changed_paths.push(format!("{}/{}_{}", r.to, m.timestamp, m.name));
        }
    }

    for name in deleted_datatables {
        if let Err(e) = remote_forget_datatable_migrations(db, w_id, name).await {
            tracing::warn!("Failed to clear _wm_migrations for deleted data table {name}: {e}");
        }
    }
    for (i, r) in renames.iter().enumerate() {
        let tmp = format!("__wm_rename_tmp/{i}");
        if let Err(e) = remote_rename_datatable_migrations(db, w_id, &r.from, &r.from, &tmp).await {
            tracing::warn!(
                "Failed to stage _wm_migrations rename {} -> {}: {e}",
                r.from,
                r.to
            );
        }
    }
    for (i, r) in renames.iter().enumerate() {
        let tmp = format!("__wm_rename_tmp/{i}");
        if let Err(e) = remote_rename_datatable_migrations(db, w_id, &r.from, &tmp, &r.to).await {
            tracing::warn!(
                "Failed to finish _wm_migrations rename {} -> {}: {e}",
                r.from,
                r.to
            );
        }
    }

    Ok(changed_paths)
}

/// Deploy the migration paths a data table rename/delete moved or removed, so a
/// linked git repo drops the stale files and picks up the renamed ones. Call
/// after the config transaction commits — the deploy reads the workspace export.
pub(crate) async fn record_datatable_cascade_deployments(
    authed: &ApiAuthed,
    db: &DB,
    w_id: &str,
    changed_paths: Vec<String>,
) -> Result<()> {
    if changed_paths.is_empty() {
        return Ok(());
    }
    let objs = changed_paths
        .into_iter()
        .map(|path| DeployedObject::DatatableMigration { path })
        .collect();
    handle_deployment_metadata_batch(
        &authed.email,
        &authed.username,
        db,
        w_id,
        objs,
        Some("Data table migrations updated by a data table rename or deletion".to_string()),
    )
    .await
}

/// Copy a workspace's migration definitions to another workspace, so a fork
/// inherits the same per-data-table migration history as its parent.
pub(crate) async fn clone_datatable_migrations(
    tx: &mut Transaction<'_, Postgres>,
    source_workspace_id: &str,
    target_workspace_id: &str,
) -> Result<()> {
    sqlx::query!(
        "INSERT INTO datatable_migrations (workspace_id, datatable, timestamp, name, code_up, code_down)
         SELECT $2, datatable, timestamp, name, code_up, code_down
         FROM datatable_migrations WHERE workspace_id = $1",
        source_workspace_id,
        target_workspace_id,
    )
    .execute(&mut **tx)
    .await?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn validate_migration_name_accepts_safe_names() {
        for name in ["initial", "add_index_to_customers", "fix-bug_2", "ABC123"] {
            assert!(
                validate_migration_name(name).is_ok(),
                "{name} should be valid"
            );
        }
    }

    #[test]
    fn validate_migration_name_rejects_unsafe_names() {
        for name in [
            "",
            "add index",
            "a/b",
            "a\\b",
            "a..b",
            "a.b",
            "naïve",
            "a/../b",
        ] {
            assert!(
                validate_migration_name(name).is_err(),
                "{name} should be rejected"
            );
        }
    }

    #[test]
    fn validate_datatable_path_segment_accepts_and_rejects() {
        for ok in ["mydt", "my-dt", "main", "a b"] {
            assert!(
                validate_datatable_path_segment(ok).is_ok(),
                "{ok} should be ok"
            );
        }
        for bad in ["", "a/b", "a\\b", "..", "a..b", "../etc", "x/.."] {
            assert!(
                validate_datatable_path_segment(bad).is_err(),
                "{bad} should be rejected"
            );
        }
    }

    // A new name must survive the git-sync path matchers; an existing one is only
    // held to the escape rule, so a workspace carrying a legacy name can still save
    // its settings and rename its way out.
    #[test]
    fn validate_new_datatable_name_requires_a_sync_safe_charset() {
        for ok in ["mydt", "my-dt", "my_dt.v2", "main"] {
            assert!(validate_new_datatable_name(ok).is_ok(), "{ok} should be ok");
        }
        for bad in [
            "a b", "a*b", "a[b]", "a{b}", "a?b", "café", ".staging", "-dt", "a/b", "",
        ] {
            assert!(
                validate_new_datatable_name(bad).is_err(),
                "{bad} should be rejected"
            );
            if bad != "a/b" && !bad.is_empty() {
                assert!(
                    validate_datatable_path_segment(bad).is_ok(),
                    "{bad} should still be storable once persisted"
                );
            }
        }
    }

    #[test]
    fn parse_datatable_migration_diff_path_roundtrips() {
        assert_eq!(
            parse_datatable_migration_diff_path("mydt/20260101000001_create_users.up.sql"),
            Some(("mydt".to_string(), 20260101000001))
        );
        // up and down map to the same (datatable, timestamp) record.
        assert_eq!(
            parse_datatable_migration_diff_path("mydt/20260101000001_create_users.down.sql"),
            Some(("mydt".to_string(), 20260101000001))
        );
        // datatable names may themselves be hyphenated.
        assert_eq!(
            parse_datatable_migration_diff_path("my-dt/42_x.up.sql"),
            Some(("my-dt".to_string(), 42))
        );
    }

    #[test]
    fn parse_datatable_migration_diff_path_rejects_malformed() {
        // no slash → not a migration path
        assert_eq!(parse_datatable_migration_diff_path("nofile"), None);
        // filename not starting with digits → no timestamp
        assert_eq!(
            parse_datatable_migration_diff_path("mydt/create_users.up.sql"),
            None
        );
        // empty filename
        assert_eq!(parse_datatable_migration_diff_path("mydt/"), None);
    }

    async fn seed_migration(pool: &DB, w_id: &str, datatable: &str, timestamp: i64, name: &str) {
        sqlx::query(
            "INSERT INTO datatable_migrations (workspace_id, datatable, timestamp, name, code_up) \
             VALUES ($1, $2, $3, $4, 'select 1;')",
        )
        .bind(w_id)
        .bind(datatable)
        .bind(timestamp)
        .bind(name)
        .execute(pool)
        .await
        .unwrap();
    }

    async fn migration_keys(pool: &DB, w_id: &str) -> Vec<(String, String)> {
        sqlx::query_as::<_, (String, String)>(
            "SELECT datatable, name FROM datatable_migrations WHERE workspace_id = $1 \
             ORDER BY datatable, timestamp",
        )
        .bind(w_id)
        .fetch_all(pool)
        .await
        .unwrap()
    }

    // The cascade keeps each data table's migrations attached to its name when a
    // data table is renamed, drops them when it is deleted, survives a swap
    // (A->B, B->A) at a shared timestamp without a primary-key collision, and
    // reports both ends of every move so git sync drops the stale files.
    #[sqlx::test(migrations = "../migrations")]
    async fn cascade_renames_and_deletes_datatable_migrations(pool: DB) {
        let w_id = format!("dtmig{}", uuid::Uuid::new_v4().simple());
        sqlx::query("INSERT INTO workspace (id, name, owner) VALUES ($1, $1, 'test@windmill.dev')")
            .bind(&w_id)
            .execute(&pool)
            .await
            .unwrap();

        // rename a -> a2, delete d, and swap sa <-> sb (both at timestamp 5000)
        seed_migration(&pool, &w_id, "a", 1, "a_mig").await;
        seed_migration(&pool, &w_id, "d", 1, "d_mig").await;
        seed_migration(&pool, &w_id, "sa", 5000, "sa_mig").await;
        seed_migration(&pool, &w_id, "sb", 5000, "sb_mig").await;

        let mut tx = pool.begin().await.unwrap();
        let mut changed = cascade_datatable_migration_renames_and_deletes(
            &pool,
            &mut tx,
            &w_id,
            &[
                DatatableRename { from: "a".to_string(), to: "a2".to_string() },
                DatatableRename { from: "sa".to_string(), to: "sb".to_string() },
                DatatableRename { from: "sb".to_string(), to: "sa".to_string() },
            ],
            &["d".to_string()],
        )
        .await
        .unwrap();
        tx.commit().await.unwrap();

        assert_eq!(
            migration_keys(&pool, &w_id).await,
            vec![
                ("a2".to_string(), "a_mig".to_string()),
                ("sa".to_string(), "sb_mig".to_string()),
                ("sb".to_string(), "sa_mig".to_string()),
            ]
        );

        changed.sort();
        assert_eq!(
            changed,
            vec![
                "a/1_a_mig".to_string(),
                "a2/1_a_mig".to_string(),
                "d/1_d_mig".to_string(),
                "sa/5000_sa_mig".to_string(),
                "sa/5000_sb_mig".to_string(),
                "sb/5000_sa_mig".to_string(),
                "sb/5000_sb_mig".to_string(),
            ]
        );
    }

    // A fork inherits its parent's migration definitions unchanged.
    #[sqlx::test(migrations = "../migrations")]
    async fn clone_copies_datatable_migrations_to_target(pool: DB) {
        let src = format!("src{}", uuid::Uuid::new_v4().simple());
        let dst = format!("dst{}", uuid::Uuid::new_v4().simple());
        for w in [&src, &dst] {
            sqlx::query(
                "INSERT INTO workspace (id, name, owner) VALUES ($1, $1, 'test@windmill.dev')",
            )
            .bind(w)
            .execute(&pool)
            .await
            .unwrap();
        }
        seed_migration(&pool, &src, "customers", 1, "create_customers").await;
        seed_migration(&pool, &src, "customers", 2, "add_index").await;
        seed_migration(&pool, &src, "orders", 3, "create_orders").await;

        let mut tx = pool.begin().await.unwrap();
        clone_datatable_migrations(&mut tx, &src, &dst)
            .await
            .unwrap();
        tx.commit().await.unwrap();

        // the target ends up with an identical set, and the source is untouched.
        let expected = vec![
            ("customers".to_string(), "create_customers".to_string()),
            ("customers".to_string(), "add_index".to_string()),
            ("orders".to_string(), "create_orders".to_string()),
        ];
        assert_eq!(migration_keys(&pool, &dst).await, expected);
        assert_eq!(migration_keys(&pool, &src).await, expected);
    }
}