Files
windmill/backend/windmill-api-workspaces/src/workspaces_extra.rs
T
0e807fb1dd feat: put a data table's connection under Postgres roles (#11020)
* feat(datatables): put a data table's connection under Postgres roles

A data table backed by the instance database resolved to exactly one Postgres connection,
`custom_instance_user`, for everyone who could reach it at all. There was no way to say
this job reads, that one writes, this one never sees the salaries table.

A data table role is now a real Postgres login on the cluster, defined once for the
instance by a superadmin and named exactly as they named it. A script that declares
`-- role analytics` connects as `analytics`, and Postgres decides what it may touch —
grants are ordinary SQL. Windmill answers only "may this caller ask for this role", from
the tenant lists on the data table entry: `u/alice`, `g/analysts`, `f/finance` or `*`.
A data table with no `permissions` block behaves exactly as before.

Everything that opens a connection on someone's behalf goes through one chokepoint,
`get_datatable_resource_from_db`, which takes the identity explicitly and fails closed when
there is none. The role logs in as itself — never `SET ROLE`, which a script could
`RESET ROLE` its way out of.

A fork's data table entry becomes a pointer at the workspace that governs it rather than a
copy of it. The settings clone used to hand a fork a byte-identical entry naming the
parent's database, which a fork admin could edit to grant themselves `admin` there; a
pointer has nothing local to edit, and its tenants are evaluated as a member of the
governing workspace, by email. `permissions` is stripped from the workspace export and
ignored on import: tenants name principals of one workspace, and a settings push is not
where an access decision should be made.

Operations that see the whole database whatever the roles grant stay with the governing
workspace's admins: editing the roles, a migration that declares none, and opening a
replication stream for a Postgres trigger or capture.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012ti5HyeTikPMYyW8YSdiHR

* fix(datatables): gate the paths that reach a whole database as admin

Auditing what still resolved through the unchecked resolver turned up three that act for a
caller and hand back the admin connection: `resolve_pg_source_checked` (behind schema
export, the full-schema read, database creation, import and the forked-database drop), the
connection test, and the schema snapshot a fork clone takes of its parent. On a data table
under roles each let any workspace member — or a fork admin who is nobody in the governing
workspace — read or copy the whole database whatever its roles grant.

All three now require admin reach on the governing workspace. A dump taken under a
restricted role would be a silently truncated copy rather than an error, so refusing is the
only right answer for the copy paths.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012ti5HyeTikPMYyW8YSdiHR

* fix(datatables): confine roles to the instance database, and stop a fork reaching the parent's bookkeeping

A data table role is a login on Windmill's own Postgres. Nothing stopped a workspace admin
putting a *resource-backed* data table under roles, at which point the executor dialled the
host that resource names — one the admin chose — with the role's real cluster password, and
`CONNECT` is granted to every registered instance database. Both ends now refuse: the
permissions endpoint rejects the save, and the chokepoint refuses to substitute credentials
on a non-instance entry rather than trusting the record it read.

Two more places reached the governing database without answering to it. The initial-migration
generator returned a `pg_dump` of the whole schema to any member. And the migration
rename/delete cascade followed a fork's pointer into the parent, so a fork admin renaming or
removing their own local entry relabelled or wiped the parent's `_wm_migrations` — after
which the parent re-runs every migration from zero. The remote half is now skipped when the
entry resolves into another workspace, which is also just correct: a fork renaming what it
calls a data table changes nothing about the data table.

Also: revoking a tenant now bounces the replication streams of every workspace holding an
entry that resolves here, not only the governing one, so a fork's trigger stops rather than
living on inside its open connection; the instance role catalog and the governing workspace's
tenant lists are no longer returned to someone who cannot edit them; and the tenant rename
dedup collapses non-adjacent duplicates, per role rather than once any role changed.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012ti5HyeTikPMYyW8YSdiHR

* fix(datatables): fail loudly where a role or a pointer can be left half-recorded

Three ways the feature could end up in a state nobody could see or undo.

Creating a role writes the cluster first and the catalog second, but the catalog write was an
`UPDATE` that matched nothing when the instance Postgres settings row was absent — leaving a
live login with a password nobody recorded: invisible to the catalog, un-recreatable because
the name is taken, and un-deletable because there is no entry to delete. It now errors, so
the operation is retryable once the row is restored.

Deleting a workspace only nulls the fork lineage; the data table entries pointing at it are
left resolving to nothing. Sweeping them is not an option — turning a pointer back into a copy
would hand each fork the database outright — so the delete now names the data tables it
stranded, and resolving one says which workspace is missing rather than reporting a data table
this workspace never had.

`InstanceDatatableRole` derived `Debug` while holding a Postgres password; it is now
hand-written so `{:?}` on the catalog cannot put a live credential in a log line.

Adds the two branches the reviews found unpinned: a caller who is not a member of the
governing workspace at all, and `NoIdentity` — the compatibility path for an agent worker that
predates this and sends no job id.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012ti5HyeTikPMYyW8YSdiHR

* fix(datatables): unbreak two operator messages and two comments that described other code

The two strings this branch added for states an operator hits once — the catalog write that
matched nothing, and the delete that stranded a pointer — were collapsed from their multi-line
form with the indentation left in, so both rendered with a fourteen-space gap mid-sentence.

`list_datatables` claimed to report a chain it cannot follow and then dropped it; it does drop
it, and the comment now says why that is the right place to stay quiet. The non-superadmin
check in `edit_datatable_config` was introduced as also covering references, which it does not
and need not: `reference` is overwritten from the stored entry for every caller before the
check runs.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012ti5HyeTikPMYyW8YSdiHR

* fix(datatables): serialize role catalog mutations, and state each helper's authorization contract

The catalog is one JSON document, so create, rename, enable and delete are all
read-modify-write. Two concurrent creates read the same snapshot, both succeed in the
cluster, and the second write drops the first — leaving a live Postgres login with a password
nobody recorded, which is the exact state the delete path exists to prevent. Every mutation
now runs in one transaction holding an advisory lock across the read, the cluster DDL and the
write, so a lost update cannot happen and a failure rolls the whole thing back. The DDL
helpers take that transaction rather than the pool, which is what makes the lock cover them.

Their statements moved off `sqlx::raw_sql`: the simple protocol is only needed for genuinely
multi-statement SQL, and its future is not `Send`, which an axum handler holding the
transaction requires. Each of these is one statement anyway.

The new cross-crate surface now says what callers must do. `read_role_catalog` returns
plaintext credentials; `create`/`rename`/`set_login`/`drop_instance_role` and
`converge_connect_grants` mutate cluster-wide state; `read_datatable_entry` reads a workspace's
raw config. All of them are superadmin-gated by their current handlers, but nothing said so at
the definition, which is where the next caller looks.

Also: the roles table reloads after a failed login toggle instead of leaving it claiming a flip
that did not land; the rename affordance is the design-system `Button`, not a raw one; and
`resolve_datatable_pg_as_caller` drops a `role` parameter no caller ever filled — browsing
resolves as the data table's default until the database manager grows a picker.

Why role passwords stay a plain `String` while the instance user's password beside them is a
`StringOrSecretRef`, asked three times across reviews: that one is a secret ref because an
operator supplies it and may want it from their own backend, while these are minted here and
never entered by anyone, so there is nothing for a ref to point at. Encrypting generated
secrets at rest is a separate change that would take the replication password with it. Now
said at the field.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012ti5HyeTikPMYyW8YSdiHR

* fix(datatables): give the role catalog its own row, out of reach of the config machinery

Putting it inside `custom_instance_pg_databases` was the wrong call, and it cost two ways.
The catalog serializes a generated Postgres password per role, and that row is the
operator-facing instance config, so the passwords reached `get_instance_config` and its YAML
editor — a live cluster credential in a response body, a UI field and any log of either.
Worse in the other direction: `to_settings_map` strips the catalog, so a full-row upsert of
that key writes the row back without it and the catalog is gone, while the cluster keeps every
login it described.

`custom_instance_replication_pwd` is the precedent and says exactly why — a generated secret,
written only by the server, never operator-authored, hidden so the config machinery cannot
read, rewrite or drop it. The catalog is the same thing, so it now has the same shape:
`datatable_roles`, in `HIDDEN_SETTINGS`, `PROTECTED_SETTINGS` and the agent-worker denylist.
No redaction to keep in step with three code paths, and no way for a neighbouring write to
take it out.

Two races on the same shared documents. `edit_datatable_config` read the stored data tables
outside its transaction and then wrote the whole `datatable` document, so a permissions save
committing in between was silently rolled back; it now reads under `FOR UPDATE`. And
`set_datatable_permissions` validated role ids against the catalog before opening its
transaction, so a deletion in between let it write a deleted role back — including as the
default, which every later job then fails on; it now holds the catalog lock and the settings
row across validation and write.

Completes the authorization contracts the previous commit claimed but did not finish:
`read_datatable_entry` (which it named and missed), `resolve_governing_datatable`, whose whole
job is to answer for a workspace the caller may not belong to, and
`converge_connect_grants_with`, which had not inherited its wrapper's.

Also the generic Python SDK reference: `_format_py_params` learned the bare `*` last time, but
`extract_py_functions` is a second formatter and still rendered `datatable(name, role)`, so
code written from that page passed a keyword-only argument positionally.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012ti5HyeTikPMYyW8YSdiHR

* fix(datatables): make the concurrency test pin the handlers, and the contracts describe what is enforced

The concurrency test reimplemented the read-modify-write inline, so deleting the lock from all
three handlers left it green — it pinned Postgres, not the code it was written for. It now
drives `create_datatable_role` twice concurrently and asserts the catalog kept both names.
Checked the way the last one should have been: removing the lock from the handler makes it
fail with "wmtest_a_… is a live cluster login the catalog forgot".

The contracts added last commit were stricter than this PR's own callers, which is worse than
none — the next reader sees a rule already broken and learns to ignore it.
`read_role_catalog` said superadmin-only while two of its four callers are open to any
workspace member, and `converge_connect_grants` said superadmin while
`set_datatable_permissions` reaches it as a workspace admin. Both were fine on substance: the
rule that actually holds is about the credential never reaching a response, log, audit record
or export, not about who may call. They now say that. `read_datatable_entry` gets the same
treatment rather than the one the earlier message claimed for it: it is the primitive every
resolution goes through, so it is deliberately open, and what must not escape is `permissions`
— it names the governing workspace's users, groups and folders.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012ti5HyeTikPMYyW8YSdiHR

* fix(datatables): close the last ways a role or a pointer can be left pointing at nothing

The raw settings readers hand back whatever is in the row, so moving the catalog into its own
`global_settings` key protected the config machinery and left `GET /settings/global/datatable_roles`
and the settings listing returning every live password. Both now filter that one key. The
neighbouring `custom_instance_replication_pwd` has the same shape and is not touched here: it
predates this and widening the fix to it is a decision about an operator workflow, not a
consequence of this change.

Three ways a save could leave something resolving to nothing:

A permissioned data table could be moved to a PostgreSQL resource. The block was carried across
as a server-owned field, the runtime refuses roles on a resource-backed table, so the save
succeeded and every job afterwards failed. Refused instead — turning roles off first is one step,
and it keeps discarding an access decision something somebody chose.

Renaming a governing data table left every fork pointing at the old name: the data table
disappears from their pickers and their jobs stop, with nothing in the renaming workspace to
suggest why. The rename now follows into the pointers in the same transaction.

Deleting one cannot be followed the same way, so it is reported instead — the response names what
it stranded, the way deleting a workspace does, and the fork's own error already says which
workspace is gone.

Also: `ensure_instance_db_grant_options_unchecked` claimed superadmin while the permissions
handler reaches it as a workspace admin (the same class fixed last commit, one instance missed);
the role entry kept an `instance_config_schema` derive it no longer needs; `write_role_catalog`
was the one writer of that table not stamping `updated_at`; and the concurrency test dropped its
roles only on success — a failing run is exactly the one that creates them without recording them.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012ti5HyeTikPMYyW8YSdiHR

