starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-18 16:02:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(forks): handle triggers and schedules in workspace forks (#8976)

Browse Source 
* feat(forks): strip operational state from triggers/schedules on git-sync export

When the source workspace is a fork (`wm-fork-*`), the tarball export now
omits `mode` from triggers and `enabled` from schedules. The trigger update
handler also preserves the existing DB `mode` when both fields are absent
from the request, instead of falling back to the BaseTriggerData default.

This prevents a fork's git-sync round-trip from flipping the parent
workspace's enabled/disabled state when a merge applies the fork's YAML
back to main.

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

* feat(forks): opt-in fork_triggers flag clones triggers/schedules disabled

Adds `workspace.fork_triggers` (default false) and a matching field on
CreateWorkspaceFork. When the user opts in, fork creation also runs
clone_triggers_and_schedules: every row in schedule and the ten
*_trigger tables is copied to the fork with mode='disabled' /
enabled=false. Listener identifiers (group_id, replication_slot_name,
subscription_name, …) are copied verbatim — the runtime suffix that
prevents the fork from competing with the parent ships in a follow-up
PR.

native_trigger is intentionally skipped: those triggers manage external
webhook state we don't want duplicated.

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

* feat(forks): warn before enabling triggers/schedules that conflict with parent

set_trigger_mode and schedule's set_enabled now check whether the parent
workspace has the same path actively enabled. If so, the call is rejected
with a `fork-conflict:<kind>:<parent_id>` error unless the request includes
`force=true`. The frontend interprets the prefix to surface a confirm-to-
proceed dialog.

This is the placeholder safety net until the Phase 3 listener-suffix work
removes the conflict for the namespaceable kinds (Kafka/MQTT/NATS/Postgres/
Azure/GCP-CreateNew). For SQS, GCP-Existing, and schedules — where there's
no namespacing fix — the warning is the durable solution.

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

* feat(forks): UI: opt-in clone-triggers checkbox + confirm-on-fork-conflict

Adds the user-facing surface for the fork-trigger work:

- CreateWorkspaceInner: new "Clone triggers and schedules" toggle in the
  fork-creation dialog (default off). Sends fork_triggers in the request.

- forkConflict utility: detects the `fork-conflict:<kind>:<parent_id>`
  error string from the backend, shows a confirm() dialog explaining
  why the action is blocked, retries with `force: true` if accepted.

- Wires withForkConflictRetry into every trigger setMode and the
  schedule setEnabled call, both in the per-kind editor components and
  the +page.svelte list views (HTTP, websocket, kafka, NATS, SQS, MQTT,
  GCP, Azure, Postgres, email, schedule).

OpenAPI spec gains the `force` field on each setmode/setenabled body.

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

* feat(forks): CLI --fork-triggers flag, fork-trigger docs, skill update

- Adds --fork-triggers boolean to wmill workspace fork; passes
  fork_triggers through to the create_fork API call.
- New docs/fork-triggers.md describing the model end-to-end (default,
  opt-in clone, merge-direction filter, conflict warning, future
  runtime-suffix work).
- Updates the adding-a-trigger SKILL.md to mention the fork-export
  ignore-keys participation and the clone_triggers_and_schedules
  block that new trigger kinds must extend.

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

* chore: regenerate sqlx offline query cache for fork-trigger SQL

* fix(forks): replace browser confirm() with ConfirmationModal for fork conflict

The fork-conflict warning previously used the browser's native confirm()
which doesn't match Windmill's design system. Switches to a singleton
ConfirmationModal mounted at the (logged) layout root, driven by a new
forkConflictModal store. The withForkConflictRetry helper now sets the
store and awaits the user's choice via a Promise, instead of blocking
on window.confirm.

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

* fix(forks): filter unchanged triggers in merge UI, add diff view, surface parent-only ones

The fork merge UI listed every trigger from the fork as a deployable item
regardless of whether it differed from the parent — so a fork created with
fork_triggers=true (which clones triggers in disabled state, otherwise
identical) showed every trigger as a "Fork-only" change. The 'Update
current' tab also missed triggers newly created in the parent that the
fork hadn't pulled yet.

This refactor:

- fetchAllTriggers now lists both fork and parent in parallel for each
  trigger kind, then merges by path.
- Computes a per-trigger `changeKind` (new / modified / deleted-in-source)
  using a JSON comparison that strips runtime + fork-local fields
  (mode/enabled/server_id/last_server_ping/edited_at/edited_by/etc.) so
  the disabled-on-clone difference doesn't show up as a change.
- Filters the trigger items in deployableItems by the current direction:
  Deploy mode shows fork-side new/modified, Update mode shows parent-side
  new/modified.
- Replaces the always-on "Fork-only" badge with proper New/Modified
  badges and surfaces a Diff button (modal Drawer + Monaco DiffEditor)
  for modified triggers — the diff strips the same ignored fields so
  users see only the meaningful config differences.

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

* fix(forks): always clone triggers/schedules disabled, drop opt-in flag

Disabled triggers and schedules are inert — no listener attaches, no cron
fires — so cloning them by default is safe by construction. Drops the
fork_triggers opt-in flag introduced earlier in this PR:

- Drops workspace.fork_triggers column (migration removed)
- Removes fork_triggers from CreateWorkspaceFork (API + OpenAPI)
- Removes the conditional in create_workspace_fork — clone always runs
- Removes the toggle from the fork-creation dialog
- Removes --fork-triggers from `wmill workspace fork`
- Updates docs/fork-triggers.md and adding-a-trigger SKILL.md

The merge UI continues to exclude triggers from the deploy/update default
selection, so a routine merge from a fork doesn't accidentally push
trigger config the user hasn't intentionally changed.

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

* fix(http-triggers): scope route exists check by workspace, skip non-workspaced clones in forks

The non-CLOUD branch of `route_path_key_exists` self-excluded by trigger
path alone, which silently masked cross-workspace collisions once forks
started cloning trigger rows verbatim. Tighten it to exclude only the
exact `(workspace_id, path)` row.

Fork creation also now skips non-workspaced HTTP triggers — their URL
has no workspace prefix, so a clone collides with the parent at the
matchit router (which silently drops one of two duplicates) and there is
no namespacing escape hatch. The clone copies all rows when CLOUD_HOSTED
or HTTP_ROUTE_WORKSPACED_ROUTE forces every route workspaced regardless.

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

* fix(forks-ui): silent cancel on enable conflict, clean up trigger rows in compare view

forkConflict helper now returns undefined when the user dismisses the
modal instead of throwing, so the redundant 'Cannot enable: undefined'
toast no longer appears.

CompareWorkspaces trigger rows now mirror the script row layout: drop
the redundant Disabled badge and the Trash/Details buttons (both belong
on the dedicated trigger pages, not in the deploy/compare view); pass
triggerKind through so RowIcon picks the right kind-specific icon; move
extraLabel into the summary line; replace the yellow Modified badge
with the same green ↗ ahead / blue ↘ behind treatment scripts use.

Trigger diff drawer: switch JSON → YAML for parity with DiffDrawer, fix
zero-height monaco render with className=!h-full, drop the redundant
Original/Modified label banner above the diff.

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

* fix(email-trigger): scope local_part exists check, skip non-workspaced clones in forks

Mirrors the HTTP route fix for the email-trigger non-CLOUD `email_exists`
check (in EE) which had the same path-only self-exclusion bug, and the
fork clone of `email_trigger` rows which copied non-workspaced
`local_part` verbatim. Skip non-workspaced rows in the clone unless the
instance is CLOUD_HOSTED (where lookup is workspace-scoped natively).

EE companion change in windmill-trigger-email/src/handler_ee.rs.

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

* chore: update ee-repo-ref to 78512dd73b4a1c9f70574cff863374179e3a621b

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

Previous ee-repo-ref: 1ac77f50747b58e720a11162dfd309bc252a24ab

New ee-repo-ref: 78512dd73b4a1c9f70574cff863374179e3a621b

Automated by sync-ee-ref workflow.

* fix(forks): always-warn on parent row, kind-specific modal copy, cancel-aware toggles

- Conflict check now fires whenever the parent has the path (regardless of
  parent's mode), since the cloned upstream identifier is shared by
  construction; closes the Postgres slot-takeover gap when the parent is
  disabled. Schedule's set_schedule_enabled gets the same treatment.
- Skip the warning entirely for HTTP and Email via a new
  TriggerCrud::FORK_CONFLICT_ON_ENABLE const — both kinds are workspace-
  scoped at runtime so cloned rows can't collide with the parent.
- Modal copy branches by failure family: split-events (Kafka/NATS/MQTT/SQS/
  GCP/Azure), duplicate-firing (Websocket/Schedule), slot-takeover
  (Postgres). Generic fallback for unknown kinds.
- withForkConflictRetry now returns boolean (true=committed, false=
  cancelled). TriggerModeToggle reuses its existing innerTriggerMode local
  state via a function binding for the regular Toggle, snapping back to
  the prop when onToggleMode signals a cancel — needed because the native
  bind:checked diverges from the parent's prop after a click and Svelte's
  reactivity won't re-push a same-valued prop down. Schedule list page
  uses {#key} on a reset version since it renders Toggle directly.
- Editor inners revert mode = previousMode on cancel; list pages skip the
  re-fetch (loadTriggers/loadSchedules) on cancel to avoid pointless
  network traffic and the schedule "Job stats loading..." flash.
- Drop withForkConflictRetry from HTTP and Email editors + list pages
  since the backend never emits the conflict for those kinds.

* fix(forks-ui): widen onToggleMode types, scope schedule toggle reset by path

- TriggerEditorToolbar and TriggerSuspendedJobsModal forwarded
  onToggleMode as `(mode) => void`, dropping the new boolean return so
  any caller wired through them would silently no-op the cancel-revert.
  Match the wider TriggerModeToggle signature.
- Schedule list page used a single resetVersion counter for every row's
  {#key}, so cancelling on any one schedule remounted every <Toggle> on
  the page. Switch to a per-path Record<string, number> bumped only for
  the affected row.

* chore: bump ee-repo-ref to c3a4553 (email FORK_CONFLICT_ON_ENABLE override)

* fix(forks): include Suspended in conflict gate, use parent_workspace_id for fork detection

Three fixes from the Claude review on PR #8976:

- Suspended mode still attaches the listener (it just pauses auto-run of
  queued jobs); two suspended fork+parent listeners would still split
  Kafka events / share a PG slot. Gate set_trigger_mode on
  `mode != Disabled` instead of `mode == Enabled` so Suspended also
  surfaces the warning.
- workspaces_export.rs::fork_*_ignore_keys keyed off the wm-fork-* prefix
  while set_trigger_mode and set_schedule_enabled key off
  parent_workspace_id. Switch the export filter to query
  parent_workspace_id once at the top of tarball_workspace and pass
  is_fork through. The column is the contract; the prefix is a
  creation-time naming convention that could in principle drift.
- TriggerModeToggle's suspend-dropdown action reassigned the non-bindable
  `triggerMode` prop instead of the local `innerTriggerMode` mirror,
  leaking inconsistent state if the dispatch was cancelled. Now writes
  to innerTriggerMode like the Toggle's on:change handler does.

* fix(cli): skip setScheduleEnabled when local YAML lacks `enabled`

Tarball export from a fork strips `enabled` from schedules so the
fork→parent git-sync round-trip can't flip the parent's operational
state. The CLI's pushSchedule called setScheduleEnabled whenever
`localSchedule.enabled != schedule.enabled`, which evaluates truthy
when local is undefined (fork-pulled YAML) and remote is true/false —
sending `{ enabled: undefined }` that serializes to `{}` and gets
rejected by the backend (`SetEnabled.enabled` is required).

Skip the call when `localSchedule.enabled === undefined` so a sync push
of fork-pulled YAMLs preserves the target's existing enabled state
instead of erroring out. Trigger updates were already safe — the
backend's update_trigger preserves `mode` when the request omits it.

* Revert "fix(cli): skip setScheduleEnabled when local YAML lacks `enabled`"

This reverts commit 23ba7e72fc.

* feat(cli): --force flag and friendlier error on fork-conflict for schedule enable

`wmill schedule enable foo/bar` against a fork whose parent has the same
path used to surface the raw `fork-conflict:schedule:<parent>` error
body. The CLI now:

- accepts `--force` to bypass the warning (mirrors the API field and the
  UI's "Enable anyway" confirmation),
- detects the `fork-conflict:` prefix on errors and prints a one-screen
  explanation pointing at --force instead of the raw body.

Disable doesn't trigger the warning (the gate fires only on transitions
to listener-attaching modes), so no flag there. Trigger enable/disable
isn't exposed as a standalone CLI command — sync push goes through
updateTrigger which has its own backend mode-preservation, so no
fork-conflict surfaces from the CLI for those.

* chore: regenerate cli-commands docs after adding --force to schedule enable

* chore: update ee-repo-ref to 967f961f0a88b027d894aebd03977181129477a8

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

Previous ee-repo-ref: c3a4553296473932e15392a06415dd7fb9aa6591

New ee-repo-ref: 967f961f0a88b027d894aebd03977181129477a8

Automated by sync-ee-ref workflow.

* fix(forks): address CI dead-code, claude/cubic review feedback

- backend: cfg-gate `fork_trigger_ignore_keys` to match its already-gated
  callsite. CI compiles with `-D warnings`, so the unused-fn under feature
  combos that disable all trigger crates was breaking check_oss/check_ee/
  cargo_test/test-linux/test-windows.
- cli: re-apply the `pushSchedule` undefined-skip (originally 23ba7e7,
  reverted in 4d172a1). Tarball export from forks strips `enabled`, so
  fork-pulled YAMLs that get sync-pushed back via `wmill schedule push`
  would otherwise serialize `{ enabled: undefined }` → `{}` and the
  backend's required `SetEnabled.enabled` rejects the body. Skipping
  preserves the target's existing flag, which is the round-trip-safe
  behavior. (`wmill workspace merge` extension to triggers/schedules is
  tracked in #9001 — until then sync push is the only CLI path.)
- TriggerModeToggle suspend-dropdown action awaits onToggleMode and
  resets `innerTriggerMode = triggerMode` on cancel, matching the Toggle
  on:change handler. Without this, dismissing the fork-conflict modal on
  a Suspend transition leaves the toggle stuck in 'suspended'.
- forkConflict: when a new modal opens with a previous resolver still
  pending, resolve the older promise to false. Avoids a dangling promise
  if the user clicks toggles on two rows in quick succession.
- schedules list: bump `toggleResetVersions[path]` on the
  permission-denied branch so the Toggle re-mounts back to the prop's
  `enabled` value. Without this, a user without write permission could
  click the toggle and have it stick visually flipped.
- docs/fork-triggers.md: switch the merge-direction filter description
  from `wm-fork-*` prefix to `parent_workspace_id IS NOT NULL` (matches
  the code after 4dd38fe). Drop the misleading "merge-direction filter
  strips identifier columns too" line in Future Work — the runtime
  suffix is applied at listener attach, the stored column never carries
  it, so no export filtering is needed there.

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
This commit is contained in:
hugocasa
2026-05-01 22:55:24 +02:00
committed by GitHub
parent 45d959a49e
commit d60dd745e4
59 changed files with 1782 additions and 348 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.claude/skills/adding-a-trigger/SKILL.md
+3 -1
View File
@@ -111,7 +111,9 @@ Regenerate frontend client: `npm run generate-backend-client` from `frontend/`.
**`backend/windmill-api-workspaces/src/workspaces.rs`** — add `{kind}_used: bool` to the `UsedTriggers` struct and add an `EXISTS(SELECT 1 FROM {kind}_trigger …)` to the `get_used_triggers` query.
**`backend/windmill-api/src/workspaces_export.rs`** — add export block mirroring gcp's (export lists all triggers, serializes them to YAML/JSON).
**`backend/windmill-api/src/workspaces_export.rs`** — add export block mirroring gcp's (export lists all triggers, serializes them to YAML/JSON). The block re-uses the `trigger_ignore_keys` variable so the new kind automatically participates in fork-export stripping (`mode` field is omitted when the source workspace is a fork — keeps fork→parent merges from flipping the parent's enabled state).
**Fork cloning (`clone_triggers_and_schedules` in workspaces.rs)** — add an `INSERT INTO {kind}_trigger ... SELECT ...` block that copies all rows from the parent workspace, forcing `mode = 'disabled'::TRIGGER_MODE`. Always runs at fork creation; forgetting this means users can't carry `{kind}` triggers into their forks.
## 6.5 Hardcoded trigger-kind arrays (silent-failure hotspots)
backend/.sqlx/query-0b347b021123e66ffb6b7eb690f7619daf34795410505609a1ff3bf0be953550.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO sqs_trigger (\n path, queue_url, aws_resource_path, message_attributes, script_path,\n is_flow, workspace_id, edited_by, edited_at, extra_perms, error,\n server_id, last_server_ping, aws_auth_resource_type, error_handler_path,\n error_handler_args, retry, mode, permissioned_as, labels\n )\n SELECT\n path, queue_url, aws_resource_path, message_attributes, script_path,\n is_flow, $1, edited_by, edited_at, extra_perms, NULL,\n NULL, NULL, aws_auth_resource_type, error_handler_path,\n error_handler_args, retry, 'disabled'::TRIGGER_MODE, permissioned_as, labels\n FROM sqs_trigger WHERE workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text"
]
},
"nullable": []
},
"hash": "0b347b021123e66ffb6b7eb690f7619daf34795410505609a1ff3bf0be953550"
}
backend/.sqlx/query-149b645af2324fc3140bf2662e75e579dcc8b928be3a2cc051e62aa2ddc09b1e.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO schedule (\n workspace_id, path, edited_by, edited_at, schedule, enabled, script_path,\n args, extra_perms, is_flow, email, error, timezone, on_failure,\n on_recovery, on_failure_times, on_failure_exact, on_failure_extra_args,\n on_recovery_times, on_recovery_extra_args, ws_error_handler_muted, retry,\n summary, no_flow_overlap, tag, paused_until, on_success, on_success_extra_args,\n cron_version, description, dynamic_skip, permissioned_as, labels\n )\n SELECT\n $1, path, edited_by, edited_at, schedule, FALSE, script_path,\n args, extra_perms, is_flow, email, error, timezone, on_failure,\n on_recovery, on_failure_times, on_failure_exact, on_failure_extra_args,\n on_recovery_times, on_recovery_extra_args, ws_error_handler_muted, retry,\n summary, no_flow_overlap, tag, paused_until, on_success, on_success_extra_args,\n cron_version, description, dynamic_skip, permissioned_as, labels\n FROM schedule WHERE workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text"
]
},
"nullable": []
},
"hash": "149b645af2324fc3140bf2662e75e579dcc8b928be3a2cc051e62aa2ddc09b1e"
}
backend/.sqlx/query-16e8d3f75ce4b5b18cefc25ebb670b506677681394606426964491488d64c62c.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO azure_trigger (\n azure_resource_path, azure_mode, scope_resource_id, topic_name,\n subscription_name, event_type_filters, push_auth_config, path, script_path,\n is_flow, workspace_id, edited_by, email, edited_at, extra_perms, server_id,\n last_server_ping, error, mode, permissioned_as, error_handler_path,\n error_handler_args, retry, labels\n )\n SELECT\n azure_resource_path, azure_mode, scope_resource_id, topic_name,\n subscription_name, event_type_filters, push_auth_config, path, script_path,\n is_flow, $1, edited_by, email, edited_at, extra_perms, NULL,\n NULL, NULL, 'disabled'::TRIGGER_MODE, permissioned_as, error_handler_path,\n error_handler_args, retry, labels\n FROM azure_trigger WHERE workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text"
]
},
"nullable": []
},
"hash": "16e8d3f75ce4b5b18cefc25ebb670b506677681394606426964491488d64c62c"
}
backend/.sqlx/query-1a66a5a9c2b7b75e783c59b4c2ed3f7f84adc67318fd0bc5cdee78b2c406ae81.json
Generated
+16
View File
@@ -0,0 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO email_trigger (\n path, local_part, workspaced_local_part, script_path, is_flow,\n workspace_id, edited_by, edited_at, extra_perms, error_handler_path,\n error_handler_args, retry, mode, permissioned_as, labels\n )\n SELECT\n path, local_part, workspaced_local_part, script_path, is_flow,\n $1, edited_by, edited_at, extra_perms, error_handler_path,\n error_handler_args, retry, 'disabled'::TRIGGER_MODE, permissioned_as, labels\n FROM email_trigger\n WHERE workspace_id = $2\n AND (workspaced_local_part IS TRUE OR $3)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "1a66a5a9c2b7b75e783c59b4c2ed3f7f84adc67318fd0bc5cdee78b2c406ae81"
}
backend/.sqlx/query-22699056871c6306f689d4bc6f9a67070baa4ff83de3b4a04c4d8d6054a59319.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT EXISTS(SELECT 1 FROM schedule WHERE workspace_id = $1 AND path = $2)",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "exists",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "22699056871c6306f689d4bc6f9a67070baa4ff83de3b4a04c4d8d6054a59319"
}
backend/.sqlx/query-2e67fa50d5d66cbca4ef74111f0ff9b51a6168adad3775f8d6cf40d02cb29d14.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO gcp_trigger (\n gcp_resource_path, topic_id, subscription_id, delivery_type,\n delivery_config, path, script_path, is_flow, workspace_id, edited_by,\n edited_at, extra_perms, server_id, last_server_ping, error,\n subscription_mode, error_handler_path, error_handler_args, retry,\n auto_acknowledge_msg, ack_deadline, mode, permissioned_as, labels\n )\n SELECT\n gcp_resource_path, topic_id, subscription_id, delivery_type,\n delivery_config, path, script_path, is_flow, $1, edited_by,\n edited_at, extra_perms, NULL, NULL, NULL,\n subscription_mode, error_handler_path, error_handler_args, retry,\n auto_acknowledge_msg, ack_deadline, 'disabled'::TRIGGER_MODE, permissioned_as, labels\n FROM gcp_trigger WHERE workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text"
]
},
"nullable": []
},
"hash": "2e67fa50d5d66cbca4ef74111f0ff9b51a6168adad3775f8d6cf40d02cb29d14"
}
backend/.sqlx/query-4d64c962e219f7cda8b93c3875acc78a4e17aae2c702ec1af6c3a4ae3f1e2716.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO nats_trigger (\n path, nats_resource_path, subjects, stream_name, consumer_name,\n use_jetstream, script_path, is_flow, workspace_id, edited_by, edited_at,\n extra_perms, server_id, last_server_ping, error, error_handler_path,\n error_handler_args, retry, mode, permissioned_as, labels\n )\n SELECT\n path, nats_resource_path, subjects, stream_name, consumer_name,\n use_jetstream, script_path, is_flow, $1, edited_by, edited_at,\n extra_perms, NULL, NULL, NULL, error_handler_path,\n error_handler_args, retry, 'disabled'::TRIGGER_MODE, permissioned_as, labels\n FROM nats_trigger WHERE workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text"
]
},
"nullable": []
},
"hash": "4d64c962e219f7cda8b93c3875acc78a4e17aae2c702ec1af6c3a4ae3f1e2716"
}
backend/.sqlx/query-5fd769bde29e88eb43f5dc0ecd7f4a1fe20d216d818d69eea4ddbaace5233137.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO websocket_trigger (\n path, url, script_path, is_flow, workspace_id, edited_by, edited_at,\n extra_perms, server_id, last_server_ping, error, filters, initial_messages,\n url_runnable_args, can_return_message, error_handler_path, error_handler_args,\n retry, can_return_error_result, mode, permissioned_as, filter_logic, labels,\n heartbeat\n )\n SELECT\n path, url, script_path, is_flow, $1, edited_by, edited_at,\n extra_perms, NULL, NULL, NULL, filters, initial_messages,\n url_runnable_args, can_return_message, error_handler_path, error_handler_args,\n retry, can_return_error_result, 'disabled'::TRIGGER_MODE, permissioned_as, filter_logic, labels,\n heartbeat\n FROM websocket_trigger WHERE workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text"
]
},
"nullable": []
},
"hash": "5fd769bde29e88eb43f5dc0ecd7f4a1fe20d216d818d69eea4ddbaace5233137"
}
backend/.sqlx/query-fe464b8b3ade86743d82c5e3fb14f457e07f07e44c7b693d5d755899d4210dee.json → backend/.sqlx/query-aa0a2f90d15a642ad3caaa3876d9cb4a5391ff8663da61b48bfeb73bfa005bbe.json
Generated
+3 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT EXISTS(\n SELECT 1\n FROM http_trigger\n WHERE\n ((workspaced_route IS TRUE AND workspace_id || '/' || route_path_key = $1)\n OR (workspaced_route IS FALSE AND route_path_key = $1))\n AND http_method = $2\n AND ($3::TEXT IS NULL OR path != $3)\n )\n ",
"query": "\n SELECT EXISTS(\n SELECT 1\n FROM http_trigger\n WHERE\n ((workspaced_route IS TRUE AND workspace_id || '/' || route_path_key = $1)\n OR (workspaced_route IS FALSE AND route_path_key = $1))\n AND http_method = $2\n AND ($3::TEXT IS NULL OR NOT (workspace_id = $4 AND path = $3))\n )\n ",
"describe": {
"columns": [
{
@@ -26,6 +26,7 @@
}
}
},
"Text",
"Text"
]
},
@@ -33,5 +34,5 @@
null
]
},
"hash": "fe464b8b3ade86743d82c5e3fb14f457e07f07e44c7b693d5d755899d4210dee"
"hash": "aa0a2f90d15a642ad3caaa3876d9cb4a5391ff8663da61b48bfeb73bfa005bbe"
}
backend/.sqlx/query-aee9bf16e37a5361f96d6b35dddffa6cc37346cdb6b38ecbf6e1530eef3a57bd.json
Generated
+16
View File
@@ -0,0 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO http_trigger (\n path, route_path, route_path_key, script_path, is_flow, workspace_id,\n edited_by, edited_at, extra_perms, authentication_method, http_method,\n static_asset_config, is_static_website, workspaced_route, wrap_body,\n raw_string, authentication_resource_path, summary, description,\n error_handler_path, error_handler_args, retry, request_type, mode,\n permissioned_as, labels\n )\n SELECT\n path, route_path, route_path_key, script_path, is_flow, $1,\n edited_by, edited_at, extra_perms, authentication_method, http_method,\n static_asset_config, is_static_website, workspaced_route, wrap_body,\n raw_string, authentication_resource_path, summary, description,\n error_handler_path, error_handler_args, retry, request_type, 'disabled'::TRIGGER_MODE,\n permissioned_as, labels\n FROM http_trigger\n WHERE workspace_id = $2\n AND (workspaced_route IS TRUE OR $3)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "aee9bf16e37a5361f96d6b35dddffa6cc37346cdb6b38ecbf6e1530eef3a57bd"
}
backend/.sqlx/query-b20e4486e2038ac8138f5d7435db8cbd78ac3d68674335db79df7fd8a4ba91fd.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO postgres_trigger (\n path, script_path, is_flow, workspace_id, edited_by, edited_at,\n extra_perms, postgres_resource_path, error, server_id, last_server_ping,\n replication_slot_name, publication_name, error_handler_path,\n error_handler_args, retry, mode, permissioned_as, labels\n )\n SELECT\n path, script_path, is_flow, $1, edited_by, edited_at,\n extra_perms, postgres_resource_path, NULL, NULL, NULL,\n replication_slot_name, publication_name, error_handler_path,\n error_handler_args, retry, 'disabled'::TRIGGER_MODE, permissioned_as, labels\n FROM postgres_trigger WHERE workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text"
]
},
"nullable": []
},
"hash": "b20e4486e2038ac8138f5d7435db8cbd78ac3d68674335db79df7fd8a4ba91fd"
}
backend/.sqlx/query-c8c8c457ee80125938af97993c4ca759b66507071163f8d98a778f417b930413.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO mqtt_trigger (\n mqtt_resource_path, subscribe_topics, client_version, v5_config, v3_config,\n client_id, path, script_path, is_flow, workspace_id, edited_by, edited_at,\n extra_perms, server_id, last_server_ping, error, error_handler_path,\n error_handler_args, retry, mode, permissioned_as, labels\n )\n SELECT\n mqtt_resource_path, subscribe_topics, client_version, v5_config, v3_config,\n client_id, path, script_path, is_flow, $1, edited_by, edited_at,\n extra_perms, NULL, NULL, NULL, error_handler_path,\n error_handler_args, retry, 'disabled'::TRIGGER_MODE, permissioned_as, labels\n FROM mqtt_trigger WHERE workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text"
]
},
"nullable": []
},
"hash": "c8c8c457ee80125938af97993c4ca759b66507071163f8d98a778f417b930413"
}
backend/.sqlx/query-c9c0b92c1fea9b4bdafba32da60e5579a4c2940bed63097ffec7f757d9882667.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO kafka_trigger (\n path, kafka_resource_path, topics, group_id, script_path, is_flow,\n workspace_id, edited_by, edited_at, extra_perms, server_id,\n last_server_ping, error, error_handler_path, error_handler_args, retry,\n mode, filters, auto_offset_reset, reset_offset, auto_commit,\n permissioned_as, filter_logic, labels\n )\n SELECT\n path, kafka_resource_path, topics, group_id, script_path, is_flow,\n $1, edited_by, edited_at, extra_perms, NULL,\n NULL, NULL, error_handler_path, error_handler_args, retry,\n 'disabled'::TRIGGER_MODE, filters, auto_offset_reset, reset_offset, auto_commit,\n permissioned_as, filter_logic, labels\n FROM kafka_trigger WHERE workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text"
]
},
"nullable": []
},
"hash": "c9c0b92c1fea9b4bdafba32da60e5579a4c2940bed63097ffec7f757d9882667"
}
backend/.sqlx/query-f57ff370f6775602d2c200d78650d65ffa5bfc5f10e8bd2a3162894c93283259.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT parent_workspace_id IS NOT NULL FROM workspace WHERE id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
null
]
},
"hash": "f57ff370f6775602d2c200d78650d65ffa5bfc5f10e8bd2a3162894c93283259"
}
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
26184ab7a4aadfc529dcedf038aa08d36c7ad381
967f961f0a88b027d894aebd03977181129477a8
backend/windmill-api-schedule/src/lib.rs
+37
View File
@@ -855,6 +855,38 @@ pub async fn set_enabled(
let mut tx = user_db.begin(&authed).await?;
let path = path.to_path();
check_scopes(&authed, || format!("schedules:write:{}", path))?;
// Block enabling a schedule in a fork when the parent has the same path
// (regardless of parent's enabled flag), unless force=true. Two enabled
// crons fire in lockstep; even when the parent is currently disabled the
// user is likely to re-enable it later, at which point both fire — better
// to surface that risk at every fork-side enable. There's no namespacing
// fix for schedules (Phase 3 doesn't help cron); the user has to confirm
// or point the script at fork-only side effects.
if payload.enabled && !payload.force {
let parent_id: Option<String> = sqlx::query_scalar!(
"SELECT parent_workspace_id FROM workspace WHERE id = $1",
&w_id
)
.fetch_optional(&mut *tx)
.await?
.flatten();
if let Some(parent_id) = parent_id {
let exists: Option<bool> = sqlx::query_scalar!(
"SELECT EXISTS(SELECT 1 FROM schedule WHERE workspace_id = $1 AND path = $2)",
&parent_id,
path,
)
.fetch_one(&mut *tx)
.await?;
if exists == Some(true) {
return Err(Error::BadRequest(format!(
"fork-conflict:schedule:{}",
parent_id
)));
}
}
}
// email is still written for backwards compat with old workers that don't know about permissioned_as
let schedule_o = sqlx::query_as!(
Schedule,
@@ -1281,6 +1313,11 @@ pub use windmill_queue::schedule::clear_schedule;
#[derive(Deserialize)]
pub struct SetEnabled {
pub enabled: bool,
/// Bypass the parent-state warning when enabling a schedule in a fork
/// whose parent has the same path enabled. The frontend sets this after
/// the user confirms the duplicate-firing dialog.
#[serde(default)]
pub force: bool,
}
// #[derive(Deserialize)]
backend/windmill-api-workspaces/src/workspaces.rs
+265
View File
@@ -30,6 +30,7 @@ use uuid::Uuid;
use windmill_audit::audit_oss::{audit_log, AuditAuthorable};
use windmill_audit::ActionKind;
use windmill_common::db::UserDB;
use windmill_common::global_settings::HTTP_ROUTE_WORKSPACED_ROUTE;
use windmill_common::users::username_to_permissioned_as;
use windmill_common::variables::{
build_crypt, decrypt, encrypt, SECRET_SALT, WORKSPACE_CRYPT_CACHE,
@@ -3857,6 +3858,264 @@ async fn clone_workspace_data(
Ok(())
}
/// Clone every trigger and schedule from the parent workspace, forcing
/// `mode='disabled'` / `enabled=false`. Always runs at fork creation —
/// disabled rows have no side effects, so cloning them is safe and lets
/// users re-enable selectively in the fork. Listener identifiers
/// (group_id, replication_slot_name, subscription_name, …) are copied
/// verbatim — the runtime suffix that prevents the fork from competing with
/// the parent ships in a follow-up PR.
async fn clone_triggers_and_schedules(
tx: &mut Transaction<'_, Postgres>,
source_workspace_id: &str,
target_workspace_id: &str,
) -> Result<()> {
sqlx::query!(
r#"INSERT INTO schedule (
workspace_id, path, edited_by, edited_at, schedule, enabled, script_path,
args, extra_perms, is_flow, email, error, timezone, on_failure,
on_recovery, on_failure_times, on_failure_exact, on_failure_extra_args,
on_recovery_times, on_recovery_extra_args, ws_error_handler_muted, retry,
summary, no_flow_overlap, tag, paused_until, on_success, on_success_extra_args,
cron_version, description, dynamic_skip, permissioned_as, labels
)
SELECT
$1, path, edited_by, edited_at, schedule, FALSE, script_path,
args, extra_perms, is_flow, email, error, timezone, on_failure,
on_recovery, on_failure_times, on_failure_exact, on_failure_extra_args,
on_recovery_times, on_recovery_extra_args, ws_error_handler_muted, retry,
summary, no_flow_overlap, tag, paused_until, on_success, on_success_extra_args,
cron_version, description, dynamic_skip, permissioned_as, labels
FROM schedule WHERE workspace_id = $2"#,
target_workspace_id,
source_workspace_id,
)
.execute(&mut **tx)
.await?;
// Skip non-workspaced HTTP triggers: their URL has no workspace prefix, so
// a clone would collide with the parent's row at runtime (matchit::Router
// silently drops one of two duplicates) and `route_path_key_exists` would
// also fail to spot the cross-workspace conflict cleanly. The instance
// settings `CLOUD_HOSTED` and `HTTP_ROUTE_WORKSPACED_ROUTE` force every
// route to be workspace-prefixed regardless of the column, so when either
// is on we clone everything.
let force_workspaced =
*CLOUD_HOSTED || HTTP_ROUTE_WORKSPACED_ROUTE.load(std::sync::atomic::Ordering::Relaxed);
sqlx::query!(
r#"INSERT INTO http_trigger (
path, route_path, route_path_key, script_path, is_flow, workspace_id,
edited_by, edited_at, extra_perms, authentication_method, http_method,
static_asset_config, is_static_website, workspaced_route, wrap_body,
raw_string, authentication_resource_path, summary, description,
error_handler_path, error_handler_args, retry, request_type, mode,
permissioned_as, labels
)
SELECT
path, route_path, route_path_key, script_path, is_flow, $1,
edited_by, edited_at, extra_perms, authentication_method, http_method,
static_asset_config, is_static_website, workspaced_route, wrap_body,
raw_string, authentication_resource_path, summary, description,
error_handler_path, error_handler_args, retry, request_type, 'disabled'::TRIGGER_MODE,
permissioned_as, labels
FROM http_trigger
WHERE workspace_id = $2
AND (workspaced_route IS TRUE OR $3)"#,
target_workspace_id,
source_workspace_id,
force_workspaced,
)
.execute(&mut **tx)
.await?;
sqlx::query!(
r#"INSERT INTO websocket_trigger (
path, url, script_path, is_flow, workspace_id, edited_by, edited_at,
extra_perms, server_id, last_server_ping, error, filters, initial_messages,
url_runnable_args, can_return_message, error_handler_path, error_handler_args,
retry, can_return_error_result, mode, permissioned_as, filter_logic, labels,
heartbeat
)
SELECT
path, url, script_path, is_flow, $1, edited_by, edited_at,
extra_perms, NULL, NULL, NULL, filters, initial_messages,
url_runnable_args, can_return_message, error_handler_path, error_handler_args,
retry, can_return_error_result, 'disabled'::TRIGGER_MODE, permissioned_as, filter_logic, labels,
heartbeat
FROM websocket_trigger WHERE workspace_id = $2"#,
target_workspace_id,
source_workspace_id,
)
.execute(&mut **tx)
.await?;
sqlx::query!(
r#"INSERT INTO kafka_trigger (
path, kafka_resource_path, topics, group_id, script_path, is_flow,
workspace_id, edited_by, edited_at, extra_perms, server_id,
last_server_ping, error, error_handler_path, error_handler_args, retry,
mode, filters, auto_offset_reset, reset_offset, auto_commit,
permissioned_as, filter_logic, labels
)
SELECT
path, kafka_resource_path, topics, group_id, script_path, is_flow,
$1, edited_by, edited_at, extra_perms, NULL,
NULL, NULL, error_handler_path, error_handler_args, retry,
'disabled'::TRIGGER_MODE, filters, auto_offset_reset, reset_offset, auto_commit,
permissioned_as, filter_logic, labels
FROM kafka_trigger WHERE workspace_id = $2"#,
target_workspace_id,
source_workspace_id,
)
.execute(&mut **tx)
.await?;
sqlx::query!(
r#"INSERT INTO nats_trigger (
path, nats_resource_path, subjects, stream_name, consumer_name,
use_jetstream, script_path, is_flow, workspace_id, edited_by, edited_at,
extra_perms, server_id, last_server_ping, error, error_handler_path,
error_handler_args, retry, mode, permissioned_as, labels
)
SELECT
path, nats_resource_path, subjects, stream_name, consumer_name,
use_jetstream, script_path, is_flow, $1, edited_by, edited_at,
extra_perms, NULL, NULL, NULL, error_handler_path,
error_handler_args, retry, 'disabled'::TRIGGER_MODE, permissioned_as, labels
FROM nats_trigger WHERE workspace_id = $2"#,
target_workspace_id,
source_workspace_id,
)
.execute(&mut **tx)
.await?;
sqlx::query!(
r#"INSERT INTO postgres_trigger (
path, script_path, is_flow, workspace_id, edited_by, edited_at,
extra_perms, postgres_resource_path, error, server_id, last_server_ping,
replication_slot_name, publication_name, error_handler_path,
error_handler_args, retry, mode, permissioned_as, labels
)
SELECT
path, script_path, is_flow, $1, edited_by, edited_at,
extra_perms, postgres_resource_path, NULL, NULL, NULL,
replication_slot_name, publication_name, error_handler_path,
error_handler_args, retry, 'disabled'::TRIGGER_MODE, permissioned_as, labels
FROM postgres_trigger WHERE workspace_id = $2"#,
target_workspace_id,
source_workspace_id,
)
.execute(&mut **tx)
.await?;
sqlx::query!(
r#"INSERT INTO mqtt_trigger (
mqtt_resource_path, subscribe_topics, client_version, v5_config, v3_config,
client_id, path, script_path, is_flow, workspace_id, edited_by, edited_at,
extra_perms, server_id, last_server_ping, error, error_handler_path,
error_handler_args, retry, mode, permissioned_as, labels
)
SELECT
mqtt_resource_path, subscribe_topics, client_version, v5_config, v3_config,
client_id, path, script_path, is_flow, $1, edited_by, edited_at,
extra_perms, NULL, NULL, NULL, error_handler_path,
error_handler_args, retry, 'disabled'::TRIGGER_MODE, permissioned_as, labels
FROM mqtt_trigger WHERE workspace_id = $2"#,
target_workspace_id,
source_workspace_id,
)
.execute(&mut **tx)
.await?;
sqlx::query!(
r#"INSERT INTO sqs_trigger (
path, queue_url, aws_resource_path, message_attributes, script_path,
is_flow, workspace_id, edited_by, edited_at, extra_perms, error,
server_id, last_server_ping, aws_auth_resource_type, error_handler_path,
error_handler_args, retry, mode, permissioned_as, labels
)
SELECT
path, queue_url, aws_resource_path, message_attributes, script_path,
is_flow, $1, edited_by, edited_at, extra_perms, NULL,
NULL, NULL, aws_auth_resource_type, error_handler_path,
error_handler_args, retry, 'disabled'::TRIGGER_MODE, permissioned_as, labels
FROM sqs_trigger WHERE workspace_id = $2"#,
target_workspace_id,
source_workspace_id,
)
.execute(&mut **tx)
.await?;
sqlx::query!(
r#"INSERT INTO gcp_trigger (
gcp_resource_path, topic_id, subscription_id, delivery_type,
delivery_config, path, script_path, is_flow, workspace_id, edited_by,
edited_at, extra_perms, server_id, last_server_ping, error,
subscription_mode, error_handler_path, error_handler_args, retry,
auto_acknowledge_msg, ack_deadline, mode, permissioned_as, labels
)
SELECT
gcp_resource_path, topic_id, subscription_id, delivery_type,
delivery_config, path, script_path, is_flow, $1, edited_by,
edited_at, extra_perms, NULL, NULL, NULL,
subscription_mode, error_handler_path, error_handler_args, retry,
auto_acknowledge_msg, ack_deadline, 'disabled'::TRIGGER_MODE, permissioned_as, labels
FROM gcp_trigger WHERE workspace_id = $2"#,
target_workspace_id,
source_workspace_id,
)
.execute(&mut **tx)
.await?;
sqlx::query!(
r#"INSERT INTO azure_trigger (
azure_resource_path, azure_mode, scope_resource_id, topic_name,
subscription_name, event_type_filters, push_auth_config, path, script_path,
is_flow, workspace_id, edited_by, email, edited_at, extra_perms, server_id,
last_server_ping, error, mode, permissioned_as, error_handler_path,
error_handler_args, retry, labels
)
SELECT
azure_resource_path, azure_mode, scope_resource_id, topic_name,
subscription_name, event_type_filters, push_auth_config, path, script_path,
is_flow, $1, edited_by, email, edited_at, extra_perms, NULL,
NULL, NULL, 'disabled'::TRIGGER_MODE, permissioned_as, error_handler_path,
error_handler_args, retry, labels
FROM azure_trigger WHERE workspace_id = $2"#,
target_workspace_id,
source_workspace_id,
)
.execute(&mut **tx)
.await?;
// Skip non-workspaced email triggers: same shape as the non-workspaced
// HTTP route case — a clone would share the same `local_part@domain`
// address as the parent, and incoming mail would arbitrarily land in one
// or the other. CLOUD_HOSTED scopes email lookup by workspace_id natively,
// so on cloud we clone everything.
sqlx::query!(
r#"INSERT INTO email_trigger (
path, local_part, workspaced_local_part, script_path, is_flow,
workspace_id, edited_by, edited_at, extra_perms, error_handler_path,
error_handler_args, retry, mode, permissioned_as, labels
)
SELECT
path, local_part, workspaced_local_part, script_path, is_flow,
$1, edited_by, edited_at, extra_perms, error_handler_path,
error_handler_args, retry, 'disabled'::TRIGGER_MODE, permissioned_as, labels
FROM email_trigger
WHERE workspace_id = $2
AND (workspaced_local_part IS TRUE OR $3)"#,
target_workspace_id,
source_workspace_id,
*CLOUD_HOSTED,
)
.execute(&mut **tx)
.await?;
Ok(())
}
async fn update_workspace_settings(
tx: &mut Transaction<'_, Postgres>,
source_workspace_id: &str,
@@ -4724,6 +4983,12 @@ async fn create_workspace_fork(
// Clone all data from the parent workspace using Rust implementation
clone_workspace_data(&mut tx, &parent_workspace_id, &forked_id).await?;
// Clone triggers and schedules unconditionally, always with mode='disabled' /
// enabled=false. Disabled rows have no side effects (no listener
// attaches, no cron fires) so this is safe by construction. The user
// re-enables in the fork, with parent-conflict warnings on enable.
clone_triggers_and_schedules(&mut tx, &parent_workspace_id, &forked_id).await?;
// Update forked datatable settings to point to new databases
for fdt in &nw.forked_datatables {
apply_forked_datatable(&db, &mut tx, &parent_workspace_id, &forked_id, fdt).await?;
backend/windmill-api/openapi.yaml
+55
View File
@@ -13217,6 +13217,11 @@ paths:
properties:
enabled:
type: boolean
force:
type: boolean
description: >
Bypass the parent-state conflict warning when enabling a
schedule in a fork whose parent has the same path enabled.
required:
- enabled
@@ -13663,6 +13668,11 @@ paths:
properties:
mode:
$ref: "#/components/schemas/TriggerMode"
force:
type: boolean
description: >
Bypass the parent-state conflict warning when enabling a
trigger in a fork whose parent has the same path enabled.
required:
- mode
responses:
@@ -13830,6 +13840,11 @@ paths:
properties:
mode:
$ref: "#/components/schemas/TriggerMode"
force:
type: boolean
description: >
Bypass the parent-state conflict warning when enabling a
trigger in a fork whose parent has the same path enabled.
required:
- mode
responses:
@@ -14030,6 +14045,11 @@ paths:
properties:
mode:
$ref: "#/components/schemas/TriggerMode"
force:
type: boolean
description: >
Bypass the parent-state conflict warning when enabling a
trigger in a fork whose parent has the same path enabled.
required:
- mode
responses:
@@ -14271,6 +14291,11 @@ paths:
properties:
mode:
$ref: "#/components/schemas/TriggerMode"
force:
type: boolean
description: >
Bypass the parent-state conflict warning when enabling a
trigger in a fork whose parent has the same path enabled.
required:
- mode
responses:
@@ -14466,6 +14491,11 @@ paths:
properties:
mode:
$ref: "#/components/schemas/TriggerMode"
force:
type: boolean
description: >
Bypass the parent-state conflict warning when enabling a
trigger in a fork whose parent has the same path enabled.
required:
- mode
responses:
@@ -15254,6 +15284,11 @@ paths:
properties:
mode:
$ref: "#/components/schemas/TriggerMode"
force:
type: boolean
description: >
Bypass the parent-state conflict warning when enabling a
trigger in a fork whose parent has the same path enabled.
required:
- mode
responses:
@@ -15449,6 +15484,11 @@ paths:
properties:
mode:
$ref: "#/components/schemas/TriggerMode"
force:
type: boolean
description: >
Bypass the parent-state conflict warning when enabling a
trigger in a fork whose parent has the same path enabled.
required:
- mode
responses:
@@ -15703,6 +15743,11 @@ paths:
properties:
mode:
$ref: "#/components/schemas/TriggerMode"
force:
type: boolean
description: >
Bypass the parent-state conflict warning when enabling a
trigger in a fork whose parent has the same path enabled.
required:
- mode
responses:
@@ -16249,6 +16294,11 @@ paths:
properties:
mode:
$ref: "#/components/schemas/TriggerMode"
force:
type: boolean
description: >
Bypass the parent-state conflict warning when enabling a
trigger in a fork whose parent has the same path enabled.
required:
- mode
responses:
@@ -16475,6 +16525,11 @@ paths:
properties:
mode:
$ref: "#/components/schemas/TriggerMode"
force:
type: boolean
description: >
Bypass the parent-state conflict warning when enabling a
trigger in a fork whose parent has the same path enabled.
required:
- mode
responses:
backend/windmill-api/src/workspaces_export.rs
+112 -12
View File
@@ -119,6 +119,45 @@ pub fn is_none_or_false(val: &Option<bool>) -> bool {
}
}
/// Returns the keys to strip from trigger/schedule serialization when the
/// source workspace is a fork. Stripping these keys avoids propagating
/// fork-local operational state (enabled flag, runtime listener identifiers)
/// back to the parent workspace through the git-sync round-trip.
#[cfg(any(
feature = "http_trigger",
feature = "websocket",
feature = "postgres_trigger",
feature = "mqtt_trigger",
feature = "native_trigger",
all(
feature = "enterprise",
any(
feature = "kafka",
feature = "sqs_trigger",
feature = "gcp_trigger",
feature = "azure_trigger",
feature = "nats",
feature = "smtp",
),
feature = "private"
)
))]
fn fork_trigger_ignore_keys(is_fork: bool) -> Option<Vec<&'static str>> {
if is_fork {
Some(vec!["mode", "enabled"])
} else {
None
}
}
fn fork_schedule_ignore_keys(is_fork: bool) -> Option<Vec<&'static str>> {
if is_fork {
Some(vec!["enabled"])
} else {
None
}
}
enum ArchiveImpl {
#[cfg(feature = "zip")]
Zip(async_zip::tokio::write::ZipFileWriter<tokio::fs::File>),
@@ -415,6 +454,19 @@ pub(crate) async fn tarball_workspace(
let mut tx = user_db.begin(&authed).await?;
// Source-of-truth check for fork-ness: the workspace's parent_workspace_id
// column. The wm-fork-* prefix is a creation-time naming convention that
// could in principle drift (rename, manual SQL); the column is the
// contract that matches what the conflict-warning gates read.
let is_fork: bool = sqlx::query_scalar!(
"SELECT parent_workspace_id IS NOT NULL FROM workspace WHERE id = $1",
&w_id
)
.fetch_optional(&mut *tx)
.await?
.flatten()
.unwrap_or(false);
let tmp_dir = TempDir::new_in(&*WINDMILL_DIR)?;
let name = match archive_type.as_deref() {
@@ -682,8 +734,11 @@ pub(crate) async fn tarball_workspace(
.fetch_all(&mut *tx)
.await?;
let schedule_ignore_keys = fork_schedule_ignore_keys(is_fork);
for schedule in schedules {
let app_str = &to_string_without_metadata(&schedule, false, None).unwrap();
let app_str =
&to_string_without_metadata(&schedule, false, schedule_ignore_keys.clone())
.unwrap();
archive
.write_to_archive(&app_str, &format!("{}.schedule.json", schedule.path))
.await?;
@@ -691,6 +746,27 @@ pub(crate) async fn tarball_workspace(
}
if include_triggers.unwrap_or(false) {
#[cfg(any(
feature = "http_trigger",
feature = "websocket",
feature = "postgres_trigger",
feature = "mqtt_trigger",
feature = "native_trigger",
all(
feature = "enterprise",
any(
feature = "kafka",
feature = "sqs_trigger",
feature = "gcp_trigger",
feature = "azure_trigger",
feature = "nats",
feature = "smtp",
),
feature = "private"
)
))]
let trigger_ignore_keys = fork_trigger_ignore_keys(is_fork);
#[cfg(feature = "http_trigger")]
{
use crate::triggers::http::HttpTrigger;
@@ -698,7 +774,9 @@ pub(crate) async fn tarball_workspace(
let http_triggers = handler.list_triggers(&mut *tx, &w_id, None).await?;
for trigger in http_triggers {
let trigger_str = &to_string_without_metadata(&trigger, false, None).unwrap();
let trigger_str =
&to_string_without_metadata(&trigger, false, trigger_ignore_keys.clone())
.unwrap();
archive
.write_to_archive(
&trigger_str,
@@ -715,7 +793,9 @@ pub(crate) async fn tarball_workspace(
let websocket_triggers = handler.list_triggers(&mut *tx, &w_id, None).await?;
for trigger in websocket_triggers {
let trigger_str = &to_string_without_metadata(&trigger, false, None).unwrap();
let trigger_str =
&to_string_without_metadata(&trigger, false, trigger_ignore_keys.clone())
.unwrap();
archive
.write_to_archive(
&trigger_str,
@@ -732,7 +812,9 @@ pub(crate) async fn tarball_workspace(
let kafka_triggers = handler.list_triggers(&mut *tx, &w_id, None).await?;
for trigger in kafka_triggers {
let trigger_str = &to_string_without_metadata(&trigger, false, None).unwrap();
let trigger_str =
&to_string_without_metadata(&trigger, false, trigger_ignore_keys.clone())
.unwrap();
archive
.write_to_archive(
&trigger_str,
@@ -749,7 +831,9 @@ pub(crate) async fn tarball_workspace(
let sqs_triggers = handler.list_triggers(&mut *tx, &w_id, None).await?;
for trigger in sqs_triggers {
let trigger_str = &to_string_without_metadata(&trigger, false, None).unwrap();
let trigger_str =
&to_string_without_metadata(&trigger, false, trigger_ignore_keys.clone())
.unwrap();
archive
.write_to_archive(
&trigger_str,
@@ -766,7 +850,9 @@ pub(crate) async fn tarball_workspace(
let gcp_triggers = handler.list_triggers(&mut *tx, &w_id, None).await?;
for trigger in gcp_triggers {
let trigger_str = &to_string_without_metadata(&trigger, false, None).unwrap();
let trigger_str =
&to_string_without_metadata(&trigger, false, trigger_ignore_keys.clone())
.unwrap();
archive
.write_to_archive(
&trigger_str,
@@ -783,7 +869,9 @@ pub(crate) async fn tarball_workspace(
let azure_triggers = handler.list_triggers(&mut *tx, &w_id, None).await?;
for trigger in azure_triggers {
let trigger_str = &to_string_without_metadata(&trigger, false, None).unwrap();
let trigger_str =
&to_string_without_metadata(&trigger, false, trigger_ignore_keys.clone())
.unwrap();
archive
.write_to_archive(
&trigger_str,
@@ -801,7 +889,8 @@ pub(crate) async fn tarball_workspace(
for trigger in nats_triggers {
let trigger_str: &String =
&to_string_without_metadata(&trigger, false, None).unwrap();
&to_string_without_metadata(&trigger, false, trigger_ignore_keys.clone())
.unwrap();
archive
.write_to_archive(
&trigger_str,
@@ -818,7 +907,9 @@ pub(crate) async fn tarball_workspace(
let postgres_triggers = handler.list_triggers(&mut *tx, &w_id, None).await?;
for trigger in postgres_triggers {
let trigger_str = &to_string_without_metadata(&trigger, false, None).unwrap();
let trigger_str =
&to_string_without_metadata(&trigger, false, trigger_ignore_keys.clone())
.unwrap();
archive
.write_to_archive(
&trigger_str,
@@ -835,7 +926,9 @@ pub(crate) async fn tarball_workspace(
let mqtt_triggers = handler.list_triggers(&mut *tx, &w_id, None).await?;
for trigger in mqtt_triggers {
let trigger_str = &to_string_without_metadata(&trigger, false, None).unwrap();
let trigger_str =
&to_string_without_metadata(&trigger, false, trigger_ignore_keys.clone())
.unwrap();
archive
.write_to_archive(
&trigger_str,
@@ -852,7 +945,9 @@ pub(crate) async fn tarball_workspace(
let email_triggers = handler.list_triggers(&mut *tx, &w_id, None).await?;
for trigger in email_triggers {
let trigger_str = &to_string_without_metadata(&trigger, false, None).unwrap();
let trigger_str =
&to_string_without_metadata(&trigger, false, trigger_ignore_keys.clone())
.unwrap();
archive
.write_to_archive(
&trigger_str,
@@ -872,11 +967,16 @@ pub(crate) async fn tarball_workspace(
list_native_triggers(&mut *tx, &w_id, service_name, None, None, None, None)
.await?;
let mut native_ignore_keys = vec!["webhook_token_hash"];
if let Some(ref extra) = trigger_ignore_keys {
native_ignore_keys.extend_from_slice(extra);
}
for trigger in native_triggers {
let trigger_str = &to_string_without_metadata(
&trigger,
false,
Some(vec!["webhook_token_hash"]),
Some(native_ignore_keys.clone()),
)
.unwrap();
archive
backend/windmill-trigger-http/src/handler.rs
+17 -5
View File
@@ -62,7 +62,8 @@ pub async fn route_path_key_exists(
.await?
.unwrap_or(false)
} else {
let http_route_workspaced = HTTP_ROUTE_WORKSPACED_ROUTE.load(std::sync::atomic::Ordering::Relaxed);
let http_route_workspaced =
HTTP_ROUTE_WORKSPACED_ROUTE.load(std::sync::atomic::Ordering::Relaxed);
let effective_workspaced = workspaced_route.unwrap_or(false) || http_route_workspaced;
let route_path_key = if effective_workspaced {
std::borrow::Cow::Owned(format!("{}/{}", w_id, route_path_key.trim_matches('/')))
@@ -70,6 +71,10 @@ pub async fn route_path_key_exists(
std::borrow::Cow::Borrowed(route_path_key)
};
// Self-exclusion is by `(workspace_id, path)` not just `path`: workspace
// forks clone trigger rows verbatim, so the same trigger path can exist
// in multiple workspaces. Excluding by path alone would silently mask a
// real collision against the parent's row.
sqlx::query_scalar!(
r#"
SELECT EXISTS(
@@ -79,12 +84,13 @@ pub async fn route_path_key_exists(
((workspaced_route IS TRUE AND workspace_id || '/' || route_path_key = $1)
OR (workspaced_route IS FALSE AND route_path_key = $1))
AND http_method = $2
AND ($3::TEXT IS NULL OR path != $3)
AND ($3::TEXT IS NULL OR NOT (workspace_id = $4 AND path = $3))
)
"#,
&route_path_key,
http_method as &HttpMethod,
trigger_path
trigger_path,
w_id
)
.fetch_one(db)
.await?
@@ -145,7 +151,8 @@ async fn require_admin_for_instance_wide_route(
is_admin: bool,
workspaced_route: Option<bool>,
) -> Result<bool> {
let http_route_workspaced = HTTP_ROUTE_WORKSPACED_ROUTE.load(std::sync::atomic::Ordering::Relaxed);
let http_route_workspaced =
HTTP_ROUTE_WORKSPACED_ROUTE.load(std::sync::atomic::Ordering::Relaxed);
let effective_workspaced = workspaced_route.unwrap_or(false) || http_route_workspaced;
if !is_admin && !effective_workspaced {
return Err(Error::NotAuthorized(
@@ -371,6 +378,10 @@ impl TriggerCrud for HttpTrigger {
const ROUTE_PREFIX: &'static str = "/http_triggers";
const DEPLOYMENT_NAME: &'static str = "HTTP trigger";
const IS_ALLOWED_ON_CLOUD: bool = true;
// Cloned HTTP triggers are always workspaced (the clone filter excludes
// workspaced_route=false rows), so fork and parent live at distinct URLs
// and never collide.
const FORK_CONFLICT_ON_ENABLE: bool = false;
const ADDITIONAL_SELECT_FIELDS: &[&'static str] = &[
"route_path",
"route_path_key",
@@ -465,7 +476,8 @@ impl TriggerCrud for HttpTrigger {
let resolved_edited_by = trigger.base.resolve_edited_by(authed);
let resolved_permissioned_as = trigger.base.resolve_permissioned_as(authed);
let http_route_workspaced = HTTP_ROUTE_WORKSPACED_ROUTE.load(std::sync::atomic::Ordering::Relaxed);
let http_route_workspaced =
HTTP_ROUTE_WORKSPACED_ROUTE.load(std::sync::atomic::Ordering::Relaxed);
let effective_workspaced =
trigger.config.workspaced_route.unwrap_or(false) || http_route_workspaced;
backend/windmill-trigger/src/handler.rs
+96 -1
View File
@@ -60,6 +60,15 @@ pub trait TriggerCrud: Send + Sync + 'static {
const DEPLOYMENT_NAME: &'static str;
const ADDITIONAL_SELECT_FIELDS: &[&'static str] = &[];
const IS_ALLOWED_ON_CLOUD: bool;
/// Whether enabling this trigger in a fork while the parent has the same
/// path enabled is a real conflict (shared upstream resource). True for
/// listener-based kinds where two consumers compete (Kafka group, PG slot,
/// SQS queue, etc.) and for Websocket where both subscribers fire on every
/// broadcast. False for kinds whose upstream identifier is implicitly
/// workspace-scoped at runtime (HTTP routes, Email local_part — clones for
/// the non-workspaced sub-case are filtered out, so any cloned row is
/// already collision-free vs. the parent).
const FORK_CONFLICT_ON_ENABLE: bool = true;
fn get_deployed_object(path: String, parent_path: Option<String>) -> DeployedObject;
@@ -557,7 +566,7 @@ async fn update_trigger<T: TriggerCrud>(
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
Path((workspace_id, path)): Path<(String, StripPath)>,
Json(edit_trigger): Json<TriggerData<T::TriggerConfigRequest>>,
Json(mut edit_trigger): Json<TriggerData<T::TriggerConfigRequest>>,
) -> Result<String> {
let path = path.to_path();
check_scopes(&authed, || {
@@ -574,6 +583,24 @@ async fn update_trigger<T: TriggerCrud>(
let mut tx = user_db.begin(&authed).await?;
// When the request omits `mode`/`enabled`, preserve the existing DB value
// instead of falling back to the BaseTriggerData default (Enabled). This
// keeps fork→parent git-sync round-trips from flipping the parent's
// operational state — see fork_trigger_ignore_keys in workspaces_export.rs.
if edit_trigger.base.is_mode_unspecified() {
let existing_mode: Option<TriggerMode> = sqlx::query_scalar(&format!(
"SELECT mode FROM {} WHERE workspace_id = $1 AND path = $2",
T::TABLE_NAME
))
.bind(&workspace_id)
.bind(path)
.fetch_optional(&mut *tx)
.await?;
if let Some(m) = existing_mode {
edit_trigger.base.set_mode(m);
}
}
let new_path = edit_trigger.base.path.to_string();
let labels = edit_trigger.base.labels.clone();
let on_behalf_of_info = windmill_common::check_on_behalf_of_preservation(
@@ -732,6 +759,52 @@ async fn exists_trigger<T: TriggerCrud>(
#[derive(serde::Deserialize)]
struct SetTriggerModePayload {
mode: TriggerMode,
/// When true, bypass the parent-state warning that would otherwise reject
/// enabling a trigger that's already enabled in the parent workspace.
/// The frontend sets this after the user confirms the duplicate-execution
/// dialog. See windmill-trigger/src/handler.rs::set_trigger_mode for the
/// full check.
#[serde(default)]
force: bool,
}
/// Returns the parent workspace id when this workspace is a fork *and* the
/// parent has a row at the same trigger path. Used to gate enabling a trigger
/// in a fork behind an explicit `force=true` confirmation: the fork's row was
/// cloned from the parent, so its upstream identifier (Kafka group, PG slot,
/// SQS queue URL, etc.) is shared by construction. The risk is independent of
/// the parent's current `mode`: if the parent is enabled, the two listeners
/// compete; if it's disabled, the fork can destructively take over shared
/// state (e.g. advance the PG WAL, claim an MQTT client_id) before the parent
/// re-enables. Either way, the user should be asked to confirm.
async fn parent_has_trigger(
tx: &mut PgConnection,
table_name: &str,
workspace_id: &str,
path: &str,
) -> Result<Option<String>> {
let parent: Option<String> =
sqlx::query_scalar("SELECT parent_workspace_id FROM workspace WHERE id = $1")
.bind(workspace_id)
.fetch_optional(&mut *tx)
.await?
.flatten();
let Some(parent_id) = parent else {
return Ok(None);
};
let exists: Option<bool> = sqlx::query_scalar(&format!(
"SELECT EXISTS(SELECT 1 FROM {} WHERE workspace_id = $1 AND path = $2)",
table_name
))
.bind(&parent_id)
.bind(path)
.fetch_one(&mut *tx)
.await?;
Ok(if exists == Some(true) {
Some(parent_id)
} else {
None
})
}
async fn set_trigger_mode<T: TriggerCrud>(
@@ -746,6 +819,28 @@ async fn set_trigger_mode<T: TriggerCrud>(
check_scopes(&authed, || format!("{}:write", T::scope_domain_name()))?;
let mut tx = user_db.begin(&authed).await?;
// Block transitioning a trigger in a fork to any mode that attaches a
// listener (Enabled or Suspended) when the parent has the same path,
// unless the caller passes force=true. Suspended still keeps the
// listener attached — it just stops auto-running queued jobs — so a
// suspended fork would still split Kafka events / share a PG slot
// with the parent. The cloned upstream identifier is shared by
// construction; the risk is independent of the parent's current mode.
// Skipped for kinds where the upstream identifier is already
// workspace-scoped at runtime (HTTP, Email).
if T::FORK_CONFLICT_ON_ENABLE && payload.mode != TriggerMode::Disabled && !payload.force {
if let Some(parent_id) =
parent_has_trigger(&mut *tx, T::TABLE_NAME, &workspace_id, path).await?
{
return Err(Error::BadRequest(format!(
"fork-conflict:{}:{}",
T::TRIGGER_TYPE,
parent_id
)));
}
}
let updated = handler
.set_trigger_mode(&authed, &mut *tx, &workspace_id, path, &payload.mode)
.await?;
backend/windmill-trigger/src/types.rs
+15
View File
@@ -128,6 +128,21 @@ impl BaseTriggerData {
)
}
/// True when neither `mode` nor the legacy `enabled` field was provided in
/// the request. Used by the update path to distinguish "explicitly Enabled"
/// from "missing — preserve existing value", which matters for git-sync
/// round-trips through fork workspaces (see workspaces_export.rs).
pub fn is_mode_unspecified(&self) -> bool {
#[allow(deprecated)]
{
self.mode.is_none() && self.enabled.is_none()
}
}
pub fn set_mode(&mut self, mode: TriggerMode) {
self.mode = Some(mode);
}
pub fn resolve_permissioned_as(&self, authed: &impl Authable) -> String {
if let Some(ref permissioned_as) = self.permissioned_as {
if self.preserve_permissioned_as.unwrap_or(false)
cli/src/commands/schedule/schedule.ts
+45 -7
View File
@@ -153,7 +153,17 @@ export async function pushSchedule(
...preserveFields,
},
});
if (localSchedule.enabled != schedule.enabled) {
// Tarball export from a fork strips `enabled` from schedule YAMLs so
// the fork→parent git-sync round-trip can't flip the parent's state.
// Skip the secondary setScheduleEnabled call when the local YAML
// doesn't carry `enabled` — sending `{ enabled: undefined }` would
// serialize to `{}` and the backend (`SetEnabled.enabled` is required)
// would reject the request. Preserving the target's existing flag is
// exactly the round-trip-safe behavior.
if (
localSchedule.enabled !== undefined &&
localSchedule.enabled !== schedule.enabled
) {
log.info(colors.bold.yellow(
`Schedule ${path} is ${localSchedule.enabled ? "enabled" : "disabled"} locally but not on remote, updating remote`
));
@@ -187,20 +197,44 @@ export async function pushSchedule(
}
}
async function enable(opts: GlobalOptions, path: string) {
async function enable(opts: GlobalOptions & { force?: boolean }, path: string) {
opts = await mergeConfigWithConfigFile(opts);
const workspace = await resolveWorkspace(opts);
await requireLogin(opts);
await wmill.setScheduleEnabled({
workspace: workspace.workspaceId,
path,
requestBody: { enabled: true },
});
try {
await wmill.setScheduleEnabled({
workspace: workspace.workspaceId,
path,
requestBody: { enabled: true, force: opts.force },
});
} catch (e) {
const conflict = parseForkConflict(e);
if (conflict) {
log.error(
`Cannot enable schedule '${path}': the parent workspace '${conflict.parentWorkspaceId}' has the same path configured. ` +
`Both crons would fire on every tick and the script would run twice per scheduled time.\n` +
`Re-run with --force to enable anyway.`
);
process.exit(1);
}
throw e;
}
log.info(colors.green(`Schedule ${path} enabled.`));
}
/** Parse a backend error body of the shape `fork-conflict:<kind>:<parent_workspace_id>`. */
function parseForkConflict(
e: unknown
): { kind: string; parentWorkspaceId: string } | undefined {
const body = (e as any)?.body;
const raw = typeof body === "string" ? body : (e as any)?.message ?? "";
const m = String(raw).match(/fork-conflict:([^:]+):(.+)/);
if (!m) return undefined;
return { kind: m[1], parentWorkspaceId: m[2].trim() };
}
async function disable(opts: GlobalOptions, path: string) {
opts = await mergeConfigWithConfigFile(opts);
const workspace = await resolveWorkspace(opts);
@@ -260,6 +294,10 @@ const command = new Command()
.arguments("<file_path:string> <remote_path:string>")
.action(push as any)
.command("enable", "Enable a schedule")
.option(
"--force",
"Bypass the fork-conflict warning when the parent workspace has the same schedule (acknowledges that both crons will fire)"
)
.arguments("<path:string>")
.action(enable as any)
.command("disable", "Disable a schedule")
cli/src/guidance/skills.gen.ts
+1
View File
@@ -6756,6 +6756,7 @@ schedule related commands
- \`schedule new <path:string>\` - create a new schedule locally
- \`schedule push <file_path:string> <remote_path:string>\` - push a local schedule spec. This overrides any remote versions.
- \`schedule enable <path:string>\` - Enable a schedule
- \`--force\` - Bypass the fork-conflict warning when the parent workspace has the same schedule (acknowledges that both crons will fire)
- \`schedule disable <path:string>\` - Disable a schedule
- \`schedule set-permissioned-as <path:string> <email:string>\` - Set the email (run-as user) for a schedule (requires admin or wm_deployers group)
docs/fork-triggers.md
+173
View File
@@ -0,0 +1,173 @@
# Triggers and schedules in workspace forks
A workspace fork is a developer-controlled copy of a parent workspace, used to
test changes before merging back via git sync. Triggers and schedules in a
fork need special handling for two reasons:
1. **Listener take-over**: most trigger kinds (Kafka, Postgres, MQTT, NATS,
SQS, GCP, Azure) attach to a stateful upstream resource. Two listeners
sharing the same identifier compete for events.
2. **Merge round-trip**: every change made in a fork can flow back to the
parent through git sync. If the fork sets a trigger to `disabled`, that
value would otherwise overwrite the parent's `enabled` state on merge.
## Cloning model — always cloned, always disabled
Fork creation always runs `clone_triggers_and_schedules`. Every row in each
`*_trigger` table and in `schedule` is copied from the parent into the fork
with two invariants:
- **Always disabled.** The clone forces `mode='disabled'::TRIGGER_MODE` on
triggers and `enabled=false` on schedules, regardless of the parent's
state. Disabled rows have **no side effects** — no listener attaches to
the upstream, no cron fires — so this clone is safe by construction.
The user re-enables manually in the fork.
- **Listener identifiers copied verbatim.** Stored values for `group_id`,
`replication_slot_name`, `subscription_name`, `client_id`,
`consumer_name`, etc. are copied 1:1. Until the runtime-suffix work
ships (see below), enabling a cloned listener in the fork would compete
with the parent — the conflict warning below catches that case.
`native_trigger` (Nextcloud, Google Drive, GitHub) is intentionally **not
cloned**. Those triggers manage external webhook state we don't want
duplicated.
**Non-workspaced HTTP triggers are also skipped.** A row with
`workspaced_route=false` (and where neither `CLOUD_HOSTED` nor the
`HTTP_ROUTE_WORKSPACED_ROUTE` instance setting is on) has a runtime URL
without any workspace prefix. A clone would collide with the parent's row at
the matchit router level, where duplicate inserts are silently dropped — one
trigger would invisibly hijack the other. There is no namespacing escape
hatch for these (the whole point of `workspaced_route=false` is to skip the
prefix), so the clone filter excludes them. The fork user can re-create one
manually if they need it. When `CLOUD_HOSTED` or `HTTP_ROUTE_WORKSPACED_ROUTE`
is on, every route is workspace-prefixed at runtime regardless of the column,
and the clone copies all rows.
**Non-workspaced email triggers are skipped on the same grounds.** A row
with `workspaced_local_part=false` exposes a bare `local_part@domain`
address shared instance-wide; a clone would share the address with the
parent and incoming mail would be delivered arbitrarily. The clone filter
copies email triggers only when `workspaced_local_part IS TRUE` (or
`CLOUD_HOSTED`, since cloud scopes email lookup by `workspace_id` natively).
## Merge-direction filter (always on)
Whenever the source workspace has `parent_workspace_id IS NOT NULL` (i.e.
it's a fork), the tarball export at `/api/w/{workspace}/workspaces/tarball`
strips fork-local fields:
- `mode` from every `*_trigger` row
- `enabled` from every `schedule` row
The fork-detection key is the column, not the `wm-fork-*` naming convention,
so it stays consistent with the conflict-warning gates in `set_trigger_mode`
and `set_schedule_enabled` and survives any future ID rename.
The trigger update handler complements this: when an incoming `update_trigger`
request omits both `mode` and `enabled`, the existing DB value is preserved
instead of falling back to the BaseTriggerData default of `Enabled`. This
means the fork→parent merge cannot flip the parent's operational state, even
if the fork has an explicit (locally-disabled) state for that path.
The schedule `EditSchedule` payload already lacks an `enabled` field, so its
update path is naturally safe.
## Conflict warning on enable
The `set_*_trigger_mode` endpoint fires the warning whenever a fork transitions
to a mode that *attaches a listener* — `Enabled` or `Suspended`. Suspended is
not "off": the listener still attaches and consumes events; only the auto-run
of queued jobs is paused. Two suspended forks would still split Kafka events
or share a Postgres slot with the parent. `Disabled` is the only mode that
fully detaches.
The check fires whenever the parent workspace has a row at the same trigger
path — **regardless of the parent's current `mode`/`enabled`**. If so, the
endpoint rejects the request with an error string of the shape:
```
fork-conflict:<kind>:<parent_workspace_id>
```
The frontend's `withForkConflictRetry` helper detects this prefix, asks the
user to confirm via a dialog (a `ConfirmationModal` mounted at the logged
layout root, driven by the `forkConflictModal` store), and re-issues the
call with `force: true` if the user agrees. The CLI sees the raw error.
The check fires whenever the parent has the row because the fork's row was
*cloned* from the parent — the upstream identifier (Kafka group, PG slot,
SQS queue URL, GCP/Azure subscription, …) is shared by construction. That
sharing is a risk independent of the parent's current state:
- Both enabled → the listeners compete (split events) or fire twice.
- Parent disabled → the fork can destructively claim shared state (PG WAL
advance, Azure secret_hash reuse, MQTT client_id race) before the parent
re-enables.
The check is opt-out per kind via `TriggerCrud::FORK_CONFLICT_ON_ENABLE`
(default `true`). It is **skipped** for kinds whose upstream identifier is
already workspace-scoped at runtime — fork and parent there can never share
a real upstream:
- **HTTP** — routes are `/r/<workspace_id>/...`; cloned rows always have
`workspaced_route=true` (non-workspaced are filtered out at clone time).
- **Email** — addresses are workspace-prefixed; cloned rows always have
`workspaced_local_part=true`.
The check **fires** for every other kind. The frontend modal copy splits the
conflict into three families so the user can act on the right risk:
- **Split events** (Kafka, NATS, MQTT, SQS, GCP, Azure) — events split
between the two listeners; each side receives a fraction of its traffic.
- **Duplicate firing** (Websocket, Schedule) — every event fires the script
twice (once in fork, once in parent).
- **Slot takeover** (Postgres) — the replication slot is exclusive *and*
destructive: enabling either errors with "slot already active" (parent
enabled) or hijacks the WAL position (parent disabled).
This warning is the *durable* solution for trigger kinds where the conflict
cannot be eliminated by namespacing alone:
- **SQS** — the queue *is* the event source; two consumers will compete for
messages no matter what.
- **GCP-Existing subscription** — same as SQS.
- **Schedule** — same wall-clock firing.
For the kinds that *can* be auto-namespaced (see below), the warning is the
short-term placeholder until that work lands.
## Merge UI behavior
The merge UI (`CompareWorkspaces.svelte`) lists triggers and schedules
side-by-side from both workspaces, computes a per-row change check that
ignores runtime fields (`mode`, `enabled`, `server_id`, `last_server_ping`,
`edited_at`/`edited_by`, `extra_perms`, `permissioned_as`), and only shows
rows that differ in actual config. A fresh clone (only `mode` differs) is
filtered out.
Triggers and schedules are **never auto-selected** in the default deploy /
update selection — only diff items (scripts/flows/apps/etc.) are. The user
opts in by clicking individual trigger rows. This keeps a routine
`Deploy to parent` flow from accidentally pushing trigger config the fork
hasn't intentionally changed.
## Future work — runtime listener suffix
A follow-up PR will append a fork-specific suffix to the upstream identifier
at runtime for the kinds that support it:
| Kind | Identifier | Notes |
|---|---|---|
| Kafka | `group_id` | Two consumer groups never share messages. |
| MQTT | `client_id` | Brokers reject duplicate client_ids; suffix avoids that. |
| NATS | durable consumer name | Fork consumes independently. |
| Postgres | `replication_slot_name` + `publication_name` | Fork auto-creates its own publication on enable, drops on disable / fork delete. |
| Azure Event Grid | `subscription_name` | `manage_azure_subscription` creates the suffixed sub in Azure. |
| GCP Pub/Sub (CreateNew) | `subscription_id` | `manage_google_subscription` creates the suffixed sub. |
The suffix is applied at runtime by the listener — the *stored* identifier
column never carries the suffix, so nothing extra needs to be filtered on
export. The follow-up also adds cleanup-on-fork-delete hooks for the
upstream resources (Azure / GCP / Postgres publication) so deleted forks
don't leak external state.
frontend/src/lib/components/CompareWorkspaces.svelte
+222 -181
View File
@@ -10,12 +10,10 @@
CircleCheck,
CircleX,
DiffIcon,
Eye,
FileJson,
FlaskConical,
GitFork,
Loader2,
Trash2,
UserPlus
} from 'lucide-svelte'
import type { CiTestResult } from '$lib/gen'
@@ -42,20 +40,11 @@
type WorkspaceItemDiff
} from '$lib/gen'
import Button from './common/button/Button.svelte'
import ConfirmationModal from './common/confirmationModal/ConfirmationModal.svelte'
import DiffDrawer from './DiffDrawer.svelte'
import DiffEditor from './DiffEditor.svelte'
import Drawer from './common/drawer/Drawer.svelte'
import DrawerContent from './common/drawer/DrawerContent.svelte'
import ParentWorkspaceProtectionAlert from './ParentWorkspaceProtectionAlert.svelte'
import ScheduleEditor from './triggers/schedules/ScheduleEditor.svelte'
import RouteEditor from './triggers/http/RouteEditor.svelte'
import WebsocketTriggerEditor from './triggers/websocket/WebsocketTriggerEditor.svelte'
import KafkaTriggerEditor from './triggers/kafka/KafkaTriggerEditor.svelte'
import PostgresTriggerEditor from './triggers/postgres/PostgresTriggerEditor.svelte'
import NatsTriggerEditor from './triggers/nats/NatsTriggerEditor.svelte'
import MqttTriggerEditor from './triggers/mqtt/MqttTriggerEditor.svelte'
import SqsTriggerEditor from './triggers/sqs/SqsTriggerEditor.svelte'
import GcpTriggerEditor from './triggers/gcp/GcpTriggerEditor.svelte'
import AzureTriggerEditor from './triggers/azure/AzureTriggerEditor.svelte'
import EmailTriggerEditor from './triggers/email/EmailTriggerEditor.svelte'
import { userWorkspaces, workspaceStore } from '$lib/stores'
import type { Kind } from '$lib/utils_deployable'
@@ -74,6 +63,7 @@
} from './OnBehalfOfSelector.svelte'
import { sendUserToast } from '$lib/toast'
import { deepEqual } from 'fast-equals'
import { orderedJsonStringify, orderedYamlStringify } from '$lib/utils'
import WorkspaceDeployLayout from './WorkspaceDeployLayout.svelte'
import DeploymentRequestPanel from './deploymentRequest/DeploymentRequestPanel.svelte'
import { userStore } from '$lib/stores'
@@ -551,6 +541,7 @@
}))
const triggerItems = forkTriggers
.filter((t) => {
if (!isTriggerRelevantForDirection(t, mergeIntoParent)) return false
const key = getTriggerKey(t)
return deploymentStatus[key]?.status !== 'deployed'
})
@@ -558,6 +549,7 @@
key: getTriggerKey(trigger),
path: trigger.path,
kind: 'trigger' as Kind,
triggerKind: trigger.triggerKind,
diff: undefined as WorkspaceItemDiff | undefined,
trigger
}))
@@ -571,8 +563,21 @@
triggerKind: TriggerKind
scriptPath: string
isFlow: boolean
enabled?: boolean
extraLabel?: string
/** Raw row as returned by the listX endpoint, used to detect changes
* between the fork and parent workspaces. */
raw?: any
/** True only when both workspaces have the trigger and the relevant
* fields differ. False when the trigger only exists on one side. */
hasChanges?: boolean
/** Source value for this row in a diff view (raw object from the
* workspace whose state we'd push). */
sourceRaw?: any
/** Target value (the row that would be overwritten). */
targetRaw?: any
/** "new" when only in source, "modified" when both differ,
* "deleted-in-source" when only in target. */
changeKind?: 'new' | 'modified' | 'deleted-in-source'
}
let ciTestResults = $state<Record<string, CiTestResult[]>>({})
@@ -627,222 +632,222 @@
})
let forkTriggers = $state<ForkTrigger[]>([])
let triggerToDelete = $state<ForkTrigger | undefined>(undefined)
let deploymentRequestPanel: DeploymentRequestPanel | undefined = $state(undefined)
let hasOpenDeploymentRequest = $state(false)
// Trigger detail drawer refs — one per trigger kind. Each is lazy-mounted
// on first openEdit() call, so having them all sit here is cheap.
let scheduleEditor: ScheduleEditor | undefined = $state()
let routeEditor: RouteEditor | undefined = $state()
let websocketEditor: WebsocketTriggerEditor | undefined = $state()
let kafkaEditor: KafkaTriggerEditor | undefined = $state()
let postgresEditor: PostgresTriggerEditor | undefined = $state()
let natsEditor: NatsTriggerEditor | undefined = $state()
let mqttEditor: MqttTriggerEditor | undefined = $state()
let sqsEditor: SqsTriggerEditor | undefined = $state()
let gcpEditor: GcpTriggerEditor | undefined = $state()
let azureEditor: AzureTriggerEditor | undefined = $state()
let emailEditor: EmailTriggerEditor | undefined = $state()
function openTriggerDetails(trigger: ForkTrigger) {
const isFlow = trigger.isFlow
switch (trigger.triggerKind) {
case 'schedules':
scheduleEditor?.openEdit(trigger.path, isFlow)
break
case 'routes':
routeEditor?.openEdit(trigger.path, isFlow)
break
case 'websockets':
websocketEditor?.openEdit(trigger.path, isFlow)
break
case 'kafka':
kafkaEditor?.openEdit(trigger.path, isFlow)
break
case 'postgres':
postgresEditor?.openEdit(trigger.path, isFlow)
break
case 'nats':
natsEditor?.openEdit(trigger.path, isFlow)
break
case 'mqtt':
mqttEditor?.openEdit(trigger.path, isFlow)
break
case 'sqs':
sqsEditor?.openEdit(trigger.path, isFlow)
break
case 'gcp':
gcpEditor?.openEdit(trigger.path, isFlow)
break
case 'azure':
azureEditor?.openEdit(trigger.path, isFlow)
break
case 'emails':
emailEditor?.openEdit(trigger.path, isFlow)
break
}
}
/** Deployable trigger kinds and their list+delete services */
/** Deployable trigger kinds and their list services */
const triggerServices = {
schedules: {
list: (ws: string) => ScheduleService.listSchedules({ workspace: ws }),
delete: (ws: string, path: string) => ScheduleService.deleteSchedule({ workspace: ws, path }),
normalize: (item: any): ForkTrigger => ({
path: item.path,
triggerKind: 'schedules',
scriptPath: item.script_path,
isFlow: item.is_flow,
enabled: item.enabled,
extraLabel: item.schedule
extraLabel: item.schedule,
raw: item
})
},
routes: {
list: (ws: string) => HttpTriggerService.listHttpTriggers({ workspace: ws }),
delete: (ws: string, path: string) =>
HttpTriggerService.deleteHttpTrigger({ workspace: ws, path }),
normalize: (item: any): ForkTrigger => ({
path: item.path,
triggerKind: 'routes',
scriptPath: item.script_path,
isFlow: item.is_flow,
enabled: item.mode === 'enabled',
extraLabel: `${(item.http_method ?? 'get').toUpperCase()} ${item.route_path ?? ''}`
})
},
websockets: {
list: (ws: string) => WebsocketTriggerService.listWebsocketTriggers({ workspace: ws }),
delete: (ws: string, path: string) =>
WebsocketTriggerService.deleteWebsocketTrigger({ workspace: ws, path }),
normalize: (item: any): ForkTrigger => ({
path: item.path,
triggerKind: 'websockets',
scriptPath: item.script_path,
isFlow: item.is_flow,
enabled: item.mode === 'enabled',
extraLabel: item.url
extraLabel: item.url,
raw: item
})
},
kafka: {
list: (ws: string) => KafkaTriggerService.listKafkaTriggers({ workspace: ws }),
delete: (ws: string, path: string) =>
KafkaTriggerService.deleteKafkaTrigger({ workspace: ws, path }),
normalize: (item: any): ForkTrigger => ({
path: item.path,
triggerKind: 'kafka',
scriptPath: item.script_path,
isFlow: item.is_flow,
enabled: item.mode === 'enabled',
extraLabel: item.topics?.join(', ')
extraLabel: item.topics?.join(', '),
raw: item
})
},
postgres: {
list: (ws: string) => PostgresTriggerService.listPostgresTriggers({ workspace: ws }),
delete: (ws: string, path: string) =>
PostgresTriggerService.deletePostgresTrigger({ workspace: ws, path }),
normalize: (item: any): ForkTrigger => ({
path: item.path,
triggerKind: 'postgres',
scriptPath: item.script_path,
isFlow: item.is_flow,
enabled: item.mode === 'enabled'
raw: item
})
},
nats: {
list: (ws: string) => NatsTriggerService.listNatsTriggers({ workspace: ws }),
delete: (ws: string, path: string) =>
NatsTriggerService.deleteNatsTrigger({ workspace: ws, path }),
normalize: (item: any): ForkTrigger => ({
path: item.path,
triggerKind: 'nats',
scriptPath: item.script_path,
isFlow: item.is_flow,
enabled: item.mode === 'enabled',
extraLabel: item.subjects?.join(', ')
extraLabel: item.subjects?.join(', '),
raw: item
})
},
mqtt: {
list: (ws: string) => MqttTriggerService.listMqttTriggers({ workspace: ws }),
delete: (ws: string, path: string) =>
MqttTriggerService.deleteMqttTrigger({ workspace: ws, path }),
normalize: (item: any): ForkTrigger => ({
path: item.path,
triggerKind: 'mqtt',
scriptPath: item.script_path,
isFlow: item.is_flow,
enabled: item.mode === 'enabled'
raw: item
})
},
sqs: {
list: (ws: string) => SqsTriggerService.listSqsTriggers({ workspace: ws }),
delete: (ws: string, path: string) =>
SqsTriggerService.deleteSqsTrigger({ workspace: ws, path }),
normalize: (item: any): ForkTrigger => ({
path: item.path,
triggerKind: 'sqs',
scriptPath: item.script_path,
isFlow: item.is_flow,
enabled: item.mode === 'enabled',
extraLabel: item.queue_url
extraLabel: item.queue_url,
raw: item
})
},
gcp: {
list: (ws: string) => GcpTriggerService.listGcpTriggers({ workspace: ws }),
delete: (ws: string, path: string) =>
GcpTriggerService.deleteGcpTrigger({ workspace: ws, path }),
normalize: (item: any): ForkTrigger => ({
path: item.path,
triggerKind: 'gcp',
scriptPath: item.script_path,
isFlow: item.is_flow,
enabled: item.mode === 'enabled',
extraLabel: item.topic_id
extraLabel: item.topic_id,
raw: item
})
},
azure: {
list: (ws: string) => AzureTriggerService.listAzureTriggers({ workspace: ws }),
delete: (ws: string, path: string) =>
AzureTriggerService.deleteAzureTrigger({ workspace: ws, path }),
normalize: (item: any): ForkTrigger => ({
path: item.path,
triggerKind: 'azure',
scriptPath: item.script_path,
isFlow: item.is_flow,
enabled: item.mode === 'enabled',
extraLabel: item.topic_name ?? item.scope_resource_id
extraLabel: item.topic_name ?? item.scope_resource_id,
raw: item
})
},
emails: {
list: (ws: string) => EmailTriggerService.listEmailTriggers({ workspace: ws }),
delete: (ws: string, path: string) =>
EmailTriggerService.deleteEmailTrigger({ workspace: ws, path }),
normalize: (item: any): ForkTrigger => ({
path: item.path,
triggerKind: 'emails',
scriptPath: item.script_path,
isFlow: item.is_flow,
enabled: item.mode === 'enabled',
extraLabel: getEmailAddress(
item.local_part,
item.workspaced_local_part,
currentWorkspaceId,
emailDomain ?? ''
)
),
raw: item
})
}
} as const
let emailDomain = $state<string | undefined>(undefined)
/**
* Fields that should not count as a "change" between the parent and the
* fork. Mode/enabled are forced to 'disabled'/false on clone and stripped
* by the fork-export filter; the rest are runtime state or per-row
* metadata that always diverges. Comparing without these matches the
* semantics of "is this trigger configured the same way?" rather than
* "are these two rows byte-identical?".
*/
const TRIGGER_COMPARE_IGNORE = new Set([
'workspace_id',
'mode',
'enabled',
'edited_at',
'edited_by',
'last_server_ping',
'server_id',
'error',
'extra_perms',
'permissioned_as'
])
function stripIgnoredFields(row: any): any {
if (!row || typeof row !== 'object') return row
const out: Record<string, any> = {}
for (const [k, v] of Object.entries(row)) {
if (!TRIGGER_COMPARE_IGNORE.has(k)) out[k] = v
}
return out
}
function rowsHaveSameConfig(a: any, b: any): boolean {
return (
orderedJsonStringify(stripIgnoredFields(a)) === orderedJsonStringify(stripIgnoredFields(b))
)
}
async function fetchAllTriggers() {
try {
emailDomain = await getEmailDomain()
const entries = Object.values(triggerServices)
const entries = Object.entries(triggerServices) as Array<
[TriggerKind, (typeof triggerServices)[keyof typeof triggerServices]]
>
// Fetch fork + parent in parallel for each kind. Either side may
// fail (e.g. permission denied on parent) — fall back to empty.
const results = await Promise.allSettled(
entries.map(async (svc) => {
const items = await svc.list(currentWorkspaceId)
return items.map(svc.normalize)
entries.map(async ([kind, svc]) => {
const [forkItems, parentItems] = await Promise.all([
svc.list(currentWorkspaceId).catch(() => [] as any[]),
svc.list(parentWorkspaceId).catch(() => [] as any[])
])
const byPath = new Map<string, { fork?: any; parent?: any }>()
for (const item of forkItems) {
byPath.set(item.path, { fork: item })
}
for (const item of parentItems) {
const entry = byPath.get(item.path) ?? {}
entry.parent = item
byPath.set(item.path, entry)
}
const merged: ForkTrigger[] = []
for (const [path, entry] of byPath) {
const sourceItem = entry.fork ?? entry.parent
const normalized = svc.normalize(sourceItem)
let changeKind: ForkTrigger['changeKind']
let hasChanges = false
if (entry.fork && !entry.parent) {
changeKind = 'new'
} else if (!entry.fork && entry.parent) {
changeKind = 'deleted-in-source'
} else if (entry.fork && entry.parent) {
hasChanges = !rowsHaveSameConfig(entry.fork, entry.parent)
if (hasChanges) changeKind = 'modified'
}
merged.push({
...normalized,
raw: sourceItem,
hasChanges,
changeKind,
sourceRaw: entry.fork,
targetRaw: entry.parent,
path
})
}
return merged
})
)
forkTriggers = results.flatMap((r) => (r.status === 'fulfilled' ? r.value : []))
@@ -852,28 +857,21 @@
}
}
function deleteTrigger(trigger: ForkTrigger) {
triggerToDelete = trigger
}
async function confirmDeleteTrigger() {
const trigger = triggerToDelete
if (!trigger) return
triggerToDelete = undefined
const triggerType = triggerKindToTriggerType(trigger.triggerKind)
const displayName = triggerType ? triggerDisplayNamesMap[triggerType] : trigger.triggerKind
try {
const svc = triggerServices[trigger.triggerKind as keyof typeof triggerServices]
if (!svc) {
throw new Error(`No service for trigger kind: ${trigger.triggerKind}`)
}
await svc.delete(currentWorkspaceId, trigger.path)
forkTriggers = forkTriggers.filter(
(t) => !(t.path === trigger.path && t.triggerKind === trigger.triggerKind)
)
sendUserToast(`Deleted ${displayName} trigger '${trigger.path}'`)
} catch (e: any) {
sendUserToast(`Failed to delete trigger '${trigger.path}': ${e.body || e.message}`, true)
/**
* Triggers worth showing in the merge UI given the current direction.
* - "Deploy to parent": rows that exist in fork and either don't exist in
* parent or have config differences.
* - "Update current" (pull from parent): mirror.
* Triggers that exist on both sides with identical config are filtered
* out — they would generate a no-op deploy and only add noise.
*/
function isTriggerRelevantForDirection(t: ForkTrigger, deployingToParent: boolean): boolean {
const existsInFork = !!t.sourceRaw
const existsInParent = !!t.targetRaw
if (deployingToParent) {
return existsInFork && (!existsInParent || !!t.hasChanges)
} else {
return existsInParent && (!existsInFork || !!t.hasChanges)
}
}
@@ -882,6 +880,38 @@
return triggerType ? triggerDisplayNamesMap[triggerType] : triggerKind
}
let triggerDiffOpen = $state(false)
let triggerDiffPayload = $state<
| {
kindLabel: string
path: string
originalLabel: string
modifiedLabel: string
original: string
modified: string
}
| undefined
>(undefined)
function openTriggerDiff(t: ForkTrigger) {
// `sourceRaw` is the fork row, `targetRaw` is the parent row regardless
// of direction (set in fetchAllTriggers). The diff reads from the
// destination (left) to the source (right), matching the deploy arrow.
const sourceWorkspace = mergeIntoParent ? currentWorkspaceId : parentWorkspaceId
const targetWorkspace = mergeIntoParent ? parentWorkspaceId : currentWorkspaceId
const fromRow = mergeIntoParent ? t.sourceRaw : t.targetRaw
const toRow = mergeIntoParent ? t.targetRaw : t.sourceRaw
triggerDiffPayload = {
kindLabel: getTriggerDisplayName(t.triggerKind),
path: t.path,
originalLabel: `${targetWorkspace} (target)`,
modifiedLabel: `${sourceWorkspace} (source)`,
original: orderedYamlStringify(stripIgnoredFields(toRow ?? {})),
modified: orderedYamlStringify(stripIgnoredFields(fromRow ?? {}))
}
triggerDiffOpen = true
}
// Fetch triggers when workspace is available
$effect(() => {
if (currentWorkspaceId) {
@@ -1089,7 +1119,10 @@
{#if item.trigger}
{@const t = item.trigger as ForkTrigger}
<span class="text-emphasis">{getTriggerDisplayName(t.triggerKind)}</span>
<span class="text-secondary mx-1">&rarr;</span>
{#if t.extraLabel}
<span class="text-secondary ml-1">{t.extraLabel}</span>
{/if}
<span class="text-tertiary mx-1">&rarr;</span>
<span class="text-secondary">{t.scriptPath}</span>
{:else}
{@const diff = item.diff as WorkspaceItemDiff}
@@ -1120,30 +1153,38 @@
{#if item.trigger}
{@const t = item.trigger as ForkTrigger}
{@const key = item.key}
<Badge color="indigo" size="xs">Fork-only</Badge>
{#if t.changeKind === 'new'}
<Badge
title={mergeIntoParent
? `Only exists in '${currentWorkspaceId}'`
: `Only exists in '${parentWorkspaceId}'`}
color="indigo"
size="xs">New</Badge
>
{/if}
{#if t.isFlow}
<Badge color="blue" size="xs">flow</Badge>
{/if}
{#if t.extraLabel}
<span class="text-tertiary text-xs">({t.extraLabel})</span>
{/if}
{#if t.enabled != null}
<Badge color={t.enabled ? 'green' : 'gray'} size="xs">
{t.enabled ? 'Enabled' : 'Disabled'}
</Badge>
{/if}
{#if !deploymentStatus[key] || deploymentStatus[key].status != 'deployed'}
<Button
size="xs"
variant="subtle"
startIcon={{ icon: Eye }}
onclick={() => openTriggerDetails(t)}
>
Details
</Button>
<Button size="xs" variant="subtle" color="red" onclick={() => deleteTrigger(t)}>
<Trash2 size={12} />
</Button>
{#if mergeIntoParent}
<Badge color="green" size="xs">
<ArrowUpRight class="w-3 h-3 inline" />
1 ahead
</Badge>
{:else}
<Badge color="blue" size="xs">
<ArrowDownRight class="w-3 h-3 inline" />
1 behind
</Badge>
{/if}
{#if t.changeKind === 'modified'}
<div>
<Button size="xs" variant="subtle" onclick={() => openTriggerDiff(t)}>
<DiffIcon class="w-3 h-3" />
Show diff
</Button>
</div>
{/if}
{/if}
{:else}
{@const diff = item.diff as WorkspaceItemDiff}
@@ -1326,30 +1367,30 @@
<DiffDrawer bind:this={diffDrawer} {isFlow} />
<ScheduleEditor bind:this={scheduleEditor} />
<RouteEditor bind:this={routeEditor} />
<WebsocketTriggerEditor bind:this={websocketEditor} />
<KafkaTriggerEditor bind:this={kafkaEditor} />
<PostgresTriggerEditor bind:this={postgresEditor} />
<NatsTriggerEditor bind:this={natsEditor} />
<MqttTriggerEditor bind:this={mqttEditor} />
<SqsTriggerEditor bind:this={sqsEditor} />
<GcpTriggerEditor bind:this={gcpEditor} />
<AzureTriggerEditor bind:this={azureEditor} />
<EmailTriggerEditor bind:this={emailEditor} />
<ConfirmationModal
title="Delete trigger"
confirmationText="Delete"
open={!!triggerToDelete}
onConfirmed={confirmDeleteTrigger}
onCanceled={() => (triggerToDelete = undefined)}
>
{#if triggerToDelete}
Are you sure you want to delete the {getTriggerDisplayName(triggerToDelete.triggerKind)} trigger
'{triggerToDelete.path}'?
{/if}
</ConfirmationModal>
<Drawer bind:open={triggerDiffOpen} size="900px">
<DrawerContent
title={triggerDiffPayload
? `${triggerDiffPayload.kindLabel} ${triggerDiffPayload.path}`
: 'Trigger diff'}
on:close={() => (triggerDiffOpen = false)}
>
{#if triggerDiffPayload}
<div class="flex flex-col h-full">
<div class="flex-1 min-h-0">
<DiffEditor
open={triggerDiffOpen}
className="!h-full"
defaultLang="yaml"
defaultOriginal={triggerDiffPayload.original}
defaultModified={triggerDiffPayload.modified}
readOnly
inlineDiff={false}
/>
</div>
</div>
{/if}
</DrawerContent>
</Drawer>
{:else}
<div class="flex items-center justify-center h-full">
<div class="text-gray-500">No comparison data available</div>
frontend/src/lib/components/ForkConflictModal.svelte
+64
View File
@@ -0,0 +1,64 @@
<script lang="ts">
import ConfirmationModal from '$lib/components/common/confirmationModal/ConfirmationModal.svelte'
import { forkConflictModal } from '$lib/stores'
const state = $derived(forkConflictModal.val)
// Three failure shapes we want to describe accurately:
// - split-events: shared upstream subscription/queue, two listeners halve each other's traffic
// - duplicate-firing: independent triggers on shared input, every event fires twice
// - slot-takeover: PG replication slot is exclusive AND its position is destructively shared
const SPLIT_KINDS = new Set(['kafka', 'nats', 'mqtt', 'sqs', 'gcp', 'azure'])
const DUPLICATE_KINDS = new Set(['websocket', 'schedule'])
const family = $derived(
!state
? 'unknown'
: state.kind === 'postgres'
? 'slot'
: SPLIT_KINDS.has(state.kind)
? 'split'
: DUPLICATE_KINDS.has(state.kind)
? 'duplicate'
: 'unknown'
)
function close(confirmed: boolean) {
const resolve = state?.resolve
forkConflictModal.val = undefined
resolve?.(confirmed)
}
</script>
<ConfirmationModal
open={!!state}
title="Enable in fork conflicts with parent"
confirmationText="Enable anyway"
onConfirmed={() => close(true)}
onCanceled={() => close(false)}
>
{#if state}
<p>
The parent workspace (<span class="font-mono">{state.parentWorkspaceId}</span>) has the same
{state.kindLabel} configured at this path. Because this fork's row was cloned from it, the upstream
identifier is shared.
</p>
<p class="mt-2">
{#if family === 'split'}
If both are enabled, the two listeners will compete on the same upstream and each side will
receive only a fraction of its events.
{:else if family === 'duplicate'}
If both are enabled, every event will fire the script twice — once in the fork and once in
the parent.
{:else if family === 'slot'}
The cloned <span class="font-mono">replication_slot_name</span> points at the same Postgres slot,
which only allows one consumer at a time. Enabling here will either fail with "slot already active"
if the parent is enabled, or hijack the slot's WAL position if it isn't — causing the parent
to lose events when re-enabled.
{:else}
Enabling it here may compete for the same upstream events or duplicate side effects.
{/if}
</p>
<p class="mt-2">Enable in this fork anyway?</p>
{/if}
</ConfirmationModal>
frontend/src/lib/components/triggers/TriggerEditorToolbar.svelte
+1 -1
View File
@@ -21,7 +21,7 @@
extra?: Snippet
onDelete?: () => void
onReset?: () => void
onToggleMode: (mode: TriggerMode) => void
onToggleMode: (mode: TriggerMode) => void | boolean | Promise<void | boolean>
onUpdate?: () => void
cloudDisabled?: boolean
trigger?: Trigger
frontend/src/lib/components/triggers/TriggerModeToggle.svelte
+42 -8
View File
@@ -11,7 +11,12 @@
interface Props {
triggerMode: TriggerMode
onToggleMode: (mode: TriggerMode) => void
// Optionally returns false to signal that the change was cancelled
// (e.g. user dismissed a fork-conflict modal). When the parent didn't
// optimistically update `triggerMode` (list-page rows fall in this
// bucket), we resync the local toggle state from the prop after the
// dispatch resolves.
onToggleMode: (mode: TriggerMode) => void | boolean | Promise<void | boolean>
canWrite: boolean
hideToggleLabels?: boolean
hideDropdown?: boolean
@@ -33,7 +38,18 @@
includeModalConfig
}: Props = $props()
let innerTriggerMode = $derived(triggerMode)
// Local writable state mirroring `triggerMode`. Both inner controls bind
// to it: the suspended ToggleButtonGroup directly (it already needed a
// writable to revert to 'suspended' when there are queued jobs), and the
// regular Toggle via a function binding that maps 'enabled'/'disabled'
// to a boolean. Either control can write back here, and the $effect
// re-syncs us from the prop whenever the parent updates `triggerMode`.
// This lets the parent reset us in cases where the user click can't go
// through (e.g. cancelled fork-conflict modal).
let innerTriggerMode = $state(triggerMode)
$effect(() => {
innerTriggerMode = triggerMode
})
let suspendedJobsModal = $derived(passedSuspendedJobsModal ?? null)
</script>
@@ -75,9 +91,18 @@
<Toggle
disabled={!canWrite}
options={hideToggleLabels ? undefined : { right: 'enable', left: 'disable' }}
checked={triggerMode === 'enabled'}
on:change={(e) => {
onToggleMode(e.detail ? 'enabled' : 'disabled')
bind:checked={
() => innerTriggerMode === 'enabled', (v) => (innerTriggerMode = v ? 'enabled' : 'disabled')
}
on:change={async (e) => {
const result = await onToggleMode(e.detail ? 'enabled' : 'disabled')
if (result === false) {
// Cancelled: snap the toggle back to the prop. Needed for
// list-page rows where the parent doesn't optimistically
// flip `triggerMode` (so $effect won't re-run from a no-op
// prop change).
innerTriggerMode = triggerMode
}
}}
/>
{#if !hideDropdown}
@@ -87,9 +112,18 @@
{
displayName: 'Suspend job execution',
icon: Pause,
action: () => {
triggerMode = 'suspended'
onToggleMode?.('suspended')
action: async () => {
// Optimistically flip the local mirror, not the
// non-bindable `triggerMode` prop. The parent will
// echo the new mode back via $effect on success;
// on cancel (e.g. user dismisses the fork-conflict
// modal), reset to whatever the prop says — same
// shape as the Toggle's on:change handler.
innerTriggerMode = 'suspended'
const result = await onToggleMode?.('suspended')
if (result === false) {
innerTriggerMode = triggerMode
}
},
tooltip:
'When a trigger is in suspended mode, it will continue to accept payloads and queue jobs, but those jobs will not run automatically. You can review the list of suspended jobs, and resume or cancel them individually.'
frontend/src/lib/components/triggers/TriggerSuspendedJobsModal.svelte
+1 -1
View File
@@ -43,7 +43,7 @@
triggerPath: string
triggerKind: JobTriggerKind
hasChanged: boolean
onToggleMode: (mode: TriggerMode) => void
onToggleMode: (mode: TriggerMode) => void | boolean | Promise<void | boolean>
runnableConfig: TriggerRunnableConfig
}
frontend/src/lib/components/triggers/azure/AzureTriggerEditorInner.svelte
+15 -5
View File
@@ -5,6 +5,7 @@
import Path from '$lib/components/Path.svelte'
import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores'
import { canWrite, capitalize, emptyString, sendUserToast } from '$lib/utils'
import { withForkConflictRetry } from '$lib/utils/forkConflict'
import { Loader2 } from 'lucide-svelte'
import Label from '$lib/components/Label.svelte'
import {
@@ -263,13 +264,22 @@
}
async function handleToggleMode(newMode: TriggerMode) {
const previousMode = mode
mode = newMode
if (!trigger?.draftConfig) {
await AzureTriggerService.setAzureTriggerMode({
path: initialPath,
workspace: $workspaceStore ?? '',
requestBody: { mode: newMode }
})
const ok = await withForkConflictRetry(
(force) =>
AzureTriggerService.setAzureTriggerMode({
path: initialPath,
workspace: $workspaceStore ?? '',
requestBody: { mode: newMode, force }
}),
'Azure trigger'
)
if (!ok) {
mode = previousMode
return
}
sendUserToast(`${capitalize(newMode)} Azure trigger ${initialPath}`)
onUpdate?.(initialPath)
}
frontend/src/lib/components/triggers/email/EmailTriggerEditorInner.svelte
+2
View File
@@ -257,6 +257,8 @@
async function handleToggleMode(newMode: TriggerMode) {
mode = newMode
if (!trigger?.draftConfig) {
// Email addresses are always workspace-prefixed (clone filter
// excludes workspaced_local_part=false) — no fork-conflict warning.
await EmailTriggerService.setEmailTriggerMode({
path: initialPath,
workspace: $workspaceStore ?? '',
frontend/src/lib/components/triggers/gcp/GcpTriggerEditorInner.svelte
+15 -6
View File
@@ -5,6 +5,7 @@
import Path from '$lib/components/Path.svelte'
import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores'
import { canWrite, capitalize, emptyString, sendUserToast } from '$lib/utils'
import { withForkConflictRetry } from '$lib/utils/forkConflict'
import { Loader2 } from 'lucide-svelte'
import Label from '$lib/components/Label.svelte'
import {
@@ -278,15 +279,23 @@
}
async function handleToggleMode(newMode: TriggerMode) {
const previousMode = mode
mode = newMode
if (!trigger?.draftConfig) {
await GcpTriggerService.setGcpTriggerMode({
path: initialPath,
workspace: $workspaceStore ?? '',
requestBody: { mode: newMode }
})
const ok = await withForkConflictRetry(
(force) =>
GcpTriggerService.setGcpTriggerMode({
path: initialPath,
workspace: $workspaceStore ?? '',
requestBody: { mode: newMode, force }
}),
'GCP Pub/Sub trigger'
)
if (!ok) {
mode = previousMode
return
}
sendUserToast(`${capitalize(newMode)} GCP Pub/Sub trigger ${initialPath}`)
onUpdate?.(initialPath)
}
if (originalConfig) {
frontend/src/lib/components/triggers/http/RouteEditorInner.svelte
+2
View File
@@ -408,6 +408,8 @@
async function handleToggleMode(newMode: TriggerMode) {
mode = newMode
if (!trigger?.draftConfig) {
// HTTP routes are always workspace-prefixed at runtime, so fork
// and parent live at distinct URLs — no fork-conflict warning.
await HttpTriggerService.setHttpTriggerMode({
path: initialPath,
workspace: $workspaceStore ?? '',
frontend/src/lib/components/triggers/kafka/KafkaTriggerEditorInner.svelte
+15 -6
View File
@@ -9,6 +9,7 @@
import { KafkaTriggerService, type ErrorHandler, type Retry, type TriggerMode } from '$lib/gen'
import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores'
import { canWrite, capitalize, emptyString, sendUserToast } from '$lib/utils'
import { withForkConflictRetry } from '$lib/utils/forkConflict'
import Section from '$lib/components/Section.svelte'
import { Loader2, RotateCcw } from 'lucide-svelte'
import Label from '$lib/components/Label.svelte'
@@ -316,15 +317,23 @@
}
async function handleToggleMode(newMode: TriggerMode) {
const previousMode = mode
mode = newMode
if (!trigger?.draftConfig) {
await KafkaTriggerService.setKafkaTriggerMode({
path: initialPath,
workspace: $workspaceStore ?? '',
requestBody: { mode: newMode }
})
const ok = await withForkConflictRetry(
(force) =>
KafkaTriggerService.setKafkaTriggerMode({
path: initialPath,
workspace: $workspaceStore ?? '',
requestBody: { mode: newMode, force }
}),
'Kafka trigger'
)
if (!ok) {
mode = previousMode
return
}
sendUserToast(`${capitalize(newMode)} Kafka trigger ${initialPath}`)
onUpdate?.(initialPath)
}
if (originalConfig) {
frontend/src/lib/components/triggers/mqtt/MqttTriggerEditorInner.svelte
+15 -6
View File
@@ -8,6 +8,7 @@
import ScriptPicker from '$lib/components/ScriptPicker.svelte'
import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores'
import { canWrite, capitalize, emptyString, sendUserToast } from '$lib/utils'
import { withForkConflictRetry } from '$lib/utils/forkConflict'
import Section from '$lib/components/Section.svelte'
import { Loader2 } from 'lucide-svelte'
import Label from '$lib/components/Label.svelte'
@@ -302,15 +303,23 @@
}
async function handleToggleMode(newMode: TriggerMode) {
const previousMode = mode
mode = newMode
if (!trigger?.draftConfig) {
await MqttTriggerService.setMqttTriggerMode({
path: initialPath,
workspace: $workspaceStore ?? '',
requestBody: { mode: newMode }
})
const ok = await withForkConflictRetry(
(force) =>
MqttTriggerService.setMqttTriggerMode({
path: initialPath,
workspace: $workspaceStore ?? '',
requestBody: { mode: newMode, force }
}),
'MQTT trigger'
)
if (!ok) {
mode = previousMode
return
}
sendUserToast(`${capitalize(newMode)} MQTT trigger ${initialPath}`)
onUpdate?.(initialPath)
}
if (originalConfig) {
frontend/src/lib/components/triggers/nats/NatsTriggerEditorInner.svelte
+15 -6
View File
@@ -8,6 +8,7 @@
import { NatsTriggerService, type ErrorHandler, type Retry, type TriggerMode } from '$lib/gen'
import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores'
import { canWrite, capitalize, emptyString, sendUserToast } from '$lib/utils'
import { withForkConflictRetry } from '$lib/utils/forkConflict'
import Section from '$lib/components/Section.svelte'
import { Loader2 } from 'lucide-svelte'
import Label from '$lib/components/Label.svelte'
@@ -282,15 +283,23 @@
}
async function handleToggleMode(newMode: TriggerMode) {
const previousMode = mode
mode = newMode
if (!trigger?.draftConfig) {
await NatsTriggerService.setNatsTriggerMode({
path: initialPath,
workspace: $workspaceStore ?? '',
requestBody: { mode: newMode }
})
const ok = await withForkConflictRetry(
(force) =>
NatsTriggerService.setNatsTriggerMode({
path: initialPath,
workspace: $workspaceStore ?? '',
requestBody: { mode: newMode, force }
}),
'NATS trigger'
)
if (!ok) {
mode = previousMode
return
}
sendUserToast(`${capitalize(newMode)} NATS trigger ${initialPath}`)
onUpdate?.(initialPath)
}
if (originalConfig) {
frontend/src/lib/components/triggers/postgres/PostgresTriggerEditorInner.svelte
+15 -6
View File
@@ -15,6 +15,7 @@
} from '$lib/gen'
import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores'
import { canWrite, emptyString, emptyStringTrimmed, sendUserToast } from '$lib/utils'
import { withForkConflictRetry } from '$lib/utils/forkConflict'
import Section from '$lib/components/Section.svelte'
import { Loader2 } from 'lucide-svelte'
import Label from '$lib/components/Label.svelte'
@@ -443,15 +444,23 @@
}
async function handleToggleMode(newMode: TriggerMode) {
const previousMode = mode
mode = newMode
if (!trigger?.draftConfig) {
await PostgresTriggerService.setPostgresTriggerMode({
path: initialPath,
workspace: $workspaceStore ?? '',
requestBody: { mode: newMode }
})
const ok = await withForkConflictRetry(
(force) =>
PostgresTriggerService.setPostgresTriggerMode({
path: initialPath,
workspace: $workspaceStore ?? '',
requestBody: { mode: newMode, force }
}),
'postgres trigger'
)
if (!ok) {
mode = previousMode
return
}
sendUserToast(`${capitalize(newMode)} postgres trigger ${initialPath}`)
onUpdate?.(initialPath)
}
if (originalConfig) {
frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte
+15 -5
View File
@@ -37,6 +37,7 @@
import WorkerTagPicker from '$lib/components/WorkerTagPicker.svelte'
import { runScheduleNow } from '../scheduled/utils'
import { handleConfigChange } from '../utils'
import { withForkConflictRetry } from '$lib/utils/forkConflict'
import TextInput from '$lib/components/text_input/TextInput.svelte'
import { twMerge } from 'tailwind-merge'
import PermissionedAsLine from '../PermissionedAsLine.svelte'
@@ -627,13 +628,22 @@
}
async function handleToggleEnabled(nEnabled: boolean) {
const previousEnabled = enabled
enabled = nEnabled
if (!trigger?.draftConfig) {
await ScheduleService.setScheduleEnabled({
path: initialPath,
workspace: $workspaceStore ?? '',
requestBody: { enabled: nEnabled }
})
const ok = await withForkConflictRetry(
(force) =>
ScheduleService.setScheduleEnabled({
path: initialPath,
workspace: $workspaceStore ?? '',
requestBody: { enabled: nEnabled, force }
}),
'schedule'
)
if (!ok) {
enabled = previousEnabled
return
}
sendUserToast(`${nEnabled ? 'enabled' : 'disabled'} schedule ${initialPath}`)
onUpdate?.(initialPath)
}
frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte
+15 -6
View File
@@ -5,6 +5,7 @@
import Path from '$lib/components/Path.svelte'
import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores'
import { canWrite, capitalize, emptyString, sendUserToast } from '$lib/utils'
import { withForkConflictRetry } from '$lib/utils/forkConflict'
import { Loader2 } from 'lucide-svelte'
import Label from '$lib/components/Label.svelte'
import {
@@ -240,15 +241,23 @@
}
async function handleToggleMode(newMode: TriggerMode) {
const previousMode = mode
mode = newMode
if (!trigger?.draftConfig) {
await SqsTriggerService.setSqsTriggerMode({
path: initialPath,
workspace: $workspaceStore ?? '',
requestBody: { mode: newMode }
})
const ok = await withForkConflictRetry(
(force) =>
SqsTriggerService.setSqsTriggerMode({
path: initialPath,
workspace: $workspaceStore ?? '',
requestBody: { mode: newMode, force }
}),
'SQS trigger'
)
if (!ok) {
mode = previousMode
return
}
sendUserToast(`${capitalize(newMode)} SQS trigger ${initialPath}`)
onUpdate?.(initialPath)
}
if (originalConfig) {
frontend/src/lib/components/triggers/websocket/WebsocketTriggerEditorInner.svelte
+15 -6
View File
@@ -21,6 +21,7 @@
} from '$lib/gen'
import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores'
import { canWrite, emptySchema, emptyString, sendUserToast } from '$lib/utils'
import { withForkConflictRetry } from '$lib/utils/forkConflict'
import Section from '$lib/components/Section.svelte'
import { Loader2, X, Plus } from 'lucide-svelte'
import Label from '$lib/components/Label.svelte'
@@ -370,15 +371,23 @@
}
async function handleToggleMode(newMode: TriggerMode) {
const previousMode = mode
mode = newMode
if (!trigger?.draftConfig) {
await WebsocketTriggerService.setWebsocketTriggerMode({
path: initialPath,
workspace: $workspaceStore ?? '',
requestBody: { mode: newMode }
})
const ok = await withForkConflictRetry(
(force) =>
WebsocketTriggerService.setWebsocketTriggerMode({
path: initialPath,
workspace: $workspaceStore ?? '',
requestBody: { mode: newMode, force }
}),
'websocket trigger'
)
if (!ok) {
mode = previousMode
return
}
sendUserToast(`${capitalize(newMode)} websocket trigger ${initialPath}`)
onUpdate?.(initialPath)
}
if (originalConfig) {
frontend/src/lib/stores.ts
+14 -1
View File
@@ -132,12 +132,25 @@ export const codeCompletionSessionEnabled = writable<boolean>(
export const usedTriggerKinds = writable<string[]>([])
export let globalDbManagerDrawer: StateStore<DbManagerUriState | undefined> = { val: undefined }
export let globalForkModal: StateStore<GlobalForkModalState | undefined> = createState({ val: undefined })
export let globalForkModal: StateStore<GlobalForkModalState | undefined> = createState({
val: undefined
})
export type GlobalForkModalState = {
opened: true
}
export type ForkConflictModalState = {
kind: string
kindLabel: string
parentWorkspaceId: string
resolve: (proceed: boolean) => void
}
export let forkConflictModal: StateStore<ForkConflictModalState | undefined> = createState({
val: undefined
})
type SQLBaseSchema = {
[schemaKey: string]: {
[tableKey: string]: {
frontend/src/lib/utils/forkConflict.ts
+75
View File
@@ -0,0 +1,75 @@
import { forkConflictModal } from '$lib/stores'
/**
* The backend rejects "enable" requests on triggers/schedules in a fork when
* the parent workspace has the same path enabled. The error body is shaped as
* `fork-conflict:<kind>:<parent_workspace_id>`
* so the UI can show a tailored confirm-to-proceed dialog and re-issue the
* call with `force: true` if the user agrees.
*/
export interface ForkConflict {
kind: string
parentWorkspaceId: string
}
export function detectForkConflict(e: unknown): ForkConflict | null {
const body = (e as any)?.body
const raw =
typeof body === 'string'
? body
: ((body as any)?.error?.message ?? (body as any)?.message ?? (e as any)?.message ?? '')
const m = String(raw).match(/fork-conflict:([^:]+):(.+)/)
if (!m) return null
return { kind: m[1], parentWorkspaceId: m[2].trim() }
}
/**
* Opens the global ForkConflictModal and awaits the user's choice. Resolves
* to true when the user clicks "Enable anyway", false when they cancel or
* dismiss. If a previous modal is still pending (e.g. user clicked toggles
* on two rows in quick succession), resolve the older promise to false so
* the prior caller doesn't hang.
*/
function askForkConflictConfirm(kind: string, kindLabel: string, parentWorkspaceId: string) {
return new Promise<boolean>((resolve) => {
const previous = forkConflictModal.val
previous?.resolve(false)
forkConflictModal.val = { kind, kindLabel, parentWorkspaceId, resolve }
})
}
/**
* Catches a fork-conflict error from `fn(false)`, shows the confirmation
* dialog, and retries with `fn(true)` when the user accepts. Re-throws every
* other error.
*
* Returns `true` when the call committed (no conflict, or user confirmed and
* retry succeeded) and `false` when the user dismissed the modal. Callers
* should bail on `false` to skip success toasts and revert any optimistic UI
* state.
*
* `kindLabel` is shown to the user — pass a friendly name like "kafka trigger"
* or "schedule" so the dialog reads naturally.
*/
export async function withForkConflictRetry(
fn: (force: boolean) => Promise<unknown>,
kindLabel: string
): Promise<boolean> {
try {
await fn(false)
return true
} catch (e) {
const conflict = detectForkConflict(e)
if (!conflict) throw e
const proceed = await askForkConflictConfirm(
conflict.kind,
kindLabel,
conflict.parentWorkspaceId
)
// User explicitly dismissed the modal — treat as a silent no-op so the
// caller's catch block doesn't pop a redundant error toast.
if (!proceed) return false
await fn(true)
return true
}
}
frontend/src/routes/(root)/(logged)/+layout.svelte
+3
View File
@@ -15,6 +15,7 @@
import WorkspaceMenu from '$lib/components/sidebar/WorkspaceMenu.svelte'
import SidebarContent from '$lib/components/sidebar/SidebarContent.svelte'
import CriticalAlertModal from '$lib/components/sidebar/CriticalAlertModal.svelte'
import ForkConflictModal from '$lib/components/ForkConflictModal.svelte'
import {
enterpriseLicense,
isPremiumStore,
@@ -837,6 +838,8 @@
<DBManagerDrawer uriState={globalDbManagerDrawer.val} />
{/if}
<ForkConflictModal />
<Modal2
title="Forking {$workspaceStore}"
target="#content"
frontend/src/routes/(root)/(logged)/azure_triggers/+page.svelte
+15 -7
View File
@@ -17,6 +17,7 @@
storeLocalSetting,
removeTriggerKindIfUnused
} from '$lib/utils'
import { withForkConflictRetry } from '$lib/utils/forkConflict'
import { base } from '$app/paths'
import { page } from '$app/stores'
import CenteredPage from '$lib/components/CenteredPage.svelte'
@@ -111,21 +112,28 @@
clearInterval(interval)
})
async function onToggleMode(path: string, mode: TriggerMode): Promise<void> {
async function onToggleMode(path: string, mode: TriggerMode): Promise<boolean> {
let committed = false
try {
await AzureTriggerService.setAzureTriggerMode({
path,
workspace: $workspaceStore!,
requestBody: { mode }
})
const ok = await withForkConflictRetry(
(force) =>
AzureTriggerService.setAzureTriggerMode({
path,
workspace: $workspaceStore!,
requestBody: { mode, force }
}),
'Azure trigger'
)
committed = ok
if (ok) loadTriggers()
} catch (err) {
sendUserToast(
`Cannot ${mode === 'enabled' ? 'enable' : mode === 'disabled' ? 'disable' : 'suspend'} Azure Event Grid trigger: ${err.body}`,
true
)
} finally {
loadTriggers()
}
return committed
}
run(() => {
frontend/src/routes/(root)/(logged)/email_triggers/+page.svelte
+2
View File
@@ -227,6 +227,8 @@
async function onToggleMode(path: string, mode: TriggerMode): Promise<void> {
try {
// Email addresses are always workspace-prefixed (clone filter
// excludes workspaced_local_part=false) — no fork-conflict warning.
await EmailTriggerService.setEmailTriggerMode({
path,
workspace: $workspaceStore!,
frontend/src/routes/(root)/(logged)/gcp_triggers/+page.svelte
+15 -7
View File
@@ -17,6 +17,7 @@
storeLocalSetting,
removeTriggerKindIfUnused
} from '$lib/utils'
import { withForkConflictRetry } from '$lib/utils/forkConflict'
import { base } from '$app/paths'
import { page } from '$app/stores'
import CenteredPage from '$lib/components/CenteredPage.svelte'
@@ -111,21 +112,28 @@
clearInterval(interval)
})
async function onToggleMode(path: string, mode: TriggerMode): Promise<void> {
async function onToggleMode(path: string, mode: TriggerMode): Promise<boolean> {
let committed = false
try {
await GcpTriggerService.setGcpTriggerMode({
path,
workspace: $workspaceStore!,
requestBody: { mode }
})
const ok = await withForkConflictRetry(
(force) =>
GcpTriggerService.setGcpTriggerMode({
path,
workspace: $workspaceStore!,
requestBody: { mode, force }
}),
'GCP Pub/Sub trigger'
)
committed = ok
if (ok) loadTriggers()
} catch (err) {
sendUserToast(
`Cannot ${mode === 'enabled' ? 'enable' : mode === 'disabled' ? 'disable' : 'suspend'} GCP Pub/Sub trigger: ${err.body}`,
true
)
} finally {
loadTriggers()
}
return committed
}
run(() => {
frontend/src/routes/(root)/(logged)/kafka_triggers/+page.svelte
+15 -7
View File
@@ -16,6 +16,7 @@
storeLocalSetting,
removeTriggerKindIfUnused
} from '$lib/utils'
import { withForkConflictRetry } from '$lib/utils/forkConflict'
import { base } from '$app/paths'
import { page } from '$app/stores'
import CenteredPage from '$lib/components/CenteredPage.svelte'
@@ -104,13 +105,20 @@
clearInterval(interval)
})
async function onToggleMode(path: string, mode: TriggerMode): Promise<void> {
async function onToggleMode(path: string, mode: TriggerMode): Promise<boolean> {
let committed = false
try {
await KafkaTriggerService.setKafkaTriggerMode({
path,
workspace: $workspaceStore!,
requestBody: { mode }
})
const ok = await withForkConflictRetry(
(force) =>
KafkaTriggerService.setKafkaTriggerMode({
path,
workspace: $workspaceStore!,
requestBody: { mode, force }
}),
'Kafka trigger'
)
committed = ok
if (ok) loadTriggers()
} catch (err) {
sendUserToast(
`Cannot ` +
@@ -118,9 +126,9 @@
` Kafka trigger: ${err.body}`,
true
)
} finally {
loadTriggers()
}
return committed
}
run(() => {
frontend/src/routes/(root)/(logged)/mqtt_triggers/+page.svelte
+18 -8
View File
@@ -17,6 +17,7 @@
storeLocalSetting,
removeTriggerKindIfUnused
} from '$lib/utils'
import { withForkConflictRetry } from '$lib/utils/forkConflict'
import { base } from '$app/paths'
import { page } from '$app/stores'
import CenteredPage from '$lib/components/CenteredPage.svelte'
@@ -99,22 +100,31 @@
clearInterval(interval)
})
async function onToggleMode(path: string, mode: TriggerMode): Promise<void> {
async function onToggleMode(path: string, mode: TriggerMode): Promise<boolean> {
let committed = false
try {
await MqttTriggerService.setMqttTriggerMode({
path,
workspace: $workspaceStore!,
requestBody: { mode }
})
sendUserToast(`${capitalize(mode)} MQTT trigger ${path}`)
const ok = await withForkConflictRetry(
(force) =>
MqttTriggerService.setMqttTriggerMode({
path,
workspace: $workspaceStore!,
requestBody: { mode, force }
}),
'MQTT trigger'
)
if (ok) {
sendUserToast(`${capitalize(mode)} MQTT trigger ${path}`)
loadTriggers()
}
committed = ok
} catch (err) {
sendUserToast(
`Cannot ${mode === 'enabled' ? 'enable' : mode === 'disabled' ? 'disable' : 'suspend'} mqtt trigger: ${err.body}`,
true
)
} finally {
loadTriggers()
}
return committed
}
run(() => {
frontend/src/routes/(root)/(logged)/nats_triggers/+page.svelte
+15 -7
View File
@@ -16,6 +16,7 @@
storeLocalSetting,
removeTriggerKindIfUnused
} from '$lib/utils'
import { withForkConflictRetry } from '$lib/utils/forkConflict'
import { base } from '$app/paths'
import { page } from '$app/stores'
import CenteredPage from '$lib/components/CenteredPage.svelte'
@@ -103,13 +104,20 @@
clearInterval(interval)
})
async function onToggleMode(path: string, mode: TriggerMode): Promise<void> {
async function onToggleMode(path: string, mode: TriggerMode): Promise<boolean> {
let committed = false
try {
await NatsTriggerService.setNatsTriggerMode({
path,
workspace: $workspaceStore!,
requestBody: { mode }
})
const ok = await withForkConflictRetry(
(force) =>
NatsTriggerService.setNatsTriggerMode({
path,
workspace: $workspaceStore!,
requestBody: { mode, force }
}),
'NATS trigger'
)
committed = ok
if (ok) loadTriggers()
} catch (err) {
sendUserToast(
`Cannot ` +
@@ -117,9 +125,9 @@
` NATS trigger: ${err.body}`,
true
)
} finally {
loadTriggers()
}
return committed
}
run(() => {
frontend/src/routes/(root)/(logged)/postgres_triggers/+page.svelte
+18 -8
View File
@@ -17,6 +17,7 @@
removeTriggerKindIfUnused,
capitalize
} from '$lib/utils'
import { withForkConflictRetry } from '$lib/utils/forkConflict'
import { base } from '$app/paths'
import { page } from '$app/stores'
import CenteredPage from '$lib/components/CenteredPage.svelte'
@@ -110,19 +111,28 @@
clearInterval(interval)
})
async function onToggleMode(path: string, mode: TriggerMode): Promise<void> {
async function onToggleMode(path: string, mode: TriggerMode): Promise<boolean> {
let committed = false
try {
await PostgresTriggerService.setPostgresTriggerMode({
path,
workspace: $workspaceStore!,
requestBody: { mode }
})
sendUserToast(`${capitalize(mode)} postgres trigger ${path}`)
const ok = await withForkConflictRetry(
(force) =>
PostgresTriggerService.setPostgresTriggerMode({
path,
workspace: $workspaceStore!,
requestBody: { mode, force }
}),
'postgres trigger'
)
if (ok) {
sendUserToast(`${capitalize(mode)} postgres trigger ${path}`)
loadTriggers()
}
committed = ok
} catch (err) {
sendUserToast(`Cannot change postgres trigger mode: ${err.body}`, true)
} finally {
loadTriggers()
}
return committed
}
run(() => {
frontend/src/routes/(root)/(logged)/routes/+page.svelte
+2
View File
@@ -243,6 +243,8 @@
async function onToggleMode(path: string, mode: TriggerMode): Promise<void> {
try {
// HTTP routes are always workspace-prefixed at runtime, so fork
// and parent live at distinct URLs — no fork-conflict warning.
await HttpTriggerService.setHttpTriggerMode({
path,
workspace: $workspaceStore!,
frontend/src/routes/(root)/(logged)/schedules/+page.svelte
+50 -16
View File
@@ -6,6 +6,7 @@
WorkspaceService
} from '$lib/gen'
import { canWrite, displayDate, getLocalSetting, storeLocalSetting } from '$lib/utils'
import { withForkConflictRetry } from '$lib/utils/forkConflict'
import { base } from '$app/paths'
import CenteredPage from '$lib/components/CenteredPage.svelte'
import { Badge, Button, Skeleton } from '$lib/components/common'
@@ -140,16 +141,41 @@
loadingSchedulesWithJobStats = false
}
// Per-path counter bumped when a schedule toggle is cancelled or errors,
// to force-remount that row's <Toggle>. Toggle uses `bind:checked` on
// its native input; once the user clicks, the local checkbox state
// diverges from the parent's prop expression, and Svelte 5 prop
// reactivity won't push a same-valued prop back down. Re-mounting
// re-initializes from the prop. List-page rows don't optimistically
// flip `enabled`, so they need this nudge — but only the affected row,
// not all rows on the page.
let toggleResetVersions = $state<Record<string, number>>({})
function bumpToggleReset(path: string) {
toggleResetVersions[path] = (toggleResetVersions[path] ?? 0) + 1
}
async function setScheduleEnabled(path: string, enabled: boolean): Promise<void> {
try {
await ScheduleService.setScheduleEnabled({
path,
workspace: $workspaceStore!,
requestBody: { enabled }
})
loadSchedules()
const ok = await withForkConflictRetry(
(force) =>
ScheduleService.setScheduleEnabled({
path,
workspace: $workspaceStore!,
requestBody: { enabled, force }
}),
'schedule'
)
if (ok) {
loadSchedules()
} else {
// Cancelled — nothing changed on the server, skip the reload
// (which would re-fetch job stats and flash the loading flag)
// and just nudge the toggle back to the prop value.
bumpToggleReset(path)
}
} catch (err) {
sendUserToast(`Cannot ` + (enabled ? 'enable' : 'disable') + ` schedule: ${err.body}`, true)
bumpToggleReset(path)
loadSchedules()
}
}
@@ -416,16 +442,24 @@
{/if}
</div>
<Toggle
checked={enabled}
on:change={(e) => {
if (canWrite) {
setScheduleEnabled(path, e.detail)
} else {
sendUserToast('not enough permission', true)
}
}}
/>
{#key toggleResetVersions[path] ?? 0}
<Toggle
checked={enabled}
on:change={(e) => {
if (canWrite) {
setScheduleEnabled(path, e.detail)
} else {
sendUserToast('not enough permission', true)
// Permission denied — bump the row's reset
// counter so the Toggle remounts back to the
// prop value. Without this, the local
// `bind:checked` flip from the user's click
// stays stuck on.
bumpToggleReset(path)
}
}}
/>
{/key}
<div class="flex gap-2 items-center justify-end">
<Button
href={`${base}/runs/?schedule_path=${path}&show_schedules=true&show_future_jobs=true`}
frontend/src/routes/(root)/(logged)/sqs_triggers/+page.svelte
+15 -7
View File
@@ -16,6 +16,7 @@
storeLocalSetting,
removeTriggerKindIfUnused
} from '$lib/utils'
import { withForkConflictRetry } from '$lib/utils/forkConflict'
import { base } from '$app/paths'
import { page } from '$app/stores'
import CenteredPage from '$lib/components/CenteredPage.svelte'
@@ -97,13 +98,20 @@
clearInterval(interval)
})
async function onToggleMode(path: string, mode: TriggerMode): Promise<void> {
async function onToggleMode(path: string, mode: TriggerMode): Promise<boolean> {
let committed = false
try {
await SqsTriggerService.setSqsTriggerMode({
path,
workspace: $workspaceStore!,
requestBody: { mode }
})
const ok = await withForkConflictRetry(
(force) =>
SqsTriggerService.setSqsTriggerMode({
path,
workspace: $workspaceStore!,
requestBody: { mode, force }
}),
'SQS trigger'
)
committed = ok
if (ok) loadTriggers()
} catch (err) {
sendUserToast(
`Cannot ` +
@@ -111,9 +119,9 @@
` sqs trigger: ${err.body}`,
true
)
} finally {
loadTriggers()
}
return committed
}
run(() => {
frontend/src/routes/(root)/(logged)/websocket_triggers/+page.svelte
+18 -8
View File
@@ -17,6 +17,7 @@
removeTriggerKindIfUnused,
capitalize
} from '$lib/utils'
import { withForkConflictRetry } from '$lib/utils/forkConflict'
import { base } from '$app/paths'
import { page } from '$app/stores'
import CenteredPage from '$lib/components/CenteredPage.svelte'
@@ -97,14 +98,23 @@
clearInterval(interval)
})
async function onToggleMode(path: string, mode: TriggerMode): Promise<void> {
async function onToggleMode(path: string, mode: TriggerMode): Promise<boolean> {
let committed = false
try {
await WebsocketTriggerService.setWebsocketTriggerMode({
path,
workspace: $workspaceStore!,
requestBody: { mode }
})
sendUserToast(`${capitalize(mode)} websocket trigger ${path}`)
const ok = await withForkConflictRetry(
(force) =>
WebsocketTriggerService.setWebsocketTriggerMode({
path,
workspace: $workspaceStore!,
requestBody: { mode, force }
}),
'websocket trigger'
)
if (ok) {
sendUserToast(`${capitalize(mode)} websocket trigger ${path}`)
loadTriggers()
}
committed = ok
} catch (err) {
sendUserToast(
`Cannot ` +
@@ -112,9 +122,9 @@
` websocket trigger: ${err.body}`,
true
)
} finally {
loadTriggers()
}
return committed
}
run(() => {
system_prompts/auto-generated/cli/cli-commands.md
+1
View File
@@ -388,6 +388,7 @@ schedule related commands
- `schedule new <path:string>` - create a new schedule locally
- `schedule push <file_path:string> <remote_path:string>` - push a local schedule spec. This overrides any remote versions.
- `schedule enable <path:string>` - Enable a schedule
- `--force` - Bypass the fork-conflict warning when the parent workspace has the same schedule (acknowledges that both crons will fire)
- `schedule disable <path:string>` - Disable a schedule
- `schedule set-permissioned-as <path:string> <email:string>` - Set the email (run-as user) for a schedule (requires admin or wm_deployers group)
system_prompts/auto-generated/prompts.ts
+1
View File
@@ -2536,6 +2536,7 @@ schedule related commands
- \`schedule new <path:string>\` - create a new schedule locally
- \`schedule push <file_path:string> <remote_path:string>\` - push a local schedule spec. This overrides any remote versions.
- \`schedule enable <path:string>\` - Enable a schedule
- \`--force\` - Bypass the fork-conflict warning when the parent workspace has the same schedule (acknowledges that both crons will fire)
- \`schedule disable <path:string>\` - Disable a schedule
- \`schedule set-permissioned-as <path:string> <email:string>\` - Set the email (run-as user) for a schedule (requires admin or wm_deployers group)
system_prompts/auto-generated/skills/cli-commands/SKILL.md
+1
View File
@@ -393,6 +393,7 @@ schedule related commands
- `schedule new <path:string>` - create a new schedule locally
- `schedule push <file_path:string> <remote_path:string>` - push a local schedule spec. This overrides any remote versions.
- `schedule enable <path:string>` - Enable a schedule
- `--force` - Bypass the fork-conflict warning when the parent workspace has the same schedule (acknowledges that both crons will fire)
- `schedule disable <path:string>` - Disable a schedule
- `schedule set-permissioned-as <path:string> <email:string>` - Set the email (run-as user) for a schedule (requires admin or wm_deployers group)