backend: v2 schedule_path -> trigger and trigger_kind

backend/migrations/20250117124431_v2_job.down.sql

@@ -17,7 +17,8 @@ ALTER TABLE v2_job
     DROP COLUMN flow_step CASCADE,
     DROP COLUMN flow_step_id CASCADE,
     DROP COLUMN flow_root_job CASCADE,
-    DROP COLUMN schedule_path CASCADE,
+    DROP COLUMN trigger CASCADE,
+    DROP COLUMN trigger_kind CASCADE,
     DROP COLUMN same_worker CASCADE,
     DROP COLUMN visible_to_owner CASCADE,
     DROP COLUMN concurrent_limit CASCADE,
@@ -27,3 +28,4 @@ ALTER TABLE v2_job
     DROP COLUMN priority CASCADE,
     DROP COLUMN args CASCADE,
     DROP COLUMN pre_run_error CASCADE;
+DROP TYPE job_trigger_kind;

backend/migrations/20250117124431_v2_job.up.sql

@@ -1,4 +1,6 @@
 -- Add up migration script here
+CREATE TYPE job_trigger_kind AS ENUM ('webhook', 'http', 'websocket', 'kafka', 'email', 'nats', 'schedule', 'app', 'ui');
+
 ALTER TABLE v2_job
     ADD COLUMN IF NOT EXISTS created_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,
     ADD COLUMN IF NOT EXISTS created_by VARCHAR(255) DEFAULT 'missing' NOT NULL,
@@ -12,7 +14,8 @@ ALTER TABLE v2_job
     ADD COLUMN IF NOT EXISTS flow_step INTEGER,
     ADD COLUMN IF NOT EXISTS flow_step_id VARCHAR(255),
     ADD COLUMN IF NOT EXISTS flow_root_job UUID,
-    ADD COLUMN IF NOT EXISTS schedule_path VARCHAR(255),
+    ADD COLUMN IF NOT EXISTS trigger VARCHAR(255),
+    ADD COLUMN IF NOT EXISTS trigger_kind job_trigger_kind,
     ADD COLUMN IF NOT EXISTS same_worker BOOLEAN DEFAULT FALSE NOT NULL,
     ADD COLUMN IF NOT EXISTS visible_to_owner BOOLEAN DEFAULT TRUE NOT NULL,
     ADD COLUMN IF NOT EXISTS concurrent_limit INTEGER,

backend/migrations/20250117124743_v2_job_queue_sync.up.sql

@@ -18,7 +18,7 @@ BEGIN
     -- __last_ping
     NEW.__job_kind := job.kind;
     NEW.__env_id := 0xcafe; -- Magic used bellow.
-    NEW.__schedule_path := job.schedule_path;
+    NEW.__schedule_path := CASE WHEN job.trigger_kind = 'schedule'::job_trigger_kind THEN job.trigger END;
     NEW.__permissioned_as := job.permissioned_as;
     -- __flow_status
     NEW.__raw_flow := job.raw_flow;
@@ -55,7 +55,7 @@ CREATE OR REPLACE FUNCTION v2_job_queue_after_insert() RETURNS TRIGGER AS $$ BEG
         kind, runnable_id, runnable_path, parent_job,
         script_lang,
         flow_step, flow_step_id, flow_root_job,
-        schedule_path,
+        trigger, trigger_kind,
         tag, same_worker, visible_to_owner, concurrent_limit, concurrency_time_window_s, cache_ttl, timeout, priority,
         args, pre_run_error,
         raw_code, raw_lock, raw_flow
@@ -64,7 +64,7 @@ CREATE OR REPLACE FUNCTION v2_job_queue_after_insert() RETURNS TRIGGER AS $$ BEG
         NEW.__job_kind, NEW.__script_hash, NEW.__script_path, NEW.__parent_job,
         NEW.__language,
         NULL, NEW.__flow_step_id, NEW.__root_job,
-        NEW.__schedule_path,
+        NEW.__schedule_path, CASE WHEN NEW.__schedule_path IS NOT NULL THEN 'schedule'::job_trigger_kind END,
         NEW.tag, NEW.__same_worker, NEW.__visible_to_owner, NEW.__concurrent_limit, NEW.__concurrency_time_window_s,
         NEW.__cache_ttl, NEW.__timeout, NEW.priority,
         NEW.__args, NEW.__pre_run_error,
@@ -83,7 +83,8 @@ CREATE OR REPLACE FUNCTION v2_job_queue_after_insert() RETURNS TRIGGER AS $$ BEG
         flow_step = EXCLUDED.flow_step,
         flow_step_id = EXCLUDED.flow_step_id,
         flow_root_job = EXCLUDED.flow_root_job,