* refactor(datatables): put the role catalog in its own table, not in global_settings

Five findings across three rounds were all the same choice. A set of live Postgres credentials
was living in `global_settings`, which has generic read, list, write, config-export and CLI
round-trip paths that know nothing about what they carry: the passwords reached the instance
config and its YAML editor, a full-row upsert of a neighbouring key erased the catalog,
`GET /settings/global/{key}` and the settings listing returned them raw, and this round the
redaction that fixed the last two turned `wmill instance push` into something that wipes every
password — a fix breaking the assumption the previous fix made. `POST /settings/global/datatable_roles`
could also empty it outside the lock.

The approved plan offered a table or `global_settings`, so this is the other option it already
allowed rather than a new design. `datatable_role` is a table: no generic settings path can read
it, list it, export it, write it or round-trip it, so none of the five needs a guard. The
redaction, the hidden/protected/agent-denylist entries and the JSON document all go with it.

One row per role also removes the read-modify-write the concurrency work was about: two
concurrent creates are two inserts, and the unique index on `name` is what settles a collision.
The advisory lock stays for the one window rows do not cover — `CREATE ROLE` is invisible to
another transaction until commit, so without it both creates pass their `pg_roles` check.

Also from this round: rename mappings are checked against the configuration they claim to
describe, since fork pointers are rewritten from them — a caller could otherwise submit
`main -> missing` against an unchanged config and repoint every fork of `main` at a name nothing
has, and `A -> B` plus `B -> C` moved what pointed at `A` all the way to `C`. And the warning
naming forks a delete stranded reached the response but not the screen: both the data table
settings save and the workspace delete now show it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012ti5HyeTikPMYyW8YSdiHR

* fix(datatables): validate a rename against the save it describes, and re-check under the locks

Three from the round, all about deciding on state that could already have moved.

A permission save resolved the data table and checked it was instance-backed before taking any
lock, then wrote under one. A config save committing in between could move the table onto a
PostgreSQL resource — recreating exactly what the transition guard refuses — or rename it, in
which case the write targeted a key that no longer existed and reported success having changed
nothing. It now re-resolves and re-checks on the locked state.

Rename validation checked that the source existed before and the target existed after, which
still accepts `main -> decoy` against a save that keeps both: every fork of `main` then follows
onto a different data table, silently, because it keeps resolving. The rule is now the actual
old-to-new key transition — a source may only survive if another rename took its name, and a
target may only pre-exist if another rename freed it. That also stops two sources sharing one
target, and it admits a swap, which the previous guard refused: `datatables` is keyed by name, so
a swap cannot be done one save at a time, and refusing it was a regression against main. The
pointer cascade now runs in two passes through a temporary name, the way the migration cascade
one layer down already handles the same shape, so `A -> B` with `B -> C` moves each pointer once
from what it named before the save.

The tenant mutators say what they are for: they write an access decision for any workspace named,
with an arbitrary mutation, and exist for the transaction that frees or renames a principal.
Editing a decision on purpose belongs in the permissions endpoint.

Carried in the same change: the stranded-fork list is a field rather than a phrase to grep out of
a success string; the pointer cascade matches with `EXISTS` instead of a `LIKE` over the whole
document, so a workspace whose pointers name something else is not rewritten to a byte-identical
value under an exclusive lock; and `InstanceDatatableRole` drops the serde derives left over from
the JSON document, one of which would emit `pwd`.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012ti5HyeTikPMYyW8YSdiHR

* fix(datatables): cascade on the leave route that is used, gate migrations before the admin connection, and drop a role atomically

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

* fix(datatables): refuse to copy a data table that is under roles

pg_dump carries no roles and the import runs with --no-privileges, so a copied
data table arrives owned by the admin connection with no GRANT for any role.
The settings clone brings `permissions` across, so the fork's tenants pass
Windmill's check, connect as the role they were given, and are denied by
Postgres on everything: an entry that reads as configured and answers nothing.

Refuse the copy — in the import endpoint before any data moves, and in the fork
path the CLI takes. Replaying the source's owners and ACLs into the clone is
what lifts this, and is a change of its own. Dropping `permissions` from the
copy instead would be the unsafe half, since the copy holds the parent's rows.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012ti5HyeTikPMYyW8YSdiHR

* fix(datatables): refuse the clone's database too, not only its data

A clone is two endpoints: `create_pg_database` then `import_pg_database`. Only
the second refused a data table under roles, so a fork asking to clone one
created and registered an empty `wm_fork_…` instance database and then failed —
and nothing collects it, since `drop_forked_datatable_databases` only drops
entries carrying `forked_from` and no entry names this one.

Refuse in both, so the clone stops before a database exists.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012ti5HyeTikPMYyW8YSdiHR

* nit worker error msg

* fix pg_dump stuck on version 17 on nix

* fix(datatables): refuse a malformed role annotation instead of ignoring it

`-- Role operator`, `-- role operator;` and `-- role operator -- why` all failed
the annotation parser's exact-match rule, so the query fell through to the data
table's default role and ran, silently, under a login the author did not choose.
Naming a role exists precisely to not do that.

A leading comment whose first word is `role` is now an annotation attempt: the
keyword matches case-insensitively, one trailing `;` is tolerated, and anything
else is an error naming the line. Only callers that already know the target is a
`datatable://` reference ever run this, so ordinary SQL keeps its comments.

Also bumps the dev shell's postgres client to 18 — it trailed the server the dev
database runs, which takes out every data table export, clone and fork-with-data.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012ti5HyeTikPMYyW8YSdiHR

* fix(datatables): refuse a malformed role query string instead of ignoring it

`?Role=analytics`, `?role=` and `?x=1&role=…` all fell through the reference
parser's exact-match rule, so the connection resolved to the data table's default
role and ran under a login the caller never asked for — the URI half of the same
trap as a malformed `-- role` annotation.

The key now matches case-insensitively, and anything else in the query string is
an error naming it; `role` is the only parameter a reference takes. Callers that
only need the entry keep a lenient `datatable_ref_name`, since they never act on
the role. The DuckDB `ATTACH` parser propagates it rather than attaching under
the default.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012ti5HyeTikPMYyW8YSdiHR

* fix(datatables): carry the role annotation into the row_to_json retry

The retry rebuilds its SQL from `pruneComments(code)`, so the leading comment
block never reached the second attempt — and with it the `-- role <name>` line
that decides which login the query runs as. The retry connected as the data
table's default role instead, so a query the first attempt was denied could
succeed on the second, reported as "recovered with the row_to_json fix".

Carry the leading comment block over. The retry itself is unchanged.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BjfMkJyKzodxkobqGZ6Lqb

* chore(datatables): don't mount the roles UI until the ACL editor lands

Enforcement ships first. The permissions drawer is what turns roles on, and the
catalog section is what creates them — both are only useful once there is a way
to grant a role the privileges it needs, which arrives with the ACL editor. Left
mounted they would offer a feature whose other half does not exist.

The two components are complete and reviewed; only their call sites here are
commented out, with a note pointing the follow-up PRs at them.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BjfMkJyKzodxkobqGZ6Lqb

* fix(datatables): honour `-- role: x`, and fix the DuckDB attach test

Two review findings, both real.

`attach_datatable_parses_name_and_role` never compiled: `parse_attach_datatable`
returns `Result<Option<_>>` now and one call site kept a single `unwrap`. Its
`?Role=analytics` case also asserted a refusal, contradicting the parser in the
same commit, which matches the key case-insensitively. Replaced with the cases
that are genuinely malformed, and a positive one for the cased key.

`-- role: analytics` fell through to the default role — the silent fallback the
strict parser exists to remove, for the spelling most likely to be typed. The
keyword now accepts an optional colon, attached or spaced, while a word that
merely starts with it (`rolebased`) is still not an attempt.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BjfMkJyKzodxkobqGZ6Lqb

* fix(datatables): clone a fork's pointer instead of failing after the copy

Forking a fork with cloning left an orphan database. The preflight resolves the
pointer and sees the governing entry, so both endpoints ran and filled the new
database; `apply_forked_datatable` then refused the inherited pointer and rolled
the fork back, stranding a registered `wm_fork_*` that no entry names and whose
name blocks the retry.

Refusing earlier would have been the smaller change, but forking a fork and
cloning worked before pointers existed, so it would trade an orphan for a
regression. Resolve what the pointer names and write the terminal entry the
clone needs: the whole `database` object rather than a patch of its
`resource_path`, since a pointer has none, and `reference` removed with it.

Also accepts `-- role=x` and `-- Role = x`, two more spellings that fell through
to the default role.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BjfMkJyKzodxkobqGZ6Lqb

* fix(datatables): refuse to roll back the catalog while roles exist

The down migration dropped the table and left every role behind: live Postgres
logins whose passwords only that table carried, so after a revert Windmill could
neither use, disable nor delete them, and re-applying could not recreate them
because the names were taken. Cleaning up here is not possible either — dropping
a role means reassigning what it owns in every instance database, and a
migration runs in one — so it now refuses while the catalog is non-empty and
says to delete the roles through instance settings, which does the cluster work.

Also enforces the instance-only invariant the resolved-pointer clone relies on
rather than only asserting it in a comment.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BjfMkJyKzodxkobqGZ6Lqb

* refactor(datatables): settle clonability in one place, before anything is created

A clone is three stages a workspace apart — `create_pg_database`, then
`import_pg_database`, then `apply_forked_datatable` inside the fork transaction.
Only the third can roll back, and `CREATE DATABASE` is not transactional, so any
refusal that lives there strands a registered `wm_fork_*` that no entry names
and whose name blocks the retry.

That orphan has now been fixed three times, most recently reintroduced by a
guard added one commit ago. Patching each new refusal into the first endpoint is
not the fix; having two places that can refuse is. `ensure_datatable_is_clonable`
now answers every reason a copy can be refused and returns what it resolved, and
the stage that writes the entry only does the work.

Also takes an ACCESS EXCLUSIVE lock before the rollback guard counts, so a role
created concurrently cannot slip between the check and the drop.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BjfMkJyKzodxkobqGZ6Lqb

* fix(datatables): let a retried clone reclaim its own leftover database

A clone creates its target database one request before it copies into it, and
the fork that would name it is written a request after that. Any failure in
between — a pg_dump error, a bad restore, a dropped connection, the source's
roles changing mid-flow — left a registered `wm_fork_*` that no entry names,
and every retry then failed on its name. This predates data table roles.

`create_pg_database` now reclaims such a leftover before creating: only a
`wm_fork_*` database Windmill registered as a data table database and that no
data table or ducklake entry names, in any workspace, archived ones included.
The drop never terminates connections, so a clone still copying into it makes
the reclaim fail instead of being cut off. It is limited to callers who
administer the source — reaching it is not enough, since on a data table
without roles every member reaches it — and anyone else gets the refusal an
existing database always got.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* Revert "fix(datatables): let a retried clone reclaim its own leftover database"

This reverts commit 7dd3275a10.

The reclaim tied the caller to the source they administer, but not to the
database it dropped. Between another workspace's import and its final fork
request, that workspace's target is full, registered, unnamed and has no open
connection, so an admin of any instance data table could name it and have it
dropped and recreated empty. The victim's fork would then commit pointing at
the empty copy. Safe reclaim needs durable clone ownership and serialization
with the request that names the database; until then the leftover stays, as it
did before this PR.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* docs(datatables): record the stale clone database as a known limitation

A clone is three requests and `CREATE DATABASE` is not transactional, so a
failure after the first leaves a registered `wm_fork_*` behind, as it did
before data table roles. Accepted for this PR: it is harmless to data and goes
away once the clone is a single server-side operation.

