mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-09-22 16:02:38 +00:00
Diego ImbertandClaude Opus 5
030254af47
The tenant cascade on leaving went onto `/users/leave`. The UI and the generated client call `/workspaces/leave` — a different handler in a different crate with the same name — which deleted the membership and left `u/<username>` in the tenant lists. Leaving and rejoining therefore restored the access the leave was supposed to end, and a later account taking the username would have inherited it. The regression test drives the route the client actually calls; without the fix it fails with "leaving kept the tenant". The migration endpoints authorized too late. `run_datatable_migrations` opened the data table's admin connection, created `_wm_migrations` and read it before reaching the per-migration role check — so with nothing pending, nothing was checked at all. Rollback returned before its check when nothing was applied, and the status endpoint had none. All three now ask, before any connection is opened, whether the caller can reach the data table as any role at all; which role a given migration runs as is still decided per migration, and by the executor after that. Deleting a role committed the cluster drop and the catalog row, then swept the tenant lists in separate transactions. A sweep failing part-way left workspaces naming a role nothing can connect as, while the retry answered `NotFound` because the catalog entry was already gone. The sweep now runs in the same transaction, so the drop, the row and every tenant list commit together. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012ti5HyeTikPMYyW8YSdiHR
2143 lines
75 KiB
Rust
2143 lines
75 KiB
Rust
/*
|
|
* 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::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::datatable_roles::ADMIN_DATATABLE_ROLE;
|
|
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.
|
|
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(×tamp) {
|
|
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);
|
|
}
|
|
}
|