-        schedule_path = EXCLUDED.schedule_path,
+        trigger = EXCLUDED.trigger,
+        trigger_kind = EXCLUDED.trigger_kind,
         tag = EXCLUDED.tag,
         same_worker = EXCLUDED.same_worker,
         visible_to_owner = EXCLUDED.visible_to_owner,

backend/migrations/20250117124744_v2_job_completed_sync.up.sql

@@ -13,21 +13,21 @@ BEGIN
         NEW.__parent_job := job.parent_job;
         NEW.__created_by := job.created_by;
         NEW.__created_at := job.created_at;
-        NEW.__success := NEW.status = 'success';
+        NEW.__success := NEW.status = 'success'::job_status;
         NEW.__script_hash := job.runnable_id;
         NEW.__script_path := job.runnable_path;
         NEW.__args := job.args;
         -- __logs
         NEW.__raw_code := job.raw_code;
-        NEW.__canceled := NEW.status = 'canceled';
+        NEW.__canceled := NEW.status = 'canceled'::job_status;
         NEW.__job_kind := job.kind;
         -- __env_id
-        NEW.__schedule_path := job.schedule_path;
+        NEW.__schedule_path := CASE WHEN job.trigger_kind = 'schedule'::job_trigger_kind THEN job.trigger END;
         NEW.__permissioned_as := job.permissioned_as;
         NEW.__raw_flow := job.raw_flow;
         NEW.__is_flow_step := job.flow_step_id IS NOT NULL;
         NEW.__language := job.script_lang;
-        NEW.__is_skipped := NEW.status = 'skipped';
+        NEW.__is_skipped := NEW.status = 'skipped'::job_status;
         NEW.__raw_lock := job.raw_lock;
         NEW.__email := job.permissioned_as_email;
         NEW.__visible_to_owner := job.visible_to_owner;

backend/migrations/20250117124748_v2_migrate_from_v1.up.sql

@@ -16,7 +16,7 @@ INSERT INTO v2_job (
     kind, runnable_id, runnable_path, parent_job,
     script_lang,
     flow_step_id, flow_root_job,
-    schedule_path,
+    trigger, trigger_kind,
     tag, same_worker, visible_to_owner, concurrent_limit, concurrency_time_window_s, cache_ttl, timeout, priority,
     args, pre_run_error,
     raw_code, raw_lock, raw_flow
@@ -25,7 +25,7 @@ INSERT INTO v2_job (
     __job_kind, __script_hash, __script_path, __parent_job,
     __language,
     __flow_step_id, __root_job,
-    __schedule_path,
+    __schedule_path, CASE WHEN __schedule_path IS NOT NULL THEN 'schedule'::job_trigger_kind END,
     tag, __same_worker, __visible_to_owner, __concurrent_limit, __concurrency_time_window_s,
     __cache_ttl, __timeout, priority,
     __args, __pre_run_error,
@@ -38,15 +38,15 @@ INSERT INTO v2_job (
     id, workspace_id, created_at, created_by, permissioned_as, permissioned_as_email,
     kind, runnable_id, runnable_path, parent_job,
     script_lang,
-    schedule_path,
+    trigger, trigger_kind,
     tag, visible_to_owner, priority,
     args,
     raw_code, raw_lock, raw_flow
 ) SELECT
     id, workspace_id, __created_at, __created_by, __permissioned_as, __email,
     __job_kind, __script_hash, __script_path, __parent_job,
     __language,
-    __schedule_path,
+    __schedule_path, CASE WHEN __schedule_path IS NOT NULL THEN 'schedule'::job_trigger_kind END,
     __tag, __visible_to_owner, __priority,
     __args,
     __raw_code, __raw_lock, __raw_flow
@@ -66,7 +66,8 @@ UPDATE v2_job SET
     script_lang = v2_job_queue.__language,
     flow_step_id = v2_job_queue.__flow_step_id,
     flow_root_job = v2_job_queue.__root_job,
-    schedule_path = v2_job_queue.__schedule_path,
+    trigger = v2_job_queue.__schedule_path,
+    trigger_kind = CASE WHEN v2_job_queue.__schedule_path IS NOT NULL THEN 'schedule'::job_trigger_kind END,
     tag = v2_job_queue.tag,
     same_worker = v2_job_queue.__same_worker,
     visible_to_owner = v2_job_queue.__visible_to_owner,
@@ -94,7 +95,8 @@ UPDATE v2_job SET
     runnable_path = v2_job_completed.__script_path,
     parent_job = v2_job_completed.__parent_job,
     script_lang = v2_job_completed.__language,
-    schedule_path = v2_job_completed.__schedule_path,
+    trigger = v2_job_completed.__schedule_path,
+    trigger_kind = CASE WHEN v2_job_completed.__schedule_path IS NOT NULL THEN 'schedule'::job_trigger_kind END,
     tag = v2_job_completed.__tag,
     visible_to_owner = v2_job_completed.__visible_to_owner,
     priority = v2_job_completed.__priority,