The comment also records why the obvious fix is wrong: reclaiming the leftover
on retry, without durable clone ownership, can drop another workspace's fully
copied database between its import and its final fork request.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(datatables): bounce the streams reading a data table when it is deleted

Deleting a governing data table, or the workspace that holds it, only collected
the fork pointers it stranded, for the warning. A Postgres trigger or capture
already streaming through one of those pointers kept the replication connection
it opened while the pointer still resolved, so it went on dispatching the
governing database's rows after the fork lost access — until its connection
happened to restart. The governing workspace's own streams on a deleted entry
did the same.

Both deletion paths now bounce the affected listeners inside their own
transaction, through the helper a permission change already uses, so a
listener that reconnects re-resolves the entry and finds it gone. The helper is
split so a caller can pass the (workspace, local name) pairs it already holds.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(datatables): keep the fork schema baseline, and bounce streams on every removal

Three fixes from review.

`edit_datatable_config` took `forked_from` wholesale from the stored entry, so
the fork schema diff's save of an advanced baseline was silently discarded and
an applied change was offered again. Whether an entry carries a clone stamp is
still carried from the store, since that is what marks its database droppable,
but the baseline inside it is now taken from the request.

The stranded-pointer warning and the stream bounce ran over the optional
`deleted_datatables` hint, which the settings-sync CLI never sends, so removing
a governing data table through `wmill` bounced nothing. Removals are now derived
from the stored configuration against the saved one.

`delete_workspace` read the pointers to bounce before its transaction, so a fork
committing a pointer during the deletion was missed. The read now happens inside
the transaction, after the workspace row is deleted: a fork's insert key-share
locks that row through its parent foreign key, so it is either seen or fails on
the missing parent.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* refactor(datatables): keep Postgres triggers and data table roles apart

A replication stream reads every row of every table whatever the data table's
roles grant, and its listener checks access only when it connects. Rather than
chase every way access can change and bounce the streams each one affects, a
data table now carries one or the other:

- a Postgres trigger or capture cannot be created on, or connect to, a data
  table under roles;
- roles cannot be turned on while an enabled trigger or a live capture reads
  the data table, its own or a fork's through its pointer. The refusal names
  each one to disable.

This removes the stream bounces on roles edits and on data table and workspace
deletion, and the trigger gate that admitted admins. The fork schema baseline
fix from the same review round is kept.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(datatables): refuse a Postgres trigger on a data table under roles when it is saved

Creating or editing a trigger that points at a data table under roles was
accepted, and its listener then retried the refused connection every 30
seconds forever. The save is now refused, and a trigger that reaches such a
data table anyway (re-enabled, or cloned into a fork) is disabled by its
listener with the reason, as a missing replication slot is.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(datatables): disable a data table role before deleting it

Deleting a role reassigns and drops what it owns in each registered
database on its own connection, and each of those passes commits as it
goes. A database failing part-way left the role enabled in the catalog and
able to log in, but already stripped in the databases reached before it.
The role is now disabled in its own commit first, so a failed delete
leaves a disabled role to retry.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(datatables): serialize roles going on with a stream starting

Turning roles on looked for enabled triggers and live captures once,
without a lock anything starting a stream also took. A trigger enabled in
that window could have its listener connect before roles committed, and a
healthy listener never checks again. Both transitions now serialize on one
advisory lock: roles going on hold it exclusive while they look, and
trigger create, edit and enable, and capture setup and ping hold it shared
while they commit. Either the look sees the stream, or the listener
connects after roles are committed and refuses.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(datatables): wait out live listeners, and resolve stored names containing `?`

Turning roles on counted a trigger as gone once disabled, and a capture
once its client stopped pinging, but the listener keeps its replication
connection until its next heartbeat notices. A trigger or capture whose
listener pinged in the last 15 seconds, the window a server holds a
listener for, now still counts as streaming.

Data table names could contain `?` before they were restricted, and such
entries are still stored. Splitting `?role=` off a reference misread them:
`a?b` became `a` with an unknown parameter, and the clone checks looked at
a different entry than the one copied. An entry stored under the whole
reference is now looked up first, in the Postgres executor, DuckDB ATTACH
and the clone checks. Agent workers cannot read the workspace and keep
the strict parse, which refuses such a name rather than misreading it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(datatables): warn when a settings sync strands fork pointers

A settings save reported the fork pointers left resolving to nothing only
for the names in `deleted_datatables`, which `wmill sync push` never sends.
The save now works out what it removed from the locked entries, and the
CLI prints the stranded pointers it returns.

Also correct the replication helper's contract: no role or admin check
makes a replication connection safe, so a data table under roles is
refused outright rather than gated as an admin operation.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BjfMkJyKzodxkobqGZ6Lqb

* fix(datatables): refuse a save that drops a data table's roles through an undeclared rename

A data table's roles follow its entry only through a declared rename. A
settings sync sends the whole map and never declares one, so renaming a
data table under roles there read as a delete and a new entry on the same
database: the new entry carried no roles, and every caller connected as
admin. Such a save is now refused, naming both entries.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BjfMkJyKzodxkobqGZ6Lqb

* fix(datatables): no entry without roles may newly reach a database under roles

The previous guard only caught a new name replacing an entry under roles.
A whole-map save could also repoint an existing entry without roles at
that database, or another workspace could point one there, and every
caller of that entry would connect as admin. The rule is now stated on
the saved entries: one that carries no roles and newly points at an
instance database any entry under roles uses, in this workspace or
another, is refused. A declared rename carries its roles and passes.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BjfMkJyKzodxkobqGZ6Lqb

* feat(datatables): move data table role catalog and resolution to the enterprise edition

Roles are an Enterprise Edition feature. The catalog, the Postgres logins,
CONNECT convergence, tenant evaluation and the role half of connection
resolution move to windmill-ee-private. Every public function keeps its path
and signature and forwards through datatable_roles_oss, which re-exports the
enterprise implementation or, without it, refuses.

Without the enterprise edition a data table under roles, or a caller naming a
role, is refused a connection rather than resolved as admin, and the reach and
admin-access checks refuse one under roles. A data table not under roles
resolves as before in every edition, and an instance database keeps the
CONNECT grants it was created with. The catalog lock, the stream lock, the
tenant cascades and the permissions stripping stay in OSS: they only restrict.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BjfMkJyKzodxkobqGZ6Lqb

* feat(datatables): move the data table permissions endpoints to the enterprise edition

The permissions read, save and usable-roles handlers move to
windmill-ee-private; the routes stay registered and, without the enterprise
edition, answer that data table roles are an Enterprise Edition feature.
ensure_governs_datatable and ensure_reaches_datatable keep their paths: the
first refuses, the second passes a data table not under roles.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BjfMkJyKzodxkobqGZ6Lqb

* feat(datatables): move the data table role catalog endpoints to the enterprise edition

The superadmin list, create, update and delete handlers move to
windmill-ee-private. The routes stay registered and, without the enterprise
edition, refuse after authentication.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BjfMkJyKzodxkobqGZ6Lqb

* test(datatables): run the roles tests on the enterprise edition, refusals without it

Each test that exercises roles runs with private and enterprise. Two tests run
without them: every roles route answers the Enterprise refusal, and a data
table saved under roles, or a named role, is refused a connection while one
not under roles resolves as before.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BjfMkJyKzodxkobqGZ6Lqb

* feat(datatables): gate the roles UI mount sites on an enterprise license

Both mount sites are still commented out; the gate travels with them.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BjfMkJyKzodxkobqGZ6Lqb

* test(datatables): run the tenant matcher test on the enterprise edition

The matcher it covers is enterprise code now, so without the enterprise
edition the test hit the stub and failed the default windmill-common run. It
runs with private and enterprise, and a counterpart without them asserts that
no tenant list covers anyone, the wildcard and a workspace admin included.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BjfMkJyKzodxkobqGZ6Lqb

* chore: update ee-repo-ref to a1873dbb67f2302b85ff5362f8387b48eccdb607

This commit updates the EE repository reference after PR #783 was merged in windmill-ee-private.

Previous ee-repo-ref: 5c853e2c20eca6b748415fc0d6862a6ebfb5fec4

New ee-repo-ref: a1873dbb67f2302b85ff5362f8387b48eccdb607

Automated by sync-ee-ref workflow.

* fix(datatables): refuse roles while a same-workspace alias reaches the database

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(datatables): let CE migrations connect as an explicitly named admin

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(datatables): serialize roles going on with aliases saved from other workspaces

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* docs(datatables): note that legacy names with ? cannot be migrated

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(datatables): drop a DuckDB data table secret once its ATTACH has used it

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* perf(datatables): resolve a workspace's data tables per pointer hop, not per entry

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(datatables): hold the parent's settings while a fork points at its data tables

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* chore: update ee-repo-ref to 7e338e4dabf91689bfd7fb0333c6534040b17b59

This commit updates the EE repository reference after PR #787 was merged in windmill-ee-private.

Previous ee-repo-ref: 38d6fcf2aeb39cfdac21814bbdbbcc02911e566a

New ee-repo-ref: 7e338e4dabf91689bfd7fb0333c6534040b17b59

Automated by sync-ee-ref workflow.

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
2026-09-18 13:56:41 +02:00

2073 lines
76 KiB
Rust

