feat: reduce memory usage (#3768)

* all

* all

* optimize deserialize

* all

* all

* fix issue

* fix compile

* fix compile

* fix compile

* fix compile

* fix compile

* fix compile

* fix compile

* fix compile

* fix compile

* fix compile
This commit is contained in:
Ruben Fiszel
2024-05-19 21:58:38 +02:00
committed by GitHub
parent 584594aed6
commit c43d04b95d
43 changed files with 977 additions and 2615 deletions
@@ -1,70 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT app.id, app.path, app.summary, app.versions, app.policy,\n app.extra_perms, app_version.value, \n app_version.created_at, app_version.created_by from app, app_version \n WHERE app.workspace_id = $1 AND app_version.id = app.versions[array_upper(app.versions, 1)]",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Int8"
},
{
"ordinal": 1,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "summary",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "versions",
"type_info": "Int8Array"
},
{
"ordinal": 4,
"name": "policy",
"type_info": "Jsonb"
},
{
"ordinal": 5,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 6,
"name": "value",
"type_info": "Json"
},
{
"ordinal": 7,
"name": "created_at",
"type_info": "Timestamptz"
},
{
"ordinal": 8,
"name": "created_by",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false,
false,
false,
false,
false,
false,
false,
false,
false
]
},
"hash": "023fffd0a042a28b5be991169a506aff92f64f84e49b4c041cd369b045c31e73"
}
@@ -1,71 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT app.id, app.path, app.summary, app.versions, app.policy,\n app.extra_perms, app_version.value, \n app_version.created_at, app_version.created_by from app, app_version \n WHERE app_version.id = $1 AND app.id = app_version.app_id AND app.workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Int8"
},
{
"ordinal": 1,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "summary",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "versions",
"type_info": "Int8Array"
},
{
"ordinal": 4,
"name": "policy",
"type_info": "Jsonb"
},
{
"ordinal": 5,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 6,
"name": "value",
"type_info": "Json"
},
{
"ordinal": 7,
"name": "created_at",
"type_info": "Timestamptz"
},
{
"ordinal": 8,
"name": "created_by",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Int8",
"Text"
]
},
"nullable": [
false,
false,
false,
false,
false,
false,
false,
false,
false
]
},
"hash": "438fb925ee90d5115bd3c3be8ae48b56ba86017af3ca519bd3a15829edaa7d1b"
}
@@ -1,38 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT success, result, started_at FROM completed_job WHERE workspace_id = $1 AND schedule_path = $2 AND script_path = $3 AND id != $4 ORDER BY created_at DESC LIMIT $5",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "success",
"type_info": "Bool"
},
{
"ordinal": 1,
"name": "result",
"type_info": "Jsonb"
},
{
"ordinal": 2,
"name": "started_at",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Uuid",
"Int8"
]
},
"nullable": [
false,
true,
false
]
},
"hash": "4a3917318d967d96c4b6e70786a034302f04da9d1e9eec59b1eb6052c2d947be"
}
@@ -1,187 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO schedule (workspace_id, path, schedule, timezone, edited_by, script_path, is_flow, args, enabled, email, on_failure, on_failure_times, on_failure_exact, on_failure_extra_args, on_recovery, on_recovery_times, on_recovery_extra_args, ws_error_handler_muted, retry, summary, no_flow_overlap, tag ) VALUES ( $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22 ) RETURNING *",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "edited_by",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "edited_at",
"type_info": "Timestamptz"
},
{
"ordinal": 4,
"name": "schedule",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "enabled",
"type_info": "Bool"
},
{
"ordinal": 6,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 7,
"name": "args",
"type_info": "Jsonb"
},
{
"ordinal": 8,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 9,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 10,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 11,
"name": "error",
"type_info": "Text"
},
{
"ordinal": 12,
"name": "timezone",
"type_info": "Varchar"
},
{
"ordinal": 13,
"name": "on_failure",
"type_info": "Varchar"
},
{
"ordinal": 14,
"name": "on_recovery",
"type_info": "Varchar"
},
{
"ordinal": 15,
"name": "on_failure_times",
"type_info": "Int4"
},
{
"ordinal": 16,
"name": "on_failure_exact",
"type_info": "Bool"
},
{
"ordinal": 17,
"name": "on_failure_extra_args",
"type_info": "Json"
},
{
"ordinal": 18,
"name": "on_recovery_times",
"type_info": "Int4"
},
{
"ordinal": 19,
"name": "on_recovery_extra_args",
"type_info": "Json"
},
{
"ordinal": 20,
"name": "ws_error_handler_muted",
"type_info": "Bool"
},
{
"ordinal": 21,
"name": "retry",
"type_info": "Jsonb"
},
{
"ordinal": 22,
"name": "summary",
"type_info": "Varchar"
},
{
"ordinal": 23,
"name": "no_flow_overlap",
"type_info": "Bool"
},
{
"ordinal": 24,
"name": "tag",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Varchar",
"Varchar",
"Varchar",
"Varchar",
"Bool",
"Jsonb",
"Bool",
"Varchar",
"Varchar",
"Int4",
"Bool",
"Json",
"Varchar",
"Int4",
"Json",
"Bool",
"Jsonb",
"Varchar",
"Bool",
"Varchar"
]
},
"nullable": [
false,
false,
false,
false,
false,
false,
false,
true,
false,
false,
false,
true,
false,
true,
true,
true,
true,
true,
true,
true,
false,
true,
true,
false,
true
]
},
"hash": "597335a4a1eda7799303e0d930b5468af2364f5d2075aab48787b93e5774336d"
}
@@ -1,29 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT path, app_version.value from app LEFT JOIN app_version ON app_version.id = versions[array_upper(versions, 1)] WHERE workspace_id = $1 LIMIT $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "value",
"type_info": "Json"
}
],
"parameters": {
"Left": [
"Text",
"Int8"
]
},
"nullable": [
false,
false
]
},
"hash": "644335d376b6554ab222d25dfa8722234661f2456b469f21b175e8607584614e"
}
@@ -5,7 +5,7 @@
"columns": [
{
"ordinal": 0,
"name": "bool",
"name": "?column?",
"type_info": "Bool"
}
],
@@ -1,182 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE schedule SET schedule = $1, timezone = $2, args = $3, on_failure = $4, on_failure_times = $5, on_failure_exact = $6, on_failure_extra_args = $7, on_recovery = $8, on_recovery_times = $9, on_recovery_extra_args = $10, ws_error_handler_muted = $11, retry = $12, summary = $13, no_flow_overlap = $14, tag = $15\n WHERE path = $16 AND workspace_id = $17 RETURNING *",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "edited_by",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "edited_at",
"type_info": "Timestamptz"
},
{
"ordinal": 4,
"name": "schedule",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "enabled",
"type_info": "Bool"
},
{
"ordinal": 6,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 7,
"name": "args",
"type_info": "Jsonb"
},
{
"ordinal": 8,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 9,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 10,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 11,
"name": "error",
"type_info": "Text"
},
{
"ordinal": 12,
"name": "timezone",
"type_info": "Varchar"
},
{
"ordinal": 13,
"name": "on_failure",
"type_info": "Varchar"
},
{
"ordinal": 14,
"name": "on_recovery",
"type_info": "Varchar"
},
{
"ordinal": 15,
"name": "on_failure_times",
"type_info": "Int4"
},
{
"ordinal": 16,
"name": "on_failure_exact",
"type_info": "Bool"
},
{
"ordinal": 17,
"name": "on_failure_extra_args",
"type_info": "Json"
},
{
"ordinal": 18,
"name": "on_recovery_times",
"type_info": "Int4"
},
{
"ordinal": 19,
"name": "on_recovery_extra_args",
"type_info": "Json"
},
{
"ordinal": 20,
"name": "ws_error_handler_muted",
"type_info": "Bool"
},
{
"ordinal": 21,
"name": "retry",
"type_info": "Jsonb"
},
{
"ordinal": 22,
"name": "summary",
"type_info": "Varchar"
},
{
"ordinal": 23,
"name": "no_flow_overlap",
"type_info": "Bool"
},
{
"ordinal": 24,
"name": "tag",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Jsonb",
"Varchar",
"Int4",
"Bool",
"Json",
"Varchar",
"Int4",
"Json",
"Bool",
"Jsonb",
"Varchar",
"Bool",
"Varchar",
"Text",
"Text"
]
},
"nullable": [
false,
false,
false,
false,
false,
false,
false,
true,
false,
false,
false,
true,
false,
true,
true,
true,
true,
true,
true,
true,
false,
true,
true,
false,
true
]
},
"hash": "90435687152ccd37db7e0a6a5baeb558db50f5f96b1e8eff0b10e4398566f325"
}
@@ -1,168 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE schedule SET path = $1, script_path = $1 WHERE path = $2 AND workspace_id = $3 AND is_flow IS true RETURNING *",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "edited_by",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "edited_at",
"type_info": "Timestamptz"
},
{
"ordinal": 4,
"name": "schedule",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "enabled",
"type_info": "Bool"
},
{
"ordinal": 6,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 7,
"name": "args",
"type_info": "Jsonb"
},
{
"ordinal": 8,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 9,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 10,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 11,
"name": "error",
"type_info": "Text"
},
{
"ordinal": 12,
"name": "timezone",
"type_info": "Varchar"
},
{
"ordinal": 13,
"name": "on_failure",
"type_info": "Varchar"
},
{
"ordinal": 14,
"name": "on_recovery",
"type_info": "Varchar"
},
{
"ordinal": 15,
"name": "on_failure_times",
"type_info": "Int4"
},
{
"ordinal": 16,
"name": "on_failure_exact",
"type_info": "Bool"
},
{
"ordinal": 17,
"name": "on_failure_extra_args",
"type_info": "Json"
},
{
"ordinal": 18,
"name": "on_recovery_times",
"type_info": "Int4"
},
{
"ordinal": 19,
"name": "on_recovery_extra_args",
"type_info": "Json"
},
{
"ordinal": 20,
"name": "ws_error_handler_muted",
"type_info": "Bool"
},
{
"ordinal": 21,
"name": "retry",
"type_info": "Jsonb"
},
{
"ordinal": 22,
"name": "summary",
"type_info": "Varchar"
},
{
"ordinal": 23,
"name": "no_flow_overlap",
"type_info": "Bool"
},
{
"ordinal": 24,
"name": "tag",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text"
]
},
"nullable": [
false,
false,
false,
false,
false,
false,
false,
true,
false,
false,
false,
true,
false,
true,
true,
true,
true,
true,
true,
true,
false,
true,
true,
false,
true
]
},
"hash": "911b1e1f2a5ba6d5159916e5598020e680c45043b0736ad0153ee261a151dd90"
}
@@ -1,169 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE schedule SET enabled = $1, email = $2 WHERE path = $3 AND workspace_id = $4 RETURNING *",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "edited_by",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "edited_at",
"type_info": "Timestamptz"
},
{
"ordinal": 4,
"name": "schedule",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "enabled",
"type_info": "Bool"
},
{
"ordinal": 6,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 7,
"name": "args",
"type_info": "Jsonb"
},
{
"ordinal": 8,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 9,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 10,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 11,
"name": "error",
"type_info": "Text"
},
{
"ordinal": 12,
"name": "timezone",
"type_info": "Varchar"
},
{
"ordinal": 13,
"name": "on_failure",
"type_info": "Varchar"
},
{
"ordinal": 14,
"name": "on_recovery",
"type_info": "Varchar"
},
{
"ordinal": 15,
"name": "on_failure_times",
"type_info": "Int4"
},
{
"ordinal": 16,
"name": "on_failure_exact",
"type_info": "Bool"
},
{
"ordinal": 17,
"name": "on_failure_extra_args",
"type_info": "Json"
},
{
"ordinal": 18,
"name": "on_recovery_times",
"type_info": "Int4"
},
{
"ordinal": 19,
"name": "on_recovery_extra_args",
"type_info": "Json"
},
{
"ordinal": 20,
"name": "ws_error_handler_muted",
"type_info": "Bool"
},
{
"ordinal": 21,
"name": "retry",
"type_info": "Jsonb"
},
{
"ordinal": 22,
"name": "summary",
"type_info": "Varchar"
},
{
"ordinal": 23,
"name": "no_flow_overlap",
"type_info": "Bool"
},
{
"ordinal": 24,
"name": "tag",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Bool",
"Varchar",
"Text",
"Text"
]
},
"nullable": [
false,
false,
false,
false,
false,
false,
false,
true,
false,
false,
false,
true,
false,
true,
true,
true,
true,
true,
true,
true,
false,
true,
true,
false,
true
]
},
"hash": "96dc1cd308f3c21cfb50b88048054dc03f93e261d25969d66aa48e9d0502960f"
}
@@ -1,71 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT app.id, app.path, app.summary, app.versions, app.policy,\n app.extra_perms, app_version.value, \n app_version.created_at, app_version.created_by from app, app_version \n WHERE app.id = $1 AND app.workspace_id = $2 AND app_version.id = app.versions[array_upper(app.versions, 1)]",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Int8"
},
{
"ordinal": 1,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "summary",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "versions",
"type_info": "Int8Array"
},
{
"ordinal": 4,
"name": "policy",
"type_info": "Jsonb"
},
{
"ordinal": 5,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 6,
"name": "value",
"type_info": "Json"
},
{
"ordinal": 7,
"name": "created_at",
"type_info": "Timestamptz"
},
{
"ordinal": 8,
"name": "created_by",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Int8",
"Text"
]
},
"nullable": [
false,
false,
false,
false,
false,
false,
false,
false,
false
]
},
"hash": "9ae98fbcea508dfc7113621c00b856f98f38dfb701a4660ea1a4058a6d7564f2"
}
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT concurrency_key FROM script WHERE hash = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "concurrency_key",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Int8",
"Text"
]
},
"nullable": [
true
]
},
"hash": "a3ccf362b4f6df400b3c7a084795dbf541eb14c5c374656ffb96da7283a2a6f1"
}
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT value->>'concurrency_key' FROM flow WHERE path = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "a875cb56485b812e9d4739afd0915067f7e5abe0ca0adf264b792fccf21e005b"
}
@@ -1,167 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT * FROM schedule WHERE path = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "edited_by",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "edited_at",
"type_info": "Timestamptz"
},
{
"ordinal": 4,
"name": "schedule",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "enabled",
"type_info": "Bool"
},
{
"ordinal": 6,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 7,
"name": "args",
"type_info": "Jsonb"
},
{
"ordinal": 8,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 9,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 10,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 11,
"name": "error",
"type_info": "Text"
},
{
"ordinal": 12,
"name": "timezone",
"type_info": "Varchar"
},
{
"ordinal": 13,
"name": "on_failure",
"type_info": "Varchar"
},
{
"ordinal": 14,
"name": "on_recovery",
"type_info": "Varchar"
},
{
"ordinal": 15,
"name": "on_failure_times",
"type_info": "Int4"
},
{
"ordinal": 16,
"name": "on_failure_exact",
"type_info": "Bool"
},
{
"ordinal": 17,
"name": "on_failure_extra_args",
"type_info": "Json"
},
{
"ordinal": 18,
"name": "on_recovery_times",
"type_info": "Int4"
},
{
"ordinal": 19,
"name": "on_recovery_extra_args",
"type_info": "Json"
},
{
"ordinal": 20,
"name": "ws_error_handler_muted",
"type_info": "Bool"
},
{
"ordinal": 21,
"name": "retry",
"type_info": "Jsonb"
},
{
"ordinal": 22,
"name": "summary",
"type_info": "Varchar"
},
{
"ordinal": 23,
"name": "no_flow_overlap",
"type_info": "Bool"
},
{
"ordinal": 24,
"name": "tag",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false,
false,
false,
false,
false,
false,
false,
true,
false,
false,
false,
true,
false,
true,
true,
true,
true,
true,
true,
true,
false,
true,
true,
false,
true
]
},
"hash": "aa2800113a8a8805f47cdc1dd0f29d94c546fe531e7edd3e91da4978af5442fb"
}
@@ -1,83 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT app.id, app.path, app.summary, app.versions, app.policy,\n app.extra_perms, app_version.value, \n app_version.created_at, app_version.created_by,\n app.draft_only, draft.value as \"draft?\"\n from app\n INNER JOIN app_version ON\n app_version.id = app.versions[array_upper(app.versions, 1)]\n LEFT JOIN draft ON \n app.path = draft.path AND draft.workspace_id = $2 AND draft.typ = 'app' \n WHERE app.path = $1 AND app.workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Int8"
},
{
"ordinal": 1,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "summary",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "versions",
"type_info": "Int8Array"
},
{
"ordinal": 4,
"name": "policy",
"type_info": "Jsonb"
},
{
"ordinal": 5,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 6,
"name": "value",
"type_info": "Json"
},
{
"ordinal": 7,
"name": "created_at",
"type_info": "Timestamptz"
},
{
"ordinal": 8,
"name": "created_by",
"type_info": "Varchar"
},
{
"ordinal": 9,
"name": "draft_only",
"type_info": "Bool"
},
{
"ordinal": 10,
"name": "draft?",
"type_info": "Json"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false,
false,
false,
false,
false,
false,
false,
false,
false,
true,
false
]
},
"hash": "aedba9cff117a7069dfd56602110b7f66903d6afb4f87fbffe0175982ea8741f"
}
@@ -18,8 +18,8 @@
"Left": []
},
"nullable": [
false,
true
true,
false
]
},
"hash": "b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76"
@@ -1,168 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE schedule SET script_path = $1 WHERE script_path = $2 AND path != $2 AND workspace_id = $3 AND is_flow IS false RETURNING *",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "edited_by",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "edited_at",
"type_info": "Timestamptz"
},
{
"ordinal": 4,
"name": "schedule",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "enabled",
"type_info": "Bool"
},
{
"ordinal": 6,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 7,
"name": "args",
"type_info": "Jsonb"
},
{
"ordinal": 8,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 9,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 10,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 11,
"name": "error",
"type_info": "Text"
},
{
"ordinal": 12,
"name": "timezone",
"type_info": "Varchar"
},
{
"ordinal": 13,
"name": "on_failure",
"type_info": "Varchar"
},
{
"ordinal": 14,
"name": "on_recovery",
"type_info": "Varchar"
},
{
"ordinal": 15,
"name": "on_failure_times",
"type_info": "Int4"
},
{
"ordinal": 16,
"name": "on_failure_exact",
"type_info": "Bool"
},
{
"ordinal": 17,
"name": "on_failure_extra_args",
"type_info": "Json"
},
{
"ordinal": 18,
"name": "on_recovery_times",
"type_info": "Int4"
},
{
"ordinal": 19,
"name": "on_recovery_extra_args",
"type_info": "Json"
},
{
"ordinal": 20,
"name": "ws_error_handler_muted",
"type_info": "Bool"
},
{
"ordinal": 21,
"name": "retry",
"type_info": "Jsonb"
},
{
"ordinal": 22,
"name": "summary",
"type_info": "Varchar"
},
{
"ordinal": 23,
"name": "no_flow_overlap",
"type_info": "Bool"
},
{
"ordinal": 24,
"name": "tag",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text"
]
},
"nullable": [
false,
false,
false,
false,
false,
false,
false,
true,
false,
false,
false,
true,
false,
true,
true,
true,
true,
true,
true,
true,
false,
true,
true,
false,
true
]
},
"hash": "c10348d26e3e3dc9e345d9044db35db9906617eb9c98aaf58a55d9681ffb85e3"
}
@@ -1,71 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT app.id, app.path, app.summary, app.versions, app.policy,\n app.extra_perms, app_version.value, \n app_version.created_at, app_version.created_by from app, app_version \n WHERE app.path = $1 AND app.workspace_id = $2 AND app_version.id = app.versions[array_upper(app.versions, 1)]",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Int8"
},
{
"ordinal": 1,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "summary",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "versions",
"type_info": "Int8Array"
},
{
"ordinal": 4,
"name": "policy",
"type_info": "Jsonb"
},
{
"ordinal": 5,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 6,
"name": "value",
"type_info": "Json"
},
{
"ordinal": 7,
"name": "created_at",
"type_info": "Timestamptz"
},
{
"ordinal": 8,
"name": "created_by",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false,
false,
false,
false,
false,
false,
false,
false,
false
]
},
"hash": "d07faf1c9501d1bc5f6fd9588bb3da7b7f4897fcfd26dcb3b8ce406b1f653077"
}
@@ -1,168 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE schedule SET path = $1, script_path = $1 WHERE path = $2 AND workspace_id = $3 AND is_flow IS false RETURNING *",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "edited_by",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "edited_at",
"type_info": "Timestamptz"
},
{
"ordinal": 4,
"name": "schedule",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "enabled",
"type_info": "Bool"
},
{
"ordinal": 6,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 7,
"name": "args",
"type_info": "Jsonb"
},
{
"ordinal": 8,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 9,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 10,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 11,
"name": "error",
"type_info": "Text"
},
{
"ordinal": 12,
"name": "timezone",
"type_info": "Varchar"
},
{
"ordinal": 13,
"name": "on_failure",
"type_info": "Varchar"
},
{
"ordinal": 14,
"name": "on_recovery",
"type_info": "Varchar"
},
{
"ordinal": 15,
"name": "on_failure_times",
"type_info": "Int4"
},
{
"ordinal": 16,
"name": "on_failure_exact",
"type_info": "Bool"
},
{
"ordinal": 17,
"name": "on_failure_extra_args",
"type_info": "Json"
},
{
"ordinal": 18,
"name": "on_recovery_times",
"type_info": "Int4"
},
{
"ordinal": 19,
"name": "on_recovery_extra_args",
"type_info": "Json"
},
{
"ordinal": 20,
"name": "ws_error_handler_muted",
"type_info": "Bool"
},
{
"ordinal": 21,
"name": "retry",
"type_info": "Jsonb"
},
{
"ordinal": 22,
"name": "summary",
"type_info": "Varchar"
},
{
"ordinal": 23,
"name": "no_flow_overlap",
"type_info": "Bool"
},
{
"ordinal": 24,
"name": "tag",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text"
]
},
"nullable": [
false,
false,
false,
false,
false,
false,
false,
true,
false,
false,
false,
true,
false,
true,
true,
true,
true,
true,
true,
true,
false,
true,
true,
false,
true
]
},
"hash": "dd74fa9468b5fe8c8ad657ded06076c11a78d0206af2af20685c691d4d9520bb"
}
@@ -1,168 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE schedule SET script_path = $1 WHERE script_path = $2 AND path != $2 AND workspace_id = $3 AND is_flow IS true RETURNING *",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "edited_by",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "edited_at",
"type_info": "Timestamptz"
},
{
"ordinal": 4,
"name": "schedule",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "enabled",
"type_info": "Bool"
},
{
"ordinal": 6,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 7,
"name": "args",
"type_info": "Jsonb"
},
{
"ordinal": 8,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 9,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 10,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 11,
"name": "error",
"type_info": "Text"
},
{
"ordinal": 12,
"name": "timezone",
"type_info": "Varchar"
},
{
"ordinal": 13,
"name": "on_failure",
"type_info": "Varchar"
},
{
"ordinal": 14,
"name": "on_recovery",
"type_info": "Varchar"
},
{
"ordinal": 15,
"name": "on_failure_times",
"type_info": "Int4"
},
{
"ordinal": 16,
"name": "on_failure_exact",
"type_info": "Bool"
},
{
"ordinal": 17,
"name": "on_failure_extra_args",
"type_info": "Json"
},
{
"ordinal": 18,
"name": "on_recovery_times",
"type_info": "Int4"
},
{
"ordinal": 19,
"name": "on_recovery_extra_args",
"type_info": "Json"
},
{
"ordinal": 20,
"name": "ws_error_handler_muted",
"type_info": "Bool"
},
{
"ordinal": 21,
"name": "retry",
"type_info": "Jsonb"
},
{
"ordinal": 22,
"name": "summary",
"type_info": "Varchar"
},
{
"ordinal": 23,
"name": "no_flow_overlap",
"type_info": "Bool"
},
{
"ordinal": 24,
"name": "tag",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text"
]
},
"nullable": [
false,
false,
false,
false,
false,
false,
false,
true,
false,
false,
false,
true,
false,
true,
true,
true,
true,
true,
true,
true,
false,
true,
true,
false,
true
]
},
"hash": "eced0a09ba547ce1dccb54a5419b22373603c9d01f77047b3553bde125bf71e8"
}
@@ -1,29 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT path, value from flow WHERE workspace_id = $1 LIMIT $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "value",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Text",
"Int8"
]
},
"nullable": [
false,
false
]
},
"hash": "efbd9dc28ab5e53d070684b2e99d945872791171a1c8cef6088ea662f3b2cebb"
}
@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT payload\n FROM capture\n WHERE workspace_id = $1\n AND path = $2\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "payload",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false
]
},
"hash": "f4960efbb4189b595541608c9ab5cbbd56eb42059e961b6338b685105626a5c7"
}
@@ -1,166 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT * FROM schedule\n WHERE workspace_id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "edited_by",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "edited_at",
"type_info": "Timestamptz"
},
{
"ordinal": 4,
"name": "schedule",
"type_info": "Varchar"
},
{
"ordinal": 5,
"name": "enabled",
"type_info": "Bool"
},
{
"ordinal": 6,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 7,
"name": "args",
"type_info": "Jsonb"
},
{
"ordinal": 8,
"name": "extra_perms",
"type_info": "Jsonb"
},
{
"ordinal": 9,
"name": "is_flow",
"type_info": "Bool"
},
{
"ordinal": 10,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 11,
"name": "error",
"type_info": "Text"
},
{
"ordinal": 12,
"name": "timezone",
"type_info": "Varchar"
},
{
"ordinal": 13,
"name": "on_failure",
"type_info": "Varchar"
},
{
"ordinal": 14,
"name": "on_recovery",
"type_info": "Varchar"
},
{
"ordinal": 15,
"name": "on_failure_times",
"type_info": "Int4"
},
{
"ordinal": 16,
"name": "on_failure_exact",
"type_info": "Bool"
},
{
"ordinal": 17,
"name": "on_failure_extra_args",
"type_info": "Json"
},
{
"ordinal": 18,
"name": "on_recovery_times",
"type_info": "Int4"
},
{
"ordinal": 19,
"name": "on_recovery_extra_args",
"type_info": "Json"
},
{
"ordinal": 20,
"name": "ws_error_handler_muted",
"type_info": "Bool"
},
{
"ordinal": 21,
"name": "retry",
"type_info": "Jsonb"
},
{
"ordinal": 22,
"name": "summary",
"type_info": "Varchar"
},
{
"ordinal": 23,
"name": "no_flow_overlap",
"type_info": "Bool"
},
{
"ordinal": 24,
"name": "tag",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false,
false,
false,
false,
false,
false,
false,
true,
false,
false,
false,
true,
false,
true,
true,
true,
true,
true,
true,
true,
false,
true,
true,
false,
true
]
},
"hash": "fdd3710a381dac33ef5ee5ec5564a4874e6d7807d43fa6ea7b8408cad9e97480"
}
+1 -1
View File
@@ -1 +1 @@
e0b0494c26b63efa4aa3f41a9fb42c7733dcca2e
081d34e693a9a69fe821d3f3baa672da5705a1cb
+24 -21
View File
@@ -8,7 +8,6 @@ use futures::StreamExt;
use futures::{stream, Stream};
use serde::Deserialize;
use serde_json::json;
use sqlx::types::Json;
use sqlx::{postgres::PgListener, types::Uuid, Pool, Postgres};
#[cfg(feature = "enterprise")]
@@ -868,13 +867,17 @@ impl RunJob {
async fn push(self, db: &Pool<Postgres>) -> Uuid {
let RunJob { payload, args } = self;
let mut hm_args = std::collections::HashMap::new();
for (k, v) in args {
hm_args.insert(k, windmill_common::worker::to_raw_value(&v));
}
let tx = PushIsolationLevel::IsolatedRoot(db.clone(), None);
let (uuid, tx) = windmill_queue::push::<_, rsmq_async::MultiplexedRsmq>(
let (uuid, tx) = windmill_queue::push::<rsmq_async::MultiplexedRsmq>(
&db,
tx,
"test-workspace",
payload,
Json(args),
hm_args.into(),
/* user */ "test-user",
/* email */ "test@windmill.dev",
/* permissioned_as */ "u/test-user".to_string(),
@@ -1070,7 +1073,7 @@ async fn test_deno_flow(db: Pool<Postgres>) {
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
},
}.into(),
stop_after_if: Default::default(),
summary: Default::default(),
suspend: Default::default(),
@@ -1108,7 +1111,7 @@ async fn test_deno_flow(db: Pool<Postgres>) {
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
},
}.into(),
stop_after_if: Default::default(),
summary: Default::default(),
suspend: Default::default(),
@@ -1121,7 +1124,7 @@ async fn test_deno_flow(db: Pool<Postgres>) {
delete_after_use: None,
continue_on_error: None,
}],
},
}.into(),
stop_after_if: Default::default(),
summary: Default::default(),
suspend: Default::default(),
@@ -1207,12 +1210,12 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
input_transforms: [
(
"loop".to_string(),
InputTransform::Static { value: json!(false) },
InputTransform::Static { value: windmill_common::worker::to_raw_value(&false) },
),
("i".to_string(), InputTransform::Static { value: json!(1) }),
("i".to_string(), InputTransform::Static { value: windmill_common::worker::to_raw_value(&1) }),
(
"path".to_string(),
InputTransform::Static { value: json!("outer.txt") },
InputTransform::Static { value: windmill_common::worker::to_raw_value(&"outer.txt") },
),
]
.into(),
@@ -1224,7 +1227,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
},
}.into(),
stop_after_if: Default::default(),
summary: Default::default(),
suspend: Default::default(),
@@ -1240,7 +1243,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
FlowModule {
id: "b".to_string(),
value: FlowModuleValue::ForloopFlow {
iterator: InputTransform::Static { value: json!([1, 2, 3]) },
iterator: InputTransform::Static { value: windmill_common::worker::to_raw_value(&[1, 2, 3]) },
skip_failures: false,
parallel: false,
parallelism: None,
@@ -1257,11 +1260,11 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
),
(
"loop".to_string(),
InputTransform::Static { value: json!(true) },
InputTransform::Static { value: windmill_common::worker::to_raw_value(&true) },
),
(
"path".to_string(),
InputTransform::Static { value: json!("inner.txt") },
InputTransform::Static { value: windmill_common::worker::to_raw_value(&"inner.txt") },
),
]
.into(),
@@ -1273,7 +1276,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
},
}.into(),
stop_after_if: Default::default(),
summary: Default::default(),
suspend: Default::default(),
@@ -1291,10 +1294,10 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
value: FlowModuleValue::RawScript {
input_transforms: [(
"path".to_string(),
InputTransform::Static { value: json!("inner.txt") },
InputTransform::Static { value: windmill_common::worker::to_raw_value(&"inner.txt") },
), (
"path2".to_string(),
InputTransform::Static { value: json!("outer.txt") },
InputTransform::Static { value: windmill_common::worker::to_raw_value(&"outer.txt") },
)]
.into(),
language: ScriptLang::Deno,
@@ -1308,7 +1311,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
},
}.into(),
stop_after_if: Default::default(),
summary: Default::default(),
suspend: Default::default(),
@@ -1323,7 +1326,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
},
],
},
}.into(),
stop_after_if: Default::default(),
summary: Default::default(),
suspend: Default::default(),
@@ -1346,11 +1349,11 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
),
(
"path".to_string(),
InputTransform::Static { value: json!("outer.txt") },
InputTransform::Static { value: windmill_common::worker::to_raw_value(&"outer.txt") },
),
(
"path2".to_string(),
InputTransform::Static { value: json!("inner.txt") },
InputTransform::Static { value: windmill_common::worker::to_raw_value(&"inner.txt") },
),
]
.into(),
@@ -1365,7 +1368,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
},
}.into(),
stop_after_if: Default::default(),
summary: Default::default(),
suspend: Default::default(),
+36 -42
View File
@@ -39,6 +39,7 @@ use windmill_common::{
http_get_from_hub, not_found_if_none, paginate, query_elems_from_hub, Pagination, StripPath,
},
variables::build_crypt,
worker::to_raw_value,
HUB_BASE_URL,
};
@@ -94,37 +95,37 @@ pub struct ListableApp {
pub struct AppVersion {
pub id: i64,
pub app_id: Uuid,
pub value: serde_json::Value,
pub value: sqlx::types::Json<Box<RawValue>>,
pub created_by: String,
pub created_at: chrono::DateTime<chrono::Utc>,
}
#[derive(Serialize, Deserialize)]
#[derive(Serialize, Deserialize, FromRow)]
pub struct AppWithLastVersion {
pub id: i64,
pub path: String,
pub summary: String,
pub policy: serde_json::Value,
pub versions: Vec<i64>,
pub value: serde_json::Value,
pub value: sqlx::types::Json<Box<RawValue>>,
pub created_by: String,
pub created_at: chrono::DateTime<chrono::Utc>,
pub extra_perms: serde_json::Value,
}
#[derive(Serialize, Deserialize)]
#[derive(Serialize, Deserialize, FromRow)]
pub struct AppWithLastVersionAndDraft {
pub id: i64,
pub path: String,
pub summary: String,
pub policy: serde_json::Value,
pub versions: Vec<i64>,
pub value: serde_json::Value,
pub value: sqlx::types::Json<Box<RawValue>>,
pub created_by: String,
pub created_at: chrono::DateTime<chrono::Utc>,
pub extra_perms: serde_json::Value,
#[serde(skip_serializing_if = "Option::is_none")]
pub draft: Option<serde_json::Value>,
pub draft: Option<sqlx::types::Json<Box<RawValue>>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub draft_only: Option<bool>,
}
@@ -178,7 +179,7 @@ pub struct Policy {
pub struct CreateApp {
pub path: String,
pub summary: String,
pub value: serde_json::Value,
pub value: sqlx::types::Json<Box<RawValue>>,
pub policy: Policy,
pub draft_only: Option<bool>,
pub deployment_message: Option<String>,
@@ -188,7 +189,7 @@ pub struct CreateApp {
pub struct EditApp {
pub path: Option<String>,
pub summary: Option<String>,
pub value: Option<serde_json::Value>,
pub value: Option<sqlx::types::Json<Box<RawValue>>>,
pub policy: Option<Policy>,
pub deployment_message: Option<String>,
}
@@ -196,7 +197,7 @@ pub struct EditApp {
#[derive(Serialize, FromRow)]
pub struct SearchApp {
path: String,
value: serde_json::Value,
value: sqlx::types::Json<Box<RawValue>>,
}
async fn list_search_apps(
authed: ApiAuthed,
@@ -210,12 +211,11 @@ async fn list_search_apps(
let n = 3;
let mut tx = user_db.begin(&authed).await?;
let rows = sqlx::query_as!(
SearchApp,
let rows = sqlx::query_as::<_, SearchApp>(
"SELECT path, app_version.value from app LEFT JOIN app_version ON app_version.id = versions[array_upper(versions, 1)] WHERE workspace_id = $1 LIMIT $2",
&w_id,
n
)
.bind(&w_id)
.bind(n)
.fetch_all(&mut *tx)
.await?
.into_iter()
@@ -293,15 +293,14 @@ async fn get_app(
let path = path.to_path();
let mut tx = user_db.begin(&authed).await?;
let app_o = sqlx::query_as!(
AppWithLastVersion,
let app_o = sqlx::query_as::<_, AppWithLastVersion>(
"SELECT app.id, app.path, app.summary, app.versions, app.policy,
app.extra_perms, app_version.value,
app_version.created_at, app_version.created_by from app, app_version
WHERE app.path = $1 AND app.workspace_id = $2 AND app_version.id = app.versions[array_upper(app.versions, 1)]",
path.to_owned(),
&w_id
)
.bind(path.to_owned())
.bind(&w_id)
.fetch_optional(&mut *tx)
.await?;
tx.commit().await?;
@@ -318,21 +317,20 @@ async fn get_app_w_draft(
let path = path.to_path();
let mut tx = user_db.begin(&authed).await?;
let app_o = sqlx::query_as!(
AppWithLastVersionAndDraft,
let app_o = sqlx::query_as::<_, AppWithLastVersionAndDraft>(
r#"SELECT app.id, app.path, app.summary, app.versions, app.policy,
app.extra_perms, app_version.value,
app_version.created_at, app_version.created_by,
app.draft_only, draft.value as "draft?"
app.draft_only, draft.value as "draft"
from app
INNER JOIN app_version ON
app_version.id = app.versions[array_upper(app.versions, 1)]
LEFT JOIN draft ON
app.path = draft.path AND draft.workspace_id = $2 AND draft.typ = 'app'
WHERE app.path = $1 AND app.workspace_id = $2"#,
path.to_owned(),
&w_id
)
.bind(path.to_owned())
.bind(&w_id)
.fetch_optional(&mut *tx)
.await?;
tx.commit().await?;
@@ -406,15 +404,14 @@ async fn get_app_by_id(
) -> JsonResult<AppWithLastVersion> {
let mut tx = user_db.begin(&authed).await?;
let app_o = sqlx::query_as!(
AppWithLastVersion,
let app_o = sqlx::query_as::<_, AppWithLastVersion>(
"SELECT app.id, app.path, app.summary, app.versions, app.policy,
app.extra_perms, app_version.value,
app_version.created_at, app_version.created_by from app, app_version
WHERE app_version.id = $1 AND app.id = app_version.app_id AND app.workspace_id = $2",
id,
&w_id
)
.bind(&id)
.bind(&w_id)
.fetch_optional(&mut *tx)
.await?;
tx.commit().await?;
@@ -438,15 +435,13 @@ async fn get_public_app_by_secret(
let id: i64 = bytes.parse().map_err(to_anyhow)?;
let app_o = sqlx::query_as!(
AppWithLastVersion,
let app_o = sqlx::query_as::<_, AppWithLastVersion>(
"SELECT app.id, app.path, app.summary, app.versions, app.policy,
app.extra_perms, app_version.value,
app_version.created_at, app_version.created_by from app, app_version
WHERE app.id = $1 AND app.workspace_id = $2 AND app_version.id = app.versions[array_upper(app.versions, 1)]",
id,
&w_id
)
WHERE app.id = $1 AND app.workspace_id = $2 AND app_version.id = app.versions[array_upper(app.versions, 1)]")
.bind(&id)
.bind(&w_id)
.fetch_optional(&mut *tx)
.await?;
tx.commit().await?;
@@ -599,9 +594,9 @@ async fn create_app(
)
.await?;
let mut args: HashMap<String, serde_json::Value> = HashMap::new();
let mut args: HashMap<String, Box<serde_json::value::RawValue>> = HashMap::new();
if let Some(dm) = app.deployment_message {
args.insert("deployment_message".to_string(), json!(dm));
args.insert("deployment_message".to_string(), to_raw_value(&dm));
}
let tx = PushIsolationLevel::Transaction(tx);
@@ -610,7 +605,7 @@ async fn create_app(
tx,
&w_id,
JobPayload::AppDependencies { path: app.path.clone(), version: v_id },
args,
PushArgs { args, extra: HashMap::new() },
&authed.username,
&authed.email,
windmill_common::users::username_to_permissioned_as(&authed.username),
@@ -889,18 +884,18 @@ async fn update_app(
let tx: PushIsolationLevel<'_, rsmq_async::MultiplexedRsmq> =
PushIsolationLevel::Transaction(tx);
let mut args: HashMap<String, serde_json::Value> = HashMap::new();
let mut args: HashMap<String, Box<serde_json::value::RawValue>> = HashMap::new();
if let Some(dm) = ns.deployment_message {
args.insert("deployment_message".to_string(), json!(dm));
args.insert("deployment_message".to_string(), to_raw_value(&dm));
}
args.insert("parent_path".to_string(), json!(path));
args.insert("parent_path".to_string(), to_raw_value(&path));
let (dependency_job_uuid, new_tx) = push(
&db,
tx,
&w_id,
JobPayload::AppDependencies { path: npath.clone(), version: v_id },
args,
PushArgs { args, extra: HashMap::new() },
&authed.username,
&authed.email,
windmill_common::users::username_to_permissioned_as(&authed.username),
@@ -1160,8 +1155,7 @@ fn build_args(
component: &str,
path: String,
args: HashMap<String, Box<RawValue>>,
) -> Result<PushArgs<HashMap<String, Box<RawValue>>>> {
) -> Result<PushArgs> {
let key = format!("{}:{}", component, &path);
let (static_inputs, one_of_inputs) = match policy {
Policy { triggerables_v2: Some(t), .. } => {
@@ -1276,5 +1270,5 @@ fn build_args(
for (k, v) in static_inputs {
extra.insert(k.to_string(), v.to_owned());
}
Ok(PushArgs { extra, args: sqlx::types::Json(safe_args) })
Ok(PushArgs { extra, args: safe_args })
}
+11 -10
View File
@@ -6,15 +6,12 @@
* LICENSE-AGPL for a copy of the license.
*/
use std::collections::HashMap;
use axum::{
extract::{Extension, Path},
routing::{get, post, put},
Router,
};
use hyper::StatusCode;
use serde_json::value::RawValue;
use sqlx::types::Json;
use windmill_common::{
db::UserDB,
@@ -89,7 +86,7 @@ pub async fn new_payload(
pub async fn update_payload(
Extension(db): Extension<DB>,
Path((w_id, path)): Path<(String, StripPath)>,
args: PushArgs<HashMap<String, Box<RawValue>>>,
args: PushArgs,
) -> Result<StatusCode> {
let mut tx = db.begin().await?;
@@ -102,7 +99,7 @@ pub async fn update_payload(
",
&w_id,
&path.to_path(),
Json(args) as Json<PushArgs<HashMap<String, Box<RawValue>>>>,
Json(args) as Json<PushArgs>,
)
.execute(&mut *tx)
.await?;
@@ -112,27 +109,31 @@ pub async fn update_payload(
Ok(StatusCode::NO_CONTENT)
}
#[derive(sqlx::FromRow)]
struct Payload {
payload: sqlx::types::Json<Box<serde_json::value::RawValue>>,
}
pub async fn get_payload(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Path((w_id, path)): Path<(String, StripPath)>,
) -> JsonResult<serde_json::Value> {
) -> JsonResult<Box<serde_json::value::RawValue>> {
let mut tx = user_db.begin(&authed).await?;
let payload = sqlx::query_scalar!(
let payload = sqlx::query_as::<_, Payload>(
"
SELECT payload
FROM capture
WHERE workspace_id = $1
AND path = $2
",
&w_id,
&path.to_path(),
)
.bind(&w_id)
.bind(&path.to_path())
.fetch_optional(&mut *tx)
.await?;
tx.commit().await?;
not_found_if_none(payload, "capture", path.to_path()).map(axum::Json)
not_found_if_none(payload.map(|x| x.payload.0), "capture", path.to_path()).map(axum::Json)
}
+1 -1
View File
@@ -38,7 +38,7 @@ pub enum DraftType {
#[derive(Deserialize, Serialize, Debug)]
pub struct Draft {
pub path: String,
pub value: serde_json::Value,
pub value: sqlx::types::Json<Box<serde_json::value::Value>>,
pub typ: DraftType,
}
+38 -37
View File
@@ -25,12 +25,12 @@ use axum::{
use hyper::StatusCode;
use serde::{Deserialize, Serialize};
use serde_json::json;
use sql_builder::prelude::*;
use sqlx::{FromRow, Postgres, Transaction};
use windmill_audit::audit_ee::audit_log;
use windmill_audit::ActionKind;
use windmill_common::utils::query_elems_from_hub;
use windmill_common::worker::to_raw_value;
use windmill_common::HUB_BASE_URL;
use windmill_common::{
db::UserDB,
@@ -71,7 +71,7 @@ pub fn global_service() -> Router {
#[derive(Serialize, FromRow)]
pub struct SearchFlow {
path: String,
value: serde_json::Value,
value: sqlx::types::Json<Box<serde_json::value::RawValue>>,
}
async fn list_search_flows(
authed: ApiAuthed,
@@ -85,12 +85,11 @@ async fn list_search_flows(
let n = 3;
let mut tx = user_db.begin(&authed).await?;
let rows = sqlx::query_as!(
SearchFlow,
let rows = sqlx::query_as::<_, SearchFlow>(
"SELECT path, value from flow WHERE workspace_id = $1 LIMIT $2",
&w_id,
n
)
.bind(&w_id)
.bind(n)
.fetch_all(&mut *tx)
.await?
.into_iter()
@@ -363,9 +362,9 @@ async fn create_flow(
)
.await?;
let mut args: HashMap<String, serde_json::Value> = HashMap::new();
let mut args: HashMap<String, Box<serde_json::value::RawValue>> = HashMap::new();
if let Some(dm) = nf.deployment_message {
args.insert("deployment_message".to_string(), json!(dm));
args.insert("deployment_message".to_string(), to_raw_value(&dm));
}
let tx = PushIsolationLevel::Transaction(tx);
@@ -377,7 +376,7 @@ async fn create_flow(
path: nf.path.clone(),
dedicated_worker: nf.dedicated_worker,
},
args,
args.into(),
&authed.username,
&authed.email,
windmill_common::users::username_to_permissioned_as(&authed.username),
@@ -515,22 +514,19 @@ async fn update_flow(
}
}
let mut schedulables: Vec<Schedule> = sqlx::query_as!(
Schedule,
"UPDATE schedule SET script_path = $1 WHERE script_path = $2 AND path != $2 AND workspace_id = $3 AND is_flow IS true RETURNING *",
nf.path,
flow_path,
w_id,
)
let mut schedulables: Vec<Schedule> = sqlx::query_as::<_, Schedule>(
"UPDATE schedule SET script_path = $1 WHERE script_path = $2 AND path != $2 AND workspace_id = $3 AND is_flow IS true RETURNING *")
.bind(&nf.path)
.bind(&flow_path)
.bind(&w_id)
.fetch_all(&mut tx)
.await?;
let schedule = sqlx::query_as!(Schedule,
"UPDATE schedule SET path = $1, script_path = $1 WHERE path = $2 AND workspace_id = $3 AND is_flow IS true RETURNING *",
nf.path,
flow_path,
w_id,
)
let schedule = sqlx::query_as::<_, Schedule>(
"UPDATE schedule SET path = $1, script_path = $1 WHERE path = $2 AND workspace_id = $3 AND is_flow IS true RETURNING *")
.bind(&nf.path)
.bind(&flow_path)
.bind(&w_id)
.fetch_optional(&mut tx)
.await?;
@@ -582,11 +578,11 @@ async fn update_flow(
let tx = PushIsolationLevel::Transaction(tx);
let mut args: HashMap<String, serde_json::Value> = HashMap::new();
let mut args: HashMap<String, Box<serde_json::value::RawValue>> = HashMap::new();
if let Some(dm) = nf.deployment_message {
args.insert("deployment_message".to_string(), json!(dm));
args.insert("deployment_message".to_string(), to_raw_value(&dm));
}
args.insert("parent_path".to_string(), json!(flow_path));
args.insert("parent_path".to_string(), to_raw_value(&flow_path));
let (dependency_job_uuid, mut new_tx) = push(
&db,
@@ -596,7 +592,7 @@ async fn update_flow(
path: nf.path.clone(),
dedicated_worker: nf.dedicated_worker,
},
args,
windmill_queue::PushArgs { args, extra: HashMap::new() },
&authed.username,
&authed.email,
windmill_common::users::username_to_permissioned_as(&authed.username),
@@ -663,10 +659,10 @@ pub struct FlowWDraft {
pub summary: String,
pub description: String,
pub schema: Option<Schema>,
pub value: serde_json::Value,
pub value: sqlx::types::Json<Box<serde_json::value::RawValue>>,
pub extra_perms: serde_json::Value,
#[serde(skip_serializing_if = "Option::is_none")]
pub draft: Option<serde_json::Value>,
pub draft: Option<sqlx::types::Json<Box<serde_json::value::RawValue>>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub draft_only: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
@@ -881,16 +877,18 @@ mod tests {
modules: vec![
FlowModule {
id: "a".to_string(),
value: FlowModuleValue::Script {
value: windmill_common::worker::to_raw_value(&FlowModuleValue::Script {
path: "test".to_string(),
input_transforms: [(
"test".to_string(),
InputTransform::Static { value: serde_json::json!("test2") },
InputTransform::Static {
value: windmill_common::worker::to_raw_value(&"test2".to_string()),
},
)]
.into(),
hash: None,
tag_override: None,
},
}),
stop_after_if: None,
summary: None,
suspend: Default::default(),
@@ -905,7 +903,7 @@ mod tests {
},
FlowModule {
id: "b".to_string(),
value: FlowModuleValue::RawScript {
value: windmill_common::worker::to_raw_value(&FlowModuleValue::RawScript {
input_transforms: HashMap::new(),
content: "test".to_string(),
language: scripts::ScriptLang::Deno,
@@ -915,7 +913,7 @@ mod tests {
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
},
}),
stop_after_if: Some(StopAfterIf {
expr: "foo = 'bar'".to_string(),
skip_if_stopped: false,
@@ -933,13 +931,15 @@ mod tests {
},
FlowModule {
id: "c".to_string(),
value: FlowModuleValue::ForloopFlow {
iterator: InputTransform::Static { value: serde_json::json!([1, 2, 3]) },
value: windmill_common::worker::to_raw_value(&FlowModuleValue::ForloopFlow {
iterator: InputTransform::Static {
value: windmill_common::worker::to_raw_value(&[1, 2, 3]),
},
modules: vec![],
skip_failures: true,
parallel: false,
parallelism: None,
},
}),
stop_after_if: Some(StopAfterIf {
expr: "previous.isEmpty()".to_string(),
skip_if_stopped: false,
@@ -963,7 +963,8 @@ mod tests {
input_transforms: HashMap::new(),
hash: None,
tag_override: None,
},
}
.into(),
stop_after_if: Some(StopAfterIf {
expr: "previous.isEmpty()".to_string(),
skip_if_stopped: false,
+9 -7
View File
@@ -47,7 +47,7 @@ pub struct InputRow {
pub runnable_id: String,
pub runnable_type: RunnableType,
pub name: String,
pub args: Value,
pub args: sqlx::types::Json<Box<serde_json::value::RawValue>>,
pub created_at: DateTime<Utc>,
pub created_by: String,
pub is_public: bool,
@@ -95,12 +95,12 @@ pub struct RunnableParams {
pub runnable_type: RunnableType,
}
#[derive(Debug, Serialize, Deserialize)]
#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
pub struct Input {
id: Uuid,
name: String,
created_at: chrono::DateTime<chrono::Utc>,
args: serde_json::Value,
args: sqlx::types::Json<Box<serde_json::value::RawValue>>,
created_by: String,
is_public: bool,
success: bool,
@@ -110,7 +110,7 @@ pub struct Input {
pub struct CompletedJobMini {
id: Uuid,
created_at: chrono::DateTime<chrono::Utc>,
args: Option<serde_json::Value>,
args: Option<sqlx::types::Json<Box<serde_json::value::RawValue>>>,
created_by: String,
success: bool,
}
@@ -161,7 +161,9 @@ async fn get_input_history(
row.created_by
),
created_at: row.created_at,
args: row.args.unwrap_or(serde_json::json!({})),
args: row.args.unwrap_or(sqlx::types::Json(
serde_json::value::RawValue::from_string("null".to_string()).unwrap(),
)),
created_by: row.created_by,
is_public: true,
success: row.success,
@@ -240,7 +242,7 @@ async fn list_saved_inputs(
#[derive(Debug, Serialize, Deserialize)]
pub struct CreateInput {
name: String,
args: serde_json::Value,
args: Box<serde_json::value::RawValue>,
}
async fn create_input(
@@ -262,7 +264,7 @@ async fn create_input(
.bind(&r.runnable_id)
.bind(&r.runnable_type)
.bind(&input.name)
.bind(&input.args)
.bind(sqlx::types::Json(&input.args))
.bind(&authed.username)
.execute(&mut *tx)
.await?;
+33 -25
View File
@@ -903,7 +903,11 @@ pub struct ListQueueQuery {
pub is_not_schedule: Option<bool>,
}
pub fn filter_list_queue_query(mut sqlb: SqlBuilder, lq: &ListQueueQuery, w_id: &str) -> SqlBuilder {
pub fn filter_list_queue_query(
mut sqlb: SqlBuilder,
lq: &ListQueueQuery,
w_id: &str,
) -> SqlBuilder {
if w_id != "admins" || !lq.all_workspaces.is_some_and(|x| x) {
sqlb.and_where_eq("workspace_id", "?".bind(&w_id));
}
@@ -1004,7 +1008,6 @@ pub fn filter_list_queue_query(mut sqlb: SqlBuilder, lq: &ListQueueQuery, w_id:
}
sqlb
}
pub fn list_queue_jobs_query(w_id: &str, lq: &ListQueueQuery, fields: &[&str]) -> SqlBuilder {
@@ -2054,7 +2057,10 @@ impl Job {
)
}
pub async fn concurrency_key(&self, db: &Pool<Postgres>) -> Result<Option<String>, sqlx::Error> {
pub async fn concurrency_key(
&self,
db: &Pool<Postgres>,
) -> Result<Option<String>, sqlx::Error> {
sqlx::query_scalar!(
"SELECT key FROM concurrency_key WHERE job_id = $1",
self.id()
@@ -2202,7 +2208,7 @@ struct Preview {
content: Option<String>,
kind: Option<PreviewKind>,
path: Option<String>,
args: Option<Box<JsonRawValue>>,
args: Option<HashMap<String, Box<JsonRawValue>>>,
language: Option<ScriptLang>,
tag: Option<String>,
dedicated_worker: Option<bool>,
@@ -2218,7 +2224,7 @@ pub struct WorkflowTask {
struct PreviewFlow {
value: FlowValue,
path: Option<String>,
args: Option<Box<JsonRawValue>>,
args: Option<HashMap<String, Box<JsonRawValue>>>,
tag: Option<String>,
restarted_from: Option<RestartedFrom>,
}
@@ -2331,7 +2337,7 @@ pub async fn run_flow_by_path(
Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
Path((w_id, flow_path)): Path<(String, StripPath)>,
Query(run_query): Query<RunJobQuery>,
args: PushArgs<HashMap<String, Box<JsonRawValue>>>,
args: PushArgs,
) -> error::Result<(StatusCode, String)> {
#[cfg(feature = "enterprise")]
check_license_key_valid().await?;
@@ -2432,7 +2438,8 @@ pub async fn restart_flow(
let push_args = completed_job
.args
.map(|json| PushArgs { args: json.clone(), extra: json.0 });
.map(|json| PushArgs { args: json.0, extra: HashMap::new() })
.unwrap_or_else(PushArgs::empty);
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);
@@ -2476,7 +2483,7 @@ pub async fn run_script_by_path(
Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
Path((w_id, script_path)): Path<(String, StripPath)>,
Query(run_query): Query<RunJobQuery>,
args: PushArgs<HashMap<String, Box<JsonRawValue>>>,
args: PushArgs,
) -> error::Result<(StatusCode, String)> {
#[cfg(feature = "enterprise")]
check_license_key_valid().await?;
@@ -2563,7 +2570,7 @@ pub async fn run_workflow_as_code(
let mut extra = HashMap::new();
extra.insert(ENTRYPOINT_OVERRIDE.to_string(), to_raw_value(&entrypoint));
let args = PushArgs { args: sqlx::types::Json(task.args), extra };
let args = PushArgs { args: task.args.unwrap_or_else(HashMap::new), extra };
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let tag = run_query.tag.clone().or(tag).or(Some(job.tag));
@@ -2854,7 +2861,7 @@ pub async fn run_wait_result_job_by_path_get(
});
let inner_args: HashMap<String, Box<RawValue>> = HashMap::new();
let args = PushArgs { extra: payload_args, args: sqlx::types::Json(inner_args) };
let args = PushArgs { extra: payload_args, args: inner_args };
check_queue_too_long(&db, QUEUE_LIMIT_WAIT_RESULT.or(run_query.queue_limit)).await?;
let script_path = script_path.to_path();
@@ -2932,7 +2939,7 @@ pub async fn run_wait_result_flow_by_path_get(
payload_args.insert(k.to_string(), v.clone());
});
let args = PushArgs { extra: payload_args, args: sqlx::types::Json(HashMap::new()) };
let args = PushArgs { extra: payload_args, args: HashMap::new() };
run_wait_result_flow_by_path_internal(
db, run_query, flow_path, authed, rsmq, user_db, args, w_id,
@@ -2947,7 +2954,7 @@ pub async fn run_wait_result_script_by_path(
Extension(db): Extension<DB>,
Path((w_id, script_path)): Path<(String, StripPath)>,
Query(run_query): Query<RunJobQuery>,
args: PushArgs<HashMap<String, Box<JsonRawValue>>>,
args: PushArgs,
) -> error::Result<Response> {
#[cfg(feature = "enterprise")]
check_license_key_valid().await?;
@@ -2973,7 +2980,7 @@ async fn run_wait_result_script_by_path_internal(
rsmq: Option<rsmq_async::MultiplexedRsmq>,
user_db: UserDB,
w_id: String,
args: PushArgs<HashMap<String, Box<JsonRawValue>>>,
args: PushArgs,
) -> error::Result<Response> {
check_queue_too_long(&db, QUEUE_LIMIT_WAIT_RESULT.or(run_query.queue_limit)).await?;
let script_path = script_path.to_path();
@@ -3026,7 +3033,7 @@ pub async fn run_wait_result_script_by_hash(
Extension(db): Extension<DB>,
Path((w_id, script_hash)): Path<(String, ScriptHash)>,
Query(run_query): Query<RunJobQuery>,
args: PushArgs<HashMap<String, Box<JsonRawValue>>>,
args: PushArgs,
) -> error::Result<Response> {
#[cfg(feature = "enterprise")]
check_license_key_valid().await?;
@@ -3106,7 +3113,7 @@ pub async fn run_wait_result_flow_by_path(
Extension(db): Extension<DB>,
Path((w_id, flow_path)): Path<(String, StripPath)>,
Query(run_query): Query<RunJobQuery>,
args: PushArgs<HashMap<String, Box<JsonRawValue>>>,
args: PushArgs,
) -> error::Result<Response> {
#[cfg(feature = "enterprise")]
check_license_key_valid().await?;
@@ -3124,7 +3131,7 @@ async fn run_wait_result_flow_by_path_internal(
authed: ApiAuthed,
rsmq: Option<rsmq_async::MultiplexedRsmq>,
user_db: UserDB,
args: PushArgs<HashMap<String, Box<JsonRawValue>>>,
args: PushArgs,
w_id: String,
) -> error::Result<Response> {
check_queue_too_long(&db, run_query.queue_limit).await?;
@@ -3220,7 +3227,7 @@ async fn run_preview_script(
dedicated_worker: preview.dedicated_worker,
}),
},
preview.args.unwrap_or_default(),
preview.args.unwrap_or_default().into(),
authed.display_username(),
&authed.email,
username_to_permissioned_as(&authed.username),
@@ -3301,7 +3308,7 @@ async fn run_bundle_preview_script(
custom_concurrency_key: None,
}),
},
args,
args.into(),
authed.display_username(),
&authed.email,
username_to_permissioned_as(&authed.username),
@@ -3421,10 +3428,7 @@ pub async fn run_dependencies_job(
"raw_deps".to_string(),
JsonRawValue::from_string("true".to_string()).unwrap(),
);
(
PushArgs { extra: hm, args: sqlx::types::Json(HashMap::new()) },
deps,
)
(PushArgs { extra: hm, args: HashMap::new() }, deps)
} else {
(
PushArgs::empty(),
@@ -3668,7 +3672,7 @@ async fn run_preview_flow_job(
path: raw_flow.path,
restarted_from: raw_flow.restarted_from,
},
raw_flow.args.unwrap_or_default(),
raw_flow.args.unwrap_or_default().into(),
authed.display_username(),
&authed.email,
username_to_permissioned_as(&authed.username),
@@ -3700,7 +3704,7 @@ pub async fn run_job_by_hash(
Extension(rsmq): Extension<Option<rsmq_async::MultiplexedRsmq>>,
Path((w_id, script_hash)): Path<(String, ScriptHash)>,
Query(run_query): Query<RunJobQuery>,
args: PushArgs<HashMap<String, Box<JsonRawValue>>>,
args: PushArgs,
) -> error::Result<(StatusCode, String)> {
#[cfg(feature = "enterprise")]
check_license_key_valid().await?;
@@ -3895,7 +3899,11 @@ async fn get_job_update(
}
}
pub fn filter_list_completed_query(mut sqlb: SqlBuilder, lq: &ListCompletedQuery, w_id: &str) -> SqlBuilder {
pub fn filter_list_completed_query(
mut sqlb: SqlBuilder,
lq: &ListCompletedQuery,
w_id: &str,
) -> SqlBuilder {
if w_id != "admins" || !lq.all_workspaces.is_some_and(|x| x) {
sqlb.and_where_eq("workspace_id", "?".bind(&w_id));
}
+49 -55
View File
@@ -153,38 +153,36 @@ async fn create_schedule(
)
.await?;
let schedule = sqlx::query_as!(
Schedule,
let schedule = sqlx::query_as::<_, Schedule>(
"INSERT INTO schedule (workspace_id, path, schedule, timezone, edited_by, script_path, \
is_flow, args, enabled, email, on_failure, on_failure_times, on_failure_exact, \
on_failure_extra_args, on_recovery, on_recovery_times, on_recovery_extra_args, \
ws_error_handler_muted, retry, summary, no_flow_overlap, tag \
) VALUES ( \
$1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22 \
) RETURNING *",
w_id,
ns.path,
ns.schedule,
ns.timezone,
&authed.username,
ns.script_path,
ns.is_flow,
ns.args,
ns.enabled.unwrap_or(false),
&authed.email,
ns.on_failure,
ns.on_failure_times,
ns.on_failure_exact,
ns.on_failure_extra_args,
ns.on_recovery,
ns.on_recovery_times,
ns.on_recovery_extra_args,
ns.ws_error_handler_muted.unwrap_or(false),
ns.retry,
ns.summary,
ns.no_flow_overlap.unwrap_or(false),
ns.tag,
)
) RETURNING *")
.bind(&w_id)
.bind(&ns.path)
.bind(&ns.schedule)
.bind(&ns.timezone)
.bind(&authed.username)
.bind(&ns.script_path)
.bind(&ns.is_flow)
.bind(&ns.args)
.bind(&ns.enabled.unwrap_or(false))
.bind(&authed.email)
.bind(&ns.on_failure)
.bind(&ns.on_failure_times)
.bind(&ns.on_failure_exact)
.bind(&ns.on_failure_extra_args)
.bind(&ns.on_recovery)
.bind(&ns.on_recovery_times)
.bind(&ns.on_recovery_extra_args)
.bind(&ns.ws_error_handler_muted.unwrap_or(false))
.bind(&ns.retry)
.bind(&ns.summary)
.bind(&ns.no_flow_overlap.unwrap_or(false))
.bind(&ns.tag)
.fetch_one(&mut tx)
.await
.map_err(|e| Error::InternalErr(format!("inserting schedule in {w_id}: {e}")))?;
@@ -245,31 +243,29 @@ async fn edit_schedule(
cron::Schedule::from_str(&es.schedule).map_err(|e| Error::BadRequest(e.to_string()))?;
clear_schedule(tx.transaction_mut(), path, &w_id).await?;
let schedule = sqlx::query_as!(
Schedule,
let schedule = sqlx::query_as::<_, Schedule>(
"UPDATE schedule SET schedule = $1, timezone = $2, args = $3, on_failure = $4, on_failure_times = $5, \
on_failure_exact = $6, on_failure_extra_args = $7, on_recovery = $8, on_recovery_times = $9, \
on_recovery_extra_args = $10, ws_error_handler_muted = $11, retry = $12, summary = $13, \
no_flow_overlap = $14, tag = $15
WHERE path = $16 AND workspace_id = $17 RETURNING *",
es.schedule,
es.timezone,
es.args,
es.on_failure,
es.on_failure_times,
es.on_failure_exact,
es.on_failure_extra_args,
es.on_recovery,
es.on_recovery_times,
es.on_recovery_extra_args,
es.ws_error_handler_muted.unwrap_or(false),
es.retry,
es.summary,
es.no_flow_overlap.unwrap_or(false),
es.tag,
path,
w_id,
)
WHERE path = $16 AND workspace_id = $17 RETURNING *")
.bind(&es.schedule)
.bind(&es.timezone)
.bind(&es.args)
.bind(&es.on_failure)
.bind(&es.on_failure_times)
.bind(&es.on_failure_exact)
.bind(&es.on_failure_extra_args)
.bind(&es.on_recovery)
.bind(&es.on_recovery_times)
.bind(&es.on_recovery_extra_args)
.bind(&es.ws_error_handler_muted.unwrap_or(false))
.bind(&es.retry)
.bind(&es.summary)
.bind(&es.no_flow_overlap.unwrap_or(false))
.bind(&es.tag)
.bind(&path)
.bind(&w_id)
.fetch_one(&mut tx)
.await
.map_err(|e| Error::InternalErr(format!("updating schedule in {w_id}: {e}")))?;
@@ -473,14 +469,12 @@ pub async fn set_enabled(
let mut tx: QueueTransaction<'_, rsmq_async::MultiplexedRsmq> =
(rsmq.clone(), user_db.begin(&authed).await?).into();
let path = path.to_path();
let schedule_o = sqlx::query_as!(
Schedule,
"UPDATE schedule SET enabled = $1, email = $2 WHERE path = $3 AND workspace_id = $4 RETURNING *",
&payload.enabled,
authed.email,
path,
w_id
)
let schedule_o = sqlx::query_as::<_, Schedule>(
"UPDATE schedule SET enabled = $1, email = $2 WHERE path = $3 AND workspace_id = $4 RETURNING *")
.bind(&payload.enabled)
.bind(&authed.email)
.bind(&path)
.bind(&w_id)
.fetch_optional(&mut tx)
.await?;
+17 -18
View File
@@ -25,6 +25,7 @@ use axum::{
use hyper::StatusCode;
use serde::{Deserialize, Serialize};
use serde_json::json;
use serde_json::value::RawValue;
use sql_builder::prelude::*;
use sqlx::{FromRow, Postgres, Transaction};
use std::{
@@ -51,6 +52,7 @@ use windmill_common::{
utils::{
not_found_if_none, paginate, query_elems_from_hub, require_admin, Pagination, StripPath,
},
worker::to_raw_value,
HUB_BASE_URL,
};
use windmill_git_sync::{handle_deployment_metadata, DeployedObject};
@@ -70,7 +72,7 @@ pub struct ScriptWDraft {
pub kind: ScriptKind,
pub tag: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub draft: Option<serde_json::Value>,
pub draft: Option<sqlx::types::Json<Box<RawValue>>>,
pub schema: Option<Schema>,
#[serde(skip_serializing_if = "Option::is_none")]
pub draft_only: Option<bool>,
@@ -622,22 +624,19 @@ async fn create_script_internal<'c>(
.execute(&mut tx)
.await?;
let mut schedulables = sqlx::query_as!(
Schedule,
"UPDATE schedule SET script_path = $1 WHERE script_path = $2 AND path != $2 AND workspace_id = $3 AND is_flow IS false RETURNING *",
ns.path,
p_path,
w_id,
)
let mut schedulables = sqlx::query_as::<_, Schedule>(
"UPDATE schedule SET script_path = $1 WHERE script_path = $2 AND path != $2 AND workspace_id = $3 AND is_flow IS false RETURNING *")
.bind(&ns.path)
.bind(&p_path)
.bind(&w_id)
.fetch_all(&mut tx)
.await?;
let schedule = sqlx::query_as!(Schedule,
"UPDATE schedule SET path = $1, script_path = $1 WHERE path = $2 AND workspace_id = $3 AND is_flow IS false RETURNING *",
ns.path,
p_path,
w_id,
)
let schedule = sqlx::query_as::<_, Schedule>(
"UPDATE schedule SET path = $1, script_path = $1 WHERE path = $2 AND workspace_id = $3 AND is_flow IS false RETURNING *")
.bind(&ns.path)
.bind(&p_path)
.bind(&w_id)
.fetch_optional(&mut tx)
.await?;
@@ -716,12 +715,12 @@ async fn create_script_internal<'c>(
ns.tag
};
let mut args: HashMap<String, serde_json::Value> = HashMap::new();
let mut args: HashMap<String, Box<serde_json::value::RawValue>> = HashMap::new();
if let Some(dm) = ns.deployment_message {
args.insert("deployment_message".to_string(), json!(dm));
args.insert("deployment_message".to_string(), to_raw_value(&dm));
}
if let Some(ref p_path) = p_path_opt {
args.insert("parent_path".to_string(), json!(p_path));
args.insert("parent_path".to_string(), to_raw_value(&p_path));
}
let tx = PushIsolationLevel::Transaction(tx);
@@ -735,7 +734,7 @@ async fn create_script_internal<'c>(
path: ns.path,
dedicated_worker: ns.dedicated_worker,
},
args,
args.into(),
&authed.username,
&authed.email,
permissioned_as,
+13 -15
View File
@@ -41,7 +41,7 @@ use windmill_common::s3_helpers::LargeFileStorage;
use windmill_common::schedule::Schedule;
use windmill_common::users::username_to_permissioned_as;
use windmill_common::variables::build_crypt;
use windmill_common::worker::CLOUD_HOSTED;
use windmill_common::worker::{to_raw_value, CLOUD_HOSTED};
use windmill_common::workspaces::WorkspaceGitSyncSettings;
use windmill_common::{
error::{to_anyhow, Error, JsonResult, Result},
@@ -58,7 +58,7 @@ use crate::oauth2_ee::InstanceEvent;
use crate::variables::{decrypt, encrypt};
use hyper::{header, StatusCode};
use serde::{Deserialize, Serialize};
use serde_json::{json, Map, Value};
use serde_json::Value;
use sqlx::{FromRow, Postgres, Transaction};
use tempfile::TempDir;
use tokio::fs::File;
@@ -476,15 +476,15 @@ async fn run_slack_message_test_job(
Path(w_id): Path<String>,
Json(req): Json<RunSlackMessageTestJobRequest>,
) -> JsonResult<RunSlackMessageTestJobResponse> {
let mut fake_result = Map::new();
fake_result.insert("error".to_string(), json!(req.test_msg));
fake_result.insert("success_result".to_string(), json!(req.test_msg));
let mut fake_result = HashMap::new();
fake_result.insert("error".to_string(), to_raw_value(&req.test_msg));
fake_result.insert("success_result".to_string(), to_raw_value(&req.test_msg));
let mut extra_args = Map::new();
extra_args.insert("channel".to_string(), json!(req.channel));
let mut extra_args = HashMap::new();
extra_args.insert("channel".to_string(), to_raw_value(&req.channel));
extra_args.insert(
"slack".to_string(),
json!(format!("$res:{WORKSPACE_SLACK_BOT_TOKEN_PATH}")),
to_raw_value(&format!("$res:{WORKSPACE_SLACK_BOT_TOKEN_PATH}")),
);
let uuid = windmill_queue::push_error_handler(
@@ -499,7 +499,7 @@ async fn run_slack_message_test_job(
sqlx::types::Json(&fake_result),
None,
Some(Utc::now()),
Some(json!(extra_args)),
Some(sqlx::types::Json(to_raw_value(&extra_args))),
authed.email.as_str(),
false,
false,
@@ -2588,14 +2588,13 @@ async fn tarball_workspace(
}
{
let apps = sqlx::query_as!(
AppWithLastVersion,
let apps = sqlx::query_as::<_, AppWithLastVersion>(
"SELECT app.id, app.path, app.summary, app.versions, app.policy,
app.extra_perms, app_version.value,
app_version.created_at, app_version.created_by from app, app_version
WHERE app.workspace_id = $1 AND app_version.id = app.versions[array_upper(app.versions, 1)]",
&w_id
)
.bind(&w_id)
.fetch_all(&mut *tx)
.await?;
@@ -2608,12 +2607,11 @@ async fn tarball_workspace(
}
if include_schedules.unwrap_or(false) {
let schedules = sqlx::query_as!(
Schedule,
let schedules = sqlx::query_as::<_, Schedule>(
"SELECT * FROM schedule
WHERE workspace_id = $1",
&w_id
)
.bind(&w_id)
.fetch_all(&mut *tx)
.await?;
+101 -5
View File
@@ -13,7 +13,6 @@ use serde::{Deserialize, Serialize};
use uuid::Uuid;
use crate::flows::FlowValue;
use crate::more_serde::default_false;
const MINUTES: Duration = Duration::from_secs(60);
const HOURS: Duration = MINUTES.saturating_mul(60);
@@ -71,7 +70,7 @@ pub struct RestartedFrom {
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct Iterator {
pub index: usize,
pub itered: Vec<serde_json::Value>,
pub itered: Vec<Box<serde_json::value::RawValue>>,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
@@ -111,7 +110,23 @@ pub struct FlowCleanupModule {
pub flow_jobs_to_clean: Vec<Uuid>,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
#[derive(Deserialize)]
struct UntaggedFlowStatusModule {
#[serde(rename = "type")]
type_: String,
id: Option<String>,
count: Option<u16>,
job: Option<Uuid>,
iterator: Option<Iterator>,
flow_jobs: Option<Vec<Uuid>>,
branch_chosen: Option<BranchChosen>,
branchall: Option<BranchAllStatus>,
parallel: Option<bool>,
while_loop: Option<bool>,
approvers: Option<Vec<Approval>>,
}
#[derive(Serialize, Debug, Clone)]
#[serde(tag = "type")]
pub enum FlowStatusModule {
WaitingForPriorSteps {
@@ -138,10 +153,8 @@ pub enum FlowStatusModule {
#[serde(skip_serializing_if = "Option::is_none")]
branchall: Option<BranchAllStatus>,
#[serde(skip_serializing_if = "std::ops::Not::not")]
#[serde(default = "default_false")]
parallel: bool,
#[serde(skip_serializing_if = "std::ops::Not::not")]
#[serde(default = "default_false")]
while_loop: bool,
},
Success {
@@ -165,6 +178,89 @@ pub enum FlowStatusModule {
},
}
impl<'de> Deserialize<'de> for FlowStatusModule {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let untagged: UntaggedFlowStatusModule =
UntaggedFlowStatusModule::deserialize(deserializer)?;
match untagged.type_.as_str() {
"WaitingForPriorSteps" => Ok(FlowStatusModule::WaitingForPriorSteps {
id: untagged
.id
.ok_or_else(|| serde::de::Error::missing_field("id"))?,
}),
"WaitingForEvents" => Ok(FlowStatusModule::WaitingForEvents {
id: untagged
.id
.ok_or_else(|| serde::de::Error::missing_field("id"))?,
count: untagged
.count
.ok_or_else(|| serde::de::Error::missing_field("count"))?,
job: untagged
.job
.ok_or_else(|| serde::de::Error::missing_field("job"))?,
}),
"WaitingForExecutor" => Ok(FlowStatusModule::WaitingForExecutor {
id: untagged
.id
.ok_or_else(|| serde::de::Error::missing_field("id"))?,
job: untagged
.job
.ok_or_else(|| serde::de::Error::missing_field("job"))?,
}),
"InProgress" => Ok(FlowStatusModule::InProgress {
id: untagged
.id
.ok_or_else(|| serde::de::Error::missing_field("id"))?,
job: untagged
.job
.ok_or_else(|| serde::de::Error::missing_field("job"))?,
iterator: untagged.iterator,
flow_jobs: untagged.flow_jobs,
branch_chosen: untagged.branch_chosen,
branchall: untagged.branchall,
parallel: untagged.parallel.unwrap_or(false),
while_loop: untagged.while_loop.unwrap_or(false),
}),
"Success" => Ok(FlowStatusModule::Success {
id: untagged
.id
.ok_or_else(|| serde::de::Error::missing_field("id"))?,
job: untagged
.job
.ok_or_else(|| serde::de::Error::missing_field("job"))?,
flow_jobs: untagged.flow_jobs,
branch_chosen: untagged.branch_chosen,
approvers: untagged.approvers.unwrap_or_default(),
}),
"Failure" => Ok(FlowStatusModule::Failure {
id: untagged
.id
.ok_or_else(|| serde::de::Error::missing_field("id"))?,
job: untagged
.job
.ok_or_else(|| serde::de::Error::missing_field("job"))?,
flow_jobs: untagged.flow_jobs,
branch_chosen: untagged.branch_chosen,
}),
other => Err(serde::de::Error::unknown_variant(
other,
&[
"WaitingForPriorSteps",
"WaitingForEvents",
"WaitingForExecutor",
"InProgress",
"Success",
"Failure",
],
)),
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum JobResult {
SingleJob(Uuid),
+196 -19
View File
@@ -16,9 +16,7 @@ use rand::Rng;
use serde::{Deserialize, Serialize, Serializer};
use crate::{
more_serde::{
default_empty_string, default_false, default_id, default_null, default_true, is_default,
},
more_serde::{default_empty_string, default_id, default_null, default_true, is_default},
scripts::{Schema, ScriptHash, ScriptLang},
};
@@ -220,7 +218,7 @@ pub struct Mock {
pub struct FlowModule {
#[serde(default = "default_id")]
pub id: String,
pub value: FlowModuleValue,
pub value: Box<serde_json::value::RawValue>,
#[serde(skip_serializing_if = "Option::is_none")]
pub stop_after_if: Option<StopAfterIf>,
#[serde(skip_serializing_if = "Option::is_none")]
@@ -246,13 +244,82 @@ pub struct FlowModule {
pub continue_on_error: Option<bool>,
}
#[derive(Deserialize)]
pub struct FlowModuleValueType {
#[serde(rename = "type")]
pub type_: String,
}
#[derive(Deserialize)]
pub struct FlowModuleValueWithParallel {
#[serde(rename = "type")]
pub type_: String,
pub parallel: Option<bool>,
pub parallelism: Option<u16>,
}
impl FlowModule {
pub fn id_append(&mut self, s: &str) {
self.id = format!("{}-{}", self.id, s);
}
pub fn get_value(&self) -> anyhow::Result<FlowModuleValue> {
serde_json::from_str::<FlowModuleValue>(self.value.get()).map_err(crate::error::to_anyhow)
}
pub fn is_flow(&self) -> bool {
self.get_type().is_ok_and(|x| x == "flow")
}
pub fn get_value_with_parallel(&self) -> anyhow::Result<FlowModuleValueWithParallel> {
serde_json::from_str::<FlowModuleValueWithParallel>(self.value.get())
.map_err(crate::error::to_anyhow)
}
pub fn is_simple(&self) -> bool {
self.get_type()
.is_ok_and(|x| x == "flow" || x == "script" || x == "rawscript")
}
pub fn get_type(&self) -> anyhow::Result<String> {
serde_json::from_str::<FlowModuleValueType>(self.value.get())
.map_err(crate::error::to_anyhow)
.map(|x| x.type_)
}
}
#[derive(Deserialize, Serialize, Debug, Clone)]
#[derive(Deserialize)]
pub struct UntaggedInputTransform {
#[serde(rename = "type")]
pub type_: String,
pub value: Option<Box<serde_json::value::RawValue>>,
pub expr: Option<String>,
}
impl<'de> Deserialize<'de> for InputTransform {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let untagged: UntaggedInputTransform = UntaggedInputTransform::deserialize(deserializer)?;
match untagged.type_.as_str() {
"static" => {
let value = untagged.value.unwrap_or_else(default_null);
Ok(InputTransform::Static { value })
}
"javascript" => {
let expr = untagged.expr.unwrap_or_else(default_empty_string);
Ok(InputTransform::Javascript { expr })
}
other => Err(serde::de::Error::unknown_variant(
other,
&["static", "javascript"],
)),
}
}
}
#[derive(Serialize, Debug, Clone)]
#[serde(
tag = "type",
rename_all(serialize = "lowercase", deserialize = "lowercase")
@@ -260,7 +327,7 @@ impl FlowModule {
pub enum InputTransform {
Static {
#[serde(default = "default_null")]
value: serde_json::Value,
value: Box<serde_json::value::RawValue>,
},
Javascript {
#[serde(default = "default_empty_string")]
@@ -269,25 +336,19 @@ pub enum InputTransform {
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct BranchOneModules {
pub struct Branch {
#[serde(skip_serializing_if = "Option::is_none")]
pub summary: Option<String>,
#[serde(default = "default_empty_string")]
pub expr: String,
pub modules: Vec<FlowModule>,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct BranchAllModules {
#[serde(skip_serializing_if = "Option::is_none")]
pub summary: Option<String>,
pub modules: Vec<FlowModule>,
#[serde(default = "default_true")]
pub skip_failure: bool,
#[serde(default = "default_true")]
pub parallel: bool,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
#[derive(Serialize, Debug, Clone)]
#[serde(
tag = "type",
rename_all(serialize = "lowercase", deserialize = "lowercase")
@@ -313,7 +374,6 @@ pub enum FlowModuleValue {
modules: Vec<FlowModule>,
#[serde(default = "default_true")]
skip_failures: bool,
#[serde(default = "default_false")]
parallel: bool,
#[serde(skip_serializing_if = "Option::is_none")]
parallelism: Option<u16>,
@@ -324,11 +384,11 @@ pub enum FlowModuleValue {
skip_failures: bool,
},
BranchOne {
branches: Vec<BranchOneModules>,
branches: Vec<Branch>,
default: Vec<FlowModule>,
},
BranchAll {
branches: Vec<BranchAllModules>,
branches: Vec<Branch>,
#[serde(default = "default_true")]
parallel: bool,
},
@@ -354,6 +414,123 @@ pub enum FlowModuleValue {
Identity,
}
#[derive(Deserialize)]
struct UntaggedFlowModuleValue {
#[serde(rename = "type")]
type_: String,
#[serde(alias = "input_transform")]
input_transforms: Option<HashMap<String, InputTransform>>,
path: Option<String>,
hash: Option<ScriptHash>,
tag_override: Option<String>,
iterator: Option<InputTransform>,
modules: Option<Vec<FlowModule>>,
skip_failures: Option<bool>,
parallel: Option<bool>,
parallelism: Option<u16>,
branches: Option<Vec<Branch>>,
default: Option<Vec<FlowModule>>,
content: Option<String>,
lock: Option<String>,
tag: Option<String>,
language: Option<ScriptLang>,
custom_concurrency_key: Option<String>,
concurrent_limit: Option<i32>,
concurrency_time_window_s: Option<i32>,
}
impl<'de> Deserialize<'de> for FlowModuleValue {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let untagged: UntaggedFlowModuleValue = UntaggedFlowModuleValue::deserialize(deserializer)?;
match untagged.type_.as_str() {
"script" => Ok(FlowModuleValue::Script {
input_transforms: untagged.input_transforms.unwrap_or_default(),
path: untagged
.path
.ok_or_else(|| serde::de::Error::missing_field("path"))?,
hash: untagged.hash,
tag_override: untagged.tag_override,
}),
"flow" => Ok(FlowModuleValue::Flow {
input_transforms: untagged.input_transforms.unwrap_or_default(),
path: untagged
.path
.ok_or_else(|| serde::de::Error::missing_field("path"))?,
}),
"forloopflow" => Ok(FlowModuleValue::ForloopFlow {
iterator: untagged
.iterator
.ok_or_else(|| serde::de::Error::missing_field("iterator"))?,
modules: untagged
.modules
.ok_or_else(|| serde::de::Error::missing_field("modules"))?,
skip_failures: untagged.skip_failures.unwrap_or(true),
parallel: untagged.parallel.unwrap_or(false),
parallelism: untagged.parallelism,
}),
"whileloopflow" => Ok(FlowModuleValue::WhileloopFlow {
modules: untagged
.modules
.ok_or_else(|| serde::de::Error::missing_field("modules"))?,
skip_failures: untagged.skip_failures.unwrap_or(false),
}),
"branchone" => Ok(FlowModuleValue::BranchOne {
branches: untagged
.branches
.ok_or_else(|| serde::de::Error::missing_field("branches"))?,
default: untagged
.default
.ok_or_else(|| serde::de::Error::missing_field("default"))?,
}),
"branchall" => Ok(FlowModuleValue::BranchAll {
branches: untagged
.branches
.ok_or_else(|| serde::de::Error::missing_field("branches"))?,
parallel: untagged.parallel.unwrap_or(true),
}),
"rawscript" => Ok(FlowModuleValue::RawScript {
input_transforms: untagged.input_transforms.unwrap_or_default(),
content: untagged
.content
.ok_or_else(|| serde::de::Error::missing_field("content"))?,
lock: untagged.lock,
path: untagged.path,
tag: untagged.tag,
language: untagged
.language
.ok_or_else(|| serde::de::Error::missing_field("language"))?,
custom_concurrency_key: untagged.custom_concurrency_key,
concurrent_limit: untagged.concurrent_limit,
concurrency_time_window_s: untagged.concurrency_time_window_s,
}),
"identity" => Ok(FlowModuleValue::Identity),
other => Err(serde::de::Error::unknown_variant(
other,
&[
"script",
"flow",
"forloopflow",
"whileloopflow",
"branchone",
"branchall",
"rawscript",
"identity",
],
)),
}
}
}
impl Into<Box<serde_json::value::RawValue>> for FlowModuleValue {
fn into(self) -> Box<serde_json::value::RawValue> {
crate::worker::to_raw_value(&self)
}
}
impl FlowModuleValue {
pub fn is_simple(&self) -> bool {
match self {
@@ -391,7 +568,7 @@ pub fn add_virtual_items_if_necessary(modules: &mut Vec<FlowModule>) {
{
modules.push(FlowModule {
id: format!("{}-v", modules[modules.len() - 1].id),
value: FlowModuleValue::Identity,
value: crate::worker::to_raw_value(&FlowModuleValue::Identity),
stop_after_if: None,
summary: Some("Virtual module needed for suspend/sleep when last module".to_string()),
mock: None,
+6 -4
View File
@@ -137,9 +137,11 @@ impl QueuedJob {
}
pub fn parse_raw_flow(&self) -> Option<FlowValue> {
self.raw_flow
.as_ref()
.and_then(|v| serde_json::from_str::<FlowValue>((**v).get()).ok())
self.raw_flow.as_ref().and_then(|v| {
let str = (**v).get();
// tracing::error!("raw_flow: {}", str);
return serde_json::from_str::<FlowValue>(str).ok();
})
}
pub fn parse_flow_status(&self) -> Option<FlowStatus> {
@@ -327,7 +329,7 @@ pub enum JobPayload {
SingleScriptFlow {
path: String,
hash: ScriptHash,
args: HashMap<String, serde_json::Value>,
args: HashMap<String, Box<serde_json::value::RawValue>>,
retry: Retry, // for now only used to retry the script, so retry is necessarily present
custom_concurrency_key: Option<String>,
concurrent_limit: Option<i32>,
+3 -2
View File
@@ -10,6 +10,7 @@
use crate::utils::rd_string;
use serde::{Deserialize, Deserializer};
use serde_json::value::RawValue;
use std::{fmt::Display, str::FromStr};
pub fn default_true() -> bool {
@@ -20,8 +21,8 @@ pub fn default_false() -> bool {
false
}
pub fn default_null() -> serde_json::Value {
serde_json::Value::Null
pub fn default_null() -> Box<RawValue> {
RawValue::from_string("null".to_string()).unwrap()
}
pub fn default_empty_string() -> String {
+3 -3
View File
@@ -23,7 +23,7 @@ pub struct Schedule {
pub enabled: bool,
pub script_path: String,
pub is_flow: bool,
pub args: Option<serde_json::Value>,
pub args: Option<sqlx::types::Json<Box<serde_json::value::RawValue>>>,
pub extra_perms: serde_json::Value,
pub email: String,
#[serde(skip_serializing_if = "Option::is_none")]
@@ -35,13 +35,13 @@ pub struct Schedule {
#[serde(skip_serializing_if = "Option::is_none")]
pub on_failure_exact: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub on_failure_extra_args: Option<serde_json::Value>,
pub on_failure_extra_args: Option<sqlx::types::Json<Box<serde_json::value::RawValue>>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub on_recovery: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub on_recovery_times: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub on_recovery_extra_args: Option<serde_json::Value>,
pub on_recovery_extra_args: Option<sqlx::types::Json<Box<serde_json::value::RawValue>>>,
pub ws_error_handler_muted: bool,
#[serde(skip_serializing_if = "Option::is_none")]
+141 -106
View File
@@ -762,13 +762,18 @@ pub async fn add_completed_job<
}
}
if queued_job.concurrent_limit.is_some() {
let concurrency_key = concurrency_key(db, queued_job).await.unwrap_or_else(|e| {
tracing::error!(
"Could not get concurrency key for job {} defaulting to default key: {e:?}",
queued_job.id
);
return queued_job.full_path_with_workspace();
});
let concurrency_key = match concurrency_key(db, queued_job).await {
Ok(c) => c,
Err(e) => {
tracing::error!(
"Could not get concurrency key for job {} defaulting to default key: {e:?}",
queued_job.id
);
legacy_concurrency_key(db, queued_job)
.await
.unwrap_or_else(|| queued_job.full_path_with_workspace())
}
};
if let Err(e) = sqlx::query_scalar!(
"UPDATE concurrency_counter SET job_uuids = job_uuids - $2 WHERE concurrency_id = $1",
concurrency_key,
@@ -982,7 +987,11 @@ pub async fn add_completed_job<
.unwrap_or_else(|| ScriptLang::Deno),
priority: queued_job.priority,
},
queued_job.args.clone(),
queued_job
.args
.as_ref()
.map(|x| PushArgs { args: x.0.clone(), extra: HashMap::new() })
.unwrap_or_else(PushArgs::empty),
&queued_job.created_by,
&queued_job.email,
queued_job.permissioned_as.clone(),
@@ -1063,7 +1072,7 @@ pub async fn report_error_to_workspace_handler_or_critical_side_channel<
error_message: String,
) -> () {
let w_id = &queued_job.workspace_id;
let (error_handler, error_handler_extra_args) = sqlx::query_as::<_, (Option<String>, Option<serde_json::Value>)>(
let (error_handler, error_handler_extra_args) = sqlx::query_as::<_, (Option<String>, Option<Json<Box<RawValue>>>)>(
"SELECT error_handler, error_handler_extra_args FROM workspace_settings WHERE workspace_id = $1",
).bind(&w_id)
.fetch_optional(db)
@@ -1122,7 +1131,7 @@ pub async fn send_error_to_workspace_handler<
result: Json<&'a T>,
) -> Result<(), Error> {
let w_id = &queued_job.workspace_id;
let (error_handler, error_handler_extra_args, error_handler_muted_on_cancel) = sqlx::query_as::<_, (Option<String>, Option<serde_json::Value>, bool)>(
let (error_handler, error_handler_extra_args, error_handler_muted_on_cancel) = sqlx::query_as::<_, (Option<String>, Option<Json<Box<RawValue>>>, bool)>(
"SELECT error_handler, error_handler_extra_args, error_handler_muted_on_cancel FROM workspace_settings WHERE workspace_id = $1",
).bind(&w_id)
.fetch_optional(db)
@@ -1256,10 +1265,10 @@ pub async fn handle_maybe_scheduled_job<'c, R: rsmq_async::RsmqConnection + Clon
}
}
#[derive(Clone, Serialize)]
#[derive(Clone, Serialize, FromRow)]
struct CompletedJobSubset {
success: bool,
result: Option<serde_json::Value>,
result: Option<sqlx::types::Json<Box<RawValue>>>,
started_at: chrono::DateTime<chrono::Utc>,
}
async fn apply_schedule_handlers<
@@ -1285,15 +1294,15 @@ async fn apply_schedule_handlers<
let times = schedule.on_failure_times.unwrap_or(1).max(1);
let exact = schedule.on_failure_exact.unwrap_or(false);
if times > 1 || exact {
let past_jobs = sqlx::query_as!(
CompletedJobSubset,
let past_jobs = sqlx::query_as::<_, CompletedJobSubset>(
"SELECT success, result, started_at FROM completed_job WHERE workspace_id = $1 AND schedule_path = $2 AND script_path = $3 AND id != $4 ORDER BY created_at DESC LIMIT $5",
&schedule.workspace_id,
&schedule.path,
script_path,
job_id,
if exact { times } else { times - 1 } as i64,
).fetch_all(db).await?;
)
.bind(&schedule.workspace_id)
.bind(&schedule.path)
.bind(script_path)
.bind(job_id)
.bind(if exact { times } else { times - 1 } as i64,)
.fetch_all(db).await?;
let match_times = if exact {
past_jobs.len() == times as usize
@@ -1334,15 +1343,15 @@ async fn apply_schedule_handlers<
if let Some(on_recovery_path) = schedule.on_recovery.clone() {
let mut tx: QueueTransaction<'_, R> = (rsmq.clone(), db.begin().await?).into();
let times = schedule.on_recovery_times.unwrap_or(1).max(1);
let past_jobs = sqlx::query_as!(
CompletedJobSubset,
let past_jobs = sqlx::query_as::<_, CompletedJobSubset>(
"SELECT success, result, started_at FROM completed_job WHERE workspace_id = $1 AND schedule_path = $2 AND script_path = $3 AND id != $4 ORDER BY created_at DESC LIMIT $5",
&schedule.workspace_id,
&schedule.path,
script_path,
job_id,
times as i64,
).fetch_all(db).await?;
)
.bind(&schedule.workspace_id)
.bind(&schedule.path)
.bind(script_path)
.bind(job_id)
.bind(times as i64)
.fetch_all(db).await?;
if past_jobs.len() < times as usize {
return Ok(());
@@ -1399,7 +1408,7 @@ pub async fn push_error_handler<
result: Json<&'a T>,
failed_times: Option<i32>,
started_at: Option<DateTime<Utc>>,
extra_args: Option<serde_json::Value>,
extra_args: Option<Json<Box<RawValue>>>,
email: &str,
is_schedule_error_handler: bool,
is_global_error_handler: bool,
@@ -1428,10 +1437,8 @@ pub async fn push_error_handler<
}
if let Some(args_v) = extra_args {
if let serde_json::Value::Object(args_m) = args_v {
for (k, v) in args_m {
extra.insert(k, to_raw_value(&v));
}
if let Ok(args_m) = serde_json::from_str::<HashMap<String, Box<RawValue>>>(args_v.get()) {
extra.extend(args_m);
} else {
return Err(error::Error::ExecutionErr(
"args of scripts needs to be dict".to_string(),
@@ -1457,7 +1464,7 @@ pub async fn push_error_handler<
tx,
handler_w_id,
payload,
PushArgs { extra, args: Json(&result) },
PushArgs { extra, args: result },
if is_global_error_handler {
"global"
} else if is_schedule_error_handler {
@@ -1496,16 +1503,10 @@ pub async fn push_error_handler<
return Ok(uuid);
}
fn sanitize_result<T: Serialize + Send + Sync>(result: Json<&T>) -> serde_json::Value {
let result =
serde_json::from_str(&serde_json::to_string(result.0).unwrap_or_else(|_| "{}".to_string()))
.unwrap_or_else(|_| json!({}));
let result = if result.is_object() || result.is_null() {
result
} else {
json!({ "error": result })
};
result
fn sanitize_result<T: Serialize + Send + Sync>(result: Json<&T>) -> HashMap<String, Box<RawValue>> {
let as_str = serde_json::to_string(result.0).unwrap_or_else(|_| "{}".to_string());
serde_json::from_str::<HashMap<String, Box<RawValue>>>(&as_str)
.unwrap_or_else(|_| [("error".to_string(), RawValue::from_string(as_str).unwrap())].into())
}
// #[derive(Serialize)]
@@ -1534,33 +1535,33 @@ async fn handle_recovered_schedule<
successful_job_result: Json<&'a T>,
successful_times: i32,
successful_job_started_at: DateTime<Utc>,
extra_args: Option<serde_json::Value>,
extra_args: Option<Json<Box<RawValue>>>,
) -> windmill_common::error::Result<QueueTransaction<'c, R>> {
let (payload, tag) = get_payload_tag_from_prefixed_path(on_recovery_path, db, w_id).await?;
let mut args = error_job
.result
.unwrap_or(json!({}))
.as_object()
.unwrap()
.clone();
args.insert("error_started_at".to_string(), json!(error_job.started_at));
args.insert("schedule_path".to_string(), json!(schedule_path));
args.insert("path".to_string(), json!(script_path));
args.insert("is_flow".to_string(), json!(is_flow));
args.insert(
"success_result".to_string(),
serde_json::from_str(&serde_json::to_string(&successful_job_result).unwrap())
.unwrap_or_else(|_| json!("{}")),
let mut extra = HashMap::new();
extra.insert(
"error_started_at".to_string(),
to_raw_value(&error_job.started_at),
);
args.insert("success_times".to_string(), json!(successful_times));
args.insert(
extra.insert("schedule_path".to_string(), to_raw_value(&schedule_path));
extra.insert("path".to_string(), to_raw_value(&script_path));
extra.insert("is_flow".to_string(), to_raw_value(&is_flow));
extra.insert(
"success_result".to_string(),
serde_json::from_str::<Box<RawValue>>(
&serde_json::to_string(&successful_job_result).unwrap(),
)
.unwrap_or_else(|_| serde_json::value::RawValue::from_string("{}".to_string()).unwrap()),
);
extra.insert("success_times".to_string(), to_raw_value(&successful_times));
extra.insert(
"success_started_at".to_string(),
json!(successful_job_started_at),
to_raw_value(&successful_job_started_at),
);
if let Some(args_v) = extra_args {
if let serde_json::Value::Object(args_m) = args_v {
args.extend(args_m);
if let Ok(args_m) = serde_json::from_str::<HashMap<String, Box<RawValue>>>(args_v.get()) {
extra.extend(args_m);
} else {
return Err(error::Error::ExecutionErr(
"args of scripts needs to be dict".to_string(),
@@ -1574,15 +1575,21 @@ async fn handle_recovered_schedule<
{
// default slack error handler being used -> we need to inject the slack token
let slack_resource = format!("$res:{WORKSPACE_SLACK_BOT_TOKEN_PATH}");
args.insert("slack".to_string(), json!(slack_resource));
extra.insert("slack".to_string(), to_raw_value(&slack_resource));
}
let args = error_job
.result
.and_then(|x| serde_json::from_str::<HashMap<String, Box<RawValue>>>(x.0.get()).ok())
.unwrap_or_else(HashMap::new);
let tx = PushIsolationLevel::Transaction(tx);
let (uuid, tx) = push(
&db,
tx,
w_id,
payload,
args,
PushArgs { extra: extra, args: args },
SCHEDULE_RECOVERY_HANDLER_USERNAME,
SCHEDULE_RECOVERY_HANDLER_USER_EMAIL,
ERROR_HANDLER_USER_GROUP.to_string(),
@@ -2017,6 +2024,42 @@ pub async fn custom_concurrency_key(
.await
}
async fn legacy_concurrency_key(db: &Pool<Postgres>, queued_job: &QueuedJob) -> Option<String> {
let r = if queued_job.is_flow() {
sqlx::query_scalar!(
"SELECT value->>'concurrency_key' FROM flow WHERE path = $1 AND workspace_id = $2",
queued_job.script_path,
queued_job.workspace_id
)
.fetch_optional(db)
.await
} else {
sqlx::query_scalar!(
"SELECT concurrency_key FROM script WHERE hash = $1 AND workspace_id = $2",
queued_job.script_hash.unwrap_or(ScriptHash(0)).0,
queued_job.workspace_id
)
.fetch_optional(db)
.await
}
.ok()
.flatten()
.flatten();
r.map(|x| {
interpolate_args(
x,
&queued_job
.args
.clone()
.map(|x| x.0)
.unwrap_or_default()
.into(),
&queued_job.workspace_id,
)
})
}
async fn concurrency_key(
db: &Pool<Postgres>,
queued_job: &QueuedJob,
@@ -2028,25 +2071,18 @@ async fn concurrency_key(
)
}
fn interpolate_args<T: Serialize>(
x: String,
args: &T,
workspace_id: &str,
parsed_args: &mut Option<serde_json::Value>,
) -> String {
fn interpolate_args(x: String, args: &PushArgs, workspace_id: &str) -> String {
// Save this value to avoid parsing twice
if parsed_args.is_none() {
*parsed_args = Some(serde_json::to_value(args).unwrap_or_default());
}
let value = parsed_args.as_ref().unwrap();
let workspaced = x.as_str().replace("$workspace", workspace_id).to_string();
if RE_ARG_TAG.is_match(&workspaced) {
let mut interpolated = workspaced.clone();
for cap in RE_ARG_TAG.captures_iter(&workspaced) {
let arg_name = cap.get(1).unwrap().as_str();
let arg_value = value
let arg_value = args
.args
.get(arg_name)
.map(|x| serde_json::to_string(x).unwrap_or_default())
.or(args.extra.get(arg_name))
.map(|x| x.get())
.unwrap_or_default();
interpolated =
interpolated.replace(format!("$args[{}]", arg_name).as_str(), &arg_value);
@@ -2493,11 +2529,11 @@ macro_rules! fetch_scalar_isolated {
use sqlx::types::JsonRawValue;
#[derive(Serialize, Debug)]
pub struct PushArgs<T> {
pub struct PushArgs {
#[serde(flatten)]
pub extra: HashMap<String, Box<RawValue>>,
#[serde(flatten)]
pub args: Json<T>,
pub args: HashMap<String, Box<RawValue>>,
}
#[derive(Deserialize)]
@@ -2554,7 +2590,7 @@ impl DecodeQueries {
}
}
impl<T> PushArgs<T> {
impl PushArgs {
pub fn insert<K: Into<String>, V: Into<Box<RawValue>>>(&mut self, k: K, v: V) {
self.extra.insert(k.into(), v.into());
}
@@ -2567,7 +2603,7 @@ pub struct RequestQuery {
}
#[axum::async_trait]
impl<S> FromRequest<S, axum::body::Body> for PushArgs<HashMap<String, Box<RawValue>>>
impl<S> FromRequest<S, axum::body::Body> for PushArgs
where
S: Send + Sync,
{
@@ -2611,12 +2647,12 @@ where
.unwrap_or_else(|| to_raw_value(&serde_json::Value::Null));
let mut hm = HashMap::new();
hm.insert("body".to_string(), args);
Ok(PushArgs { extra, args: Json(hm) })
Ok(PushArgs { extra, args: hm })
} else {
let hm = serde_json::from_str::<Option<HashMap<String, Box<JsonRawValue>>>>(&str)
.map_err(|e| Error::BadRequest(format!("invalid json: {}", e)).into_response())?
.unwrap_or_else(HashMap::new);
Ok(PushArgs { extra, args: Json(hm) })
Ok(PushArgs { extra, args: hm })
}
} else if content_type
.unwrap()
@@ -2642,7 +2678,7 @@ where
.map(|(k, v)| (k, to_raw_value(&v)))
.collect::<HashMap<_, _>>();
return Ok(PushArgs { extra, args: Json(payload) });
return Ok(PushArgs { extra, args: payload });
} else {
Err(StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response())
}
@@ -2680,9 +2716,9 @@ pub fn build_extra(
args
}
impl PushArgs<HashMap<String, Box<RawValue>>> {
impl PushArgs {
pub fn empty() -> Self {
PushArgs { extra: HashMap::new(), args: Json(HashMap::new()) }
PushArgs { extra: HashMap::new(), args: HashMap::new() }
}
}
@@ -2690,9 +2726,9 @@ pub fn empty_result() -> Box<RawValue> {
return JsonRawValue::from_string("{}".to_string()).unwrap();
}
impl From<HashMap<String, Box<JsonRawValue>>> for PushArgs<HashMap<String, Box<JsonRawValue>>> {
impl From<HashMap<String, Box<JsonRawValue>>> for PushArgs {
fn from(value: HashMap<String, Box<JsonRawValue>>) -> Self {
PushArgs { extra: HashMap::new(), args: Json(value) }
PushArgs { extra: HashMap::new(), args: value }
}
}
@@ -2707,12 +2743,12 @@ lazy_static::lazy_static! {
}
// #[instrument(level = "trace", skip_all)]
pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection + Send + 'c>(
pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
_db: &Pool<Postgres>,
mut tx: PushIsolationLevel<'c, R>,
workspace_id: &str,
job_payload: JobPayload,
args: T,
args: PushArgs,
user: &str,
mut email: &str,
mut permissioned_as: String,
@@ -3144,12 +3180,14 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
let flow_value = FlowValue {
modules: vec![FlowModule {
id: "a".to_string(),
value: windmill_common::flows::FlowModuleValue::Script {
input_transforms: input_transforms,
path: path.clone(),
hash: Some(hash),
tag_override: tag_override,
},
value: windmill_common::worker::to_raw_value(
&windmill_common::flows::FlowModuleValue::Script {
input_transforms: input_transforms,
path: path.clone(),
hash: Some(hash),
tag_override: tag_override,
},
),
stop_after_if: None,
summary: None,
suspend: None,
@@ -3376,8 +3414,6 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
let per_workspace: bool = DEFAULT_TAGS_PER_WORKSPACE.load(std::sync::atomic::Ordering::Relaxed);
let mut parsed_args: Option<serde_json::Value> = None;
let tag = if dedicated_worker.is_some_and(|x| x) {
format!(
"{}:{}{}",
@@ -3400,8 +3436,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
tag = None;
}
let interpolated_tag =
tag.map(|x| interpolate_args(x, &args, workspace_id, &mut parsed_args));
let interpolated_tag = tag.map(|x| interpolate_args(x, &args, workspace_id));
let default = || {
let ntag = if job_kind == JobKind::Flow
@@ -3469,7 +3504,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
if concurrent_limit.is_some() {
let concurrency_key = custom_concurrency_key
.map(|x| interpolate_args(x, &args, workspace_id, &mut parsed_args))
.map(|x| interpolate_args(x, &args, workspace_id))
.unwrap_or(fullpath_with_workspace(
workspace_id,
script_path.as_ref(),
@@ -3505,7 +3540,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
script_path.clone(),
raw_code,
raw_lock,
Json(args) as Json<T>,
Json(args) as Json<PushArgs>,
job_kind.clone() as JobKind,
schedule_path,
raw_flow.map(|f| serde_json::json!(f)),
@@ -3669,8 +3704,8 @@ async fn restarted_flows_resolution(
module.id()
)))?;
match module_definition.value.clone() {
FlowModuleValue::BranchAll { branches, parallel, .. } => {
match module_definition.get_value() {
Ok(FlowModuleValue::BranchAll { branches, parallel, .. }) => {
if parallel {
return Err(Error::InternalErr(format!(
"Module {} is a parallel branchall. It can only be restarted at a given branch if it's sequential",
@@ -3702,7 +3737,7 @@ async fn restarted_flows_resolution(
while_loop: false,
});
}
FlowModuleValue::ForloopFlow { parallel, .. } => {
Ok(FlowModuleValue::ForloopFlow { parallel, .. }) => {
if parallel {
return Err(Error::InternalErr(format!(
"Module {} is not parallel loop. It can only be restarted at a given iteration if it's sequential",
+9 -8
View File
@@ -66,10 +66,12 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
return Ok(tx);
}
let mut args: serde_json::Map<String, serde_json::Value> = serde_json::Map::new();
let mut args: HashMap<String, Box<serde_json::value::RawValue>> = HashMap::new();
if let Some(args_v) = &schedule.args {
if let serde_json::Value::Object(args_m) = args_v {
if let Ok(args_m) =
serde_json::from_str::<HashMap<String, Box<serde_json::value::RawValue>>>(args_v.get())
{
args = args_m.clone()
} else {
return Err(error::Error::ExecutionErr(
@@ -121,7 +123,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
err.to_string(),
))
})?;
let mut static_args = HashMap::<String, serde_json::Value>::new();
let mut static_args = HashMap::<String, Box<serde_json::value::RawValue>>::new();
for (arg_name, arg_value) in args.clone() {
static_args.insert(arg_name, arg_value);
}
@@ -186,7 +188,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
tx,
&schedule.workspace_id,
payload,
args,
crate::PushArgs { args, extra: HashMap::new() },
&schedule_to_user(&schedule.path),
&schedule.email,
username_to_permissioned_as(&schedule.edited_by),
@@ -213,12 +215,11 @@ pub async fn get_schedule_opt<'c>(
w_id: &str,
path: &str,
) -> Result<Option<Schedule>> {
let schedule_opt = sqlx::query_as!(
Schedule,
let schedule_opt = sqlx::query_as::<_, Schedule>(
"SELECT * FROM schedule WHERE path = $1 AND workspace_id = $2",
path,
w_id
)
.bind(path)
.bind(w_id)
.fetch_optional(&mut **db)
.await?;
Ok(schedule_opt)
+7 -4
View File
@@ -399,7 +399,7 @@ async function variable(path) {{
return await Deno.core.ops.op_variable(path);
}}
async function resource(path) {{
return await Deno.core.ops.op_resource(path);
return JSON.parse(await Deno.core.ops.op_resource(path));
}}
"#,
);
@@ -518,14 +518,17 @@ async fn op_get_id(
}
#[op2(async)]
#[serde]
#[string]
async fn op_resource(
op_state: Rc<RefCell<OpState>>,
#[string] path: String,
) -> Result<serde_json::Value, anyhow::Error> {
) -> Result<Option<String>, anyhow::Error> {
let client = op_state.borrow().borrow::<OptAuthedClient>().0.clone();
if let Some(client) = client {
client.get_resource_value_interpolated(&path, None).await
client
.get_resource_value_interpolated::<Option<Box<RawValue>>>(&path, None)
.await
.map(|x| x.map(|x| x.get().to_string()))
} else {
anyhow::bail!("No client found in op state");
}
+113 -107
View File
@@ -1847,75 +1847,40 @@ async fn spawn_dedicated_workers_for_flow(
let mut workers = vec![];
let mut script_path_to_worker: HashMap<String, Sender<Arc<QueuedJob>>> = HashMap::new();
for module in modules.iter() {
match &module.value {
FlowModuleValue::Script { path, hash, .. } => {
let key = format!(
"{}:{}",
path,
hash.clone()
.map(|x| x.to_string())
.unwrap_or_else(|| "".to_string())
);
if let Some(sender) = script_path_to_worker.get(&key) {
workers.push((module.id.clone(), sender.clone(), None));
} else {
if let Some(dedi_w) = spawn_dedicated_worker(
SpawnWorker::Script { path: path.to_string(), hash: hash.clone() },
w_id,
killpill_tx.clone(),
killpill_rx,
db,
worker_dir,
base_internal_url,
worker_name,
job_completed_tx,
Some(module.id.clone()),
)
.await
{
script_path_to_worker.insert(key, dedi_w.1.clone());
workers.push(dedi_w);
let value = module.get_value();
if let Ok(value) = value {
match &value {
FlowModuleValue::Script { path, hash, .. } => {
let key = format!(
"{}:{}",
path,
hash.clone()
.map(|x| x.to_string())
.unwrap_or_else(|| "".to_string())
);
if let Some(sender) = script_path_to_worker.get(&key) {
workers.push((module.id.clone(), sender.clone(), None));
} else {
if let Some(dedi_w) = spawn_dedicated_worker(
SpawnWorker::Script { path: path.to_string(), hash: hash.clone() },
w_id,
killpill_tx.clone(),
killpill_rx,
db,
worker_dir,
base_internal_url,
worker_name,
job_completed_tx,
Some(module.id.clone()),
)
.await
{
script_path_to_worker.insert(key, dedi_w.1.clone());
workers.push(dedi_w);
}
}
}
}
FlowModuleValue::ForloopFlow { modules, .. } => {
let w = spawn_dedicated_workers_for_flow(
&modules,
path,
w_id,
killpill_tx.clone(),
killpill_rx,
db,
worker_dir,
base_internal_url,
worker_name,
job_completed_tx,
)
.await;
workers.extend(w);
}
FlowModuleValue::WhileloopFlow { modules, .. } => {
let w = spawn_dedicated_workers_for_flow(
&modules,
path,
w_id,
killpill_tx.clone(),
killpill_rx,
db,
worker_dir,
base_internal_url,
worker_name,
job_completed_tx,
)
.await;
workers.extend(w);
}
FlowModuleValue::BranchOne { branches, default } => {
for modules in branches
.iter()
.map(|x| &x.modules)
.chain(std::iter::once(default))
{
FlowModuleValue::ForloopFlow { modules, .. } => {
let w = spawn_dedicated_workers_for_flow(
&modules,
path,
@@ -1931,11 +1896,9 @@ async fn spawn_dedicated_workers_for_flow(
.await;
workers.extend(w);
}
}
FlowModuleValue::BranchAll { branches, .. } => {
for branch in branches {
FlowModuleValue::WhileloopFlow { modules, .. } => {
let w = spawn_dedicated_workers_for_flow(
&branch.modules,
&modules,
path,
w_id,
killpill_tx.clone(),
@@ -1949,32 +1912,74 @@ async fn spawn_dedicated_workers_for_flow(
.await;
workers.extend(w);
}
}
FlowModuleValue::RawScript { content, lock, path: spath, language, .. } => {
if let Some(dedi_w) = spawn_dedicated_worker(
SpawnWorker::RawScript {
path: spath.clone().unwrap_or(path.to_string()),
content: content.to_string(),
lock: lock.clone(),
lang: language.clone(),
},
w_id,
killpill_tx.clone(),
killpill_rx,
db,
worker_dir,
base_internal_url,
worker_name,
job_completed_tx,
Some(module.id.clone()),
)
.await
{
workers.push(dedi_w);
FlowModuleValue::BranchOne { branches, default } => {
for modules in branches
.iter()
.map(|x| &x.modules)
.chain(std::iter::once(default))
{
let w = spawn_dedicated_workers_for_flow(
&modules,
path,
w_id,
killpill_tx.clone(),
killpill_rx,
db,
worker_dir,
base_internal_url,
worker_name,
job_completed_tx,
)
.await;
workers.extend(w);
}
}
FlowModuleValue::BranchAll { branches, .. } => {
for branch in branches {
let w = spawn_dedicated_workers_for_flow(
&branch.modules,
path,
w_id,
killpill_tx.clone(),
killpill_rx,
db,
worker_dir,
base_internal_url,
worker_name,
job_completed_tx,
)
.await;
workers.extend(w);
}
}
FlowModuleValue::RawScript { content, lock, path: spath, language, .. } => {
if let Some(dedi_w) = spawn_dedicated_worker(
SpawnWorker::RawScript {
path: spath.clone().unwrap_or(path.to_string()),
content: content.to_string(),
lock: lock.clone(),
lang: language.clone(),
},
w_id,
killpill_tx.clone(),
killpill_rx,
db,
worker_dir,
base_internal_url,
worker_name,
job_completed_tx,
Some(module.id.clone()),
)
.await
{
workers.push(dedi_w);
}
}
FlowModuleValue::Flow { .. } => (),
FlowModuleValue::Identity => (),
}
FlowModuleValue::Flow { .. } => (),
FlowModuleValue::Identity => (),
} else {
tracing::error!("failed to get value for module: {:?}", module);
}
}
workers
@@ -3668,12 +3673,12 @@ async fn trigger_python_dependents_to_recompute_dependencies<
PushIsolationLevel::IsolatedRoot(db.clone(), rsmq.clone());
let r = get_latest_deployed_hash_for_path(db, w_id, s.as_str()).await;
if let Ok(r) = r {
let mut args: HashMap<String, serde_json::Value> = HashMap::new();
let mut args: HashMap<String, Box<RawValue>> = HashMap::new();
if let Some(ref dm) = deployment_message {
args.insert("deployment_message".to_string(), json!(dm));
args.insert("deployment_message".to_string(), to_raw_value(&dm));
}
if let Some(ref p_path) = parent_path {
args.insert("common_dependency_path".to_string(), json!(p_path));
args.insert("common_dependency_path".to_string(), to_raw_value(&p_path));
}
let (job_uuid, new_tx) = windmill_queue::push(
@@ -3686,7 +3691,7 @@ async fn trigger_python_dependents_to_recompute_dependencies<
language: r.6,
dedicated_worker: r.7,
},
args,
windmill_queue::PushArgs { args, extra: HashMap::new() },
&created_by,
email,
permissioned_as.to_string(),
@@ -3856,9 +3861,9 @@ async fn lock_modules(
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
} = e.value.clone()
} = e.get_value()?
else {
match e.value {
match e.get_value()? {
FlowModuleValue::ForloopFlow {
iterator,
modules,
@@ -3886,6 +3891,7 @@ async fn lock_modules(
parallel,
parallelism,
}
.into()
}
FlowModuleValue::BranchAll { branches, parallel } => {
let mut nbranches = vec![];
@@ -3906,7 +3912,7 @@ async fn lock_modules(
.await?;
nbranches.push(b)
}
e.value = FlowModuleValue::BranchAll { branches: nbranches, parallel }
e.value = FlowModuleValue::BranchAll { branches: nbranches, parallel }.into()
}
FlowModuleValue::BranchOne { branches, default } => {
let mut nbranches = vec![];
@@ -3941,7 +3947,7 @@ async fn lock_modules(
token,
)
.await?;
e.value = FlowModuleValue::BranchOne { branches: nbranches, default };
e.value = FlowModuleValue::BranchOne { branches: nbranches, default }.into();
}
_ => (),
};
@@ -3971,7 +3977,7 @@ async fn lock_modules(
.await;
match new_lock {
Ok(new_lock) => {
e.value = FlowModuleValue::RawScript {
e.value = windmill_common::worker::to_raw_value(&FlowModuleValue::RawScript {
lock: Some(new_lock),
path,
input_transforms,
@@ -3981,7 +3987,7 @@ async fn lock_modules(
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
};
});
new_flow_modules.push(e);
continue;
}
@@ -3993,7 +3999,7 @@ async fn lock_modules(
error = ?error,
"Failed to generate flow lock for raw script"
);
e.value = FlowModuleValue::RawScript {
e.value = windmill_common::worker::to_raw_value(&FlowModuleValue::RawScript {
lock: None,
path,
input_transforms,
@@ -4003,7 +4009,7 @@ async fn lock_modules(
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
};
});
new_flow_modules.push(e);
continue;
}
+117 -94
View File
@@ -243,7 +243,8 @@ pub async fn update_flow_status_after_job_completion_internal<
.await?
.ok_or_else(|| Error::InternalErr(format!("requiring flow to be in the queue")))?;
let module = get_module(&flow_job, module_index);
if module.is_some_and(|x| matches!(x.value, FlowModuleValue::Flow { .. })) {
if module.is_some_and(|x| x.is_flow()) {
(false, false, false)
} else {
(true, se, false)
@@ -1255,7 +1256,7 @@ pub async fn handle_flow<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
#[derive(Serialize, Debug)]
pub struct Iter {
index: i32,
value: serde_json::Value,
value: Box<RawValue>,
}
#[derive(Serialize)]
@@ -1518,7 +1519,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
if suspend.user_groups_required.is_some() {
match suspend.user_groups_required.unwrap() {
InputTransform::Static { value } => {
user_groups_required = serde_json::from_value::<Vec<String>>(value)
user_groups_required = serde_json::from_str::<Vec<String>>(value.get())
.expect("Unable to deserialize group names");
}
InputTransform::Javascript { expr } => {
@@ -1740,7 +1741,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
)
}
};
match json_value {
match json_value.and_then(|x| serde_json::from_str::<serde_json::Value>(x.get())) {
Ok(serde_json::Value::Number(n)) => {
if !n.is_u64() {
return Err(Error::ExecutionErr(format!(
@@ -1750,8 +1751,11 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
n.as_u64().map(|x| from_now(Duration::from_secs(x)))
}
_ => Err(Error::ExecutionErr(format!(
"Expected a number value, found: {json_value:?}"
Ok(x @ _) => Err(Error::ExecutionErr(format!(
"Expected an integer, found: {x:?}"
)))?,
Err(e) => Err(Error::ExecutionErr(format!(
"Expected a number value, had error instead: {e:?}",
)))?,
}
} else {
@@ -1893,10 +1897,12 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
Ok(HashMap::new())
}
} else {
match &module.value {
FlowModuleValue::Script { input_transforms, .. }
| FlowModuleValue::RawScript { input_transforms, .. }
| FlowModuleValue::Flow { input_transforms, .. } => {
match &module.get_value() {
Ok(
FlowModuleValue::Script { input_transforms, .. }
| FlowModuleValue::RawScript { input_transforms, .. }
| FlowModuleValue::Flow { input_transforms, .. },
) => {
let ctx = get_transform_context(&flow_job, &previous_id, &status).await?;
transform_context = Some(ctx);
let by_id = transform_context.as_ref().unwrap();
@@ -1912,7 +1918,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
)
.await
}
FlowModuleValue::Identity => serde_json::from_str(
Ok(FlowModuleValue::Identity) => serde_json::from_str(
&serde_json::to_string(&PreviousResult {
previous_result: Some(&arc_last_job_result),
})
@@ -1920,7 +1926,12 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
)
.map_err(|e| error::Error::InternalErr(format!("identity: {e}"))),
_ => Ok(flow_job_args),
Ok(_) => Ok(flow_job_args),
Err(e) => {
return Err(error::Error::InternalErr(format!(
"module was not convertible to acceptable value {e:?}"
)))
}
}
};
tracing::debug!(id = %flow_job.id, root_id = %job_root, "flow job args computed");
@@ -2085,10 +2096,14 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
Ok(v) => (Some(v), None),
Err(e) => (None, Some(e)),
};
let root_job = if matches!(
module.value,
FlowModuleValue::Flow { .. } | FlowModuleValue::ForloopFlow { parallel: true, .. }
) {
let value_with_parallel = module.get_value_with_parallel()?;
let root_job = if {
value_with_parallel.type_ == "flow"
|| (value_with_parallel.type_ == "forloopflow"
&& value_with_parallel.parallel.is_some_and(|x| x))
} {
None
} else {
flow_job.root_job.or_else(|| Some(flow_job.id))
@@ -2100,7 +2115,10 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
tx2,
&flow_job.workspace_id,
payload_tag.payload,
ok.unwrap_or_else(|| serde_json::from_str("{}").unwrap()),
windmill_queue::PushArgs {
args: ok.unwrap_or_else(|| serde_json::from_str("{}").unwrap()),
extra: HashMap::new(),
},
&flow_job.created_by,
&flow_job.email,
flow_job.permissioned_as.to_owned(),
@@ -2126,17 +2144,19 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
tracing::debug!(id = %flow_job.id, root_id = %job_root, "pushed next flow job: {uuid}");
if let FlowModuleValue::ForloopFlow { parallelism: Some(p), .. } = &module.value {
if i as u16 >= *p {
sqlx::query!(
"UPDATE queue
if value_with_parallel.type_ == "forloopflow" {
if let Some(p) = value_with_parallel.parallelism {
if i as u16 >= p {
sqlx::query!(
"UPDATE queue
SET suspend = $1, suspend_until = now() + interval '14 day', running = true
WHERE id = $2",
(i as u16 - p + 1) as i32,
uuid,
)
.execute(&mut inner_tx)
.await?;
(i as u16 - p + 1) as i32,
uuid,
)
.execute(&mut inner_tx)
.await?;
}
}
}
@@ -2370,14 +2390,14 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
#[derive(Debug)]
struct ForloopNextIteration {
index: usize,
itered: Vec<serde_json::Value>,
itered: Vec<Box<RawValue>>,
flow_jobs: Vec<Uuid>,
new_args: Iter,
while_loop: bool,
}
enum ForLoopStatus {
ParallelIteration { itered: Vec<serde_json::Value> },
ParallelIteration { itered: Vec<Box<RawValue>> },
NextIteration(ForloopNextIteration),
EmptyIterator,
}
@@ -2482,7 +2502,7 @@ async fn compute_next_flow_transform(
let delete_after_use = module.delete_after_use.unwrap_or(false);
tracing::debug!(id = %flow_job.id, "computing next flow transform for {:?}", &module.value);
match &module.value {
match &module.get_value()? {
FlowModuleValue::Identity => trivial_next_job(JobPayload::Identity),
FlowModuleValue::Flow { path, .. } => {
let payload = flow_to_payload(path, &delete_after_use);
@@ -2548,7 +2568,10 @@ async fn compute_next_flow_transform(
index: next_loop_idx,
itered: vec![],
flow_jobs: flow_jobs.clone(),
new_args: Iter { index: next_loop_idx as i32, value: json!(next_loop_idx) },
new_args: Iter {
index: next_loop_idx as i32,
value: windmill_common::worker::to_raw_value(&next_loop_idx),
},
while_loop: true,
},
modules,
@@ -2599,65 +2622,66 @@ async fn compute_next_flow_transform(
.await
}
ForLoopStatus::ParallelIteration { itered, .. } => {
let inner_path = Some(format!("{}/loop-parrallel", flow_job.script_path(),));
let continue_payload = if is_simple {
let payload = payload_from_simple_module(
&modules[0].value,
db,
flow_job,
module,
inner_path,
)
.await?;
ContinuePayload::ForloopJobs { n: itered.len(), payload: payload }
if modules.is_empty() {
return Ok(NextFlowTransform::EmptyInnerFlows);
} else {
let payload = {
JobPayloadWithTag {
payload: JobPayload::RawFlow {
value: FlowValue {
modules: (*modules).clone(),
failure_module: flow.failure_module.clone(),
same_worker: flow.same_worker,
concurrent_limit: None,
concurrency_time_window_s: None,
skip_expr: None,
cache_ttl: None,
priority: None,
early_return: None,
concurrency_key: None,
let inner_path =
Some(format!("{}/loop-parrallel", flow_job.script_path(),));
let value = &modules[0].get_value()?;
let continue_payload = if is_simple {
let payload =
payload_from_simple_module(value, db, flow_job, module, inner_path)
.await?;
ContinuePayload::ForloopJobs { n: itered.len(), payload: payload }
} else {
let payload = {
JobPayloadWithTag {
payload: JobPayload::RawFlow {
value: FlowValue {
modules: (*modules).clone(),
failure_module: flow.failure_module.clone(),
same_worker: flow.same_worker,
concurrent_limit: None,
concurrency_time_window_s: None,
skip_expr: None,
cache_ttl: None,
priority: None,
early_return: None,
concurrency_key: None,
},
path: Some(format!("{}/forloop", flow_job.script_path())),
restarted_from: None,
},
path: Some(format!("{}/forloop", flow_job.script_path())),
restarted_from: None,
},
tag: None,
delete_after_use: delete_after_use,
timeout: None,
}
};
ContinuePayload::ForloopJobs { n: itered.len(), payload }
};
Ok(NextFlowTransform::Continue(
continue_payload,
NextStatus::AllFlowJobs {
branchall: None,
iterator: Some(windmill_common::flow_status::Iterator {
index: 0,
itered,
}),
simple_input_transforms: if is_simple {
match &modules[0].value {
FlowModuleValue::Script { input_transforms, .. }
| FlowModuleValue::RawScript { input_transforms, .. }
| FlowModuleValue::Flow { input_transforms, .. } => {
Some(input_transforms.clone())
}
_ => None,
tag: None,
delete_after_use: delete_after_use,
timeout: None,
}
} else {
None
};
ContinuePayload::ForloopJobs { n: itered.len(), payload }
};
Ok(NextFlowTransform::Continue(
continue_payload,
NextStatus::AllFlowJobs {
branchall: None,
iterator: Some(windmill_common::flow_status::Iterator {
index: 0,
itered,
}),
simple_input_transforms: if is_simple {
match value {
FlowModuleValue::Script { input_transforms, .. }
| FlowModuleValue::RawScript { input_transforms, .. }
| FlowModuleValue::Flow { input_transforms, .. } => {
Some(input_transforms.clone())
}
_ => None,
}
} else {
None
},
},
},
))
))
}
}
}
}
@@ -2882,14 +2906,14 @@ async fn next_loop_iteration(
add_virtual_items_if_necessary(&mut modules);
let inner_path = Some(format!("{}/loop-{}", flow_job.script_path(), ns.index));
if is_simple {
let payload =
payload_from_simple_module(&modules[0].value, db, flow_job, module, inner_path).await?;
let value = &modules[0].get_value()?;
let payload = payload_from_simple_module(value, db, flow_job, module, inner_path).await?;
Ok(NextFlowTransform::Continue(
ContinuePayload::SingleJob(payload),
NextStatus::NextLoopIteration {
next: ns,
simple_input_transforms: if is_simple {
match &modules[0].value {
match value {
FlowModuleValue::Script { input_transforms, .. }
| FlowModuleValue::RawScript { input_transforms, .. }
| FlowModuleValue::Flow { input_transforms, .. } => {
@@ -2932,13 +2956,12 @@ async fn next_loop_iteration(
fn is_simple_modules(modules: &Vec<FlowModule>, flow: &FlowValue) -> bool {
let is_simple = modules.len() == 1
&& modules[0].value.is_simple()
&& modules[0].is_simple()
&& modules[0].sleep.is_none()
&& modules[0].suspend.is_none()
&& modules[0].cache_ttl.is_none()
&& (modules[0].mock.is_none()
&& modules[0].mock.as_ref().is_some_and(|m| !m.enabled)
&& flow.failure_module.is_none());
&& (modules[0].mock.is_none() || modules[0].mock.as_ref().is_some_and(|m| !m.enabled))
&& flow.failure_module.is_none();
is_simple
}
@@ -2987,7 +3010,7 @@ async fn next_forloop_status(
.await?
}
};
let itered = serde_json::from_str::<Vec<serde_json::Value>>(itered_raw.get()).map_err(
let itered = serde_json::from_str::<Vec<Box<RawValue>>>(itered_raw.get()).map_err(
|not_array| {
Error::ExecutionErr(format!(
"Expected an array value in the iterator expression, found: {not_array}"
@@ -3045,7 +3068,7 @@ async fn next_forloop_status(
.await?
}
};
serde_json::from_str::<Vec<serde_json::Value>>(itered_raw.get()).map_err(
serde_json::from_str::<Vec<Box<RawValue>>>(itered_raw.get()).map_err(
|not_array| {
Error::ExecutionErr(format!("Expected an array value, found: {not_array}"))
},