diff --git a/docs/generated/postgres-schema/README.md b/docs/generated/postgres-schema/README.md index d0aacddc1da..85b57a01a79 100644 --- a/docs/generated/postgres-schema/README.md +++ b/docs/generated/postgres-schema/README.md @@ -92,6 +92,8 @@ Auto-generated from the PostgreSQL migrations in @n8n/db. Do not edit by hand. | [public.role_mapping_rule](public.role_mapping_rule.md) | 7 | | BASE TABLE | | [public.role_mapping_rule_project](public.role_mapping_rule_project.md) | 2 | | BASE TABLE | | [public.role_scope](public.role_scope.md) | 2 | | BASE TABLE | +| [public.scheduled_job](public.scheduled_job.md) | 17 | | BASE TABLE | +| [public.scheduled_task](public.scheduled_task.md) | 16 | | BASE TABLE | | [public.scope](public.scope.md) | 3 | | BASE TABLE | | [public.secrets_provider_connection](public.secrets_provider_connection.md) | 7 | | BASE TABLE | | [public.settings](public.settings.md) | 3 | | BASE TABLE | @@ -262,6 +264,8 @@ erDiagram "public.role_mapping_rule_project" }o--|| "public.role_mapping_rule" : "FOREIGN KEY (#quot;roleMappingRuleId#quot;) REFERENCES role_mapping_rule(id) ON DELETE CASCADE" "public.role_scope" }o--|| "public.scope" : "FOREIGN KEY (#quot;scopeSlug#quot;) REFERENCES scope(slug) ON UPDATE CASCADE ON DELETE CASCADE" "public.role_scope" }o--|| "public.role" : "FOREIGN KEY (#quot;roleSlug#quot;) REFERENCES role(slug) ON UPDATE CASCADE ON DELETE CASCADE" +"public.scheduled_job" }o--o| "public.workflow_published_version" : "FOREIGN KEY (#quot;workflowId#quot;) REFERENCES workflow_published_version(#quot;workflowId#quot;) ON DELETE CASCADE" +"public.scheduled_task" }o--|| "public.scheduled_job" : "FOREIGN KEY (#quot;jobId#quot;) REFERENCES scheduled_job(id) ON DELETE CASCADE" "public.shared_credentials" }o--|| "public.credentials_entity" : "FOREIGN KEY (#quot;credentialsId#quot;) REFERENCES credentials_entity(id) ON DELETE CASCADE" "public.shared_credentials" }o--|| "public.project" : "FOREIGN KEY (#quot;projectId#quot;) REFERENCES project(id) ON DELETE CASCADE" "public.shared_workflow" }o--|| "public.workflow_entity" : "FOREIGN KEY (#quot;workflowId#quot;) REFERENCES workflow_entity(id) ON DELETE CASCADE" @@ -1090,6 +1094,43 @@ erDiagram varchar_128_ roleSlug FK varchar_128_ scopeSlug FK } +"public.scheduled_job" { + timestamp_3__with_time_zone createdAt + varchar_255_ cronExpression + boolean enabled + timestamp_3__with_time_zone fireAt + integer id + integer intervalSeconds + varchar_16_ kind + timestamp_3__with_time_zone lastFiredAt + integer maxAttempts + varchar_255_ name + timestamp_3__with_time_zone nextRunAt + varchar_36_ nodeId + json payload + varchar_128_ taskType + varchar_64_ timezone + timestamp_3__with_time_zone updatedAt + varchar_36_ workflowId FK +} +"public.scheduled_task" { + integer attempts + varchar_255_ claimedBy + timestamp_3__with_time_zone createdAt + text errorMessage + timestamp_3__with_time_zone finishedAt + bigint id + integer jobId FK + integer leaseEpoch + timestamp_3__with_time_zone leaseExpiresAt + integer maxAttempts + json payload + timestamp_3__with_time_zone runAt + timestamp_3__with_time_zone scheduledFor + timestamp_3__with_time_zone startedAt + varchar_16_ status + varchar_128_ taskType +} "public.scope" { text description text displayName diff --git a/docs/generated/postgres-schema/public.scheduled_job.md b/docs/generated/postgres-schema/public.scheduled_job.md new file mode 100644 index 00000000000..1679487f913 --- /dev/null +++ b/docs/generated/postgres-schema/public.scheduled_job.md @@ -0,0 +1,109 @@ +# public.scheduled_job + +## Columns + +| Name | Type | Default | Nullable | Children | Parents | Comment | +| ---- | ---- | ------- | -------- | -------- | ------- | ------- | +| createdAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | | +| cronExpression | varchar(255) | | true | | | Cron expression driving recurrence; set only when kind is 'cron'. | +| enabled | boolean | true | false | | | Whether the scheduler considers this job for firing. | +| fireAt | timestamp(3) with time zone | | true | | | Absolute time the job fires once; set only when kind is 'one_off'. | +| id | integer | | false | [public.scheduled_task](public.scheduled_task.md) | | | +| intervalSeconds | integer | | true | | | Gap between fires in seconds; set only when kind is 'interval'. | +| kind | varchar(16) | | false | | | Recurrence kind; selects which of the schedule columns below apply. | +| lastFiredAt | timestamp(3) with time zone | | true | | | Last time an occurrence was materialized; used to recompute nextRunAt. | +| maxAttempts | integer | 1 | false | | | Retry ceiling copied onto each occurrence this job materializes. | +| name | varchar(255) | | false | | | Human-readable job name. A well-known scheduler key for system jobs; generated for workflow trigger jobs. | +| nextRunAt | timestamp(3) with time zone | | true | | | Next time an occurrence is due; the scheduler sweep reads this to find work. NULL once disabled or a one-off has fired. | +| nodeId | varchar(36) | | true | | | Trigger node within the workflow that owns this job; NULL for non-trigger jobs. | +| payload | json | '{}'::json | false | | | Input passed to the task handler when an occurrence runs. | +| taskType | varchar(128) | | false | | | Selects which registered handler runs the task. | +| timezone | varchar(64) | | true | | | IANA timezone the cron expression is evaluated in; NULL uses the instance default. | +| updatedAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | | +| workflowId | varchar(36) | | true | | [public.workflow_published_version](public.workflow_published_version.md) | References the workflow's published version, since only published trigger nodes get scheduled; NULL for system jobs not tied to a workflow. Unpublishing the workflow deletes its jobs. | + +## Constraints + +| Name | Type | Definition | +| ---- | ---- | ---------- | +| CHK_scheduled_job_cron_expression | CHECK | CHECK ((((kind)::text <> 'cron'::text) OR ("cronExpression" IS NOT NULL))) | +| CHK_scheduled_job_fire_at | CHECK | CHECK ((((kind)::text <> 'one_off'::text) OR ("fireAt" IS NOT NULL))) | +| CHK_scheduled_job_interval_seconds | CHECK | CHECK ((((kind)::text <> 'interval'::text) OR ("intervalSeconds" IS NOT NULL))) | +| CHK_scheduled_job_kind | CHECK | CHECK (((kind)::text = ANY ((ARRAY['cron'::character varying, 'interval'::character varying, 'one_off'::character varying])::text[]))) | +| FK_scheduled_job_workflowId | FOREIGN KEY | FOREIGN KEY ("workflowId") REFERENCES workflow_published_version("workflowId") ON DELETE CASCADE | +| PK_893185383f029ca8d57bb781fa8 | PRIMARY KEY | PRIMARY KEY (id) | +| scheduled_job_createdAt_not_null | n | NOT NULL "createdAt" | +| scheduled_job_enabled_not_null | n | NOT NULL enabled | +| scheduled_job_id_not_null | n | NOT NULL id | +| scheduled_job_kind_not_null | n | NOT NULL kind | +| scheduled_job_maxAttempts_not_null | n | NOT NULL "maxAttempts" | +| scheduled_job_name_not_null | n | NOT NULL name | +| scheduled_job_payload_not_null | n | NOT NULL payload | +| scheduled_job_taskType_not_null | n | NOT NULL "taskType" | +| scheduled_job_updatedAt_not_null | n | NOT NULL "updatedAt" | + +## Indexes + +| Name | Definition | +| ---- | ---------- | +| IDX_scheduled_job_name | CREATE UNIQUE INDEX "IDX_scheduled_job_name" ON public.scheduled_job USING btree (name) | +| IDX_scheduled_job_nextRunAt | CREATE INDEX "IDX_scheduled_job_nextRunAt" ON public.scheduled_job USING btree ("nextRunAt") WHERE ((enabled = true) AND ("nextRunAt" IS NOT NULL)) | +| IDX_scheduled_job_workflowId | CREATE INDEX "IDX_scheduled_job_workflowId" ON public.scheduled_job USING btree ("workflowId") WHERE ("workflowId" IS NOT NULL) | +| PK_893185383f029ca8d57bb781fa8 | CREATE UNIQUE INDEX "PK_893185383f029ca8d57bb781fa8" ON public.scheduled_job USING btree (id) | + +## Relations + +```mermaid +erDiagram + +"public.scheduled_task" }o--|| "public.scheduled_job" : "FOREIGN KEY (#quot;jobId#quot;) REFERENCES scheduled_job(id) ON DELETE CASCADE" +"public.scheduled_job" }o--o| "public.workflow_published_version" : "FOREIGN KEY (#quot;workflowId#quot;) REFERENCES workflow_published_version(#quot;workflowId#quot;) ON DELETE CASCADE" + +"public.scheduled_job" { + timestamp_3__with_time_zone createdAt + varchar_255_ cronExpression + boolean enabled + timestamp_3__with_time_zone fireAt + integer id + integer intervalSeconds + varchar_16_ kind + timestamp_3__with_time_zone lastFiredAt + integer maxAttempts + varchar_255_ name + timestamp_3__with_time_zone nextRunAt + varchar_36_ nodeId + json payload + varchar_128_ taskType + varchar_64_ timezone + timestamp_3__with_time_zone updatedAt + varchar_36_ workflowId FK +} +"public.scheduled_task" { + integer attempts + varchar_255_ claimedBy + timestamp_3__with_time_zone createdAt + text errorMessage + timestamp_3__with_time_zone finishedAt + bigint id + integer jobId FK + integer leaseEpoch + timestamp_3__with_time_zone leaseExpiresAt + integer maxAttempts + json payload + timestamp_3__with_time_zone runAt + timestamp_3__with_time_zone scheduledFor + timestamp_3__with_time_zone startedAt + varchar_16_ status + varchar_128_ taskType +} +"public.workflow_published_version" { + timestamp_3__with_time_zone createdAt + varchar_36_ publishedVersionId FK + timestamp_3__with_time_zone updatedAt + varchar_36_ workflowId FK +} +``` + +--- + +> Generated by [tbls](https://github.com/k1LoW/tbls) diff --git a/docs/generated/postgres-schema/public.scheduled_task.md b/docs/generated/postgres-schema/public.scheduled_task.md new file mode 100644 index 00000000000..5bec8bf9682 --- /dev/null +++ b/docs/generated/postgres-schema/public.scheduled_task.md @@ -0,0 +1,102 @@ +# public.scheduled_task + +## Columns + +| Name | Type | Default | Nullable | Children | Parents | Comment | +| ---- | ---- | ------- | -------- | -------- | ------- | ------- | +| attempts | integer | 0 | false | | | Execution attempts started so far; compared against maxAttempts. | +| claimedBy | varchar(255) | | true | | | Id of the instance currently holding the lease; NULL when unclaimed. | +| createdAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | | +| errorMessage | text | | true | | | Failure detail from the last attempt. | +| finishedAt | timestamp(3) with time zone | | true | | | When the occurrence reached a terminal state; drives retention pruning. | +| id | bigint | | false | | | | +| jobId | integer | | false | | [public.scheduled_job](public.scheduled_job.md) | The scheduled_job this occurrence belongs to. | +| leaseEpoch | integer | 0 | false | | | Fencing token bumped on each claim; lets a reaped worker detect it lost ownership and not overwrite the new owner's results. | +| leaseExpiresAt | timestamp(3) with time zone | | true | | | When the current lease expires; the reaper reclaims running occurrences past this. | +| maxAttempts | integer | 1 | false | | | Attempt ceiling; once attempts reaches it, a failure is final rather than retried. | +| payload | json | '{}'::json | false | | | Handler input copied from the job. A snapshot, so editing the job later doesn't change runs already queued. | +| runAt | timestamp(3) with time zone | | false | | | Earliest time the executor may pick this up; starts at scheduledFor and is pushed out by retry backoff. | +| scheduledFor | timestamp(3) with time zone | | false | | | The logical fire time this occurrence represents; unique per job, so the same instant cannot be queued twice. | +| startedAt | timestamp(3) with time zone | | true | | | When the current attempt started running. | +| status | varchar(16) | 'pending'::character varying | false | | | Lifecycle state; drives which occurrences the claim and reaper scans consider. | +| taskType | varchar(128) | | false | | | What kind of work to run, copied from the job so a run is self-contained (no join to execute it). Also lets a run exist without a parent job in future. | + +## Constraints + +| Name | Type | Definition | +| ---- | ---- | ---------- | +| CHK_scheduled_task_running_lease | CHECK | CHECK ((((status)::text <> 'running'::text) OR ("leaseExpiresAt" IS NOT NULL))) | +| CHK_scheduled_task_status | CHECK | CHECK (((status)::text = ANY ((ARRAY['pending'::character varying, 'running'::character varying, 'succeeded'::character varying, 'failed'::character varying, 'missed'::character varying, 'cancelled'::character varying])::text[]))) | +| FK_scheduled_task_jobId | FOREIGN KEY | FOREIGN KEY ("jobId") REFERENCES scheduled_job(id) ON DELETE CASCADE | +| PK_d690af24e57e30594c1948af1e6 | PRIMARY KEY | PRIMARY KEY (id) | +| scheduled_task_attempts_not_null | n | NOT NULL attempts | +| scheduled_task_createdAt_not_null | n | NOT NULL "createdAt" | +| scheduled_task_id_not_null | n | NOT NULL id | +| scheduled_task_jobId_not_null | n | NOT NULL "jobId" | +| scheduled_task_leaseEpoch_not_null | n | NOT NULL "leaseEpoch" | +| scheduled_task_maxAttempts_not_null | n | NOT NULL "maxAttempts" | +| scheduled_task_payload_not_null | n | NOT NULL payload | +| scheduled_task_runAt_not_null | n | NOT NULL "runAt" | +| scheduled_task_scheduledFor_not_null | n | NOT NULL "scheduledFor" | +| scheduled_task_status_not_null | n | NOT NULL status | +| scheduled_task_taskType_not_null | n | NOT NULL "taskType" | + +## Indexes + +| Name | Definition | +| ---- | ---------- | +| IDX_scheduled_task_finishedAt | CREATE INDEX "IDX_scheduled_task_finishedAt" ON public.scheduled_task USING btree ("finishedAt") WHERE ("finishedAt" IS NOT NULL) | +| IDX_scheduled_task_jobId_scheduledFor | CREATE UNIQUE INDEX "IDX_scheduled_task_jobId_scheduledFor" ON public.scheduled_task USING btree ("jobId", "scheduledFor") | +| IDX_scheduled_task_leaseExpiresAt | CREATE INDEX "IDX_scheduled_task_leaseExpiresAt" ON public.scheduled_task USING btree ("leaseExpiresAt") WHERE ((status)::text = 'running'::text) | +| IDX_scheduled_task_runAt | CREATE INDEX "IDX_scheduled_task_runAt" ON public.scheduled_task USING btree ("runAt") WHERE ((status)::text = 'pending'::text) | +| PK_d690af24e57e30594c1948af1e6 | CREATE UNIQUE INDEX "PK_d690af24e57e30594c1948af1e6" ON public.scheduled_task USING btree (id) | + +## Relations + +```mermaid +erDiagram + +"public.scheduled_task" }o--|| "public.scheduled_job" : "FOREIGN KEY (#quot;jobId#quot;) REFERENCES scheduled_job(id) ON DELETE CASCADE" + +"public.scheduled_task" { + integer attempts + varchar_255_ claimedBy + timestamp_3__with_time_zone createdAt + text errorMessage + timestamp_3__with_time_zone finishedAt + bigint id + integer jobId FK + integer leaseEpoch + timestamp_3__with_time_zone leaseExpiresAt + integer maxAttempts + json payload + timestamp_3__with_time_zone runAt + timestamp_3__with_time_zone scheduledFor + timestamp_3__with_time_zone startedAt + varchar_16_ status + varchar_128_ taskType +} +"public.scheduled_job" { + timestamp_3__with_time_zone createdAt + varchar_255_ cronExpression + boolean enabled + timestamp_3__with_time_zone fireAt + integer id + integer intervalSeconds + varchar_16_ kind + timestamp_3__with_time_zone lastFiredAt + integer maxAttempts + varchar_255_ name + timestamp_3__with_time_zone nextRunAt + varchar_36_ nodeId + json payload + varchar_128_ taskType + varchar_64_ timezone + timestamp_3__with_time_zone updatedAt + varchar_36_ workflowId FK +} +``` + +--- + +> Generated by [tbls](https://github.com/k1LoW/tbls) diff --git a/docs/generated/postgres-schema/public.workflow_published_version.md b/docs/generated/postgres-schema/public.workflow_published_version.md index 63cd6bad5d6..2c110c801cc 100644 --- a/docs/generated/postgres-schema/public.workflow_published_version.md +++ b/docs/generated/postgres-schema/public.workflow_published_version.md @@ -7,7 +7,7 @@ | createdAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | | | publishedVersionId | varchar(36) | | false | | [public.workflow_history](public.workflow_history.md) | | | updatedAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | | -| workflowId | varchar(36) | | false | | [public.workflow_entity](public.workflow_entity.md) | | +| workflowId | varchar(36) | | false | [public.scheduled_job](public.scheduled_job.md) | [public.workflow_entity](public.workflow_entity.md) | | ## Constraints @@ -34,6 +34,7 @@ erDiagram "public.workflow_published_version" }o--|| "public.workflow_history" : "FOREIGN KEY (#quot;publishedVersionId#quot;) REFERENCES workflow_history(#quot;versionId#quot;) ON DELETE RESTRICT" "public.workflow_published_version" |o--|| "public.workflow_entity" : "FOREIGN KEY (#quot;workflowId#quot;) REFERENCES workflow_entity(id) ON DELETE RESTRICT" +"public.scheduled_job" }o--o| "public.workflow_published_version" : "FOREIGN KEY (#quot;workflowId#quot;) REFERENCES workflow_published_version(#quot;workflowId#quot;) ON DELETE CASCADE" "public.workflow_published_version" { timestamp_3__with_time_zone createdAt @@ -76,6 +77,25 @@ erDiagram integer versionCounter character_36_ versionId } +"public.scheduled_job" { + timestamp_3__with_time_zone createdAt + varchar_255_ cronExpression + boolean enabled + timestamp_3__with_time_zone fireAt + integer id + integer intervalSeconds + varchar_16_ kind + timestamp_3__with_time_zone lastFiredAt + integer maxAttempts + varchar_255_ name + timestamp_3__with_time_zone nextRunAt + varchar_36_ nodeId + json payload + varchar_128_ taskType + varchar_64_ timezone + timestamp_3__with_time_zone updatedAt + varchar_36_ workflowId FK +} ``` --- diff --git a/docs/generated/sqlite-schema/README.md b/docs/generated/sqlite-schema/README.md index 5606733985a..6f0e71b8a8e 100644 --- a/docs/generated/sqlite-schema/README.md +++ b/docs/generated/sqlite-schema/README.md @@ -92,6 +92,8 @@ Auto-generated from the SQLite migrations in @n8n/db. Do not edit by hand. | [role_mapping_rule](role_mapping_rule.md) | 7 | | table | | [role_mapping_rule_project](role_mapping_rule_project.md) | 2 | | table | | [role_scope](role_scope.md) | 2 | | table | +| [scheduled_job](scheduled_job.md) | 17 | | table | +| [scheduled_task](scheduled_task.md) | 16 | | table | | [scope](scope.md) | 3 | | table | | [secrets_provider_connection](secrets_provider_connection.md) | 7 | | table | | [settings](settings.md) | 3 | | table | @@ -246,6 +248,8 @@ erDiagram "role_mapping_rule_project" |o--|| "role_mapping_rule" : "FOREIGN KEY (roleMappingRuleId) REFERENCES role_mapping_rule (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" "role_scope" |o--|| "scope" : "FOREIGN KEY (scopeSlug) REFERENCES scope (slug) ON UPDATE CASCADE ON DELETE CASCADE MATCH NONE" "role_scope" |o--|| "role" : "FOREIGN KEY (roleSlug) REFERENCES role (slug) ON UPDATE CASCADE ON DELETE CASCADE MATCH NONE" +"scheduled_job" }o--o| "workflow_published_version" : "FOREIGN KEY (workflowId) REFERENCES workflow_published_version (workflowId) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" +"scheduled_task" }o--|| "scheduled_job" : "FOREIGN KEY (jobId) REFERENCES scheduled_job (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" "shared_credentials" |o--|| "project" : "FOREIGN KEY (projectId) REFERENCES project (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" "shared_credentials" |o--|| "credentials_entity" : "FOREIGN KEY (credentialsId) REFERENCES credentials_entity (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" "shared_workflow" |o--|| "project" : "FOREIGN KEY (projectId) REFERENCES project (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" @@ -1078,6 +1082,43 @@ erDiagram VARCHAR_128_ roleSlug PK VARCHAR_128_ scopeSlug PK } +"scheduled_job" { + datetime_3_ createdAt + varchar_255_ cronExpression + boolean enabled + datetime_3_ fireAt + INTEGER id + INTEGER intervalSeconds + varchar_16_ kind + datetime_3_ lastFiredAt + INTEGER maxAttempts + varchar_255_ name + datetime_3_ nextRunAt + varchar_36_ nodeId + TEXT payload + varchar_128_ taskType + varchar_64_ timezone + datetime_3_ updatedAt + varchar_36_ workflowId FK +} +"scheduled_task" { + INTEGER attempts + varchar_255_ claimedBy + datetime_3_ createdAt + TEXT errorMessage + datetime_3_ finishedAt + INTEGER id + INTEGER jobId FK + INTEGER leaseEpoch + datetime_3_ leaseExpiresAt + INTEGER maxAttempts + TEXT payload + datetime_3_ runAt + datetime_3_ scheduledFor + datetime_3_ startedAt + varchar_16_ status + varchar_128_ taskType +} "scope" { TEXT description TEXT displayName diff --git a/docs/generated/sqlite-schema/scheduled_job.md b/docs/generated/sqlite-schema/scheduled_job.md new file mode 100644 index 00000000000..f2d8d275505 --- /dev/null +++ b/docs/generated/sqlite-schema/scheduled_job.md @@ -0,0 +1,110 @@ +# scheduled_job + +## Description + +
+Table Definition + +```sql +CREATE TABLE "scheduled_job" ("id" integer PRIMARY KEY NOT NULL, "name" varchar(255) NOT NULL, "workflowId" varchar(36), "nodeId" varchar(36), "taskType" varchar(128) NOT NULL, "payload" text NOT NULL DEFAULT ('{}'), "kind" varchar(16) NOT NULL, "cronExpression" varchar(255), "timezone" varchar(64), "intervalSeconds" integer, "fireAt" datetime(3), "enabled" boolean NOT NULL DEFAULT (true), "nextRunAt" datetime(3), "lastFiredAt" datetime(3), "maxAttempts" integer NOT NULL DEFAULT (1), "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), CONSTRAINT "CHK_scheduled_job_cron_expression" CHECK ("kind" <> 'cron' OR "cronExpression" IS NOT NULL), CONSTRAINT "CHK_scheduled_job_interval_seconds" CHECK ("kind" <> 'interval' OR "intervalSeconds" IS NOT NULL), CONSTRAINT "CHK_scheduled_job_fire_at" CHECK ("kind" <> 'one_off' OR "fireAt" IS NOT NULL), CONSTRAINT "CHK_scheduled_job_kind" CHECK ("kind" IN ('cron', 'interval', 'one_off')), CONSTRAINT "FK_scheduled_job_workflowId" FOREIGN KEY ("workflowId") REFERENCES "workflow_published_version" ("workflowId") ON DELETE CASCADE) +``` + +
+ +## Columns + +| Name | Type | Default | Nullable | Children | Parents | Comment | +| ---- | ---- | ------- | -------- | -------- | ------- | ------- | +| createdAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | | +| cronExpression | varchar(255) | | true | | | | +| enabled | boolean | true | false | | | | +| fireAt | datetime(3) | | true | | | | +| id | INTEGER | | false | [scheduled_task](scheduled_task.md) | | | +| intervalSeconds | INTEGER | | true | | | | +| kind | varchar(16) | | false | | | | +| lastFiredAt | datetime(3) | | true | | | | +| maxAttempts | INTEGER | 1 | false | | | | +| name | varchar(255) | | false | | | | +| nextRunAt | datetime(3) | | true | | | | +| nodeId | varchar(36) | | true | | | | +| payload | TEXT | '{}' | false | | | | +| taskType | varchar(128) | | false | | | | +| timezone | varchar(64) | | true | | | | +| updatedAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | | +| workflowId | varchar(36) | | true | | [workflow_published_version](workflow_published_version.md) | | + +## Constraints + +| Name | Type | Definition | +| ---- | ---- | ---------- | +| - | CHECK | CHECK ("kind" <> 'cron' OR "cronExpression" IS NOT NULL) | +| - | CHECK | CHECK ("kind" <> 'interval' OR "intervalSeconds" IS NOT NULL) | +| - | CHECK | CHECK ("kind" <> 'one_off' OR "fireAt" IS NOT NULL) | +| - | CHECK | CHECK ("kind" IN ('cron', 'interval', 'one_off')) | +| - (Foreign key ID: 0) | FOREIGN KEY | FOREIGN KEY (workflowId) REFERENCES workflow_published_version (workflowId) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE | +| id | PRIMARY KEY | PRIMARY KEY (id) | + +## Indexes + +| Name | Definition | +| ---- | ---------- | +| IDX_scheduled_job_name | CREATE UNIQUE INDEX "IDX_scheduled_job_name" ON "scheduled_job" ("name") | +| IDX_scheduled_job_nextRunAt | CREATE INDEX "IDX_scheduled_job_nextRunAt" ON "scheduled_job" ("nextRunAt") WHERE "enabled" = true AND "nextRunAt" IS NOT NULL | +| IDX_scheduled_job_workflowId | CREATE INDEX "IDX_scheduled_job_workflowId" ON "scheduled_job" ("workflowId") WHERE "workflowId" IS NOT NULL | + +## Relations + +```mermaid +erDiagram + +"scheduled_task" }o--|| "scheduled_job" : "FOREIGN KEY (jobId) REFERENCES scheduled_job (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" +"scheduled_job" }o--o| "workflow_published_version" : "FOREIGN KEY (workflowId) REFERENCES workflow_published_version (workflowId) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" + +"scheduled_job" { + datetime_3_ createdAt + varchar_255_ cronExpression + boolean enabled + datetime_3_ fireAt + INTEGER id + INTEGER intervalSeconds + varchar_16_ kind + datetime_3_ lastFiredAt + INTEGER maxAttempts + varchar_255_ name + datetime_3_ nextRunAt + varchar_36_ nodeId + TEXT payload + varchar_128_ taskType + varchar_64_ timezone + datetime_3_ updatedAt + varchar_36_ workflowId FK +} +"scheduled_task" { + INTEGER attempts + varchar_255_ claimedBy + datetime_3_ createdAt + TEXT errorMessage + datetime_3_ finishedAt + INTEGER id + INTEGER jobId FK + INTEGER leaseEpoch + datetime_3_ leaseExpiresAt + INTEGER maxAttempts + TEXT payload + datetime_3_ runAt + datetime_3_ scheduledFor + datetime_3_ startedAt + varchar_16_ status + varchar_128_ taskType +} +"workflow_published_version" { + datetime_3_ createdAt + varchar_36_ publishedVersionId FK + datetime_3_ updatedAt + varchar_36_ workflowId PK +} +``` + +--- + +> Generated by [tbls](https://github.com/k1LoW/tbls) diff --git a/docs/generated/sqlite-schema/scheduled_task.md b/docs/generated/sqlite-schema/scheduled_task.md new file mode 100644 index 00000000000..2404b24028a --- /dev/null +++ b/docs/generated/sqlite-schema/scheduled_task.md @@ -0,0 +1,101 @@ +# scheduled_task + +## Description + +
+Table Definition + +```sql +CREATE TABLE "scheduled_task" ("id" integer PRIMARY KEY NOT NULL, "jobId" integer NOT NULL, "taskType" varchar(128) NOT NULL, "payload" text NOT NULL DEFAULT ('{}'), "scheduledFor" datetime(3) NOT NULL, "runAt" datetime(3) NOT NULL, "status" varchar(16) NOT NULL DEFAULT ('pending'), "attempts" integer NOT NULL DEFAULT (0), "maxAttempts" integer NOT NULL DEFAULT (1), "claimedBy" varchar(255), "leaseExpiresAt" datetime(3), "leaseEpoch" integer NOT NULL DEFAULT (0), "startedAt" datetime(3), "finishedAt" datetime(3), "errorMessage" text, "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), CONSTRAINT "CHK_scheduled_task_running_lease" CHECK ("status" <> 'running' OR "leaseExpiresAt" IS NOT NULL), CONSTRAINT "CHK_scheduled_task_status" CHECK ("status" IN ('pending', 'running', 'succeeded', 'failed', 'missed', 'cancelled')), CONSTRAINT "FK_scheduled_task_jobId" FOREIGN KEY ("jobId") REFERENCES "scheduled_job" ("id") ON DELETE CASCADE) +``` + +
+ +## Columns + +| Name | Type | Default | Nullable | Children | Parents | Comment | +| ---- | ---- | ------- | -------- | -------- | ------- | ------- | +| attempts | INTEGER | 0 | false | | | | +| claimedBy | varchar(255) | | true | | | | +| createdAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | | +| errorMessage | TEXT | | true | | | | +| finishedAt | datetime(3) | | true | | | | +| id | INTEGER | | false | | | | +| jobId | INTEGER | | false | | [scheduled_job](scheduled_job.md) | | +| leaseEpoch | INTEGER | 0 | false | | | | +| leaseExpiresAt | datetime(3) | | true | | | | +| maxAttempts | INTEGER | 1 | false | | | | +| payload | TEXT | '{}' | false | | | | +| runAt | datetime(3) | | false | | | | +| scheduledFor | datetime(3) | | false | | | | +| startedAt | datetime(3) | | true | | | | +| status | varchar(16) | 'pending' | false | | | | +| taskType | varchar(128) | | false | | | | + +## Constraints + +| Name | Type | Definition | +| ---- | ---- | ---------- | +| - | CHECK | CHECK ("status" <> 'running' OR "leaseExpiresAt" IS NOT NULL) | +| - | CHECK | CHECK ("status" IN ('pending', 'running', 'succeeded', 'failed', 'missed', 'cancelled')) | +| - (Foreign key ID: 0) | FOREIGN KEY | FOREIGN KEY (jobId) REFERENCES scheduled_job (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE | +| id | PRIMARY KEY | PRIMARY KEY (id) | + +## Indexes + +| Name | Definition | +| ---- | ---------- | +| IDX_scheduled_task_finishedAt | CREATE INDEX "IDX_scheduled_task_finishedAt" ON "scheduled_task" ("finishedAt") WHERE "finishedAt" IS NOT NULL | +| IDX_scheduled_task_jobId_scheduledFor | CREATE UNIQUE INDEX "IDX_scheduled_task_jobId_scheduledFor" ON "scheduled_task" ("jobId", "scheduledFor") | +| IDX_scheduled_task_leaseExpiresAt | CREATE INDEX "IDX_scheduled_task_leaseExpiresAt" ON "scheduled_task" ("leaseExpiresAt") WHERE "status" = 'running' | +| IDX_scheduled_task_runAt | CREATE INDEX "IDX_scheduled_task_runAt" ON "scheduled_task" ("runAt") WHERE "status" = 'pending' | + +## Relations + +```mermaid +erDiagram + +"scheduled_task" }o--|| "scheduled_job" : "FOREIGN KEY (jobId) REFERENCES scheduled_job (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" + +"scheduled_task" { + INTEGER attempts + varchar_255_ claimedBy + datetime_3_ createdAt + TEXT errorMessage + datetime_3_ finishedAt + INTEGER id + INTEGER jobId FK + INTEGER leaseEpoch + datetime_3_ leaseExpiresAt + INTEGER maxAttempts + TEXT payload + datetime_3_ runAt + datetime_3_ scheduledFor + datetime_3_ startedAt + varchar_16_ status + varchar_128_ taskType +} +"scheduled_job" { + datetime_3_ createdAt + varchar_255_ cronExpression + boolean enabled + datetime_3_ fireAt + INTEGER id + INTEGER intervalSeconds + varchar_16_ kind + datetime_3_ lastFiredAt + INTEGER maxAttempts + varchar_255_ name + datetime_3_ nextRunAt + varchar_36_ nodeId + TEXT payload + varchar_128_ taskType + varchar_64_ timezone + datetime_3_ updatedAt + varchar_36_ workflowId FK +} +``` + +--- + +> Generated by [tbls](https://github.com/k1LoW/tbls) diff --git a/docs/generated/sqlite-schema/workflow_published_version.md b/docs/generated/sqlite-schema/workflow_published_version.md index d8cf659448c..28f81bf20e1 100644 --- a/docs/generated/sqlite-schema/workflow_published_version.md +++ b/docs/generated/sqlite-schema/workflow_published_version.md @@ -18,7 +18,7 @@ CREATE TABLE "workflow_published_version" ("workflowId" varchar(36) PRIMARY KEY | createdAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | | | publishedVersionId | varchar(36) | | false | | [workflow_history](workflow_history.md) | | | updatedAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | | -| workflowId | varchar(36) | | false | | [workflow_entity](workflow_entity.md) | | +| workflowId | varchar(36) | | false | [scheduled_job](scheduled_job.md) | [workflow_entity](workflow_entity.md) | | ## Constraints @@ -46,6 +46,7 @@ erDiagram "workflow_published_version" }o--|| "workflow_history" : "FOREIGN KEY (publishedVersionId) REFERENCES workflow_history (versionId) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" "workflow_published_version" |o--|| "workflow_entity" : "FOREIGN KEY (workflowId) REFERENCES workflow_entity (id) ON UPDATE NO ACTION ON DELETE RESTRICT MATCH NONE" "workflow_published_version" |o--|| "workflow_entity" : "FOREIGN KEY (workflowId) REFERENCES workflow_entity (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" +"scheduled_job" }o--o| "workflow_published_version" : "FOREIGN KEY (workflowId) REFERENCES workflow_published_version (workflowId) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE" "workflow_published_version" { datetime_3_ createdAt @@ -88,6 +89,25 @@ erDiagram INTEGER versionCounter varchar_36_ versionId } +"scheduled_job" { + datetime_3_ createdAt + varchar_255_ cronExpression + boolean enabled + datetime_3_ fireAt + INTEGER id + INTEGER intervalSeconds + varchar_16_ kind + datetime_3_ lastFiredAt + INTEGER maxAttempts + varchar_255_ name + datetime_3_ nextRunAt + varchar_36_ nodeId + TEXT payload + varchar_128_ taskType + varchar_64_ timezone + datetime_3_ updatedAt + varchar_36_ workflowId FK +} ``` --- diff --git a/packages/@n8n/db/src/entities/index.ts b/packages/@n8n/db/src/entities/index.ts index 9a27f47f233..9f59838df37 100644 --- a/packages/@n8n/db/src/entities/index.ts +++ b/packages/@n8n/db/src/entities/index.ts @@ -28,6 +28,8 @@ import { ProjectSecretsProviderAccess } from './project-secrets-provider-access' import type { SecretsProviderAccessRole } from './project-secrets-provider-access'; import { Role } from './role'; import { RoleMappingRule } from './role-mapping-rule'; +import { ScheduledJob, ScheduledJobKind } from './scheduled-job'; +import { ScheduledTask, ScheduledTaskStatus } from './scheduled-task'; import { Scope } from './scope'; import { SecretsProviderConnection } from './secrets-provider-connection'; import { Settings } from './settings'; @@ -79,6 +81,10 @@ export { ProjectRelation, RoleMappingRule, Role, + ScheduledJob, + ScheduledJobKind, + ScheduledTask, + ScheduledTaskStatus, Scope, SharedCredentials, SharedWorkflow, @@ -154,6 +160,8 @@ export const entities = { TestCaseExecution, ExecutionEntity, Role, + ScheduledJob, + ScheduledTask, ProjectSecretsProviderAccess, SecretsProviderConnection, }; diff --git a/packages/@n8n/db/src/entities/scheduled-job.ts b/packages/@n8n/db/src/entities/scheduled-job.ts new file mode 100644 index 00000000000..32c5d7a5845 --- /dev/null +++ b/packages/@n8n/db/src/entities/scheduled-job.ts @@ -0,0 +1,139 @@ +import { Column, Entity, Index, PrimaryGeneratedColumn } from '@n8n/typeorm'; + +import { DateTimeColumn, JsonColumn, WithTimestamps } from './abstract-entity'; + +/** + * Recurrence kind. + * It selects which schedule columns apply. + */ +export const ScheduledJobKind = { + Cron: 'cron', + Interval: 'interval', + OneOff: 'one_off', +} as const; + +export type ScheduledJobKind = (typeof ScheduledJobKind)[keyof typeof ScheduledJobKind]; + +/** + * A scheduled job: the rule for when something should run, + * plus the bookkeeping the scheduler needs to act on it. + * + * A job answers "what runs, and when". + * + * For example "run workflow X every day at 9am", or "run this once at midnight". + * The row holds the timing rule (see {@link kind} and the schedule columns), + * whether it is currently active ({@link enabled}), + * and when it last fired and is next due. + * + * The job itself never executes anything. + * + * A background scheduler periodically scans the active jobs that are due and, + * for each one, inserts a concrete row into `scheduled_task` representing a single run at a specific time. + * Those task rows are what actually get picked up and run. + */ +@Entity({ name: 'scheduled_job' }) +@Index(['nextRunAt'], { + where: '"enabled" = true AND "nextRunAt" IS NOT NULL', +}) +@Index(['workflowId'], { where: '"workflowId" IS NOT NULL' }) +@Index(['name'], { unique: true }) +export class ScheduledJob extends WithTimestamps { + @PrimaryGeneratedColumn() + id: number; + + /** + * Human-readable job name, unique across all jobs. + * A well-known scheduler key for system jobs (e.g. a maintenance job); + * generated for jobs owned by a workflow trigger. + */ + @Column({ type: 'varchar', length: 255 }) + name: string; + + /** + * Workflow this job belongs to, + * referenced via its published version + * (only published trigger nodes get scheduled). + * `null` for well-known system jobs that aren't tied to a workflow. + * Unpublishing the workflow cascades its jobs away. + */ + @Column({ type: 'varchar', length: 36, nullable: true }) + workflowId: string | null; + + /** + * Trigger node within the workflow that owns this job. + * `null` for non-trigger jobs. + */ + @Column({ type: 'varchar', length: 36, nullable: true }) + nodeId: string | null; + + /** + * What kind of work this job runs. + * The scheduler is generic, so this is how it knows what to do when the job + * fires, e.g. `'scheduleTrigger'` for a workflow's schedule trigger. + * Paired with {@link payload}, which carries the handler's input. + */ + @Column({ type: 'varchar', length: 128 }) + taskType: string; + + /** + * Input handed to the task handler when an occurrence runs. + */ + @JsonColumn({ default: '{}' }) + payload: Record; + + @Column({ type: 'varchar', length: 16 }) + kind: ScheduledJobKind; + + /** + * Cron expression driving recurrence. + * Set only when {@link kind} is `cron`. + */ + @Column({ type: 'varchar', length: 255, nullable: true }) + cronExpression: string | null; + + /** + * IANA timezone the cron expression is evaluated in. + * `null` falls back to the instance default. + */ + @Column({ type: 'varchar', length: 64, nullable: true }) + timezone: string | null; + + /** + * Gap between fires in seconds. + * Set only when {@link kind} is `interval`. + */ + @Column({ type: 'int', nullable: true }) + intervalSeconds: number | null; + + /** + * Absolute time the job fires once. + * Set only when {@link kind} is `one_off`. + */ + @DateTimeColumn({ nullable: true }) + fireAt: Date | null; + + @Column({ default: true }) + enabled: boolean; + + /** + * Next time an occurrence is due to be materialized. + * The scheduler's sweep reads this to find work. + * It's set to `null` once the job is disabled or a one-off has fired, + * which drops the row out of the sweep index. + */ + @DateTimeColumn({ nullable: true }) + nextRunAt: Date | null; + + /** + * Last time an occurrence was materialized, + * used to recompute {@link nextRunAt}. + */ + @DateTimeColumn({ nullable: true }) + lastFiredAt: Date | null; + + /** + * Retry ceiling copied onto each occurrence this job materializes. + */ + @Column({ type: 'int', default: 1 }) + maxAttempts: number; +} diff --git a/packages/@n8n/db/src/entities/scheduled-task.ts b/packages/@n8n/db/src/entities/scheduled-task.ts new file mode 100644 index 00000000000..1f2d4176ee1 --- /dev/null +++ b/packages/@n8n/db/src/entities/scheduled-task.ts @@ -0,0 +1,176 @@ +import { Column, Entity, Generated, Index, PrimaryColumn } from '@n8n/typeorm'; + +import { DateTimeColumn, JsonColumn, WithCreatedAt, dbType } from './abstract-entity'; +import { idStringifier } from '../utils/transformers'; + +/** + * Where a task is in its lifecycle, from waiting to run to a final outcome. + */ +export const ScheduledTaskStatus = { + Pending: 'pending', + Running: 'running', + Succeeded: 'succeeded', + Failed: 'failed', + Missed: 'missed', + Cancelled: 'cancelled', +} as const; + +export type ScheduledTaskStatus = (typeof ScheduledTaskStatus)[keyof typeof ScheduledTaskStatus]; + +/** + * One concrete run of a {@link ScheduledJob} at a specific time. + * + * When the scheduler decides a job is due, it creates a row here. + * A worker then picks the row up, runs it, and records the outcome. + * Each row tracks + * - its own progress (see {@link status}), + * - how many times it has been tried ({@link attempts}), + * - and who is currently running it ({@link claimedBy} and the lease columns) + * + * "Claiming" a row means a worker briefly reserves it + * so two workers don't run the same task at once. + * + * That reservation (a "lease") expires after a while, + * so if the worker dies mid-run another worker can take over. + */ +@Entity({ name: 'scheduled_task' }) +@Index(['jobId', 'scheduledFor'], { unique: true }) +@Index(['runAt'], { where: '"status" = \'pending\'' }) +@Index(['leaseExpiresAt'], { where: '"status" = \'running\'' }) +@Index(['finishedAt'], { where: '"finishedAt" IS NOT NULL' }) +export class ScheduledTask extends WithCreatedAt { + /** + * 64-bit identity. + * This table is a high-churn queue: every time a job fires it inserts a row, + * and the auto-increment counter never reuses values + * (retention deletes old rows but the counter keeps climbing). + * + * Per engine: Postgres uses bigint IDENTITY; SQLite uses INTEGER PRIMARY KEY + * (the rowid alias, already 64-bit and auto-generating). + */ + @Generated() + @PrimaryColumn({ + type: dbType === 'sqlite' ? 'integer' : 'bigint', + transformer: idStringifier, + }) + id: string; + + @Column({ type: 'int' }) + jobId: number; + + /** + * What kind of work to run, copied from the job's taskType when this run is + * created rather than read back through {@link jobId}. + * + * This denormalization (together with {@link payload}) makes a run self-contained: + * - a worker has everything it needs to execute it without joining back to scheduled_job, which keeps the claim hot path simple, + * - and the run stays a stable snapshot that isn't affected if the job is later edited or removed. + * + * It also future-proofs the design: because a run already carries its own taskType and payload, + * tasks can be enqueued directly without a parent job (e.g. ad-hoc one-off work). + * + * That path isn't enabled yet ({@link jobId} is currently required), but the schema won't need reworking to allow it. + */ + @Column({ type: 'varchar', length: 128 }) + taskType: string; + + /** + * Input for the handler, copied from the job's payload when this run is + * created. + * + * Like {@link taskType} it's a snapshot, + * so editing the job later doesn't change runs that are already queued. + * + * See {@link taskType} for why these two are denormalized. + */ + @JsonColumn({ default: '{}' }) + payload: Record; + + /** + * The time this run is for (the slot the job was due). + * Together with {@link jobId} it identifies the run, + * so the same job and time can't be queued twice. + */ + @DateTimeColumn() + scheduledFor: Date; + + /** + * Earliest time a worker may pick this run up. + * Starts equal to {@link scheduledFor}. + * If a try fails it is pushed later so the retry waits a bit before running again. + */ + @DateTimeColumn() + runAt: Date; + + /** + * Where this run is in its lifecycle. + * Workers look for `pending` rows to start, + * and the cleanup pass looks at `running` rows to recover stalled ones. + */ + @Column({ type: 'varchar', length: 16, default: ScheduledTaskStatus.Pending }) + status: ScheduledTaskStatus; + + /** + * How many times a worker has started this run so far, checked against {@link maxAttempts}. + */ + @Column({ type: 'int', default: 0 }) + attempts: number; + + /** + * The most times this run may be tried before it is given up on. + * Once {@link attempts} reaches it, a failure is final instead of retried. + * + * The default of 1 means a single try. If a worker crashes mid-run, that crash + * uses up the only attempt, so the run is given up on without ever finishing. + * Where a run must not be lost this way, claim logic should raise this, or tell + * a crash apart from a real failure of the task itself. + */ + @Column({ type: 'int', default: 1 }) + maxAttempts: number; + + /** + * Which worker is currently running this; + * `null` when no one has claimed it. + */ + @Column({ type: 'varchar', length: 255, nullable: true }) + claimedBy: string | null; + + /** + * When the current worker's claim runs out. + * If this passes while the run is still `running`, + * the cleanup pass assumes the worker died and frees the run for another to take. + * + * `null` while the run is unclaimed (`pending`) or finished. A DB CHECK + * (`CHK_scheduled_task_running_lease`) enforces that a `running` row always has + * one, so the reaper never finds a claimed run with no expiry to reclaim it by. + */ + @DateTimeColumn({ nullable: true }) + leaseExpiresAt: Date | null; + + /** + * A counter bumped each time the run is claimed. + * It lets a worker notice it has lost the run: if its claim expired and someone else took over, the counter + * has moved on, so the old worker knows not to write its now-stale result. + */ + @Column({ type: 'int', default: 0 }) + leaseEpoch: number; + + /** + * When the current try started running. + */ + @DateTimeColumn({ nullable: true }) + startedAt: Date | null; + + /** + * When this run finished, whether it succeeded or failed. + * Used to clean up old rows. + */ + @DateTimeColumn({ nullable: true }) + finishedAt: Date | null; + + /** + * Why the last try failed, if it did. + */ + @Column({ type: 'text', nullable: true }) + errorMessage: string | null; +} diff --git a/packages/@n8n/db/src/migrations/common/1784000000042-CreateSchedulerTables.ts b/packages/@n8n/db/src/migrations/common/1784000000042-CreateSchedulerTables.ts new file mode 100644 index 00000000000..d1af6a3bd29 --- /dev/null +++ b/packages/@n8n/db/src/migrations/common/1784000000042-CreateSchedulerTables.ts @@ -0,0 +1,232 @@ +import type { MigrationContext, ReversibleMigration } from '../migration-types'; + +export class CreateSchedulerTables1784000000042 implements ReversibleMigration { + async up(context: MigrationContext) { + await this.createScheduledJobTable(context); + await this.createScheduledTaskTable(context); + } + + async down({ schemaBuilder: { dropTable } }: MigrationContext) { + await dropTable('scheduled_task'); + await dropTable('scheduled_job'); + } + + private async createScheduledJobTable({ + schemaBuilder: { createTable, createIndex, column }, + tablePrefix, + }: MigrationContext) { + await createTable('scheduled_job') + .withColumns( + column('id').int.primary.autoGenerate2, + column('name') + .varchar(255) + .notNull.comment( + 'Human-readable job name. A well-known scheduler key for system jobs; generated for workflow trigger jobs.', + ), + column('workflowId') + .varchar(36) + .comment( + "References the workflow's published version, since only published trigger nodes get scheduled; NULL for system jobs not tied to a workflow. Unpublishing the workflow deletes its jobs.", + ), + column('nodeId') + .varchar(36) + .comment( + 'Trigger node within the workflow that owns this job; NULL for non-trigger jobs.', + ), + column('taskType') + .varchar(128) + .notNull.comment('Selects which registered handler runs the task.'), + column('payload') + .json.notNull.default("'{}'") + .comment('Input passed to the task handler when an occurrence runs.'), + column('kind') + .varchar(16) + .notNull.withEnumCheck(['cron', 'interval', 'one_off']) + .comment('Recurrence kind; selects which of the schedule columns below apply.'), + column('cronExpression') + .varchar(255) + .comment("Cron expression driving recurrence; set only when kind is 'cron'."), + column('timezone') + .varchar(64) + .comment( + 'IANA timezone the cron expression is evaluated in; NULL uses the instance default.', + ), + column('intervalSeconds').int.comment( + "Gap between fires in seconds; set only when kind is 'interval'.", + ), + column('fireAt') + .timestampTimezone() + .comment("Absolute time the job fires once; set only when kind is 'one_off'."), + column('enabled') + .bool.notNull.default(true) + .comment('Whether the scheduler considers this job for firing.'), + column('nextRunAt') + .timestampTimezone() + .comment( + 'Next time an occurrence is due; the scheduler sweep reads this to find work. NULL once disabled or a one-off has fired.', + ), + column('lastFiredAt') + .timestampTimezone() + .comment('Last time an occurrence was materialized; used to recompute nextRunAt.'), + column('maxAttempts') + .int.notNull.default(1) + .comment('Retry ceiling copied onto each occurrence this job materializes.'), + ) + .withTimestamps.withForeignKey('workflowId', { + tableName: 'workflow_published_version', + columnName: 'workflowId', + onDelete: 'CASCADE', + name: `FK_${tablePrefix}scheduled_job_workflowId`, + }) + // Each recurrence kind requires its own schedule column to be set. + .withCheck( + `CHK_${tablePrefix}scheduled_job_cron_expression`, + '"kind" <> \'cron\' OR "cronExpression" IS NOT NULL', + ) + .withCheck( + `CHK_${tablePrefix}scheduled_job_interval_seconds`, + '"kind" <> \'interval\' OR "intervalSeconds" IS NOT NULL', + ) + .withCheck( + `CHK_${tablePrefix}scheduled_job_fire_at`, + '"kind" <> \'one_off\' OR "fireAt" IS NOT NULL', + ); + + await createIndex( + 'scheduled_job', + ['nextRunAt'], + false, + undefined, + '"enabled" = true AND "nextRunAt" IS NOT NULL', + ); + + // Index the workflowId FK so a workflow delete cascades without seq-scanning + // scheduled_job. Also serves lookups of all jobs owned by a workflow. + // Partial: system jobs (NULL workflowId) never use it, and a workflowId lookup implies + // NOT NULL so the planner can still use it for the cascade. + await createIndex( + 'scheduled_job', + ['workflowId'], + false, + undefined, + '"workflowId" IS NOT NULL', + ); + + // Names are unique across all jobs: well-known keys for system jobs, generated + // for workflow trigger jobs. + await createIndex('scheduled_job', ['name'], true); + } + + private async createScheduledTaskTable(context: MigrationContext) { + const { + schemaBuilder: { createTable, createIndex, column }, + isSqlite, + tablePrefix, + } = context; + + const idColumn = isSqlite + ? column('id').int.primary.autoGenerate2 + : column('id').bigint.primary.autoGenerate2; + + await createTable('scheduled_task') + .withColumns( + idColumn, + column('jobId').int.notNull.comment('The scheduled_job this occurrence belongs to.'), + column('taskType') + .varchar(128) + .notNull.comment( + 'What kind of work to run, copied from the job so a run is self-contained (no join to execute it). Also lets a run exist without a parent job in future.', + ), + column('payload') + .json.notNull.default("'{}'") + .comment( + "Handler input copied from the job. A snapshot, so editing the job later doesn't change runs already queued.", + ), + column('scheduledFor') + .timestampTimezone() + .notNull.comment( + 'The logical fire time this occurrence represents; unique per job, so the same instant cannot be queued twice.', + ), + column('runAt') + .timestampTimezone() + .notNull.comment( + 'Earliest time the executor may pick this up; starts at scheduledFor and is pushed out by retry backoff.', + ), + column('status') + .varchar(16) + .notNull.default("'pending'") + .withEnumCheck(['pending', 'running', 'succeeded', 'failed', 'missed', 'cancelled']) + .comment( + 'Lifecycle state; drives which occurrences the claim and reaper scans consider.', + ), + column('attempts') + .int.notNull.default(0) + .comment('Execution attempts started so far; compared against maxAttempts.'), + column('maxAttempts') + .int.notNull.default(1) + .comment( + 'Attempt ceiling; once attempts reaches it, a failure is final rather than retried.', + ), + column('claimedBy') + .varchar(255) + .comment('Id of the instance currently holding the lease; NULL when unclaimed.'), + column('leaseExpiresAt') + .timestampTimezone() + .comment( + 'When the current lease expires; the reaper reclaims running occurrences past this.', + ), + column('leaseEpoch') + .int.notNull.default(0) + .comment( + "Fencing token bumped on each claim; lets a reaped worker detect it lost ownership and not overwrite the new owner's results.", + ), + column('startedAt') + .timestampTimezone() + .comment('When the current attempt started running.'), + column('finishedAt') + .timestampTimezone() + .comment('When the occurrence reached a terminal state; drives retention pruning.'), + column('errorMessage').text.comment('Failure detail from the last attempt.'), + ) + .withCreatedAt.withForeignKey('jobId', { + tableName: 'scheduled_job', + columnName: 'id', + onDelete: 'CASCADE', + name: `FK_${tablePrefix}scheduled_task_jobId`, + }) + .withCheck( + `CHK_${tablePrefix}scheduled_task_running_lease`, + '"status" <> \'running\' OR "leaseExpiresAt" IS NOT NULL', + ); + + // Layer-1 dedup: at most one occurrence per (job, scheduled time). Its + // jobId prefix also indexes the FK for cascade deletes. + await createIndex('scheduled_task', ['jobId', 'scheduledFor'], true); + + // Pending-claim scan: the executor pulls the next due, pending occurrences. + await createIndex('scheduled_task', ['runAt'], false, undefined, '"status" = \'pending\''); + + // Lease-reaper scan: only running occurrences carry a live lease. + await createIndex( + 'scheduled_task', + ['leaseExpiresAt'], + false, + undefined, + '"status" = \'running\'', + ); + + // Retention pruning scan: prune by finishedAt cutoff. Keyed on finishedAt + // alone (not status-first): finishedAt is only set on terminal rows, so the + // partial predicate already restricts the index to prunable rows, and a + // leading timestamp range pairs with the prune LIMIT. Leading with the + // low-cardinality status instead forces a multi-range scan that Postgres + // falls back to a seq scan on. + await createIndex( + 'scheduled_task', + ['finishedAt'], + false, + undefined, + '"finishedAt" IS NOT NULL', + ); + } +} diff --git a/packages/@n8n/db/src/migrations/postgresdb/index.ts b/packages/@n8n/db/src/migrations/postgresdb/index.ts index e18733435b6..70d38024a59 100644 --- a/packages/@n8n/db/src/migrations/postgresdb/index.ts +++ b/packages/@n8n/db/src/migrations/postgresdb/index.ts @@ -215,6 +215,7 @@ import { SetChatHubEnabledFromUsage1784000000038 } from '../common/1784000000038 import { DropAgentExecutionFallbackColumns1784000000039 } from '../common/1784000000039-DropAgentExecutionFallbackColumns'; import { CreateWorkflowPublicationTriggerStatusTable1784000000040 } from '../common/1784000000040-CreateWorkflowPublicationTriggerStatusTable'; import { AddUsedPrivateCredentialsToExecutionEntity1784000000041 } from '../common/1784000000041-AddUsedPrivateCredentialsToExecutionEntity'; +import { CreateSchedulerTables1784000000042 } from '../common/1784000000042-CreateSchedulerTables'; import type { Migration } from '../migration-types'; export const postgresMigrations: Migration[] = [ @@ -435,4 +436,5 @@ export const postgresMigrations: Migration[] = [ DropAgentExecutionFallbackColumns1784000000039, CreateWorkflowPublicationTriggerStatusTable1784000000040, AddUsedPrivateCredentialsToExecutionEntity1784000000041, + CreateSchedulerTables1784000000042, ]; diff --git a/packages/@n8n/db/src/migrations/sqlite/index.ts b/packages/@n8n/db/src/migrations/sqlite/index.ts index be7784e0ba2..f8d3eac6fba 100644 --- a/packages/@n8n/db/src/migrations/sqlite/index.ts +++ b/packages/@n8n/db/src/migrations/sqlite/index.ts @@ -208,6 +208,7 @@ import { SetChatHubEnabledFromUsage1784000000038 } from '../common/1784000000038 import { DropAgentExecutionFallbackColumns1784000000039 } from '../common/1784000000039-DropAgentExecutionFallbackColumns'; import { CreateWorkflowPublicationTriggerStatusTable1784000000040 } from '../common/1784000000040-CreateWorkflowPublicationTriggerStatusTable'; import { AddUsedPrivateCredentialsToExecutionEntity1784000000041 } from '../common/1784000000041-AddUsedPrivateCredentialsToExecutionEntity'; +import { CreateSchedulerTables1784000000042 } from '../common/1784000000042-CreateSchedulerTables'; const sqliteMigrations: Migration[] = [ InitialMigration1588102412422, @@ -419,6 +420,7 @@ const sqliteMigrations: Migration[] = [ DropAgentExecutionFallbackColumns1784000000039, CreateWorkflowPublicationTriggerStatusTable1784000000040, AddUsedPrivateCredentialsToExecutionEntity1784000000041, + CreateSchedulerTables1784000000042, ]; export { sqliteMigrations }; diff --git a/packages/@n8n/db/src/repositories/index.ts b/packages/@n8n/db/src/repositories/index.ts index 8f999a19d02..5b9833f57f7 100644 --- a/packages/@n8n/db/src/repositories/index.ts +++ b/packages/@n8n/db/src/repositories/index.ts @@ -35,6 +35,8 @@ export { ProjectRelationRepository } from './project-relation.repository'; export { ProjectRepository, type ProjectListOptions } from './project.repository'; export { RoleRepository } from './role.repository'; export { RoleMappingRuleRepository } from './role-mapping-rule.repository'; +export { ScheduledJobRepository } from './scheduled-job.repository'; +export { ScheduledTaskRepository } from './scheduled-task.repository'; export { ProcessedDataRepository } from './processed-data.repository'; export { SettingsRepository } from './settings.repository'; export { TagRepository } from './tag.repository'; diff --git a/packages/@n8n/db/src/repositories/scheduled-job.repository.ts b/packages/@n8n/db/src/repositories/scheduled-job.repository.ts new file mode 100644 index 00000000000..51646445032 --- /dev/null +++ b/packages/@n8n/db/src/repositories/scheduled-job.repository.ts @@ -0,0 +1,15 @@ +import { Service } from '@n8n/di'; +import { DataSource, Repository } from '@n8n/typeorm'; + +import { ScheduledJob } from '../entities/scheduled-job'; + +@Service() +export class ScheduledJobRepository extends Repository { + constructor(dataSource: DataSource) { + super(ScheduledJob, dataSource.manager); + } + + async findAll(): Promise { + return await this.find(); + } +} diff --git a/packages/@n8n/db/src/repositories/scheduled-task.repository.ts b/packages/@n8n/db/src/repositories/scheduled-task.repository.ts new file mode 100644 index 00000000000..4efafeb822f --- /dev/null +++ b/packages/@n8n/db/src/repositories/scheduled-task.repository.ts @@ -0,0 +1,15 @@ +import { Service } from '@n8n/di'; +import { DataSource, Repository } from '@n8n/typeorm'; + +import { ScheduledTask } from '../entities/scheduled-task'; + +@Service() +export class ScheduledTaskRepository extends Repository { + constructor(dataSource: DataSource) { + super(ScheduledTask, dataSource.manager); + } + + async findAll(): Promise { + return await this.find(); + } +} diff --git a/packages/cli/test/migration/1784000000042-create-scheduler-tables.test.ts b/packages/cli/test/migration/1784000000042-create-scheduler-tables.test.ts new file mode 100644 index 00000000000..bd97dccefae --- /dev/null +++ b/packages/cli/test/migration/1784000000042-create-scheduler-tables.test.ts @@ -0,0 +1,544 @@ +import { + createTestMigrationContext, + initDbUpToMigration, + runSingleMigration, + type TestMigrationContext, +} from '@n8n/backend-test-utils'; +import { DbConnection } from '@n8n/db'; +import { Container } from '@n8n/di'; +import { DataSource } from '@n8n/typeorm'; +import { randomUUID } from 'node:crypto'; + +const MIGRATION_NAME = 'CreateSchedulerTables1784000000042'; + +describe('CreateSchedulerTables Migration', () => { + let dataSource: DataSource; + + beforeAll(async () => { + const dbConnection = Container.get(DbConnection); + await dbConnection.init(); + dataSource = Container.get(DataSource); + + const context = createTestMigrationContext(dataSource); + await context.queryRunner.clearDatabase(); + await context.queryRunner.release(); + + await initDbUpToMigration(MIGRATION_NAME); + await runSingleMigration(MIGRATION_NAME); + }); + + afterAll(async () => { + await Container.get(DbConnection).close(); + }); + + async function indexExists(context: TestMigrationContext, name: string): Promise { + if (context.isSqlite) { + const rows = await context.runQuery( + "SELECT name FROM sqlite_master WHERE type = 'index' AND name = :name", + { name }, + ); + return rows.length === 1; + } + const rows = await context.runQuery( + 'SELECT indexname FROM pg_indexes WHERE indexname = :name', + { name }, + ); + return rows.length === 1; + } + + async function insertPublishedWorkflow(context: TestMigrationContext): Promise { + const workflowId = randomUUID(); + const versionId = randomUUID(); + const now = new Date(); + await context.runQuery( + `INSERT INTO ${context.escape.tableName('workflow_entity')} ("id", "name", "active", "nodes", "connections", "triggerCount", "versionId", "createdAt", "updatedAt") + VALUES (:id, :name, :active, :nodes, :connections, :triggerCount, :versionId, :createdAt, :updatedAt)`, + { + id: workflowId, + name: `Scheduler test workflow ${workflowId}`, + active: false, + nodes: '[]', + connections: '{}', + triggerCount: 0, + versionId, + createdAt: now, + updatedAt: now, + }, + ); + await context.runQuery( + `INSERT INTO ${context.escape.tableName('workflow_history')} ("versionId", "workflowId", "nodes", "connections", "authors", "createdAt", "updatedAt") + VALUES (:versionId, :workflowId, :nodes, :connections, :authors, :createdAt, :updatedAt)`, + { + versionId, + workflowId, + nodes: '[]', + connections: '{}', + authors: 'test', + createdAt: now, + updatedAt: now, + }, + ); + await context.runQuery( + `INSERT INTO ${context.escape.tableName('workflow_published_version')} ("workflowId", "publishedVersionId", "createdAt", "updatedAt") + VALUES (:workflowId, :publishedVersionId, :createdAt, :updatedAt)`, + { workflowId, publishedVersionId: versionId, createdAt: now, updatedAt: now }, + ); + return workflowId; + } + + async function insertJob( + context: TestMigrationContext, + overrides: { workflowId?: string } = {}, + ): Promise { + const table = context.escape.tableName('scheduled_job'); + const now = new Date(); + const name = `job-${randomUUID()}`; + await context.runQuery( + `INSERT INTO ${table} ("name", "workflowId", "taskType", "kind", "cronExpression", "createdAt", "updatedAt") + VALUES (:name, :workflowId, :taskType, :kind, :cronExpression, :createdAt, :updatedAt)`, + { + name, + workflowId: overrides.workflowId ?? null, + taskType: 'scheduleTrigger', + kind: 'cron', + cronExpression: '* * * * *', + createdAt: now, + updatedAt: now, + }, + ); + const [row] = await context.runQuery>( + `SELECT "id" FROM ${table} WHERE "name" = :name`, + { name }, + ); + return row.id; + } + + describe('Up migration', () => { + it('creates the scheduled_job and scheduled_task tables', async () => { + const context = createTestMigrationContext(dataSource); + const jobs = await context.runQuery( + `SELECT * FROM ${context.escape.tableName('scheduled_job')}`, + ); + const tasks = await context.runQuery( + `SELECT * FROM ${context.escape.tableName('scheduled_task')}`, + ); + expect(jobs).toEqual([]); + expect(tasks).toEqual([]); + await context.queryRunner.release(); + }); + + it('creates all engine indexes', async () => { + const context = createTestMigrationContext(dataSource); + try { + for (const suffix of [ + 'scheduled_job_nextRunAt', + 'scheduled_job_workflowId', + 'scheduled_job_name', + 'scheduled_task_jobId_scheduledFor', + 'scheduled_task_runAt', + 'scheduled_task_leaseExpiresAt', + 'scheduled_task_finishedAt', + ]) { + expect(await indexExists(context, `IDX_${context.tablePrefix}${suffix}`)).toBe(true); + } + } finally { + await context.queryRunner.release(); + } + }); + + it('applies column defaults on insert', async () => { + const context = createTestMigrationContext(dataSource); + const jobId = await insertJob(context); + + const [job] = await context.runQuery< + Array<{ enabled: boolean | number; maxAttempts: number }> + >( + `SELECT "enabled", "maxAttempts" FROM ${context.escape.tableName('scheduled_job')} WHERE "id" = :id`, + { id: jobId }, + ); + expect(Boolean(job.enabled)).toBe(true); + expect(Number(job.maxAttempts)).toBe(1); + await context.queryRunner.release(); + }); + + it('auto-generates the scheduled_task primary key and applies status default', async () => { + const context = createTestMigrationContext(dataSource); + const jobId = await insertJob(context); + const table = context.escape.tableName('scheduled_task'); + const scheduledFor = new Date('2026-02-02T02:02:02.000Z'); + + await context.runQuery( + `INSERT INTO ${table} ("jobId", "taskType", "scheduledFor", "runAt", "createdAt") + VALUES (:jobId, :taskType, :scheduledFor, :runAt, :createdAt)`, + { + jobId, + taskType: 'scheduleTrigger', + scheduledFor, + runAt: scheduledFor, + createdAt: new Date(), + }, + ); + + const [task] = await context.runQuery>( + `SELECT "id", "status" FROM ${table} WHERE "jobId" = :jobId AND "scheduledFor" = :scheduledFor`, + { jobId, scheduledFor }, + ); + expect(task.id).not.toBeNull(); + expect(task.status).toBe('pending'); + await context.queryRunner.release(); + }); + + it('enforces the (jobId, scheduledFor) occurrence uniqueness', async () => { + const context = createTestMigrationContext(dataSource); + const jobId = await insertJob(context); + const table = context.escape.tableName('scheduled_task'); + const scheduledFor = new Date('2026-01-01T00:00:00.000Z'); + const insert = async () => + await context.runQuery( + `INSERT INTO ${table} ("jobId", "taskType", "scheduledFor", "runAt", "createdAt") + VALUES (:jobId, :taskType, :scheduledFor, :runAt, :createdAt)`, + { + jobId, + taskType: 'scheduleTrigger', + scheduledFor, + runAt: scheduledFor, + createdAt: new Date(), + }, + ); + + await insert(); + await expect(insert()).rejects.toThrow(); + await context.queryRunner.release(); + }); + + it('rejects an out-of-set kind via the CHECK constraint', async () => { + const context = createTestMigrationContext(dataSource); + const table = context.escape.tableName('scheduled_job'); + const now = new Date(); + await expect( + context.runQuery( + `INSERT INTO ${table} ("name", "taskType", "kind", "createdAt", "updatedAt") + VALUES (:name, :taskType, :kind, :createdAt, :updatedAt)`, + { + name: `job-${randomUUID()}`, + taskType: 'scheduleTrigger', + kind: 'bogus', + createdAt: now, + updatedAt: now, + }, + ), + ).rejects.toThrow(); + await context.queryRunner.release(); + }); + + it('allows multiple jobs on the same (workflowId, nodeId) trigger node', async () => { + const context = createTestMigrationContext(dataSource); + const table = context.escape.tableName('scheduled_job'); + const workflowId = await insertPublishedWorkflow(context); + const nodeId = randomUUID(); + const now = new Date(); + const insert = async () => + await context.runQuery( + `INSERT INTO ${table} ("name", "workflowId", "nodeId", "taskType", "kind", "cronExpression", "createdAt", "updatedAt") + VALUES (:name, :workflowId, :nodeId, :taskType, :kind, :cronExpression, :createdAt, :updatedAt)`, + { + name: `job-${randomUUID()}`, + workflowId, + nodeId, + taskType: 'scheduleTrigger', + kind: 'cron', + cronExpression: '* * * * *', + createdAt: now, + updatedAt: now, + }, + ); + + await insert(); + await expect(insert()).resolves.not.toThrow(); + await context.queryRunner.release(); + }); + + it('enforces name uniqueness across jobs', async () => { + const context = createTestMigrationContext(dataSource); + const table = context.escape.tableName('scheduled_job'); + const name = `job-${randomUUID()}`; + const now = new Date(); + const insert = async () => + await context.runQuery( + `INSERT INTO ${table} ("name", "taskType", "kind", "cronExpression", "createdAt", "updatedAt") + VALUES (:name, :taskType, :kind, :cronExpression, :createdAt, :updatedAt)`, + { + name, + taskType: 'scheduleTrigger', + kind: 'cron', + cronExpression: '* * * * *', + createdAt: now, + updatedAt: now, + }, + ); + + await insert(); + await expect(insert()).rejects.toThrow(); + await context.queryRunner.release(); + }); + + it('allows multiple system jobs with no workflow or node', async () => { + const context = createTestMigrationContext(dataSource); + const table = context.escape.tableName('scheduled_job'); + const now = new Date(); + const insert = async () => + await context.runQuery( + `INSERT INTO ${table} ("name", "taskType", "kind", "cronExpression", "createdAt", "updatedAt") + VALUES (:name, :taskType, :kind, :cronExpression, :createdAt, :updatedAt)`, + { + name: `job-${randomUUID()}`, + taskType: 'scheduleTrigger', + kind: 'cron', + cronExpression: '* * * * *', + createdAt: now, + updatedAt: now, + }, + ); + + await insert(); + await expect(insert()).resolves.not.toThrow(); + await context.queryRunner.release(); + }); + + it('allows multiple non-trigger jobs on the same workflow (NULL nodeId)', async () => { + const context = createTestMigrationContext(dataSource); + const table = context.escape.tableName('scheduled_job'); + const workflowId = await insertPublishedWorkflow(context); + const now = new Date(); + const insert = async () => + await context.runQuery( + `INSERT INTO ${table} ("name", "workflowId", "taskType", "kind", "cronExpression", "createdAt", "updatedAt") + VALUES (:name, :workflowId, :taskType, :kind, :cronExpression, :createdAt, :updatedAt)`, + { + name: `job-${randomUUID()}`, + workflowId, + taskType: 'scheduleTrigger', + kind: 'cron', + cronExpression: '* * * * *', + createdAt: now, + updatedAt: now, + }, + ); + + await insert(); + await expect(insert()).resolves.not.toThrow(); + await context.queryRunner.release(); + }); + + it('requires cronExpression when kind is cron', async () => { + const context = createTestMigrationContext(dataSource); + const table = context.escape.tableName('scheduled_job'); + const now = new Date(); + await expect( + context.runQuery( + `INSERT INTO ${table} ("name", "taskType", "kind", "createdAt", "updatedAt") + VALUES (:name, :taskType, :kind, :createdAt, :updatedAt)`, + { + name: `job-${randomUUID()}`, + taskType: 'scheduleTrigger', + kind: 'cron', + createdAt: now, + updatedAt: now, + }, + ), + ).rejects.toThrow(); + await context.queryRunner.release(); + }); + + it('requires intervalSeconds when kind is interval', async () => { + const context = createTestMigrationContext(dataSource); + const table = context.escape.tableName('scheduled_job'); + const now = new Date(); + await expect( + context.runQuery( + `INSERT INTO ${table} ("name", "taskType", "kind", "createdAt", "updatedAt") + VALUES (:name, :taskType, :kind, :createdAt, :updatedAt)`, + { + name: `job-${randomUUID()}`, + taskType: 'scheduleTrigger', + kind: 'interval', + createdAt: now, + updatedAt: now, + }, + ), + ).rejects.toThrow(); + await context.queryRunner.release(); + }); + + it('requires fireAt when kind is one_off', async () => { + const context = createTestMigrationContext(dataSource); + const table = context.escape.tableName('scheduled_job'); + const now = new Date(); + await expect( + context.runQuery( + `INSERT INTO ${table} ("name", "taskType", "kind", "createdAt", "updatedAt") + VALUES (:name, :taskType, :kind, :createdAt, :updatedAt)`, + { + name: `job-${randomUUID()}`, + taskType: 'scheduleTrigger', + kind: 'one_off', + createdAt: now, + updatedAt: now, + }, + ), + ).rejects.toThrow(); + await context.queryRunner.release(); + }); + + it('rejects an out-of-set status via the CHECK constraint', async () => { + const context = createTestMigrationContext(dataSource); + const jobId = await insertJob(context); + const table = context.escape.tableName('scheduled_task'); + const now = new Date(); + await expect( + context.runQuery( + `INSERT INTO ${table} ("jobId", "taskType", "scheduledFor", "runAt", "status", "createdAt") + VALUES (:jobId, :taskType, :scheduledFor, :runAt, :status, :createdAt)`, + { jobId, taskType: 't', scheduledFor: now, runAt: now, status: 'bogus', createdAt: now }, + ), + ).rejects.toThrow(); + await context.queryRunner.release(); + }); + + it('rejects a running occurrence with no lease deadline via the CHECK constraint', async () => { + const context = createTestMigrationContext(dataSource); + const jobId = await insertJob(context); + const table = context.escape.tableName('scheduled_task'); + const now = new Date(); + await expect( + context.runQuery( + `INSERT INTO ${table} ("jobId", "taskType", "scheduledFor", "runAt", "status", "leaseExpiresAt", "createdAt") + VALUES (:jobId, :taskType, :scheduledFor, :runAt, :status, :leaseExpiresAt, :createdAt)`, + { + jobId, + taskType: 't', + scheduledFor: now, + runAt: now, + status: 'running', + leaseExpiresAt: null, + createdAt: now, + }, + ), + ).rejects.toThrow(); + await context.queryRunner.release(); + }); + + it('accepts a running occurrence that carries a lease deadline', async () => { + const context = createTestMigrationContext(dataSource); + const jobId = await insertJob(context); + const table = context.escape.tableName('scheduled_task'); + const now = new Date(); + await context.runQuery( + `INSERT INTO ${table} ("jobId", "taskType", "scheduledFor", "runAt", "status", "leaseExpiresAt", "createdAt") + VALUES (:jobId, :taskType, :scheduledFor, :runAt, :status, :leaseExpiresAt, :createdAt)`, + { + jobId, + taskType: 't', + scheduledFor: now, + runAt: now, + status: 'running', + leaseExpiresAt: new Date(now.getTime() + 30_000), + createdAt: now, + }, + ); + + const [task] = await context.runQuery>( + `SELECT "status" FROM ${table} WHERE "jobId" = :jobId`, + { jobId }, + ); + expect(task.status).toBe('running'); + await context.queryRunner.release(); + }); + + it('cascades deletes from scheduled_job to its scheduled_task rows', async () => { + const context = createTestMigrationContext(dataSource); + const jobId = await insertJob(context); + const taskTable = context.escape.tableName('scheduled_task'); + const now = new Date(); + await context.runQuery( + `INSERT INTO ${taskTable} ("jobId", "taskType", "scheduledFor", "runAt", "createdAt") + VALUES (:jobId, :taskType, :scheduledFor, :runAt, :createdAt)`, + { jobId, taskType: 't', scheduledFor: now, runAt: now, createdAt: now }, + ); + + await context.runQuery( + `DELETE FROM ${context.escape.tableName('scheduled_job')} WHERE "id" = :id`, + { id: jobId }, + ); + + const remaining = await context.runQuery( + `SELECT * FROM ${taskTable} WHERE "jobId" = :jobId`, + { jobId }, + ); + expect(remaining).toEqual([]); + await context.queryRunner.release(); + }); + + it('cascades deletes from workflow_published_version to its scheduled_job rows', async () => { + const context = createTestMigrationContext(dataSource); + const workflowId = await insertPublishedWorkflow(context); + const jobId = await insertJob(context, { workflowId }); + + // Unpublishing drops the published-version row, which cascades to its jobs. + await context.runQuery( + `DELETE FROM ${context.escape.tableName('workflow_published_version')} WHERE "workflowId" = :workflowId`, + { workflowId }, + ); + + const remaining = await context.runQuery( + `SELECT * FROM ${context.escape.tableName('scheduled_job')} WHERE "id" = :id`, + { id: jobId }, + ); + expect(remaining).toEqual([]); + await context.queryRunner.release(); + }); + + it('rejects a workflow-bound job when the workflow is not published', async () => { + const context = createTestMigrationContext(dataSource); + const table = context.escape.tableName('scheduled_job'); + const now = new Date(); + // No workflow_published_version row for this id, so the FK must reject it. + await expect( + context.runQuery( + `INSERT INTO ${table} ("name", "workflowId", "taskType", "kind", "cronExpression", "createdAt", "updatedAt") + VALUES (:name, :workflowId, :taskType, :kind, :cronExpression, :createdAt, :updatedAt)`, + { + name: `job-${randomUUID()}`, + workflowId: randomUUID(), + taskType: 'scheduleTrigger', + kind: 'cron', + cronExpression: '* * * * *', + createdAt: now, + updatedAt: now, + }, + ), + ).rejects.toThrow(); + await context.queryRunner.release(); + }); + }); + + describe('Down migration', () => { + it('drops both tables and can be re-applied', async () => { + await dataSource.undoLastMigration({ transaction: 'each' }); + + const context = createTestMigrationContext(dataSource); + const jobTable = `${context.tablePrefix}scheduled_job`; + const taskTable = `${context.tablePrefix}scheduled_task`; + expect(await context.queryRunner.hasTable(jobTable)).toBe(false); + expect(await context.queryRunner.hasTable(taskTable)).toBe(false); + await context.queryRunner.release(); + + // Round-trip: up() must run cleanly again after a full revert. + await runSingleMigration(MIGRATION_NAME); + const context2 = createTestMigrationContext(dataSource); + expect(await context2.queryRunner.hasTable(jobTable)).toBe(true); + expect(await context2.queryRunner.hasTable(taskTable)).toBe(true); + await context2.queryRunner.release(); + }); + }); +});