mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-08-26 00:01:37 +00:00
68debab877
* 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>
597 lines
18 KiB
TypeScript
597 lines
18 KiB
TypeScript
import { minimatch } from 'minimatch'
|
|
import {
|
|
AzureTriggerService,
|
|
EmailTriggerService,
|
|
GcpTriggerService,
|
|
HttpTriggerService,
|
|
KafkaTriggerService,
|
|
MqttTriggerService,
|
|
AmqpTriggerService,
|
|
NatsTriggerService,
|
|
PostgresTriggerService,
|
|
ScheduleService,
|
|
SqsTriggerService,
|
|
WebsocketTriggerService,
|
|
type GcpTriggerData,
|
|
type WorkspaceDeployUISettings
|
|
} from './gen'
|
|
import type { TriggerKind } from './components/triggers'
|
|
import { base } from './base'
|
|
|
|
type DeployUIType = 'script' | 'flow' | 'app' | 'resource' | 'variable' | 'secret' | 'trigger'
|
|
|
|
export type Kind =
|
|
| 'script'
|
|
| 'resource'
|
|
| 'schedule'
|
|
| 'variable'
|
|
| 'flow'
|
|
| 'app'
|
|
| 'raw_app'
|
|
| 'resource_type'
|
|
| 'folder'
|
|
// Per-kind trigger names returned by the backend's `compareWorkspaces` API.
|
|
| 'http_trigger'
|
|
| 'websocket_trigger'
|
|
| 'kafka_trigger'
|
|
| 'nats_trigger'
|
|
| 'postgres_trigger'
|
|
| 'mqtt_trigger'
|
|
| 'amqp_trigger'
|
|
| 'sqs_trigger'
|
|
| 'gcp_trigger'
|
|
| 'azure_trigger'
|
|
| 'email_trigger'
|
|
// Data table migration, diffed per `<datatable>/<timestamp>_<name>` path.
|
|
| 'datatable_migration'
|
|
// Legacy generic kind used by the cross-workspace `DeployWorkspace` UI,
|
|
// which carries the trigger sub-kind in `additionalInformation`.
|
|
| 'trigger'
|
|
// A data-pipeline draft bundle (drafts list only — not a deployable item;
|
|
// opens the pipeline view instead).
|
|
| 'data_pipeline'
|
|
|
|
export const ALL_DEPLOYABLE: WorkspaceDeployUISettings = {
|
|
include_path: [],
|
|
include_type: ['script', 'flow', 'app', 'resource', 'variable', 'secret', 'trigger']
|
|
}
|
|
|
|
export type AdditionalInformation = {
|
|
triggers?: {
|
|
kind: TriggerKind
|
|
}
|
|
}
|
|
|
|
export function isDeployable(
|
|
type: DeployUIType,
|
|
path: string,
|
|
deployUiSettings: WorkspaceDeployUISettings | undefined
|
|
) {
|
|
if (deployUiSettings == undefined) {
|
|
return false
|
|
}
|
|
|
|
if (deployUiSettings.include_type != undefined && !deployUiSettings.include_type.includes(type)) {
|
|
return false
|
|
}
|
|
|
|
if (
|
|
deployUiSettings.include_path != undefined &&
|
|
deployUiSettings.include_path.length != 0 &&
|
|
deployUiSettings.include_path.every((x) => !minimatch(path, x))
|
|
) {
|
|
return false
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
export async function existsTrigger(
|
|
data: { workspace: string; path: string },
|
|
triggerKind: TriggerKind
|
|
) {
|
|
if (triggerKind === 'routes') {
|
|
return await HttpTriggerService.existsHttpTrigger(data)
|
|
} else if (triggerKind === 'kafka') {
|
|
return await KafkaTriggerService.existsKafkaTrigger(data)
|
|
} else if (triggerKind === 'mqtt') {
|
|
return await MqttTriggerService.existsMqttTrigger(data)
|
|
} else if (triggerKind === 'amqp') {
|
|
return await AmqpTriggerService.existsAmqpTrigger(data)
|
|
} else if (triggerKind === 'postgres') {
|
|
return await PostgresTriggerService.existsPostgresTrigger(data)
|
|
} else if (triggerKind === 'sqs') {
|
|
return await SqsTriggerService.existsSqsTrigger(data)
|
|
} else if (triggerKind === 'gcp') {
|
|
return await GcpTriggerService.existsGcpTrigger(data)
|
|
} else if (triggerKind === 'websockets') {
|
|
return await WebsocketTriggerService.existsWebsocketTrigger(data)
|
|
} else if (triggerKind === 'nats') {
|
|
return await NatsTriggerService.existsNatsTrigger(data)
|
|
} else if (triggerKind === 'azure') {
|
|
return await AzureTriggerService.existsAzureTrigger(data)
|
|
} else if (triggerKind === 'emails') {
|
|
return await EmailTriggerService.existsEmailTrigger(data)
|
|
} else if (triggerKind === 'schedules') {
|
|
return await ScheduleService.existsSchedule(data)
|
|
}
|
|
|
|
throw new Error(
|
|
`Unexpected trigger kind ${triggerKind}. Allowed kinds are: routes, kafka, mqtt, postgres, sqs, gcp, websockets, nats, azure, emails, schedules.`
|
|
)
|
|
}
|
|
|
|
/**
|
|
* Strip operational state (`mode`, `enabled`) from a trigger/schedule payload
|
|
* before sending it to an update endpoint via the merge UI. The backend's
|
|
* `update_trigger` handler preserves the target row's existing `mode` when
|
|
* both fields are absent from the request (`is_mode_unspecified()`), so
|
|
* stripping here lets a fork→parent (or parent→fork) deploy carry config
|
|
* changes without flipping the target's enabled/disabled state. Schedules'
|
|
* `EditSchedule` already lacks `enabled` on the backend, but stripping keeps
|
|
* the intent explicit and matches the YAML/CLI round-trip behavior.
|
|
*
|
|
* Used by the legacy `kind === 'trigger'` path in `utils_workspace_deploy.ts`
|
|
* (the cross-workspace deploy UI). The merge-UI deploy goes through the
|
|
* shared `deployItem` in `windmill-utils-internal`, which applies its own
|
|
* `stripOperationalStateOnUpdate` at the dispatch layer.
|
|
*/
|
|
export function stripOperationalState<T extends Record<string, any>>(
|
|
payload: T
|
|
): Omit<T, 'mode' | 'enabled'> {
|
|
const { mode: _mode, enabled: _enabled, ...rest } = payload
|
|
return rest
|
|
}
|
|
|
|
/**
|
|
* Get trigger deployment data with optional permissioned_as preservation.
|
|
* @param onBehalfOf - If set, the trigger will be deployed with this permissioned_as (u/username or g/group) and preserve_permissioned_as=true.
|
|
*/
|
|
export async function getTriggersDeployData(
|
|
kind: TriggerKind,
|
|
path: string,
|
|
workspace: string,
|
|
onBehalfOf?: string
|
|
) {
|
|
const preservePermissionedAs = onBehalfOf !== undefined
|
|
|
|
if (kind === 'sqs') {
|
|
const sqsTrigger = await SqsTriggerService.getSqsTrigger({
|
|
workspace: workspace!,
|
|
path: path
|
|
})
|
|
|
|
return {
|
|
data: {
|
|
...sqsTrigger,
|
|
permissioned_as: onBehalfOf,
|
|
preserve_permissioned_as: preservePermissionedAs
|
|
},
|
|
createFn: SqsTriggerService.createSqsTrigger,
|
|
updateFn: SqsTriggerService.updateSqsTrigger
|
|
}
|
|
} else if (kind === 'kafka') {
|
|
const kafkaTrigger = await KafkaTriggerService.getKafkaTrigger({
|
|
workspace: workspace!,
|
|
path: path
|
|
})
|
|
|
|
return {
|
|
data: {
|
|
...kafkaTrigger,
|
|
permissioned_as: onBehalfOf,
|
|
preserve_permissioned_as: preservePermissionedAs
|
|
},
|
|
createFn: KafkaTriggerService.createKafkaTrigger,
|
|
updateFn: KafkaTriggerService.updateKafkaTrigger
|
|
}
|
|
} else if (kind === 'mqtt') {
|
|
const mqttTrigger = await MqttTriggerService.getMqttTrigger({
|
|
workspace: workspace!,
|
|
path: path
|
|
})
|
|
|
|
return {
|
|
data: {
|
|
...mqttTrigger,
|
|
permissioned_as: onBehalfOf,
|
|
preserve_permissioned_as: preservePermissionedAs
|
|
},
|
|
createFn: MqttTriggerService.createMqttTrigger,
|
|
updateFn: MqttTriggerService.updateMqttTrigger
|
|
}
|
|
} else if (kind === 'amqp') {
|
|
const amqpTrigger = await AmqpTriggerService.getAmqpTrigger({
|
|
workspace: workspace!,
|
|
path: path
|
|
})
|
|
|
|
return {
|
|
data: {
|
|
...amqpTrigger,
|
|
permissioned_as: onBehalfOf,
|
|
preserve_permissioned_as: preservePermissionedAs
|
|
},
|
|
createFn: AmqpTriggerService.createAmqpTrigger,
|
|
updateFn: AmqpTriggerService.updateAmqpTrigger
|
|
}
|
|
} else if (kind === 'nats') {
|
|
const natsTrigger = await NatsTriggerService.getNatsTrigger({
|
|
workspace: workspace!,
|
|
path: path
|
|
})
|
|
|
|
return {
|
|
data: {
|
|
...natsTrigger,
|
|
permissioned_as: onBehalfOf,
|
|
preserve_permissioned_as: preservePermissionedAs
|
|
},
|
|
createFn: NatsTriggerService.createNatsTrigger,
|
|
updateFn: NatsTriggerService.updateNatsTrigger
|
|
}
|
|
} else if (kind === 'gcp') {
|
|
const gcpTrigger = await GcpTriggerService.getGcpTrigger({
|
|
workspace: workspace!,
|
|
path: path
|
|
})
|
|
|
|
gcpTrigger.subscription_id = ''
|
|
gcpTrigger.subscription_mode = 'create_update'
|
|
|
|
if (gcpTrigger.delivery_config) {
|
|
gcpTrigger.delivery_config.audience = ''
|
|
}
|
|
|
|
const data: GcpTriggerData = {
|
|
...gcpTrigger,
|
|
delivery_config: gcpTrigger.delivery_config ?? undefined,
|
|
base_endpoint:
|
|
gcpTrigger.delivery_type === 'push' ? `${window.location.origin}${base}` : undefined,
|
|
permissioned_as: onBehalfOf,
|
|
preserve_permissioned_as: preservePermissionedAs
|
|
}
|
|
|
|
return {
|
|
data,
|
|
createFn: GcpTriggerService.createGcpTrigger,
|
|
updateFn: GcpTriggerService.updateGcpTrigger
|
|
}
|
|
} else if (kind === 'postgres') {
|
|
const postgresTrigger = await PostgresTriggerService.getPostgresTrigger({
|
|
workspace: workspace!,
|
|
path: path
|
|
})
|
|
|
|
return {
|
|
data: {
|
|
...postgresTrigger,
|
|
permissioned_as: onBehalfOf,
|
|
preserve_permissioned_as: preservePermissionedAs
|
|
},
|
|
createFn: PostgresTriggerService.createPostgresTrigger,
|
|
updateFn: PostgresTriggerService.updatePostgresTrigger
|
|
}
|
|
} else if (kind === 'websockets') {
|
|
const websocketTrigger = await WebsocketTriggerService.getWebsocketTrigger({
|
|
workspace: workspace!,
|
|
path: path
|
|
})
|
|
|
|
return {
|
|
data: {
|
|
...websocketTrigger,
|
|
permissioned_as: onBehalfOf,
|
|
preserve_permissioned_as: preservePermissionedAs
|
|
},
|
|
createFn: WebsocketTriggerService.createWebsocketTrigger,
|
|
updateFn: WebsocketTriggerService.updateWebsocketTrigger
|
|
}
|
|
} else if (kind === 'routes') {
|
|
const httpTrigger = await HttpTriggerService.getHttpTrigger({
|
|
workspace: workspace!,
|
|
path: path
|
|
})
|
|
|
|
return {
|
|
data: {
|
|
...httpTrigger,
|
|
permissioned_as: onBehalfOf,
|
|
preserve_permissioned_as: preservePermissionedAs
|
|
},
|
|
createFn: HttpTriggerService.createHttpTrigger,
|
|
updateFn: HttpTriggerService.updateHttpTrigger
|
|
}
|
|
} else if (kind === 'azure') {
|
|
const azureTrigger = await AzureTriggerService.getAzureTrigger({
|
|
workspace: workspace!,
|
|
path: path
|
|
})
|
|
|
|
return {
|
|
data: {
|
|
...azureTrigger,
|
|
permissioned_as: onBehalfOf,
|
|
preserve_permissioned_as: preservePermissionedAs
|
|
},
|
|
createFn: AzureTriggerService.createAzureTrigger,
|
|
updateFn: AzureTriggerService.updateAzureTrigger
|
|
}
|
|
} else if (kind === 'emails') {
|
|
const emailTrigger = await EmailTriggerService.getEmailTrigger({
|
|
workspace: workspace!,
|
|
path: path
|
|
})
|
|
|
|
return {
|
|
data: {
|
|
...emailTrigger,
|
|
permissioned_as: onBehalfOf,
|
|
preserve_permissioned_as: preservePermissionedAs
|
|
},
|
|
createFn: EmailTriggerService.createEmailTrigger,
|
|
updateFn: EmailTriggerService.updateEmailTrigger
|
|
}
|
|
} else if (kind === 'schedules') {
|
|
const schedulesTrigger = await ScheduleService.getSchedule({
|
|
workspace: workspace!,
|
|
path: path
|
|
})
|
|
return {
|
|
data: {
|
|
...schedulesTrigger,
|
|
// permissioned_as is only set on create, not update
|
|
permissioned_as: onBehalfOf,
|
|
preserve_permissioned_as: preservePermissionedAs
|
|
},
|
|
createFn: ScheduleService.createSchedule,
|
|
updateFn: ScheduleService.updateSchedule
|
|
}
|
|
}
|
|
|
|
throw new Error(`Unexpected trigger kind got: ${kind}`)
|
|
}
|
|
|
|
/**
|
|
* Runtime fields stripped from the trigger/schedule diff so the drawer mirrors
|
|
* the backend's `compare_two_trigger_or_schedule` semantics — same set as
|
|
* `TRIGGER_COMPARE_IGNORE` and `stripTriggerOrScheduleRuntimeFields` in the
|
|
* shared deploy module.
|
|
*/
|
|
export const TRIGGER_RUNTIME_IGNORE = new Set([
|
|
'workspace_id',
|
|
'edited_by',
|
|
'edited_at',
|
|
'email',
|
|
'error',
|
|
'enabled',
|
|
'mode',
|
|
'server_id',
|
|
'last_server_ping',
|
|
'extra_perms',
|
|
'permissioned_as',
|
|
// Server-managed (kept in sync with backend `TRIGGER_COMPARE_IGNORE`).
|
|
'subscription_id',
|
|
'push_auth_config'
|
|
])
|
|
|
|
function stripTriggerRuntimeFields<T extends Record<string, any>>(row: T): Partial<T> {
|
|
const out: Record<string, any> = {}
|
|
for (const [k, v] of Object.entries(row)) {
|
|
if (!TRIGGER_RUNTIME_IGNORE.has(k)) out[k] = v
|
|
}
|
|
return out as Partial<T>
|
|
}
|
|
|
|
export async function getTriggerValue(kind: TriggerKind, path: string, workspace: string) {
|
|
let trigger: Record<string, any>
|
|
if (kind === 'sqs') {
|
|
trigger = await SqsTriggerService.getSqsTrigger({ workspace, path })
|
|
} else if (kind === 'kafka') {
|
|
trigger = await KafkaTriggerService.getKafkaTrigger({ workspace, path })
|
|
} else if (kind === 'mqtt') {
|
|
trigger = await MqttTriggerService.getMqttTrigger({ workspace, path })
|
|
} else if (kind === 'amqp') {
|
|
trigger = await AmqpTriggerService.getAmqpTrigger({ workspace, path })
|
|
} else if (kind === 'nats') {
|
|
trigger = await NatsTriggerService.getNatsTrigger({ workspace, path })
|
|
} else if (kind === 'postgres') {
|
|
trigger = await PostgresTriggerService.getPostgresTrigger({ workspace, path })
|
|
} else if (kind === 'gcp') {
|
|
trigger = await GcpTriggerService.getGcpTrigger({ workspace, path })
|
|
} else if (kind === 'websockets') {
|
|
trigger = await WebsocketTriggerService.getWebsocketTrigger({ workspace, path })
|
|
} else if (kind === 'routes') {
|
|
trigger = await HttpTriggerService.getHttpTrigger({ workspace, path })
|
|
} else if (kind === 'schedules') {
|
|
trigger = await ScheduleService.getSchedule({ workspace, path })
|
|
} else if (kind === 'azure') {
|
|
trigger = await AzureTriggerService.getAzureTrigger({ workspace, path })
|
|
} else if (kind === 'emails') {
|
|
trigger = await EmailTriggerService.getEmailTrigger({ workspace, path })
|
|
} else {
|
|
throw new Error(`Unexpected trigger kind got: ${kind}`)
|
|
}
|
|
return stripTriggerRuntimeFields(trigger)
|
|
}
|
|
|
|
/**
|
|
* Get the permissioned_as for a trigger (used for on_behalf_of during deployment).
|
|
*/
|
|
export async function getTriggerPermissionedAs(
|
|
kind: TriggerKind,
|
|
path: string,
|
|
workspace: string
|
|
): Promise<string | undefined> {
|
|
try {
|
|
if (kind === 'sqs') {
|
|
const trigger = await SqsTriggerService.getSqsTrigger({ workspace, path })
|
|
return trigger.permissioned_as
|
|
} else if (kind === 'kafka') {
|
|
const trigger = await KafkaTriggerService.getKafkaTrigger({ workspace, path })
|
|
return trigger.permissioned_as
|
|
} else if (kind === 'mqtt') {
|
|
const trigger = await MqttTriggerService.getMqttTrigger({ workspace, path })
|
|
return trigger.permissioned_as
|
|
} else if (kind === 'amqp') {
|
|
const trigger = await AmqpTriggerService.getAmqpTrigger({ workspace, path })
|
|
return trigger.permissioned_as
|
|
} else if (kind === 'nats') {
|
|
const trigger = await NatsTriggerService.getNatsTrigger({ workspace, path })
|
|
return trigger.permissioned_as
|
|
} else if (kind === 'gcp') {
|
|
const trigger = await GcpTriggerService.getGcpTrigger({ workspace, path })
|
|
return trigger.permissioned_as
|
|
} else if (kind === 'postgres') {
|
|
const trigger = await PostgresTriggerService.getPostgresTrigger({ workspace, path })
|
|
return trigger.permissioned_as
|
|
} else if (kind === 'websockets') {
|
|
const trigger = await WebsocketTriggerService.getWebsocketTrigger({ workspace, path })
|
|
return trigger.permissioned_as
|
|
} else if (kind === 'routes') {
|
|
const trigger = await HttpTriggerService.getHttpTrigger({ workspace, path })
|
|
return trigger.permissioned_as
|
|
} else if (kind === 'schedules') {
|
|
const trigger = await ScheduleService.getSchedule({ workspace, path })
|
|
return trigger.permissioned_as
|
|
} else if (kind === 'azure') {
|
|
const trigger = await AzureTriggerService.getAzureTrigger({ workspace, path })
|
|
return trigger.permissioned_as
|
|
} else if (kind === 'emails') {
|
|
const trigger = await EmailTriggerService.getEmailTrigger({ workspace, path })
|
|
return trigger.permissioned_as
|
|
}
|
|
} catch {
|
|
// Trigger may not exist in the workspace
|
|
}
|
|
return undefined
|
|
}
|
|
|
|
function retrieveScriptOrFlowKind(path: string, is_flow: boolean): { kind: Kind; path: string } {
|
|
return {
|
|
kind: is_flow ? 'flow' : 'script',
|
|
path
|
|
}
|
|
}
|
|
|
|
function retrieveKindsValues({
|
|
resource_path,
|
|
script_path,
|
|
is_flow
|
|
}: {
|
|
resource_path?: string
|
|
script_path: string
|
|
is_flow: boolean
|
|
}) {
|
|
const result: { kind: Kind; path: string }[] = []
|
|
|
|
if (resource_path) {
|
|
result.push({ kind: 'resource', path: resource_path })
|
|
}
|
|
result.push(retrieveScriptOrFlowKind(script_path, is_flow))
|
|
return result
|
|
}
|
|
|
|
export async function getTriggerDependency(kind: TriggerKind, path: string, workspace: string) {
|
|
let result: { kind: Kind; path: string }[]
|
|
if (kind === 'sqs') {
|
|
const { aws_resource_path, script_path, is_flow } = await SqsTriggerService.getSqsTrigger({
|
|
workspace: workspace!,
|
|
path: path
|
|
})
|
|
|
|
result = retrieveKindsValues({ resource_path: aws_resource_path, script_path, is_flow })
|
|
} else if (kind === 'kafka') {
|
|
const { kafka_resource_path, script_path, is_flow } = await KafkaTriggerService.getKafkaTrigger(
|
|
{
|
|
workspace: workspace!,
|
|
path: path
|
|
}
|
|
)
|
|
|
|
result = retrieveKindsValues({ resource_path: kafka_resource_path, script_path, is_flow })
|
|
} else if (kind === 'mqtt') {
|
|
const { mqtt_resource_path, script_path, is_flow } = await MqttTriggerService.getMqttTrigger({
|
|
workspace: workspace!,
|
|
path: path
|
|
})
|
|
|
|
result = retrieveKindsValues({ resource_path: mqtt_resource_path, script_path, is_flow })
|
|
} else if (kind === 'amqp') {
|
|
const { amqp_resource_path, script_path, is_flow } = await AmqpTriggerService.getAmqpTrigger({
|
|
workspace: workspace!,
|
|
path: path
|
|
})
|
|
|
|
result = retrieveKindsValues({ resource_path: amqp_resource_path, script_path, is_flow })
|
|
} else if (kind === 'nats') {
|
|
const { nats_resource_path, script_path, is_flow } = await NatsTriggerService.getNatsTrigger({
|
|
workspace: workspace!,
|
|
path: path
|
|
})
|
|
|
|
result = retrieveKindsValues({ resource_path: nats_resource_path, script_path, is_flow })
|
|
} else if (kind === 'postgres') {
|
|
const { postgres_resource_path, script_path, is_flow } =
|
|
await PostgresTriggerService.getPostgresTrigger({
|
|
workspace: workspace!,
|
|
path: path
|
|
})
|
|
|
|
result = retrieveKindsValues({ resource_path: postgres_resource_path, script_path, is_flow })
|
|
} else if (kind === 'gcp') {
|
|
const { gcp_resource_path, script_path, is_flow } = await GcpTriggerService.getGcpTrigger({
|
|
workspace: workspace!,
|
|
path: path
|
|
})
|
|
|
|
result = retrieveKindsValues({ resource_path: gcp_resource_path, script_path, is_flow })
|
|
} else if (kind === 'websockets') {
|
|
const { script_path, is_flow, url, initial_messages } =
|
|
await WebsocketTriggerService.getWebsocketTrigger({
|
|
workspace: workspace!,
|
|
path: path
|
|
})
|
|
|
|
result = retrieveKindsValues({ script_path, is_flow })
|
|
|
|
const SCRIPT_PREFIX = '$script:'
|
|
const FLOW_PREFIX = '$flow:'
|
|
|
|
if (url.startsWith(SCRIPT_PREFIX))
|
|
result.push(retrieveScriptOrFlowKind(url.substring(SCRIPT_PREFIX.length), false))
|
|
else if (url.startsWith(FLOW_PREFIX))
|
|
result.push(retrieveScriptOrFlowKind(url.substring(FLOW_PREFIX.length), true))
|
|
|
|
initial_messages?.map((message) => {
|
|
if ('runnable_result' in message) {
|
|
result.push(
|
|
retrieveScriptOrFlowKind(message.runnable_result.path, message.runnable_result.is_flow)
|
|
)
|
|
}
|
|
})
|
|
} else if (kind === 'routes') {
|
|
const { script_path, is_flow, authentication_resource_path } =
|
|
await HttpTriggerService.getHttpTrigger({
|
|
workspace: workspace!,
|
|
path: path
|
|
})
|
|
|
|
result = retrieveKindsValues({
|
|
script_path,
|
|
is_flow,
|
|
resource_path: authentication_resource_path ?? undefined
|
|
})
|
|
} else if (kind === 'schedules') {
|
|
const { script_path, is_flow } = await ScheduleService.getSchedule({
|
|
workspace: workspace!,
|
|
path: path
|
|
})
|
|
|
|
result = retrieveKindsValues({ script_path, is_flow })
|
|
} else {
|
|
throw new Error(`Unexpected trigger kind got: ${kind}`)
|
|
}
|
|
return result
|
|
}
|