mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-09-06 08:01:35 +00:00
* feat(triggers): add AMQP (RabbitMQ) trigger using the lapin library Fixes WIN-2214 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * chore(triggers): defer AMQP cross-workspace deploy pending utils-internal publish Revert the amqp_trigger additions to the shared windmill-utils-internal TriggerDeployKind and the frontend cross-workspace deploy adapter: the frontend installs the published npm package, which lacks the new kind until a release is cut. AMQP create/edit/delete/list/sync/capture are unaffected (they use local types); only cross-workspace deploy/merge of AMQP triggers waits on the package bump. Also document the at-most-once ack in the consumer loop. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(triggers): address AMQP review — at-least-once ack, workspace cascade, contracts - ack AMQP deliveries only after successful dispatch; nack+requeue on failure - add ON DELETE CASCADE workspace FK so amqp_trigger rows are cleaned on workspace deletion (and the listener stops) - fix the /amqp_triggers/test OpenAPI body and add amqp_trigger to WorkspaceDiffRow.kind - register AMQP in the generated workspace trigger tool (create_trigger) - drop banned $bindable defaults on optional props in the config section - add build_uri unit tests (encoding, ports, vhost) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(triggers): stop AMQP poison-message loop and reconnect on transient drops Chaos testing against a live RabbitMQ broker showed the previous nack(requeue) + immediate re-poll spun a tight redelivery loop (~1000 critical-error reports/sec) on a poison message, and any connection blip permanently disabled the trigger (lapin has no built-in reconnect). - on dispatch failure: nack+requeue then stop consuming; the listener framework re-lists the trigger after its ping goes stale (~15s), backing redelivery off to that cadence instead of a tight loop (verified: rate dropped from ~1000/s to ~1 per ~26s, message preserved) - on connection/stream error: stop and let the framework reconnect instead of disabling; persistent failures are still disabled via get_consumer (verified: a forced connection close now auto-reconnects and resumes) - finish the AI create-trigger action wiring for AMQP: add amqp to CreatedResourceTriggerKind, the action-card registry, and the drawer registry so the result card renders and its "Open" action works Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(triggers): complete AMQP frontend registries and defer merge rows - add amqp to capturableTriggerTypes (so AmqpCapture mounts), the Runs jobTriggerKinds filter, and CLOUD_DISABLED_TRIGGER_TYPES - wire AMQP into global AI chat mode: TRIGGER_KINDS, the request union, writeTriggerSchema, triggerServices, and the draft adapter - stop emitting actionable AMQP fork-comparison rows (revert amqp_trigger from TRIGGER_OR_SCHEDULE_TABLES) since cross-workspace deploy is deferred until windmill-utils-internal is published — avoids a deploy that fails with "Unknown kind: amqp_trigger" - use design-system TextInput instead of raw <input> in the config section Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(triggers): complete AMQP session/draft registries and constrain prefetch - add amqp to the session-deploy, draft-compare, preview-router, and copilot workspace-item registries so AMQP drafts/deploys/nav/path resolution work - include amqp_count in the MoveDrawer attached-trigger rename warning - replace the raw prefetch <input> with a design-system TextInput bounded to an integer 1-65535 (backend u16) and block save on invalid values Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(triggers): make AMQP disconnect/reconnect consistent with the Kafka trigger lapin, like rdkafka, has no transparent reconnect, so the AMQP listener now mirrors the Kafka trigger's explicit reconnect loop instead of relying on the framework re-list (which disabled the trigger once get_consumer failed on a sustained outage): - get_consumer returns cheaply; consume owns a (re)connect loop that retries with a 30s backoff, reports a critical error every 10 failed attempts, and reports a recovered critical error once it reconnects — never disabling the trigger on a connectivity failure - a consumer/stream error breaks out to reconnect rather than disabling - dispatch failure still nacks+requeues (at-least-once) with a short backoff to avoid a tight poison-message loop, keeping the connection alive Verified against a live RabbitMQ broker: killing the broker keeps the trigger enabled and retrying (attempt N), and restarting it auto-reconnects (logs "reconnected after N attempts") and resumes dispatch. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(triggers): complete AMQP capture registries and constrain prefetch contract - add the 'amqp' case to triggerKindToTriggerType so opening the AMQP editor from a capture button no longer throws "Unknown TriggerKind: amqp" - register AmqpIcon in CaptureTable's icon map and add an AMQP entry to the script/flow CaptureButton menu - bound the OpenAPI prefetch_count to an integer 1-65535 (matches the Rust u16) and regenerate clients/prompts - require a non-empty exchange name when the exchange binding is enabled - build_uri: fall back to "/" on a blank vhost and bracket IPv6 hosts (+ tests) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * feat(triggers): wire AMQP into pipeline graph, git-sync, and preprocessor types - asset_graph: discover attached amqp_trigger rows and emit an AMQP TriggerEdge so AMQP triggers render (and can be opened/deleted) on the data-pipeline canvas - frontend pipeline graph: add amqp to NativeTriggerKind, the add-trigger menu, node presentation, event-trigger set, annotation keywords, and the editor/service registrations - git-sync: add the amqp_trigger include pattern (+ test) so an AMQP git-sync deployment stages only its .amqp_trigger.* file, not an unrelated same-path object - preprocessor starters: add the AMQP event to the generated TS/Python/PHP trigger event types (kind/payload/exchange/routing_key/queue_name/redelivered/ delivery_tag) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(triggers): finish AMQP pipeline/parser wiring, prefetch validation, source lists - fix a stray edit that corrupted the pre-existing MqttTriggerEditor import ($lib/... path) in PipelineTriggerEditors.svelte - reject prefetch_count = 0 server-side in validate_config (RabbitMQ treats 0 as unlimited) and defensively skip basic_qos(0) in build_consumer (covers the capture path that bypasses CRUD validation) - recognize `// on amqp` in the canonical parser (TriggerSpec::Amqp) and add amqp to the CLI non-autorun/event-trigger sets so a pipeline cascade never runs an AMQP-only node as a manual root without an event - add amqp to the preprocessor intro lists and both pipeline AI instructions Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(triggers): reject zero AMQP prefetch in all paths and finish guidance lists - extract a shared validate_amqp_options used by both CRUD validate_config and build_consumer, so capture configs (which bypass CRUD validation) also reject prefetch 0 instead of silently connecting with an unlimited buffer (+ unit tests for 0/1/65535/None) - add AMQP to the main script-writing preprocessor-sources prompt and the CLI triggers-skill guidance list Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * docs(triggers): de-duplicate AMQP prefetch comment and fix GET response text - keep the zero-prefetch rationale only on the shared validate_amqp_options doc; drop the redundant call-site comments - correct the getAmqpTrigger OpenAPI 200 description ("deleted" -> "retrieved") Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * chore: update ee-repo-ref to aaa6cb89b05b76139252c64f057e53b94d12ac60 This commit updates the EE repository reference after PR #680 was merged in windmill-ee-private. Previous ee-repo-ref: 5da5fd65aca9594b2611837a52e4677b544b0380 New ee-repo-ref: aaa6cb89b05b76139252c64f057e53b94d12ac60 Automated by sync-ee-ref workflow. * chore(migrations): consolidate the four AMQP migrations into one The table and the three enum ADD VALUE statements (trigger_kind, job_trigger_kind, draft_kind) are one atomic feature. ALTER TYPE ... ADD VALUE runs inside the migration transaction on PG >= 14 (Windmill's minimum) since the amqp_trigger table doesn't reference those enum types, so they can share a single migration instead of four. Verified applying cleanly in a single transaction on a fresh DB. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
291 lines
10 KiB
TypeScript
291 lines
10 KiB
TypeScript
/**
|
|
* One-off migration from the localStorage UserDraft autosave to the
|
|
* DB-backed `draft` table. Reads the workspace-scoped
|
|
* `userdraft/w/{workspace}/{kind}/{path}` keys (written by the editor during
|
|
* the interim LS-backed phase, so the embedded workspace is correct), POSTing
|
|
* each to `/drafts/update` and clearing the source key only on success — so
|
|
* it's idempotent without a sentinel; failed entries retry next mount. Not
|
|
* workspace-gated: keys embed their own workspace and the token covers all of
|
|
* them, so gating would orphan other-workspace entries.
|
|
*
|
|
* Before uploading, each draft is compared against its deployed version
|
|
* (script / flow / app); a draft that's deep-equal to what's deployed carries
|
|
* no changes, so it's dropped instead of migrated (no error).
|
|
*/
|
|
|
|
import { AppService, DraftService, FlowService, ScriptService } from './gen'
|
|
import type { UserDraftItemKind } from './gen'
|
|
import type { App } from './components/apps/types'
|
|
import { migrateApp } from './components/apps/migrateApp'
|
|
import { sendUserToast } from './toast'
|
|
import { draftValuesEqual } from './userDraft.svelte'
|
|
import {
|
|
openDraftMigrationErrorModal,
|
|
reportDraftMigrationError
|
|
} from './userDraftMigrationErrors.svelte'
|
|
import { getUsernameForNamespace } from './userNamespace'
|
|
import { randomUUID } from './utils/uuid'
|
|
|
|
// Mirror of `USER_DRAFT_ITEM_KINDS`, inlined to avoid importing the reactive
|
|
// runtime. The `_Exhaustive` assertion below fails compilation if this list
|
|
// drifts from the OpenAPI schema's kinds.
|
|
const ITEM_KINDS = [
|
|
'script',
|
|
'flow',
|
|
'app',
|
|
'raw_app',
|
|
'resource',
|
|
'variable',
|
|
'trigger_schedule',
|
|
'trigger_webhook',
|
|
'trigger_default_email',
|
|
'trigger_email',
|
|
'trigger_http',
|
|
'trigger_websocket',
|
|
'trigger_postgres',
|
|
'trigger_kafka',
|
|
'trigger_nats',
|
|
'trigger_mqtt',
|
|
'trigger_amqp',
|
|
'trigger_sqs',
|
|
'trigger_gcp',
|
|
'trigger_azure',
|
|
'trigger_poll',
|
|
'trigger_cli',
|
|
'trigger_nextcloud',
|
|
'trigger_google',
|
|
'trigger_github',
|
|
'data_pipeline'
|
|
] as const satisfies readonly UserDraftItemKind[]
|
|
|
|
type _Exhaustive =
|
|
Exclude<UserDraftItemKind, (typeof ITEM_KINDS)[number]> extends never ? true : never
|
|
const _: _Exhaustive = true
|
|
void _
|
|
|
|
const KEY_PREFIX = 'userdraft/w/'
|
|
|
|
type ParsedKey = {
|
|
key: string
|
|
workspace: string
|
|
itemKind: UserDraftItemKind
|
|
path: string
|
|
}
|
|
|
|
/**
|
|
* Split a `userdraft/w/{workspace}/{kind}/{path}` key into its parts, or
|
|
* `undefined` for keys that don't match or have an unrecognized kind.
|
|
* `path` is `''` only for a legacy `/add` autosave (key ended `.../{kind}/`);
|
|
* the caller mints a fresh slot for those. An empty `path` here is always that
|
|
* case, never garbage, since a non-matching key returns `undefined`.
|
|
*/
|
|
function parseKey(key: string): ParsedKey | undefined {
|
|
if (!key.startsWith(KEY_PREFIX)) return undefined
|
|
const rest = key.slice(KEY_PREFIX.length)
|
|
const firstSlash = rest.indexOf('/')
|
|
if (firstSlash <= 0) return undefined
|
|
const workspace = rest.slice(0, firstSlash)
|
|
const afterWorkspace = rest.slice(firstSlash + 1)
|
|
for (const kind of ITEM_KINDS) {
|
|
const kindPrefix = `${kind}/`
|
|
if (afterWorkspace.startsWith(kindPrefix)) {
|
|
const path = afterWorkspace.slice(kindPrefix.length)
|
|
return { key, workspace, itemKind: kind, path }
|
|
}
|
|
}
|
|
return undefined
|
|
}
|
|
|
|
/**
|
|
* Mint a fresh `u/{user}/draft_{uuid}` slot for a pathless legacy `/add`
|
|
* autosave, matching the editors' `/add` redirect convention (`makeDraftAddLoad`).
|
|
* Underscores not dashes (path segments are `[a-zA-Z0-9_]` words); `randomUUID`
|
|
* not `crypto.randomUUID` (WebCrypto is absent on non-secure origins).
|
|
*/
|
|
function mintDraftAddPath(): string {
|
|
const username = getUsernameForNamespace()
|
|
const uuid = randomUUID().replaceAll('-', '_')
|
|
return `u/${username}/draft_${uuid}`
|
|
}
|
|
|
|
/**
|
|
* Extract `value` and `lastWrittenAt` from the LS payload (`{ value, lastWrittenAt?, ... }`).
|
|
* `lastWrittenAt` rides along as `last_sync` so a fresher server draft wins.
|
|
* `undefined` when the slot is empty / unparseable / wrong shape.
|
|
*/
|
|
function readPayload(key: string): { value: unknown; lastWrittenAt?: number } | undefined {
|
|
try {
|
|
const raw = localStorage.getItem(key)
|
|
if (raw == null || raw === 'undefined') return undefined
|
|
const parsed = JSON.parse(raw)
|
|
if (parsed == null || typeof parsed !== 'object' || !('value' in parsed)) return undefined
|
|
const lastWrittenAt = (parsed as { lastWrittenAt?: unknown }).lastWrittenAt
|
|
return {
|
|
value: (parsed as { value: unknown }).value,
|
|
lastWrittenAt: typeof lastWrittenAt === 'number' ? lastWrittenAt : undefined
|
|
}
|
|
} catch {
|
|
return undefined
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Fetch the deployed value for a draft so the migration can drop a draft that
|
|
* carries no changes (deep-equal to what's already deployed) instead of
|
|
* uploading a no-op that would light up the "unsaved" badge. Returns the
|
|
* comparable deployed payload, or `undefined` when there's nothing to compare
|
|
* against: an unsupported kind, a pathless (minted `/add`) draft, or a fetch
|
|
* miss (404 — the path is draft-only, so the draft is genuinely new). `getDraft`
|
|
* is forced off so we compare against the deployed baseline, not our own draft.
|
|
*/
|
|
async function fetchDeployedValue(
|
|
workspace: string,
|
|
kind: UserDraftItemKind,
|
|
path: string
|
|
): Promise<unknown | undefined> {
|
|
if (!path) return undefined
|
|
try {
|
|
switch (kind) {
|
|
case 'script':
|
|
return await ScriptService.getScriptByPath({ workspace, path, getDraft: false })
|
|
case 'flow':
|
|
return await FlowService.getFlowByPath({ workspace, path, getDraft: false })
|
|
case 'app': {
|
|
// The app autosave stores the inner `App`, not the `AppWithLastVersion`
|
|
// wrapper getAppByPath returns — compare against `.value`. Run
|
|
// `migrateApp` so the deployed value matches the editor-migrated draft
|
|
// (AppEditor `migrateApp`s `stateApp` on mount); without this an app
|
|
// whose deployed row predates those field migrations never dedups.
|
|
const app = await AppService.getAppByPath({ workspace, path, getDraft: false })
|
|
const value = (app as { value?: App }).value
|
|
if (value) migrateApp(value)
|
|
return value
|
|
}
|
|
default:
|
|
return undefined
|
|
}
|
|
} catch {
|
|
// No deployed item at this path (or the fetch failed) — nothing to dedup
|
|
// against, so the caller proceeds to upload the draft.
|
|
return undefined
|
|
}
|
|
}
|
|
|
|
function collectKeys(): string[] {
|
|
const keys: string[] = []
|
|
for (let i = 0; i < localStorage.length; i++) {
|
|
const k = localStorage.key(i)
|
|
if (k != null && k.startsWith(KEY_PREFIX)) keys.push(k)
|
|
}
|
|
return keys
|
|
}
|
|
|
|
/**
|
|
* Push every LS `userdraft/...` entry to `/drafts/update`, clearing each on
|
|
* success; failures stay in LS and retry next mount.
|
|
*
|
|
* `lastWrittenAt` rides as `last_sync` so the server rejects the upload when its
|
|
* own draft is fresher (the user may have edited the same path from another
|
|
* browser since this LS write). Missing `lastWrittenAt` uses epoch 0 (insert if
|
|
* absent, else yield). A `conflict` response means the server won; drop the LS
|
|
* copy. Never throws — the caller is fire-and-forget.
|
|
*/
|
|
export async function migrateUserDraftsToDb(): Promise<void> {
|
|
if (typeof localStorage === 'undefined') return
|
|
const keys = collectKeys()
|
|
if (keys.length === 0) return
|
|
|
|
// Parse up front so we only announce the migration when there's a real
|
|
// `userdraft/...` entry to upload (and can drop unparseable junk first).
|
|
const toMigrate: {
|
|
key: string
|
|
parsed: ParsedKey
|
|
path: string
|
|
value: unknown
|
|
lastWrittenAt?: number
|
|
}[] = []
|
|
for (const key of keys) {
|
|
const parsed = parseKey(key)
|
|
if (!parsed) continue
|
|
const payload = readPayload(key)
|
|
if (payload === undefined) {
|
|
// Unparseable or empty — clear so we don't keep retrying it.
|
|
try {
|
|
localStorage.removeItem(key)
|
|
} catch {
|
|
// ignore
|
|
}
|
|
continue
|
|
}
|
|
// A legacy `/add` autosave has no path — mint a fresh slot so it lands
|
|
// as a regular draft-only item instead of being dropped.
|
|
const path = parsed.path === '' ? mintDraftAddPath() : parsed.path
|
|
toMigrate.push({
|
|
key,
|
|
parsed,
|
|
path,
|
|
value: payload.value,
|
|
lastWrittenAt: payload.lastWrittenAt
|
|
})
|
|
}
|
|
if (toMigrate.length === 0) return
|
|
|
|
// Legacy drafts detected — tell the user the one-off upload is running, with
|
|
// an escape hatch to the modal where any failures show up as they happen.
|
|
sendUserToast('Migrating local storage drafts ...', 'info', [
|
|
{ label: 'See more', callback: openDraftMigrationErrorModal }
|
|
])
|
|
|
|
for (const { key, parsed, path, value, lastWrittenAt } of toMigrate) {
|
|
try {
|
|
// Dedup: if the draft is deep-equal to the deployed version it carries
|
|
// no changes — drop it (no error) instead of uploading a no-op draft.
|
|
// Fetches against `parsed.path` (the real item path); minted `/add`
|
|
// drafts have `parsed.path === ''` and so are never deduped.
|
|
const deployed = await fetchDeployedValue(parsed.workspace, parsed.itemKind, parsed.path)
|
|
if (deployed !== undefined && draftValuesEqual(value, deployed)) {
|
|
try {
|
|
localStorage.removeItem(key)
|
|
} catch {
|
|
// Best-effort; a stale LS entry is harmless — it re-dedups next mount.
|
|
}
|
|
continue
|
|
}
|
|
// Preserve the draft's original age: stamp `created_at` with the LS
|
|
// write time (epoch 0 when unknown) so migrated drafts don't all
|
|
// resurface to the top as freshly created. Same value as `last_sync`,
|
|
// which still drives the conflict check.
|
|
const writtenAt = new Date(lastWrittenAt ?? 0).toISOString()
|
|
const res = await DraftService.updateDraft({
|
|
workspace: parsed.workspace,
|
|
kind: parsed.itemKind,
|
|
path,
|
|
requestBody: { value, last_sync: writtenAt, created_at: writtenAt }
|
|
})
|
|
if (res.status === 'conflict') {
|
|
console.info(
|
|
`UserDraft LS→DB migration: server draft for ${path} is fresher, dropping LS copy`
|
|
)
|
|
}
|
|
try {
|
|
localStorage.removeItem(key)
|
|
} catch {
|
|
// Best-effort. If LS removal fails the next mount retries
|
|
// the save (the conflict rule keeps it idempotent).
|
|
}
|
|
} catch (e) {
|
|
// Leave the LS entry in place — the next mount tries again — but
|
|
// surface it so the user isn't silently stuck, with an escape
|
|
// hatch to drop the un-migratable draft.
|
|
console.error('UserDraft LS→DB migration: failed for', key, e)
|
|
reportDraftMigrationError({
|
|
key,
|
|
workspace: parsed.workspace,
|
|
itemKind: parsed.itemKind,
|
|
path,
|
|
value
|
|
})
|
|
}
|
|
}
|
|
}
|