use std::collections::HashMap;
use windmill_api_auth::{require_super_admin, ApiAuthed};
use windmill_common::DB;
use crate::workspaces::{
archive_workspace_impl, check_w_id_conflict, CREATE_WORKSPACE_REQUIRE_SUPERADMIN,
};
use axum::extract::Query;
use axum::{
extract::{Extension, Path},
Json,
};
use sqlx::{Postgres, Transaction};
use tracing::info;
use windmill_audit::audit_oss::audit_log;
use windmill_audit::ActionKind;
use windmill_common::worker::CLOUD_HOSTED;
use windmill_common::{
db::UserDB,
error::{Error, Result},
utils::require_admin,
workspaces::{DataTable, DEV_WORKSPACE_LOCK_RULE_NAME, WM_FORK_PREFIX},
};
use windmill_queue::schedule::{get_schedule_opt, push_scheduled_job};
use serde::Deserialize;
#[derive(Deserialize)]
pub(crate) struct ChangeWorkspaceId {
new_id: String,
new_name: String,
}
pub(crate) async fn change_workspace_id(
authed: ApiAuthed,
Path(old_id): Path<String>,
Extension(db): Extension<DB>,
Json(rw): Json<ChangeWorkspaceId>,
) -> Result<String> {
if *CLOUD_HOSTED && !windmill_api_auth::is_super_admin_authed(&db, &authed).await? {
return Err(Error::BadRequest(
"This feature is not available on the cloud".to_string(),
));
}
if *CREATE_WORKSPACE_REQUIRE_SUPERADMIN {
require_super_admin(&db, &authed).await?;
} else {
require_admin(authed.is_admin, &authed.username)?;
}
let mut tx = db.begin().await?;
// A rename rewrites the workspace's dev flag and reparents its children, so it decides on the
// same state the pairing handlers do: without this lock a concurrent create/attach could commit
// an active dev workspace under the shell this rename is about to archive. Both ids, since the
// rename moves the chain from one to the other.
crate::workspaces::lock_dev_pairing(&mut tx, &[&old_id, &rw.new_id]).await?;
check_w_id_conflict(&mut tx, &rw.new_id).await?;
info!(
"Changing workspace id from {} to {} (move and archive approach)",
old_id, rw.new_id
);
// Create new workspace with new id and name. Fork lineage AND the dev designation are preserved
// from the source row, not inferred from the new id's prefix: a prefix-less fork (a dev or
// detached-dev workspace) would otherwise be silently promoted to a root workspace, and a dev
// would lose its flag — leaving its prod locked with no canonical dev. Promoting out of a fork is
// a separate, explicit action — a rename never does it implicitly.
info!("Creating new workspace row");
let old = sqlx::query!(
r#"SELECT (parent_workspace_id IS NOT NULL) AS "has_parent!", is_dev_workspace
FROM workspace WHERE id = $1"#,
&old_id
)
.fetch_optional(&mut *tx)
.await?;
let new_is_fork = old.as_ref().map(|o| o.has_parent).unwrap_or(false);
let new_is_dev = new_is_fork && old.as_ref().map(|o| o.is_dev_workspace).unwrap_or(false);
// Neutralize the old row's dev flag BEFORE inserting the new one: the move-and-archive archives
// the old row only later, so without this the new dev row and the not-yet-archived old dev row
// would momentarily both be active under the same parent and trip the one-dev-per-parent index.
if new_is_dev {
sqlx::query!(
"UPDATE workspace SET is_dev_workspace = false WHERE id = $1",
&old_id
)
.execute(&mut *tx)
.await?;
}
sqlx::query!(
"INSERT INTO workspace (id, name, owner, deleted, premium, parent_workspace_id, is_dev_workspace, dev_workspace_label)
SELECT $1, $2, owner, false, premium,
CASE WHEN $4 THEN parent_workspace_id ELSE NULL END, $5,
CASE WHEN $5 THEN dev_workspace_label ELSE NULL END
FROM workspace WHERE id = $3",
&rw.new_id,
&rw.new_name,
&old_id,
new_is_fork,
new_is_dev
)
.execute(&mut *tx)
.await?;
// Duplicate workspace settings (keep copy in old workspace for reference)
info!("Duplicating workspace_settings table");
sqlx::query!(
"INSERT INTO workspace_settings (workspace_id, slack_team_id, slack_name, slack_command_script, slack_email, customer_id, plan, webhook, ai_config, large_file_storage, git_sync, default_app, default_scripts, deploy_ui, mute_critical_alerts, color, operator_settings, teams_command_script, teams_team_id, teams_team_name, git_app_installations, git_credentials, ducklake, dbt_warehouses, slack_oauth_client_id, slack_oauth_client_secret, datatable, teams_team_guid, auto_invite, error_handler, success_handler, public_app_execution_limit_per_minute, error_handler_fallback_to_instance_alerts, guest_access_enabled, guest_jwt_public_key, guest_jwt_jwks_url) SELECT $1, slack_team_id, slack_name, slack_command_script, slack_email, customer_id, plan, webhook, ai_config, large_file_storage, git_sync, default_app, default_scripts, deploy_ui, mute_critical_alerts, color, operator_settings, teams_command_script, teams_team_id, teams_team_name, git_app_installations, git_credentials, ducklake, dbt_warehouses, slack_oauth_client_id, slack_oauth_client_secret, datatable, teams_team_guid, auto_invite, error_handler, success_handler, public_app_execution_limit_per_minute, error_handler_fallback_to_instance_alerts, guest_access_enabled, guest_jwt_public_key, guest_jwt_jwks_url FROM workspace_settings WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
// The managed git-sync webhooks deliver to /api/w/{old_id}/... — a URL the
// renamed workspace no longer answers on (the old id is archived and the
// receiver skips it). Strip the webhook fields from the new row so polling
// resumes at the normal interval and the next settings save re-registers a
// hook with the new URL; the stale hooks are deleted after commit.
#[allow(unused_mut)]
let mut stale_webhooks: Vec<(String, i64)> = Vec::new();
if let Some(git_sync) = sqlx::query_scalar!(
"SELECT git_sync FROM workspace_settings WHERE workspace_id = $1",
&rw.new_id
)
.fetch_optional(&mut *tx)
.await?
.flatten()
{
if let Ok(mut settings) = serde_json::from_value::<
windmill_common::workspaces::WorkspaceGitSyncSettings,
>(git_sync)
{
let mut changed = false;
for r in settings.repositories.iter_mut() {
if let Some(ap) = r.auto_pull.as_mut() {
if let Some(hook) = ap.webhook_id {
stale_webhooks.push((r.git_repo_resource_path.clone(), hook));
}
changed |= ap.webhook_id.is_some()
|| ap.webhook_secret.is_some()
|| ap.webhook_url.is_some()
|| ap.webhook_error.is_some();
ap.webhook_id = None;
ap.webhook_secret = None;
ap.webhook_url = None;
ap.webhook_error = None;
}
}
if changed {
let serialized = serde_json::to_value(&settings)
.map_err(|e| Error::internal_err(e.to_string()))?;
sqlx::query!(
"UPDATE workspace_settings SET git_sync = $1 WHERE workspace_id = $2",
serialized,
&rw.new_id
)
.execute(&mut *tx)
.await?;
}
}
}
info!("Duplicating workspace_key table");
sqlx::query!(
"INSERT INTO workspace_key SELECT $1, kind, key FROM workspace_key WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating workspace_env table");
sqlx::query!(
"UPDATE workspace_env SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating guest_activity table");
sqlx::query("UPDATE guest_activity SET workspace_id = $1 WHERE workspace_id = $2")
.bind(&rw.new_id)
.bind(&old_id)
.execute(&mut *tx)
.await?;
info!("Updating workspace_invite table");
sqlx::query!(
"UPDATE workspace_invite SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating account table");
sqlx::query!(
"UPDATE account SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating app table");
sqlx::query!(
"UPDATE app SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating capture table");
sqlx::query!(
"UPDATE capture SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating capture_config table");
sqlx::query!(
"UPDATE capture_config SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating http_trigger table");
sqlx::query!(
"UPDATE http_trigger SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating websocket_trigger table");
sqlx::query!(
"UPDATE websocket_trigger SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating kafka_trigger table");
sqlx::query!(
"UPDATE kafka_trigger SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating nats_trigger table");
sqlx::query!(
"UPDATE nats_trigger SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating postgres_trigger table");
sqlx::query!(
"UPDATE postgres_trigger SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating mqtt_trigger table");
sqlx::query!(
"UPDATE mqtt_trigger SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating amqp_trigger table");
sqlx::query!(
"UPDATE amqp_trigger SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating gcp_trigger table");
sqlx::query!(
"UPDATE gcp_trigger SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating sqs_trigger table");
sqlx::query!(
"UPDATE sqs_trigger SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating email_trigger table");
sqlx::query!(
"UPDATE email_trigger SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating native_trigger table");
sqlx::query!(
"UPDATE native_trigger SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating dependency_map table");
sqlx::query!(
"UPDATE dependency_map SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating macro_definition table");
sqlx::query!(
"UPDATE macro_definition SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating macro_usage table");
sqlx::query!(
"UPDATE macro_usage SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating deployment_metadata table");
sqlx::query!(
"UPDATE deployment_metadata SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating draft table");
sqlx::query!(
"UPDATE draft SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating favorite table");
sqlx::query!(
"UPDATE favorite SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
// Duplicate flow table rows (FK constraint requires insert then delete)
info!("Duplicating flow table rows");
sqlx::query!(
"INSERT INTO flow
(workspace_id, path, summary, description, archived, extra_perms, dependency_job, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, on_behalf_of, on_behalf_of_email, concurrency_key, versions, value, schema, edited_by, edited_at, lock_error_logs)
SELECT $1, path, summary, description, archived, extra_perms, dependency_job, tag, ws_error_handler_muted, dedicated_worker, timeout, visible_to_runner_only, on_behalf_of, on_behalf_of_email, concurrency_key, versions, value, schema, edited_by, edited_at, lock_error_logs
FROM flow WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating flow_version table");
sqlx::query!(
"UPDATE flow_version SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating workspace_runnable_dependencies table");
sqlx::query!(
"UPDATE workspace_runnable_dependencies SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating workspace_dependencies table");
sqlx::query!(
"UPDATE workspace_dependencies SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating workspace_diff table");
sqlx::query!(
"UPDATE workspace_diff SET source_workspace_id = $1 WHERE source_workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
sqlx::query!(
"UPDATE workspace_diff SET fork_workspace_id = $1 WHERE fork_workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating workspace_diff_full_scan table");
sqlx::query!(
"UPDATE workspace_diff_full_scan SET source_workspace_id = $1 WHERE source_workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
sqlx::query!(
"UPDATE workspace_diff_full_scan SET fork_workspace_id = $1 WHERE fork_workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating workspace_fork_deployment_request table");
sqlx::query!(
"UPDATE workspace_fork_deployment_request SET source_workspace_id = $1 WHERE source_workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
sqlx::query!(
"UPDATE workspace_fork_deployment_request SET fork_workspace_id = $1 WHERE fork_workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
// Re-parent child forks: any fork whose parent_workspace_id was the old id
// must follow the renamed parent to the new id, otherwise it is left
// pointing at the soft-deleted old shell (whose data has moved here).
info!("Re-parenting child forks to the new workspace id");
let reparented_children: Vec<String> = sqlx::query_scalar!(
"UPDATE workspace SET parent_workspace_id = $1 WHERE parent_workspace_id = $2 RETURNING id",
&rw.new_id,
&old_id
)
.fetch_all(&mut *tx)
.await?;
// A fork's data table entry names the workspace that governs it by id, so the rename has to
// follow there too — anywhere, not just in the reparented children: a detached workspace can
// point at this one without being its fork. Left behind, the pointer resolves to the archived
// shell and every job through it stops.
info!("Re-pointing data table references to the new workspace id");
sqlx::query!(
r#"UPDATE workspace_settings ws
SET datatable = (
SELECT jsonb_set(ws.datatable, '{datatables}', jsonb_object_agg(
dt.key,
CASE WHEN dt.value->'reference'->>'workspace_id' = $2
THEN jsonb_set(dt.value, '{reference,workspace_id}', to_jsonb($1::text))
ELSE dt.value END
))
FROM jsonb_each(ws.datatable->'datatables') dt
)
WHERE jsonb_typeof(ws.datatable->'datatables') = 'object'
AND ws.datatable::text LIKE '%"reference"%'"#,
&rw.new_id,
&old_id,
)
.execute(&mut *tx)
.await?;
info!("Updating workspace_protection_rule table");
sqlx::query!(
"UPDATE workspace_protection_rule SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating workspace_integrations table");
sqlx::query!(
"UPDATE workspace_integrations SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating ai_agent_memory table");
sqlx::query!(
"UPDATE ai_agent_memory SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating flow_conversation table");
sqlx::query!(
"UPDATE flow_conversation SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating mcp_oauth_refresh_token table");
sqlx::query!(
"UPDATE mcp_oauth_refresh_token SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating mcp_oauth_server_code table");
sqlx::query!(
"UPDATE mcp_oauth_server_code SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Duplicating skip_workspace_diff_tally table");
sqlx::query!(
"INSERT INTO skip_workspace_diff_tally SELECT $1, added_at FROM skip_workspace_diff_tally WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating asset table");
sqlx::query!(
"UPDATE asset SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating flow_node table");
sqlx::query!(
"UPDATE flow_node SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Deleting old flow rows");
sqlx::query!("DELETE FROM flow WHERE workspace_id = $1", &old_id)
.execute(&mut *tx)
.await?;
// Duplicate group_ with new workspace id (FK constraint)
info!("Duplicating group_ table rows");
sqlx::query!(
"INSERT INTO group_ SELECT $1, name, summary, extra_perms FROM group_ WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating usr_to_group table");
sqlx::query!(
"UPDATE usr_to_group SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating group_permission_history table");
sqlx::query!(
"UPDATE group_permission_history SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Deleting old group_ rows");
sqlx::query!("DELETE FROM group_ WHERE workspace_id = $1", &old_id)
.execute(&mut *tx)
.await?;
// Duplicate folders with new workspace id (FK constraint)
info!("Duplicating folder table rows");
sqlx::query!(
"INSERT INTO folder (name, workspace_id, display_name, owners, extra_perms, summary, edited_at, created_by, default_permissioned_as, labels) \
SELECT name, $1, display_name, owners, extra_perms, summary, edited_at, created_by, default_permissioned_as, labels \
FROM folder WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating folder_permission_history table");
sqlx::query!(
"UPDATE folder_permission_history SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Deleting old folder rows");
sqlx::query!("DELETE FROM folder WHERE workspace_id = $1", &old_id)
.execute(&mut *tx)
.await?;
info!("Updating input table");
sqlx::query!(
"UPDATE input SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
// Get enabled schedules and clear their queued jobs BEFORE moving jobs
// This way we don't need to filter out scheduled jobs - they're already removed
let enabled_schedule_paths: Vec<String> = sqlx::query_scalar!(
"SELECT path FROM schedule WHERE workspace_id = $1 AND enabled = true",
&old_id
)
.fetch_all(&mut *tx)
.await?;
info!(
"Found {} enabled schedules, clearing their queued jobs",
enabled_schedule_paths.len()
);
for schedule_path in &enabled_schedule_paths {
windmill_queue::schedule::clear_schedule(&mut tx, schedule_path, &old_id).await?;
}
// Move queued jobs (not running) to new workspace using skip lock
// Scheduled jobs were already cleared above, so no need to filter them
info!("Moving v2_job_queue entries to new workspace");
sqlx::query!(
"UPDATE v2_job_queue SET workspace_id = $1
WHERE id IN (
SELECT id FROM v2_job_queue
WHERE workspace_id = $2
AND running = false
AND id IN (SELECT id FROM v2_job WHERE workspace_id = $2 AND parent_job IS NULL)
FOR UPDATE SKIP LOCKED
)",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating v2_job table for moved queue entries");
sqlx::query!(
"UPDATE v2_job SET workspace_id = $1
WHERE workspace_id = $2
AND id IN (SELECT id FROM v2_job_queue WHERE workspace_id = $1)",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating job_perms table for migrated jobs");
sqlx::query!(
"UPDATE job_perms SET workspace_id = $1
WHERE job_id IN (SELECT id FROM v2_job WHERE workspace_id = $1)",
&rw.new_id
)
.execute(&mut *tx)
.await?;
info!("Updating raw_app table");
sqlx::query!(
"UPDATE raw_app SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Deleting raw_script_temp table");
sqlx::query!(
"DELETE FROM raw_script_temp WHERE workspace_id = $1",
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating resource table");
sqlx::query!(
"UPDATE resource SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating resource_type table");
sqlx::query!(
"UPDATE resource_type SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating schedule table");
sqlx::query!(
"UPDATE schedule SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating script table");
sqlx::query!(
"UPDATE script SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
// The dbt graph tables key on (workspace_id, script_hash) and follow the
// script update by cascade. These two key on the JOB, so they follow the
// jobs that moved — only queued ones do — the way `job_perms` just did.
// Moving them wholesale would strand a completed run's retry state in a
// workspace its job is not in, which reads as "no resumable run".
info!("Updating dbt run state tables for migrated jobs");
sqlx::query!(
"UPDATE dbt_run_state SET workspace_id = $1
WHERE workspace_id = $2 AND job_id IN (SELECT id FROM v2_job WHERE workspace_id = $1)",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
sqlx::query!(
"UPDATE dbt_run_progress SET workspace_id = $1
WHERE workspace_id = $2 AND job_id IN (SELECT id FROM v2_job WHERE workspace_id = $1)",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating token table");
sqlx::query!(
"UPDATE token SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating usage table");
sqlx::query!(
"UPDATE usage SET id = $1 WHERE id = $2 AND is_workspace = true",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Duplicating usr table");
sqlx::query!(
"INSERT INTO usr SELECT $1, username, email, is_admin, created_at, operator, disabled, role FROM usr WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating variable table");
sqlx::query!(
"UPDATE variable SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
// Re-push jobs for enabled schedules in the new workspace (same transaction)
info!(
"Re-pushing jobs for {} enabled schedules",
enabled_schedule_paths.len()
);
for schedule_path in &enabled_schedule_paths {
if let Some(schedule) = get_schedule_opt(&mut *tx, &rw.new_id, schedule_path).await? {
tx = push_scheduled_job(&db, tx, &schedule, None, None).await?;
}
}
// Audit log in the same transaction as the workspace changes
audit_log(
&mut *tx,
&authed,
"workspace.change_workspace_id",
ActionKind::Update,
&rw.new_id,
Some(&authed.email),
Some(
[("old_workspace_id", old_id.as_str())]
.into_iter()
.collect::<HashMap<&str, &str>>(),
),
)
.await?;
tx.commit().await?;
// Best-effort: the hooks stripped above still exist on GitHub pointing at
// the old workspace URL; remove them (resources already live under the new id).
#[cfg(all(feature = "enterprise", feature = "private"))]
for (path, hook_id) in stale_webhooks {
if let Ok(url) =
windmill_common::git_sync_ee::resolve_repo_url_interpolated(&db, &rw.new_id, &path)
.await
{
let _ =
windmill_common::git_sync_ee::delete_repo_webhook(&db, &rw.new_id, &url, hook_id)
.await;
}
}
// A rename changes which workspace each id denotes. Workspace ids are reclaimable, so the new
// id may still carry a previous occupant's cached resolution, and the old id now names an
// archived row. Drop both rather than reason about which cached answers are still true.
windmill_queue::tags::invalidate_fork_parent_cache(&rw.new_id);
windmill_queue::tags::invalidate_fork_parent_cache(&old_id);
if let Err(e) = windmill_queue::tags::notify_fork_lineage_reset(&db).await {
tracing::warn!("failed to broadcast fork lineage change: {e:#}");
}
// The children's parent_workspace_id changed (old root -> new root); invalidate their fork-parent
// routing cache and their billing-workspace mapping so jobs route + meter under the renamed root
// rather than the old (archived) one, instead of waiting for the caches' TTLs.
for child in &reparented_children {
windmill_queue::tags::invalidate_fork_parent_cache(child);
windmill_common::workspaces::invalidate_fork_ancestor_chain_cache(child);
// Grandchildren's cached chains contain the old (renamed-away) ancestor id; a stale
// chain drops all defer ancestors in the ducklake resolver, and tag resolution walks to
// the nearest servable ancestor, so a nested fork would be tagged for the old root and
// nothing would serve it. Sweep the subtree rather than letting it wait out the TTL.
for id in windmill_common::workspaces::list_fork_descendants(&db, child)
.await
.unwrap_or_default()
{
windmill_common::workspaces::invalidate_fork_ancestor_chain_cache(&id);
windmill_queue::tags::invalidate_fork_parent_cache(&id);
}
#[cfg(feature = "cloud")]
windmill_common::workspaces::invalidate_billing_workspace_cache(child);
}
// Archive old workspace: disable schedules, cancel remaining jobs, set deleted=true
// Note: schedules were already moved to new workspace, so this will find 0 schedules
info!("Archiving old workspace");
let (_schedules_count, canceled_count, _deleted_tokens_count) =
archive_workspace_impl(&db, &old_id, &authed.username, None).await?;
info!(
"Workspace id change completed: moved {} to {}, archived old workspace",
old_id, rw.new_id
);
Ok(format!(
"Moved workspace from {} to {}, archived old workspace (canceled {} remaining jobs)",
&old_id, &rw.new_id, canceled_count
))
}
#[derive(Deserialize)]
pub(crate) struct DeleteWorkspaceQuery {
pub(crate) only_delete_forks: Option<bool>,
}
pub(crate) async fn delete_workspace(
Extension(db): Extension<DB>,
Path(w_id): Path<String>,
authed: ApiAuthed,
Query(dwq): Query<DeleteWorkspaceQuery>,
) -> Result<String> {
let w_id = match w_id.as_str() {
"starter" => Err(Error::BadRequest(
"starter workspace cannot be deleted".to_string(),
)),
"admins" => Err(Error::BadRequest(
"admins workspace cannot be deleted".to_string(),
)),
_ => Ok(w_id),
}?;
let is_fork = workspace_is_fork(&db, &w_id).await?;
if dwq.only_delete_forks.unwrap_or(false) && !is_fork {
return Err(Error::BadRequest(
"Cannot delete this workspace because it is not a workspace fork.".to_string(),
));
}
let mut tx = db.begin().await?;
if !(is_fork && is_workspace_owner(&authed, &w_id, &mut tx).await?)
&& !windmill_api_auth::is_super_admin_authed(&db, &authed).await?
{
return Err(Error::PermissionDenied(
"Deleting this workspace requires being the fork's owner or a superadmin".to_string(),
));
}
// Don't hard-delete a workspace that still has a dev workspace paired to it: the FK is
// ON DELETE SET NULL, which would orphan the (prefix-less) dev into a parentless, non-fork row
// its owner could no longer self-delete. Require detaching/deleting the dev first. Ordinary
// forks have no such guard — they keep their prefix and stay owner-deletable when orphaned.
// Archived devs (deleted = true) are included: they keep is_dev_workspace = true, so SET NULL on
// their parent would violate the `is_dev ⇒ has parent` CHECK and fail the whole delete with a 500.
if let Some(dev_id) = sqlx::query_scalar!(
"SELECT id FROM workspace WHERE parent_workspace_id = $1 AND is_dev_workspace",
&w_id
)
.fetch_optional(&mut *tx)
.await?
{
return Err(Error::BadRequest(format!(
"Cannot delete workspace '{}' because it has a dev workspace ('{}'). Detach or delete the dev workspace first.",
w_id, dev_id
)));
}
// Deleting an attached dev workspace removes the parent prod's dev_workspace_lock (below), so it
// must be a prod-admin action, not just the dev's own owner (dev ownership can diverge from
// prod's) — mirrors detach_dev_workspace, which is prod-admin gated.
require_prod_admin_for_dev_workspace(&db, &authed, &w_id).await?;
// Snapshot the fork's ducklake namespaces + RESOLVE their connection material NOW — the
// registry rows and the fork's `$res:` resources both CASCADE with the workspace row —
// but the destructive cleanup itself runs only after the commit below: a delete that
// fails mid-way must never leave a live workspace with its fork data destroyed and no
// registry row to retry from. Read-only: nothing is dropped here.
// Read before the delete: another workspace's data table entry can point at one of this
// workspace's, and deleting the workspace it names leaves that pointer resolving to nothing.
// Nothing sweeps them — turning them back into copies would hand each fork the database
// outright — so the deleter is told which data tables they just stranded.
let stranded_pointers = sqlx::query!(
r#"SELECT ws.workspace_id AS "workspace_id!", dt.key AS "datatable!"
FROM workspace_settings ws
CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt
WHERE dt.value->'reference'->>'workspace_id' = $1
ORDER BY ws.workspace_id, dt.key"#,
&w_id,
)
.fetch_all(&db)
.await
.unwrap_or_default();
let fork_ducklake_cleanups = prepare_fork_ducklake_cleanups(&db, &w_id, None)
.await
.unwrap_or_else(|e| {
tracing::warn!("deleting workspace {w_id}: preparing ducklake cleanup: {e:#}");
vec![]
});
sqlx::query!("DELETE FROM ai_agent_memory WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!(
"DELETE FROM flow_conversation WHERE workspace_id = $1",
&w_id
)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM workspace_env WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM dependency_map WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM macro_usage WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!(
"DELETE FROM macro_definition WHERE workspace_id = $1",
&w_id
)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM v2_job_queue WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
// dispatch_event / flow_conversation_message / zombie_job_counter no longer cascade from
// v2_job (see migration drop_v2_job_side_table_cascades); delete them before v2_job so the
// workspace's jobs leave no orphan side rows. One round-trip, scanning v2_job once.
sqlx::query!(
"WITH ids AS (SELECT id FROM v2_job WHERE workspace_id = $1),
_de AS (DELETE FROM dispatch_event WHERE workspace_id = $1),
_jr AS (DELETE FROM job_resolution WHERE workspace_id = $1),
_fc AS (DELETE FROM flow_conversation_message WHERE job_id IN (SELECT id FROM ids))
DELETE FROM zombie_job_counter WHERE job_id IN (SELECT id FROM ids)",
&w_id
)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM v2_job WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM capture WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
// capture_config has on delete cascade
sqlx::query!("DELETE FROM draft WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM script WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM flow WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM app WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM raw_app WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM input WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM variable WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM resource WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM schedule WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!(
"DELETE FROM v2_job_completed WHERE workspace_id = $1",
&w_id
)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM job_stats WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!(
"DELETE FROM deployment_metadata WHERE workspace_id = $1",
&w_id
)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM usr WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM resource_type WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!(
"DELETE FROM workspace_invite WHERE workspace_id = $1",
&w_id
)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM usr_to_group WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM group_ WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM folder WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM account WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM workspace_key WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!(
"DELETE FROM workspace_settings WHERE workspace_id = $1",
&w_id
)
.execute(&mut *tx)
.await?;
// Unlike the rest of this list, this also moves an instance-wide figure: the guest
// allowance and the seats past it are counted over every workspace's rows.
sqlx::query("DELETE FROM guest_activity WHERE workspace_id = $1")
.bind(&w_id)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM token WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM http_trigger WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
sqlx::query!(
"DELETE FROM websocket_trigger WHERE workspace_id = $1",
&w_id
)
.execute(&mut *tx)
.await?;
sqlx::query!("DELETE FROM kafka_trigger WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
// NATS triggers have on delete cascade
sqlx::query!("DELETE FROM raw_script_temp WHERE workspace_id = $1", &w_id)
.execute(&mut *tx)
.await?;
// workspace_diff, workspace_diff_full_scan and skip_workspace_diff_tally are keyed
// by workspace id with no FK cascade. A fork id is reused when a fork is deleted
// and recreated under the same name, so leaving these rows behind leaks the
// previous fork's cached diff verdicts onto the new fork — causing a spurious
// "changes not visible" warning that hides the deploy button.
sqlx::query!(
"DELETE FROM workspace_diff WHERE source_workspace_id = $1 OR fork_workspace_id = $1",
&w_id
)
.execute(&mut *tx)
.await?;
sqlx::query!(
"DELETE FROM workspace_diff_full_scan WHERE source_workspace_id = $1 OR fork_workspace_id = $1",
&w_id
)
.execute(&mut *tx)
.await?;
sqlx::query!(
"DELETE FROM skip_workspace_diff_tally WHERE workspace_id = $1",
&w_id
)
.execute(&mut *tx)
.await?;
// If this workspace is itself a dev workspace, deleting it dissolves the pairing, so also drop
// the parent prod's reserved dev_workspace_lock (mirrors detach_dev_workspace) — otherwise prod
// stays locked against direct deploy/forking with no dev workspace left to make changes in.
let dev_lock_parent: Option<String> = sqlx::query_scalar!(
"SELECT parent_workspace_id FROM workspace WHERE id = $1 AND is_dev_workspace",
&w_id
)
.fetch_optional(&mut *tx)
.await?
.flatten();
// Capture direct child forks before the delete: the FK is ON DELETE SET NULL, so they're about to
// be orphaned (their billing root changes from this workspace's root to themselves). We drop their
// cached mappings after commit alongside the deleted id itself.
let orphaned_children: Vec<String> = sqlx::query_scalar!(
"SELECT id FROM workspace WHERE parent_workspace_id = $1",
&w_id
)
.fetch_all(&mut *tx)
.await?;
sqlx::query!("DELETE FROM workspace WHERE id = $1", &w_id)
.execute(&mut *tx)
.await?;
if let Some(ref parent) = dev_lock_parent {
sqlx::query!(
"DELETE FROM workspace_protection_rule WHERE workspace_id = $1 AND name = $2",
parent,
DEV_WORKSPACE_LOCK_RULE_NAME
)
.execute(&mut *tx)
.await?;
}
// Record under the instance-level "admins" workspace. The per-workspace audit
// rows are deleted along with the workspace, so this instance-level entry is the
// only durable, superadmin-discoverable record of who deleted the workspace.
audit_log(
&mut *tx,
&authed,
"workspaces.delete",
ActionKind::Delete,
"admins",
Some(&w_id),
None,
)
.await?;
tx.commit().await?;
// Physical ducklake-namespace cleanup, post-commit, from the pre-read snapshot: fork
// namespaces are deterministic from (id, lake), so an orphan would silently REATTACH to a
// recreated identical fork id. Runs inline so every delete path is covered (CLI,
// force-delete dialog, direct API — not just the sidebar flow, which still calls the
// endpoint first for per-lake error toasts; the rerun is an idempotent no-op). Best
// effort: failures are logged — the workspace row is already gone, and broken storage
// credentials must not have made it undeletable.
for e in cleanup_fork_ducklake_namespaces(&db, &w_id, fork_ducklake_cleanups).await {
tracing::warn!(
"deleted workspace {w_id}: ducklake namespace cleanup: {}",
e.msg
);
}
if let Some(parent) = dev_lock_parent {
windmill_common::workspaces::invalidate_protection_rules_cache(&parent);
}
// Workspace ids are reusable after permanent deletion, so drop every cached mapping keyed by the
// deleted id (and any just-orphaned children) — otherwise a recreated id could inherit the gone
// workspace's state within the caches' lifetimes. This covers fork->parent (tag routing) and
// fork->root (billing), plus the premium/team-plan status: TEAM_PLAN_CACHE has no TTL and is only
// evicted by the premium-change NOTIFY, so without this a reused id would keep the old workspace's
// premium indefinitely (free forks/usage). Deeper (grandchild) descendants self-heal via the 60s
// billing TTL.
for id in std::iter::once(&w_id).chain(orphaned_children.iter()) {
windmill_queue::tags::invalidate_fork_parent_cache(id);
windmill_common::workspaces::invalidate_fork_ancestor_chain_cache(id);
#[cfg(feature = "cloud")]
{
windmill_common::workspaces::invalidate_billing_workspace_cache(id);
windmill_common::workspaces::invalidate_team_plan_cache(id);
}
}
// Deeper descendants' cached ancestor CHAINS still contain the deleted workspace: a stale
// chain makes the ducklake resolver drop all defer ancestors (all-or-nothing on broken
// links), and tag resolution walks to the nearest servable ancestor, so a nested fork would
// keep a tag naming the deleted workspace that nothing serves. Anchor at the orphaned
// children: the deleted row is gone, but their subtrees are intact.
for child in orphaned_children.iter() {
windmill_queue::tags::invalidate_fork_parent_cache(child);
for id in windmill_common::workspaces::list_fork_descendants(&db, child)
.await
.unwrap_or_default()
{
windmill_common::workspaces::invalidate_fork_ancestor_chain_cache(&id);
windmill_queue::tags::invalidate_fork_parent_cache(&id);
}
}
// The id is reclaimable, so its cached mapping must not outlive it anywhere: it can be claimed
// again under a different parent well inside the TTL. That is a one-entry drop, cheap enough for
// the ephemeral fork churn this endpoint sees. Orphaning descendants reshapes the tree instead,
// which no single id identifies.
let broadcast = if orphaned_children.is_empty() {
windmill_queue::tags::notify_fork_lineage_change(&db, &w_id).await
} else {
windmill_queue::tags::notify_fork_lineage_reset(&db).await
};
if let Err(e) = broadcast {
tracing::warn!("failed to broadcast fork lineage change: {e:#}");
}
if stranded_pointers.is_empty() {
Ok(format!("Deleted workspace {}", &w_id))
} else {
let stranded = stranded_pointers
.iter()
.map(|r| format!("{}/{}", r.workspace_id, r.datatable))
.collect::<Vec<_>>()
.join(", ");
Ok(format!(
concat!(
"Deleted workspace {}. These data tables were governed by it and no longer ",
"resolve: {}. Their databases still exist; a superadmin can point them at ",
"another workspace's data table."
),
&w_id, stranded
))
}
}
#[derive(Deserialize)]
pub struct DropForkedDatatableDatabasesRequest {
datatable_names: Vec<String>,
}
/// Drop forked datatable databases. Returns errors per datatable that failed.
/// Same permission as delete_workspace: fork owner or super admin.
pub async fn drop_forked_datatable_databases(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Extension(db): Extension<DB>,
Path(w_id): Path<String>,
Json(req): Json<DropForkedDatatableDatabasesRequest>,
) -> Result<Json<Vec<String>>> {
// Same permission check as delete_workspace: fork owner or super admin
let is_fork = workspace_is_fork(&db, &w_id).await?;
let mut tx = db.begin().await?;
if !(is_fork && is_workspace_owner(&authed, &w_id, &mut tx).await?)
&& !windmill_api_auth::is_super_admin_authed(&db, &authed).await?
{
return Err(Error::PermissionDenied(
"Dropping forked datatable databases requires being the fork's owner or a superadmin"
.to_string(),
));
}
tx.commit().await?;
let parent_w_id = sqlx::query_scalar!(
"SELECT parent_workspace_id FROM workspace WHERE id = $1",
&w_id
)
.fetch_optional(&db)
.await?
.flatten()
.ok_or_else(|| Error::BadRequest("No parent workspace found".to_string()))?;
let datatable_config = sqlx::query_scalar!(
"SELECT datatable->'datatables' FROM workspace_settings WHERE workspace_id = $1",
&w_id
)
.fetch_optional(&db)
.await?
.flatten()
.unwrap_or(serde_json::json!({}));
let datatables: HashMap<String, DataTable> =
serde_json::from_value(datatable_config).unwrap_or_default();
let mut errors: Vec<String> = Vec::new();
for dt_name in &req.datatable_names {
// Only a clone is droppable, and a clone is terminal by construction: a kept data table is
// a pointer at the parent's database, which this fork does not own.
let database = match datatables.get(dt_name) {
Some(dt) if dt.forked_from.is_some() => match dt.database.as_ref() {
Some(database) => database,
None => continue,
},
_ => continue,
};
if database.resource_type
== windmill_common::workspaces::DataTableCatalogResourceType::Instance
{
let db_to_drop = &database.resource_path;
if !db_to_drop.starts_with("wm_fork_") {
errors.push(format!(
"Refusing to drop instance database '{}' for datatable://{}: name does not start with 'wm_fork_'",
db_to_drop, dt_name
));
continue;
}
if let Err(e) = windmill_common::drop_custom_instance_database(&db, db_to_drop).await {
errors.push(format!(
"Could not drop instance database '{}' for datatable://{}: {}",
db_to_drop, dt_name, e
));
}
} else {
let fork_pg = match crate::workspaces::resolve_pg_source_checked(
&db,
&user_db,
&authed,
&w_id,
&format!("datatable://{}", dt_name),
)
.await
{
Ok(pg) => pg,
Err(e) => {
errors.push(format!(
"Could not resolve fork resource for datatable://{}: {}",
dt_name, e
));
continue;
}
};
// We cannot drop the current database, so we connect to the parent's version to run DROP DATABASE on
// the forked version
let parent_pg = match crate::workspaces::resolve_pg_source_checked(
&db,
&user_db,
&authed,
&parent_w_id,
&format!("datatable://{}", dt_name),
)
.await
{
Ok(pg) => pg,
Err(e) => {
errors.push(format!(
"Could not resolve parent resource for datatable://{}: {}",
dt_name, e
));
continue;
}
};
let db_to_drop = &fork_pg.dbname;
if let Err(e) = windmill_common::validate_dbname(db_to_drop) {
errors.push(format!(
"Invalid database name '{}' for datatable://{}: {}",
db_to_drop, dt_name, e
));
continue;
}
if !db_to_drop.starts_with("wm_fork_") {
errors.push(format!(
"Refusing to drop resource database '{}' for datatable://{}: name does not start with 'wm_fork_'",
db_to_drop, dt_name
));
continue;
}
match parent_pg.connect(Some(&db)).await {
Ok((client, connection)) => {
let join_handle = tokio::spawn(async move { connection.await });
if let Err(e) = client
.execute(&format!("DROP DATABASE \"{}\"", db_to_drop), &[])
.await
{
errors.push(format!(
"Could not drop database '{}' for datatable://{}: {}",
db_to_drop, dt_name, e
));
}
drop(client);
let _ = windmill_common::shutdown_pg_connection(join_handle).await;
}
Err(e) => {
errors.push(format!(
"Could not connect to drop database for datatable://{}: {}",
dt_name, e
));
}
}
}
}
Ok(Json(errors))
}
/// Drop this fork workspace's ducklake namespaces: the `wm_fork_*` metadata schema in each
/// lake's catalog database, plus (best effort) the fork's `__wm_forks/<wid>/…` data files in
/// the workspace storage. Driven by the `fork_ducklake_namespace` registry written at first
/// fork attach, so it works even after settings drift. Returns errors per lake that failed;
/// the registry row is only deleted once both cleanups succeeded, so a retry resumes.
/// Same permission as delete_workspace: fork owner or super admin. `delete_workspace` also
/// runs this inline (the UI calls this endpoint first for error visibility; the inline run
/// covers every other delete path — CLI, force-delete dialogs, direct API — whose row delete
/// would otherwise CASCADE the registry away while orphaning the physical namespaces, which a
/// recreated identical fork id would then silently reattach).
pub async fn drop_forked_ducklake_namespaces(
authed: ApiAuthed,
Extension(db): Extension<DB>,
Path(w_id): Path<String>,
) -> Result<Json<Vec<String>>> {
let is_fork = workspace_is_fork(&db, &w_id).await?;
let mut tx = db.begin().await?;
if !(is_fork && is_workspace_owner(&authed, &w_id, &mut tx).await?)
&& !windmill_api_auth::is_super_admin_authed(&db, &authed).await?
{
return Err(Error::PermissionDenied(
"Dropping forked ducklake namespaces requires being the fork's owner or a superadmin"
.to_string(),
));
}
tx.commit().await?;
require_prod_admin_for_dev_workspace(&db, &authed, &w_id).await?;
Ok(Json(
drop_forked_ducklake_namespaces_impl(&db, &w_id, None)
.await?
.into_iter()
.map(|i| i.msg)
.collect(),
))
}
/// The cleanup itself, shared by the endpoint and the post-commit `delete_workspace` run. No
/// authorization of its own — callers gate.
pub(crate) async fn drop_forked_ducklake_namespaces_impl(
db: &DB,
w_id: &str,
res_fallback_w_id: Option<&str>,
) -> Result<Vec<ForkDucklakeCleanupIssue>> {
let prepared = prepare_fork_ducklake_cleanups(db, w_id, res_fallback_w_id).await?;
Ok(cleanup_fork_ducklake_namespaces(db, w_id, prepared).await)
}
struct ForkDucklakeNamespaceRow {
ducklake_name: String,
metadata_schema: String,
catalog: String,
storage: String,
storage_ref: String,
schema_dropped: bool,
data_path: String,
}
/// A namespace row plus its RESOLVED connection material. `delete_workspace` prepares these
/// BEFORE its transaction commits — the registry rows AND the fork's `$res:` resources both
/// disappear with the workspace row — and runs the destructive cleanup only AFTER the commit:
/// a delete that fails mid-way must never leave a live workspace with its fork data
/// destroyed, and a committed delete must still be able to reach catalogs/storages whose
/// credentials lived in the (now gone) fork resources. Per-row resolution failures are
/// carried as strings so they surface with the other cleanup errors.
pub(crate) struct PreparedForkDucklakeCleanup {
ns: ForkDucklakeNamespaceRow,
catalog_pg: std::result::Result<windmill_common::PgDatabase, String>,
store: std::result::Result<(windmill_types::s3::LargeFileStorage, serde_json::Value), String>,
}
/// One row per (lake, catalog, storage, data path) ever attached — settings drift adds rows,
/// so every location the fork wrote gets cleaned, not just the first. Read-only: resolves
/// credentials but destroys nothing.
/// `res_fallback_w_id`: second workspace to resolve `$res:` catalog/storage paths against
/// when the row's own workspace no longer has them — the retry path runs AFTER the fork and
/// its resources were deleted. Fork resources are clones of a parent's, so the workspace
/// being forked again is the natural donor. None on live-workspace paths.
pub(crate) async fn prepare_fork_ducklake_cleanups(
db: &DB,
w_id: &str,
res_fallback_w_id: Option<&str>,
) -> Result<Vec<PreparedForkDucklakeCleanup>> {
let rows = sqlx::query_as!(
ForkDucklakeNamespaceRow,
r#"SELECT ducklake_name AS "ducklake_name!", metadata_schema AS "metadata_schema!",
catalog AS "catalog!", storage AS "storage!",
storage_ref AS "storage_ref!", data_path AS "data_path!",
schema_dropped AS "schema_dropped!"
FROM fork_ducklake_namespace WHERE workspace_id = $1"#,
w_id
)
.fetch_all(db)
.await?;
let mut prepared = Vec::with_capacity(rows.len());
for ns in rows {
let catalog_pg = if ns.schema_dropped {
// Schema phase already done — the retry needs no catalog connection at all
// (its credentials may be unresolvable for good with the fork's resources gone).
Err("unused: metadata schema already dropped".to_string())
} else {
resolve_fork_catalog_pg(db, w_id, res_fallback_w_id, &ns.ducklake_name, &ns.catalog)
.await
.map_err(|e| e.to_string())
};
let storage = Some(ns.storage.as_str()).filter(|s| !s.is_empty());
let store = resolve_fork_storage(db, w_id, res_fallback_w_id, storage, &ns.storage_ref)
.await
.map_err(|e| e.to_string());
prepared.push(PreparedForkDucklakeCleanup { ns, catalog_pg, store });
}
Ok(prepared)
}
/// Drop the prepared namespaces' metadata schemas + data files, deleting each registry row
/// only after both succeed (a no-op when the row already cascaded away with the workspace).
/// Returns per-namespace error strings; never fails as a whole.
/// One failed step of a fork ducklake cleanup. `blocking` = the metadata schema (or its
/// guards) failed, so the namespace is still attachable and a same-id fork must NOT be
/// created. Non-blocking = the schema is gone and only data files (or the registry-row
/// delete) failed — inert storage leftovers, tracked by the surviving row and swept by the
/// next successful cleanup of the same prefix.
pub(crate) struct ForkDucklakeCleanupIssue {
pub(crate) blocking: bool,
pub(crate) msg: String,
}
pub(crate) async fn cleanup_fork_ducklake_namespaces(
db: &DB,
w_id: &str,
prepared: Vec<PreparedForkDucklakeCleanup>,
) -> Vec<ForkDucklakeCleanupIssue> {
// The registration once-cache must not outlive the rows it mirrors: a same-id fork
// recreated within the TTL would otherwise skip re-registration, and ITS eventual
// deletion would find no rows — orphaning the deterministic namespace for the next
// same-id fork to silently reattach.
windmill_common::workspaces::invalidate_fork_ducklake_registration_cache(w_id);
let mut errors: Vec<ForkDucklakeCleanupIssue> = Vec::new();
for PreparedForkDucklakeCleanup { ns, catalog_pg, store } in prepared {
// Hard guards mirroring the forked-datatable drop: never touch a schema outside the
// fork prefix, never delete outside the fork data dir — even if a registry row was
// somehow tampered with.
if !ns
.metadata_schema
.starts_with(windmill_common::workspaces::FORK_DUCKLAKE_SCHEMA_PREFIX)
{
errors.push(ForkDucklakeCleanupIssue {
blocking: true,
msg: format!(
"Refusing to drop schema '{}' for ducklake://{}: name does not start with '{}'",
ns.metadata_schema,
ns.ducklake_name,
windmill_common::workspaces::FORK_DUCKLAKE_SCHEMA_PREFIX
),
});
continue;
}
// The fork's directory segment, NOT the raw workspace id: ids are only
// git-branch-safe and may contain `/`, which raw would let one fork's prefix nest
// inside a sibling's (`wm-fork-a/b` under `wm-fork-a`) and be swept by its cleanup.
let expected_prefix = format!(
"{}/{}/",
windmill_common::workspaces::FORK_DUCKLAKE_DATA_DIR,
windmill_common::workspaces::fork_data_dir_segment(w_id)
);
if !format!("{}/", ns.data_path.trim_end_matches('/')).starts_with(&expected_prefix) {
errors.push(ForkDucklakeCleanupIssue {
blocking: true,
msg: format!(
"Refusing to delete data path '{}' for ducklake://{}: not under '{}'",
ns.data_path, ns.ducklake_name, expected_prefix
),
});
continue;
}
let drop_res = if ns.schema_dropped {
// Recorded as already dropped by a prior partial cleanup; skipping means no
// catalog credentials are needed. Registration resets the flag when a live fork
// re-attaches (recreating the schema).
Ok(())
} else {
match catalog_pg {
Ok(pg) => drop_fork_ducklake_metadata_schema(db, pg, &ns.metadata_schema).await,
Err(e) => Err(Error::internal_err(e)),
}
};
if let Err(e) = drop_res {
errors.push(ForkDucklakeCleanupIssue {
blocking: true,
msg: format!(
"Could not drop metadata schema '{}' for ducklake://{}: {e}",
ns.metadata_schema, ns.ducklake_name
),
});
continue;
}
let delete_res = match store {
Ok((lfs, resource_value)) => {
delete_fork_ducklake_data(lfs, resource_value, &ns.data_path).await
}
Err(e) => Err(Error::internal_err(e)),
};
if let Err(e) = delete_res {
// Record the completed schema phase so retries never need catalog credentials
// again (best effort — a failed update just means the next retry re-drops an
// absent schema, which requires the catalog to be reachable).
sqlx::query!(
"UPDATE fork_ducklake_namespace SET schema_dropped = true
WHERE workspace_id = $1 AND ducklake_name = $2 AND catalog = $3
AND storage = $4 AND storage_ref = $5 AND data_path = $6",
w_id,
&ns.ducklake_name,
&ns.catalog,
&ns.storage,
&ns.storage_ref,
&ns.data_path,
)
.execute(db)
.await
.ok();
errors.push(ForkDucklakeCleanupIssue {
blocking: false,
msg: format!(
"Dropped metadata schema but could not delete data files under '{}' for ducklake://{}: {e}",
ns.data_path, ns.ducklake_name
),
});
continue;
}
sqlx::query!(
"DELETE FROM fork_ducklake_namespace
WHERE workspace_id = $1 AND ducklake_name = $2 AND catalog = $3
AND storage = $4 AND storage_ref = $5 AND data_path = $6",
w_id,
&ns.ducklake_name,
&ns.catalog,
&ns.storage,
&ns.storage_ref,
&ns.data_path,
)
.execute(db)
.await
.map_err(|e| {
errors.push(ForkDucklakeCleanupIssue {
blocking: false,
msg: format!(
"Cleaned ducklake://{} but could not delete its registry row: {e}",
ns.ducklake_name
),
})
})
.ok();
}
errors
}
/// Drop the fork's metadata schema in the catalog database recorded by the registry row —
/// NOT whatever the fork's settings point at by now: a drifted catalog resource must not make
/// cleanup drop a schema in the wrong database while orphaning the real one. The pg schema
/// holds only DuckLake metadata tables (auto-created at first fork attach), so a plain
/// `DROP SCHEMA … CASCADE` on the catalog connection removes the whole fork namespace.
/// Resolve the catalog connection recorded in the registry row (read-only): instance
/// identities rebuild instance creds; resource identities resolve their `$res:` in the fork's
/// workspace — which is why this must run BEFORE the workspace (and its resources) are
/// deleted. Mysql never registers (rejected at resolution).
async fn resolve_fork_catalog_pg(
db: &DB,
w_id: &str,
res_fallback_w_id: Option<&str>,
ducklake_name: &str,
catalog: &str,
) -> Result<windmill_common::PgDatabase> {
let (resource_type, resource_path) = catalog.split_once(':').ok_or_else(|| {
Error::internal_err(format!(
"ducklake://{ducklake_name}: malformed registry catalog identity `{catalog}`"
))
})?;
let catalog_resource = if resource_type == "instance" {
let mut pg_creds = windmill_common::PgDatabase::parse_uri(
&windmill_common::get_database_url().await?.as_str().await,
)?;
pg_creds.dbname = resource_path.to_string();
pg_creds.user = Some("custom_instance_user".to_string());
pg_creds.password =
Some(windmill_common::utils::get_custom_pg_instance_password(db).await?);
serde_json::to_value(&pg_creds)
.map_err(|e| Error::internal_err(format!("serializing pg creds: {e}")))?
} else {
resolve_res_with_fallback(db, w_id, res_fallback_w_id, resource_path).await?
};
serde_json::from_value(catalog_resource).map_err(|e| {
Error::internal_err(format!(
"ducklake://{ducklake_name}: catalog resource is not a postgres database: {e}"
))
})
}
async fn drop_fork_ducklake_metadata_schema(
db: &DB,
pg: windmill_common::PgDatabase,
metadata_schema: &str,
) -> Result<()> {
let (client, connection) = pg.connect(Some(db)).await?;
let join_handle = tokio::spawn(async move { connection.await });
let res = client
.execute(
&format!(
"DROP SCHEMA IF EXISTS \"{}\" CASCADE",
metadata_schema.replace('"', "\"\"")
),
&[],
)
.await;
drop(client);
let _ = windmill_common::shutdown_pg_connection(join_handle).await;
res.map_err(|e| Error::internal_err(format!("{e:#}")))?;
Ok(())
}
/// Resolve the storage identified by the registry's `storage_ref` (read-only) — the storage
/// that was active when the fork's data was WRITTEN, not whatever the logical storage name
/// points at by deletion time (a repointed storage must not orphan the original fork data,
/// nor get a colliding prefix deleted). `storage_ref` = '' falls back to resolving the
/// logical name against current settings (registration couldn't identify the storage — best
/// effort). Resolves the `$res:` in the fork's workspace, which is why this must run BEFORE
/// the workspace is deleted.
async fn resolve_fork_storage(
db: &DB,
w_id: &str,
res_fallback_w_id: Option<&str>,
storage: Option<&str>,
storage_ref: &str,
) -> Result<(windmill_types::s3::LargeFileStorage, serde_json::Value)> {
use windmill_types::s3::{
AzureBlobStorage, FilesystemStorage, GoogleCloudStorage, LargeFileStorage, S3Storage,
};
let lfs: LargeFileStorage = if let Some((typ, path)) = storage_ref.split_once(':') {
// Rebuild the LFS entry from the registered identity; only the variant (which
// resource parser applies) and the path matter to `lfs_to_object_store_resource`.
let s3 = |p: &str| S3Storage {
s3_resource_path: p.to_string(),
public_resource: None,
advanced_permissions: None,
};
match typ {
"S3Storage" => LargeFileStorage::S3Storage(s3(path)),
"S3AwsOidc" => LargeFileStorage::S3AwsOidc(s3(path)),
"AzureBlobStorage" => LargeFileStorage::AzureBlobStorage(AzureBlobStorage {
azure_blob_resource_path: path.to_string(),
public_resource: None,
advanced_permissions: None,
}),
"AzureWorkloadIdentity" => LargeFileStorage::AzureWorkloadIdentity(AzureBlobStorage {
azure_blob_resource_path: path.to_string(),
public_resource: None,
advanced_permissions: None,
}),
"GoogleCloudStorage" => LargeFileStorage::GoogleCloudStorage(GoogleCloudStorage {
gcs_resource_path: path.to_string(),
public_resource: None,
advanced_permissions: None,
}),
"FilesystemStorage" => LargeFileStorage::FilesystemStorage(FilesystemStorage {
root_path: path.to_string(),
public_resource: None,
advanced_permissions: None,
}),
other => {
return Err(Error::internal_err(format!(
"unknown registered storage type `{other}`"
)))
}
}
} else {
let lfs_json = sqlx::query_scalar!(
"SELECT large_file_storage FROM workspace_settings WHERE workspace_id = $1",
w_id
)
.fetch_optional(db)
.await?
.flatten()
.ok_or_else(|| Error::BadRequest("workspace has no storage configured".to_string()))?;
// Named storages live under `secondary_storage`; `None`/`_default_` is the primary.
match storage.filter(|s| *s != "_default_") {
None => serde_json::from_value(lfs_json.clone())
.map_err(|e| Error::internal_err(format!("parsing large_file_storage: {e}")))?,
Some(name) => serde_json::from_value(
lfs_json
.get("secondary_storage")
.and_then(|s| s.get(name))
.cloned()
.ok_or_else(|| {
Error::BadRequest(format!("workspace has no storage named {name}"))
})?,
)
.map_err(|e| Error::internal_err(format!("parsing storage {name}: {e}")))?,
}
};
// Filesystem storage stores a direct path (`lfs_to_object_store_resource` ignores the
// resource value); everything else references a resource whose stored path may or may not
// carry the `$res:` prefix — same normalization as `get_workspace_s3_resource_from_lfs`.
let resource_value = if matches!(lfs, LargeFileStorage::FilesystemStorage(_)) {
serde_json::Value::Null
} else {
let path = lfs.get_s3_resource_path();
let path = path.strip_prefix("$res:").unwrap_or(path);
resolve_res_with_fallback(db, w_id, res_fallback_w_id, path).await?
};
Ok((lfs, resource_value))
}
/// Resolve a `$res:` path in `w_id`, falling back to the same path in `res_fallback_w_id`
/// when the first lookup fails — retry-path cleanups run after the fork workspace (and its
/// cloned resource rows) were deleted, and the fork's resources were clones of a parent's.
async fn resolve_res_with_fallback(
db: &DB,
w_id: &str,
res_fallback_w_id: Option<&str>,
resource_path: &str,
) -> Result<serde_json::Value> {
let res = windmill_common::workspaces::transform_json_value_unchecked(
&serde_json::Value::String(format!("$res:{resource_path}")),
w_id,
db,
)
.await;
match (res, res_fallback_w_id) {
(Ok(v), _) => Ok(v),
(Err(e), None) => Err(e),
(Err(_), Some(fb)) => {
windmill_common::workspaces::transform_json_value_unchecked(
&serde_json::Value::String(format!("$res:{resource_path}")),
fb,
db,
)
.await
}
}
}
/// Delete every object under the fork's data prefix in the pre-resolved storage. Requires the
/// `parquet` (object store) feature; without it the metadata schema is still dropped and the
/// unreachable data files are left for manual cleanup.
#[cfg(feature = "parquet")]
async fn delete_fork_ducklake_data(
lfs: windmill_types::s3::LargeFileStorage,
resource_value: serde_json::Value,
data_path: &str,
) -> Result<()> {
use futures::{StreamExt, TryStreamExt};
let store = windmill_object_store::build_object_store_client(
&windmill_object_store::lfs_to_object_store_resource(&lfs, resource_value)?,
)
.await?;
let prefix = windmill_object_store::object_store_reexports::Path::from(
data_path.trim_matches('/').to_string(),
);
let locations: Vec<_> = store
.list(Some(&prefix))
.map_ok(|m| m.location)
.try_collect()
.await
.map_err(windmill_object_store::object_store_error_to_error)?;
// The object_store crate evaluates list prefixes on a path-SEGMENT basis (`a/b` does not
// match `a/bc/…`), so sibling fork segments sharing a string prefix are already excluded.
// Filter anyway — deletion must not depend on a listing implementation detail.
let boundary = format!("{}/", prefix.as_ref());
let locations: Vec<_> = locations
.into_iter()
.filter(|l| l.as_ref().starts_with(&boundary))
.collect();
// 1000-object chunks: S3 DeleteObjects caps a batch at 1000 keys.
for chunk in locations.chunks(1000) {
store
.delete_stream(futures::stream::iter(chunk.iter().cloned().map(Ok)).boxed())
.try_collect::<Vec<_>>()
.await
.map_err(windmill_object_store::object_store_error_to_error)?;
}
Ok(())
}
#[cfg(not(feature = "parquet"))]
async fn delete_fork_ducklake_data(
_lfs: windmill_types::s3::LargeFileStorage,
_resource_value: serde_json::Value,
_data_path: &str,
) -> Result<()> {
Err(Error::internal_err(
"object storage support (parquet feature) is not compiled in".to_string(),
))
}
/// Destroying an ATTACHED dev workspace (or its data environments) must be a prod-admin
/// action, not just the dev's own owner (dev ownership can diverge from prod's) — mirrors
/// detach_dev_workspace. Shared by `delete_workspace` and `drop_forked_ducklake_namespaces`
/// so the two gates cannot drift: the sidebar calls the drop endpoint BEFORE deleteWorkspace,
/// and a weaker gate on the drop would let a non-prod-admin dev owner destroy the live dev's
/// materializations and then have the deletion itself rejected. No-op for non-dev workspaces.
async fn require_prod_admin_for_dev_workspace(
db: &DB,
authed: &ApiAuthed,
w_id: &str,
) -> Result<()> {
if let Some(prod) = sqlx::query_scalar!(
"SELECT parent_workspace_id FROM workspace WHERE id = $1 AND is_dev_workspace",
w_id
)
.fetch_optional(db)
.await?
.flatten()
{
let is_prod_admin = sqlx::query_scalar!(
"SELECT is_admin FROM usr WHERE workspace_id = $1 AND email = $2",
&prod,
&authed.email
)
.fetch_optional(db)
.await?
.unwrap_or(false);
if !is_prod_admin && !windmill_api_auth::is_super_admin_authed(db, &authed).await? {
return Err(Error::PermissionDenied(format!(
"Destroying dev workspace '{w_id}' or its data requires being an admin of its parent prod workspace '{prod}' (or a superadmin)"
)));
}
}
Ok(())
}
async fn is_workspace_owner(
authed: &ApiAuthed,
w_id: &str,
tx: &mut Transaction<'_, Postgres>,
) -> Result<bool> {
let owner = sqlx::query_scalar!("SELECT owner FROM workspace WHERE id = $1", w_id)
.fetch_optional(&mut **tx)
.await?;
Ok(owner.map(|o| o == authed.email).unwrap_or(false))
}
/// Whether a workspace is a fork or dev workspace. Both forks and dev workspaces set
/// `parent_workspace_id`, but a `wm-fork-` workspace can outlive its parent (the FK is
/// `ON DELETE SET NULL`), so also treat the prefix as fork-ness — otherwise an orphaned fork would
/// lose owner-self-delete. Used to gate owner-self-delete, which is permitted for forks/dev
/// workspaces but requires superadmin otherwise.
async fn workspace_is_fork(db: &DB, w_id: &str) -> Result<bool> {
if w_id.starts_with(WM_FORK_PREFIX) {
return Ok(true);
}
Ok(sqlx::query_scalar!(
r#"SELECT (parent_workspace_id IS NOT NULL) AS "has_parent!" FROM workspace WHERE id = $1"#,
w_id
)
.fetch_optional(db)
.await?
.unwrap_or(false))
}