starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-20 08:01:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add db storage for app inline scripts (#4837)

Browse Source
This commit is contained in:
Lucas Abel
2024-12-06 13:37:45 +01:00
committed by GitHub
parent b66a31a623
commit ef08fc8e29
22 changed files with 632 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-0ad36c1598ff4ece0c325eaeb9a9177a87e1accd192402e21db5ae09c3498ab0.json
Generated
+2 -1
View File
@@ -44,7 +44,8 @@
"deploymentcallback",
"singlescriptflow",
"flowscript",
"flownode"
"flownode",
"appscript"
]
}
}
backend/.sqlx/query-0c6c80746733be8f561ab0b631854799f5e8122adaf35465cb16c3dc795bdc3b.json
Generated
+26
View File
@@ -0,0 +1,26 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO app_script (app, hash, lock, code, code_sha256)\n VALUES ($1, $2, $3, $4, $5)\n ON CONFLICT (hash) DO UPDATE SET app = EXCLUDED.app -- trivial update to return the id\n RETURNING id\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Int8",
"Bpchar",
"Text",
"Text",
"Bpchar"
]
},
"nullable": [
false
]
},
"hash": "0c6c80746733be8f561ab0b631854799f5e8122adaf35465cb16c3dc795bdc3b"
}
backend/.sqlx/query-1bae415f9440cc1334f24ce3009242cf3a6287e7b4548c7f01ad888230c27013.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT format('rawscript/%s', code_sha256) as \"path!: String\"\n FROM app_script WHERE id = $1 LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "path!: String",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Int8"
]
},
"nullable": [
null
]
},
"hash": "1bae415f9440cc1334f24ce3009242cf3a6287e7b4548c7f01ad888230c27013"
}
backend/.sqlx/query-337f31c2172194cd594042c561998a03f751b246c40daf056fced0fd91f6dd73.json
Generated
+2 -1
View File
@@ -34,7 +34,8 @@
"deploymentcallback",
"singlescriptflow",
"flowscript",
"flownode"
"flownode",
"appscript"
]
}
}
backend/.sqlx/query-5d7081a9ba0d702f63ed9d44be5ffe0ba043565790865c6a6f52fab6ce340d2c.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT policy as \"policy: sqlx::types::Json<Box<RawValue>>\"\n FROM app WHERE app.path = $1 AND app.workspace_id = $2 LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "policy: sqlx::types::Json<Box<RawValue>>",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false
]
},
"hash": "5d7081a9ba0d702f63ed9d44be5ffe0ba043565790865c6a6f52fab6ce340d2c"
}
backend/.sqlx/query-ea9bbb972217bab4d7e8f4c08e331899161e80c239d56eb657d73bbf4272939b.json
Generated
+28
View File
@@ -0,0 +1,28 @@
{
"db_name": "PostgreSQL",
"query": "SELECT app_id, value FROM app_version WHERE id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "app_id",
"type_info": "Int8"
},
{
"ordinal": 1,
"name": "value",
"type_info": "Json"
}
],
"parameters": {
"Left": [
"Int8"
]
},
"nullable": [
false,
false
]
},
"hash": "ea9bbb972217bab4d7e8f4c08e331899161e80c239d56eb657d73bbf4272939b"
}
backend/.sqlx/query-ee16199b4af456198fae062e948914fca6fc0a8787e4f8d520766b1c89f23602.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO app_version_lite (id, value) VALUES ($1, $2)\n ON CONFLICT (id) DO UPDATE SET value = EXCLUDED.value",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Int8",
"Jsonb"
]
},
"nullable": []
},
"hash": "ee16199b4af456198fae062e948914fca6fc0a8787e4f8d520766b1c89f23602"
}
backend/.sqlx/query-f9fc0084fe086ef80005bb64a8bb6b493e53583017c18e2ab44f44125c52d548.json
Generated
+2 -1
View File
@@ -48,7 +48,8 @@
"deploymentcallback",
"singlescriptflow",
"flowscript",
"flownode"
"flownode",
"appscript"
]
}
}
backend/.sqlx/query-ffa86babfcab107caffb8dda31a66b142fa628b8983b966357deb3d5e0a1df3c.json
Generated
+28
View File
@@ -0,0 +1,28 @@
{
"db_name": "PostgreSQL",
"query": "SELECT lock, code FROM app_script WHERE id = $1 LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "lock",
"type_info": "Text"
},
{
"ordinal": 1,
"name": "code",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Int8"
]
},
"nullable": [
true,
false
]
},
"hash": "ffa86babfcab107caffb8dda31a66b142fa628b8983b966357deb3d5e0a1df3c"
}
backend/migrations/20241202095902_app_script.down.sql
+3
View File
@@ -0,0 +1,3 @@
-- Add down migration script here
DROP TABLE IF EXISTS app_version_lite;
DROP TABLE IF EXISTS app_script;
backend/migrations/20241202095902_app_script.up.sql
+26
View File
@@ -0,0 +1,26 @@
-- Add up migration script here
ALTER TYPE JOB_KIND ADD VALUE IF NOT EXISTS 'appscript';
-- Same as `app_version` but with a "lite" value (w/ `inlineScript.{code,lock}`).
CREATE TABLE app_version_lite (
id BIGSERIAL PRIMARY KEY,
value JSONB,
FOREIGN KEY (id) REFERENCES app_version (id) ON DELETE CASCADE
);
GRANT ALL ON app_version_lite TO windmill_user;
GRANT ALL ON app_version_lite TO windmill_admin;
-- App `inlineScript`.
CREATE TABLE app_script (
id BIGSERIAL PRIMARY KEY,
app BIGSERIAL NOT NULL,
hash CHAR(64) NOT NULL UNIQUE, -- sha256 of `app`, `lock`, `code`.
lock TEXT,
code TEXT NOT NULL,
code_sha256 CHAR(64) NOT NULL, -- used to retrieve the policy.
FOREIGN KEY (app) REFERENCES app (id) ON DELETE CASCADE
);
GRANT ALL ON app_script TO windmill_user;
GRANT ALL ON app_script TO windmill_admin;
backend/windmill-api/openapi.yaml
+21
View File
@@ -5467,6 +5467,23 @@ paths:
schema:
$ref: "#/components/schemas/AppWithLastVersion"
/w/{workspace}/apps/get/lite/{path}:
get:
summary: get app lite by path
operationId: getAppLiteByPath
tags:
- app
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/ScriptPath"
responses:
"200":
description: app lite details
content:
application/json:
schema:
$ref: "#/components/schemas/AppWithLastVersion"
/w/{workspace}/apps/get/draft/{path}:
get:
summary: get app by path with draft
@@ -5791,6 +5808,8 @@ paths:
#flow: flow/<path>
path:
type: string
version:
type: integer
args: {}
raw_code:
type: object
@@ -5808,6 +5827,8 @@ paths:
required:
- content
- language
id:
type: integer
force_viewer_static_fields:
type: object
force_viewer_one_of_fields:
backend/windmill-api/src/apps.rs
+181 -137
View File
@@ -1,4 +1,4 @@
use std::collections::HashMap;
use std::{collections::HashMap, sync::Arc};
/*
* Author: Ruben Fiszel
@@ -31,6 +31,7 @@ use axum::{
routing::{delete, get, post},
Router,
};
use futures::future::{FutureExt, TryFutureExt};
use hyper::StatusCode;
#[cfg(feature = "parquet")]
use itertools::Itertools;
@@ -50,7 +51,8 @@ use windmill_audit::ActionKind;
#[cfg(feature = "parquet")]
use windmill_common::s3_helpers::build_object_store_client;
use windmill_common::{
apps::ListAppQuery,
apps::{AppScriptId, ListAppQuery},
cache::{self, future::FutureCachedExt},
db::UserDB,
error::{to_anyhow, Error, JsonResult, Result},
jobs::{get_payload_tag_from_prefixed_path, JobPayload, RawCode},
@@ -71,6 +73,7 @@ pub fn workspaced_service() -> Router {
.route("/list", get(list_apps))
.route("/list_search", get(list_search_apps))
.route("/get/p/*path", get(get_app))
.route("/get/lite/*path", get(get_app_lite))
.route("/get/draft/*path", get(get_app_w_draft))
.route("/secret_of/*path", get(get_secret_id))
.route("/get/v/*id", get(get_app_by_id))
@@ -191,15 +194,16 @@ pub type StaticFields = HashMap<String, Box<RawValue>>;
pub type OneOfFields = HashMap<String, Vec<Box<RawValue>>>;
pub type AllowUserResources = Vec<String>;
#[derive(Serialize, Deserialize, Debug, PartialEq, Clone)]
#[derive(Serialize, Deserialize, Debug, PartialEq, Clone, Default)]
#[serde(rename_all = "lowercase")]
pub enum ExecutionMode {
#[default]
Anonymous,
Publisher,
Viewer,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
#[derive(Serialize, Deserialize, Debug, Clone, Default)]
pub struct PolicyTriggerableInputs {
static_inputs: StaticFields,
one_of_inputs: OneOfFields,
@@ -215,7 +219,7 @@ pub struct S3Input {
file_key_regex: String,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
#[derive(Serialize, Deserialize, Debug, Clone, Default)]
pub struct Policy {
pub on_behalf_of: Option<String>,
pub on_behalf_of_email: Option<String>,
@@ -410,6 +414,33 @@ async fn get_app(
Ok(Json(app))
}
async fn get_app_lite(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Path((w_id, path)): Path<(String, StripPath)>,
) -> JsonResult<AppWithLastVersion> {
let path = path.to_path();
let mut tx = user_db.begin(&authed).await?;
let app_o = sqlx::query_as::<_, AppWithLastVersion>(
"SELECT app.id, app.path, app.summary, app.versions, app.policy,
app.extra_perms, coalesce(app_version_lite.value::json, app_version.value) as value,
app_version.created_at, app_version.created_by, NULL as starred
FROM app, app_version
LEFT JOIN app_version_lite ON app_version_lite.id = app_version.id
WHERE app.path = $1 AND app.workspace_id = $2 AND app_version.id = app.versions[array_upper(app.versions, 1)]",
)
.bind(path.to_owned())
.bind(&w_id)
.fetch_optional(&mut *tx)
.await?;
tx.commit().await?;
let app = not_found_if_none(app_o, "App", path)?;
Ok(Json(app))
}
async fn get_app_w_draft(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
@@ -583,8 +614,9 @@ async fn get_public_app_by_secret(
let app_o = sqlx::query_as::<_, AppWithLastVersion>(
"SELECT app.id, app.path, app.summary, app.versions, app.policy,
null as extra_perms, app_version.value,
null as extra_perms, coalesce(app_version_lite.value::json, app_version.value::json) as value,
app_version.created_at, app_version.created_by from app, app_version
LEFT JOIN app_version_lite ON app_version_lite.id = app_version.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)
@@ -1144,6 +1176,10 @@ async fn update_app(
#[derive(Debug, Deserialize, Clone)]
pub struct ExecuteApp {
/// The app version to execute. Fallback to `path` if not provided.
pub version: Option<i64>,
/// The app script id (from the `app_script` table) to execute.
pub id: Option<i64>,
pub args: HashMap<String, Box<RawValue>>,
// - script: script/<path>
// - flow: flow/<path>
@@ -1206,6 +1242,22 @@ async fn get_on_behalf_details_from_policy_and_authed(
Ok((username, permissioned_as, email))
}
/// Convert the triggerables from the old format to the new format.
fn empty_triggerables(mut policy: Policy) -> Policy {
use std::mem::take;
if let Some(triggerables) = take(&mut policy.triggerables) {
let mut triggerables_v2 = take(&mut policy.triggerables_v2).unwrap_or_default();
for (k, static_inputs) in triggerables.into_iter() {
triggerables_v2.insert(
k,
PolicyTriggerableInputs { static_inputs, ..Default::default() },
);
}
policy.triggerables_v2 = Some(triggerables_v2);
}
policy
}
async fn execute_component(
OptAuthed(opt_authed): OptAuthed,
Extension(db): Extension<DB>,
@@ -1228,99 +1280,136 @@ async fn execute_component(
};
let path = path.to_path();
let (arc_policy, policy): (Arc<Policy>, Policy);
let policy_triggerables_default = Default::default();
let policy = match payload.clone() {
// Two cases here:
// 1. The component is executed from the editor (i.e. in "preview" mode), then:
// - The policy is set to default (in `Viewer` execution mode).
// - The policy triggerables are built by the frontend and retrieved from the request
// payload.
// - In case of inline script, the `RawCode` from the request is pushed as is to the
// job queue.
// 2. Otherwise (i.e. "run" mode):
// - The policy and triggerables are fetched from the database.
// - In case of inline script, if an entry exists in the `app_script` table, push
// an `AppScript` job payload, as in (.1) otherwise.
let (policy, policy_triggerables) = match payload {
// 1. "preview" mode.
ExecuteApp {
force_viewer_static_fields: Some(static_fields),
force_viewer_one_of_fields: Some(one_of_fields),
force_viewer_static_fields: Some(static_inputs),
force_viewer_one_of_fields: Some(one_of_inputs),
force_viewer_allow_user_resources: Some(allow_user_resources),
..
} => {
let mut hm = HashMap::new();
if let Some(path) = payload.path.clone() {
hm.insert(
format!("{}:{path}", payload.component),
PolicyTriggerableInputs {
static_inputs: static_fields,
one_of_inputs: one_of_fields,
allow_user_resources,
},
);
} else {
hm.insert(
format!(
"{}:{}",
payload.component,
digest(payload.raw_code.clone().unwrap().content.as_str())
),
PolicyTriggerableInputs {
static_inputs: static_fields,
one_of_inputs: one_of_fields,
allow_user_resources,
},
);
}
Policy {
} => (
&Policy {
execution_mode: ExecutionMode::Viewer,
triggerables: None,
triggerables_v2: Some(hm),
on_behalf_of: None,
on_behalf_of_email: None,
s3_inputs: None,
}
}
..Default::default()
},
&PolicyTriggerableInputs {
static_inputs,
one_of_inputs,
allow_user_resources,
},
),
// 2. "run" mode.
_ => {
let policy_o = sqlx::query_scalar!(
"SELECT policy from app WHERE path = $1 AND workspace_id = $2",
// Policy is fetched from the database on app `path` and `workspace_id`.
let policy_fut = sqlx::query_scalar!(
"SELECT policy as \"policy: sqlx::types::Json<Box<RawValue>>\"
FROM app WHERE app.path = $1 AND app.workspace_id = $2 LIMIT 1",
path,
&w_id
&w_id,
)
.fetch_optional(&db)
.await?;
.map_err(Into::<Error>::into)
.map(|policy_o| Result::Ok(not_found_if_none(policy_o?, "App", path)?))
.map(|policy| Result::Ok(serde_json::from_str(policy?.get())?))
.map_ok(empty_triggerables);
let policy = not_found_if_none(policy_o, "App", path)?;
// 1. The app `version` is provided: cache the fetched policy.
// 2. Otherwise, always fetch the policy from the database.
let policy = if let Some(id) = payload.version {
let cache = cache::anon!({ u64 => Arc<Policy> } in "policy" <= 1000);
arc_policy = policy_fut
.map_ok(Arc::new)
.cached(cache, &(id as u64))
.await?;
&*arc_policy
} else {
policy = policy_fut.await?;
&policy
};
serde_json::from_value::<Policy>(policy).map_err(to_anyhow)?
// Compute the path for the triggerables map:
// - flow: `flow/<payload.path>`
// - script: `script/<payload.path>`
// - inline script: `rawscript/<sha256(raw_code.content)>`
let path = match &payload {
// flow or script: just use the `payload.path`.
ExecuteApp { path: Some(path), .. } => path,
// inline script: without entry in the `app_script` table.
ExecuteApp { raw_code: Some(raw_code), id: None, .. } => &digest(&raw_code.content),
// inline script: with an entry in the `app_script` table.
ExecuteApp { raw_code: Some(_), id: Some(id), .. } => {
let cache = cache::anon!({ u64 => Arc<String> } in "appscriptpath" <= 10000);
// `id` is unique, cache the result.
&*sqlx::query_scalar!(
"SELECT format('rawscript/%s', code_sha256) as \"path!: String\"
FROM app_script WHERE id = $1 LIMIT 1",
id
)
.fetch_one(&db)
.map_err(Into::<Error>::into)
.map_ok(Arc::new)
.cached(cache, &(*id as u64))
.await?
}
_ => unreachable!(),
};
// Retrieve the triggerables from the policy on `path` or `<component>:<path>`.
let triggerables_v2 = policy
.triggerables_v2
.as_ref()
.ok_or_else(|| Error::BadRequest(format!("Policy is missing triggerables")))?;
let policy_triggerables = triggerables_v2
.get(path) // start with `path` in case we can avoid the next` format!`.
.or_else(|| triggerables_v2.get(&format!("{}:{}", payload.component, &path)))
.or(match policy.execution_mode {
ExecutionMode::Viewer => Some(&policy_triggerables_default),
_ => None,
})
.ok_or_else(|| Error::BadRequest(format!("Path {path} forbidden by policy")))?;
(policy, policy_triggerables)
}
};
let (username, permissioned_as, email) =
get_on_behalf_details_from_policy_and_authed(&policy, &opt_authed).await?;
let (job_payload, (args, job_id), tag) = match payload {
ExecuteApp { args, component, raw_code: Some(raw_code), path: None, .. } => {
let content = &raw_code.content;
let payload = JobPayload::Code(raw_code.clone());
let path = digest(content);
let args = build_args(
policy,
&component,
path,
args,
opt_authed.as_ref(),
&user_db,
&db,
&w_id,
)
.await?;
(payload, args, None)
}
ExecuteApp { args, component, raw_code: None, path: Some(path), .. } => {
let (payload, tag) = get_payload_tag_from_prefixed_path(&path, &db, &w_id).await?;
let args = build_args(
policy,
&component,
path.to_string(),
args,
opt_authed.as_ref(),
&user_db,
&db,
&w_id,
)
.await?;
(payload, args, tag)
}
let (args, job_id) = build_args(
policy,
policy_triggerables,
payload.args,
opt_authed.as_ref(),
&user_db,
&db,
&w_id,
)
.await?;
let (job_payload, tag) = match (payload.path, payload.raw_code, payload.id) {
// flow or script:
(Some(path), None, None) => get_payload_tag_from_prefixed_path(&path, &db, &w_id).await?,
// inline script: in "preview" mode or without entry in the `app_script` table.
(None, Some(raw_code), None) => (JobPayload::Code(raw_code), None),
// inline script: in "run" mode and with an entry in the `app_script` table.
(None, Some(RawCode { language, path, cache_ttl, .. }), Some(id)) => (
JobPayload::AppScript { id: AppScriptId(id), cache_ttl, language, path },
None,
),
_ => unreachable!(),
};
let tx = windmill_queue::PushIsolationLevel::IsolatedRoot(db.clone());
@@ -1655,9 +1744,12 @@ async fn exists_app(
}
async fn build_args(
policy: Policy,
component: &str,
path: String,
policy: &Policy,
PolicyTriggerableInputs {
static_inputs,
one_of_inputs,
allow_user_resources,
}: &PolicyTriggerableInputs,
mut args: HashMap<String, Box<RawValue>>,
authed: Option<&ApiAuthed>,
user_db: &UserDB,
@@ -1665,54 +1757,6 @@ async fn build_args(
w_id: &str,
) -> Result<(PushArgsOwned, Option<Uuid>)> {
let mut job_id: Option<Uuid> = None;
let key = format!("{}:{}", component, &path);
let (static_inputs, one_of_inputs, allow_user_resources) = match policy {
Policy { triggerables_v2: Some(t), .. } => {
let PolicyTriggerableInputs { static_inputs, one_of_inputs, allow_user_resources } = t
.get(&key)
.or_else(|| t.get(&path))
.map(|x| x.clone())
.or_else(|| {
if matches!(policy.execution_mode, ExecutionMode::Viewer) {
Some(PolicyTriggerableInputs {
static_inputs: HashMap::new(),
one_of_inputs: HashMap::new(),
allow_user_resources: Vec::new(),
})
} else {
None
}
})
.ok_or_else(|| {
Error::BadRequest(format!("path {} is not allowed in the app policy", path))
})?;
(static_inputs, one_of_inputs, allow_user_resources)
}
Policy { triggerables: Some(t), .. } => {
let static_inputs = t
.get(&key)
.or_else(|| t.get(&path))
.map(|x| x.clone())
.or_else(|| {
if matches!(policy.execution_mode, ExecutionMode::Viewer) {
Some(HashMap::new())
} else {
None
}
})
.ok_or_else(|| {
Error::BadRequest(format!("path {} is not allowed in the app policy", path))
})?;
(static_inputs, HashMap::new(), Vec::new())
}
_ => Err(Error::BadRequest(format!(
"Policy is missing triggerables for {}",
key
)))?,
};
let mut safe_args = HashMap::<String, Box<RawValue>>::new();
// tracing::error!("{:?}", allow_user_resources);
@@ -1761,16 +1805,16 @@ async fn build_args(
}
for (k, v) in one_of_inputs {
if safe_args.contains_key(&k) {
if safe_args.contains_key(k) {
continue;
}
if let Some(arg_val) = args.get(&k) {
if let Some(arg_val) = args.get(k) {
let arg_str = arg_val.get();
let options_str_vec = v.iter().map(|x| x.get()).collect::<Vec<&str>>();
if options_str_vec.contains(&arg_str) {
safe_args.insert(k.to_string(), arg_val.clone());
args.remove(&k);
args.remove(k);
continue;
}
@@ -1781,7 +1825,7 @@ async fn build_args(
.all(|x| options_str_vec.contains(&x.get()))
{
safe_args.insert(k.to_string(), arg_val.clone());
args.remove(&k);
args.remove(k);
continue;
}
}
backend/windmill-common/src/apps.rs
+12 -1
View File
@@ -6,7 +6,18 @@
* LICENSE-AGPL for a copy of the license.
*/
use serde::Deserialize;
use serde::{Deserialize, Serialize};
/// Id in the `app_script` table.
#[derive(Serialize, Deserialize, Debug, Copy, Clone, Hash, Eq, PartialEq)]
#[serde(transparent)]
pub struct AppScriptId(pub i64);
impl Into<u64> for AppScriptId {
fn into(self) -> u64 {
self.0 as u64
}
}
#[derive(Deserialize)]
pub struct ListAppQuery {
backend/windmill-common/src/cache.rs
+90
View File
@@ -394,6 +394,96 @@ pub mod script {
}
}
pub mod app {
use super::*;
use crate::apps::AppScriptId;
make_static! {
/// App scripts cache.
/// FIXME: Use `Arc<Val>` for cheap cloning.
static ref CACHE: { AppScriptId => Val } in "app" <= 1000;
}
/// App app script cache value.
#[derive(Debug, Clone, Default)]
pub struct Val {
pub lock: Option<String>,
pub code: String,
}
/// Fetch the app script referenced by `id` from the cache.
/// If not present, import from the file-system cache or fetch it from the database and write
/// it to the file system and cache.
/// This should be preferred over fetching the database directly.
pub async fn fetch_script(
e: impl PgExecutor<'_>,
id: AppScriptId,
) -> error::Result<(Option<String>, String)> {
// If not present, `get_or_insert_async` will lock the key until the future completes,
// so only one thread will be able to fetch the data from the database and write it to
// the file system and cache, hence no race on the file system.
CACHE
.get_or_insert_async(&id, async {
sqlx::query!(
"SELECT lock, code FROM app_script WHERE id = $1 LIMIT 1",
id.0,
)
.fetch_one(e)
.await
.map_err(Into::into)
.map(|r| Val {
lock: r
.lock
.and_then(|x| if x.is_empty() { None } else { Some(x) }),
code: r.code,
})
})
.await
.map(|Val { lock, code }| (lock, code))
}
// ----------------------------------------------------------------------------------------------
// impl `fs::Bundle` for `Val`.
#[derive(Copy, Clone)]
pub enum Item {
Lock,
Code,
}
impl fs::Item for Item {
fn path(&self, root: &Path) -> PathBuf {
match self {
Item::Lock => root.join("lock.txt"),
Item::Code => root.join("code.txt"),
}
}
}
impl fs::Bundle for Val {
type Item = Item;
fn items() -> &'static [Self::Item] {
&[Item::Lock, Item::Code]
}
fn import(&mut self, item: Self::Item, data: Vec<u8>) -> error::Result<()> {
match item {
Item::Lock => self.lock = Some(String::from_utf8(data)?),
Item::Code => self.code = String::from_utf8(data)?,
}
Ok(())
}
fn export(&self, item: Self::Item) -> error::Result<Option<Vec<u8>>> {
match item {
Item::Lock => Ok(self.lock.as_ref().map(|s| s.as_bytes().to_vec())),
Item::Code => Ok(Some(self.code.as_bytes().to_vec())),
}
}
}
}
mod fs {
use super::*;
backend/windmill-common/src/jobs.rs
+8
View File
@@ -14,6 +14,7 @@ pub const ENTRYPOINT_OVERRIDE: &str = "_ENTRYPOINT_OVERRIDE";
pub const PREPROCESSOR_FAKE_ENTRYPOINT: &str = "__WM_PREPROCESSOR";
use crate::{
apps::AppScriptId,
error::{self, to_anyhow, Error},
flow_status::{FlowStatus, RestartedFrom},
flows::{FlowNodeId, FlowValue, Retry},
@@ -41,6 +42,7 @@ pub enum JobKind {
DeploymentCallback,
FlowScript,
FlowNode,
AppScript,
}
#[derive(sqlx::FromRow, Debug, Serialize, Clone)]
@@ -278,6 +280,12 @@ pub enum JobPayload {
id: FlowNodeId, // flow_node(id).
path: String, // flow node inner path (e.g. `outer/branchall-42`).
},
AppScript {
id: AppScriptId, // app_script(id).
path: Option<String>,
language: ScriptLang,
cache_ttl: Option<i32>,
},
Code(RawCode),
Dependencies {
path: String,
backend/windmill-queue/src/jobs.rs
+22 -1
View File
@@ -3228,7 +3228,27 @@ pub async fn push<'c, 'd>(
None,
None,
)
}
},
JobPayload::AppScript {
id, // app_script(id).
path,
language,
cache_ttl,
} => (
Some(id.0),
path,
None,
JobKind::AppScript,
None,
None,
Some(language),
None,
None,
None,
cache_ttl,
None,
None,
),
JobPayload::ScriptHub { path } => {
if path == "hub/7771/slack" || path == "hub/7836/slack" {
permissioned_as = SUPERADMIN_NOTIFICATION_EMAIL.to_string();
@@ -3989,6 +4009,7 @@ pub async fn push<'c, 'd>(
JobKind::DeploymentCallback => "jobs.run.deployment_callback",
JobKind::FlowScript => "jobs.run.flow_script",
JobKind::FlowNode => "jobs.run.flow_node",
JobKind::AppScript => "jobs.run.app_script",
};
let audit_author = if format!("u/{user}") != permissioned_as && user != permissioned_as {
backend/windmill-worker/src/worker.rs
+15
View File
@@ -7,6 +7,7 @@
*/
use windmill_common::{
apps::AppScriptId,
auth::{fetch_authed_from_permissioned_as, JWTAuthClaims, JobPerms, JWT_SECRET},
scripts::PREVIEW_IS_TAR_CODEBASE_HASH,
utils::WarnAfterExt,
@@ -2311,6 +2312,20 @@ async fn handle_code_execution_job(
codebase: None,
}
}
JobKind::AppScript => {
let (lockfile, content) = cache::app::fetch_script(
db,
AppScriptId(job.script_hash.unwrap_or(ScriptHash(0)).0),
)
.await?;
ContentReqLangEnvs {
content,
lockfile,
language: job.language.to_owned(),
envs: None,
codebase: None,
}
}
JobKind::DeploymentCallback => {
get_script_content_by_path(job.script_path.clone(), &job.workspace_id, db).await?
}
backend/windmill-worker/src/worker_lockfiles.rs
+94 -7
View File
@@ -15,6 +15,7 @@ use windmill_common::jobs::JobPayload;
use windmill_common::scripts::ScriptHash;
use windmill_common::worker::{to_raw_value, to_raw_value_owned, write_file};
use windmill_common::{
apps::AppScriptId,
error::{self, to_anyhow},
flows::{add_virtual_items_if_necessary, FlowValue},
jobs::QueuedJob,
@@ -648,7 +649,7 @@ pub async fn handle_flow_dependency_job(
// Compute a lite version of the flow value (`RawScript` => `FlowScript`).
let mut value_lite = flow.clone();
tx = reduce(
tx = reduce_flow(
tx,
&mut value_lite.modules,
&job_path,
@@ -1053,6 +1054,41 @@ async fn insert_flow_node<'c>(
Ok((tx, FlowNodeId(id)))
}
async fn insert_app_script(
db: &sqlx::Pool<sqlx::Postgres>,
app: i64,
code: String,
lock: Option<String>,
) -> Result<AppScriptId> {
let code_sha256 = format!("{:x}", sha2::Sha256::digest(&code));
let hash = {
let mut hasher = sha2::Sha256::new();
hasher.update(app.to_le_bytes());
hasher.update(&code_sha256);
hasher.update(lock.as_ref().unwrap_or(&Default::default()));
format!("{:x}", hasher.finalize())
};
// Insert the app script if it doesn't exist.
sqlx::query_scalar!(
r#"
INSERT INTO app_script (app, hash, lock, code, code_sha256)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (hash) DO UPDATE SET app = EXCLUDED.app -- trivial update to return the id
RETURNING id
"#,
app,
hash,
lock,
code,
code_sha256
)
.fetch_one(db)
.await
.map(AppScriptId)
.map_err(Into::into)
}
async fn insert_flow_modules<'c>(
mut tx: sqlx::Transaction<'c, sqlx::Postgres>,
path: &str,
@@ -1062,7 +1098,7 @@ async fn insert_flow_modules<'c>(
modules: &mut Vec<FlowModule>,
modules_node: &mut Option<FlowNodeId>,
) -> Result<sqlx::Transaction<'c, sqlx::Postgres>> {
tx = Box::pin(reduce(
tx = Box::pin(reduce_flow(
tx,
modules,
path,
@@ -1094,7 +1130,7 @@ async fn insert_flow_modules<'c>(
Ok(tx)
}
async fn reduce<'c>(
async fn reduce_flow<'c>(
mut tx: sqlx::Transaction<'c, sqlx::Postgres>,
modules: &mut Vec<FlowModule>,
path: &str,
@@ -1107,7 +1143,7 @@ async fn reduce<'c>(
let mut val =
serde_json::from_str::<FlowModuleValue>(module.value.get()).map_err(|err| {
Error::InternalErr(format!(
"reduce: Failed to parse flow module value: {}",
"reduce_flow: Failed to parse flow module value: {}",
err
))
})?;
@@ -1203,6 +1239,41 @@ async fn reduce<'c>(
Ok(tx)
}
async fn reduce_app(db: &sqlx::Pool<sqlx::Postgres>, value: &mut Value, app: i64) -> Result<()> {
match value {
Value::Object(object) => {
if let Some(Value::Object(script)) = object.get_mut("inlineScript") {
// replace `content` with an empty string:
let Some(Value::String(code)) = script.get_mut("content").map(std::mem::take)
else {
return Err(error::Error::InternalErr(
"Missing `content` in inlineScript".to_string(),
));
};
// remove `lock`:
let lock = script.remove("lock").and_then(|x| match x {
Value::String(s) => Some(s),
_ => None,
});
let id = insert_app_script(db, app, code, lock).await?;
// insert the `id` into the `script` object:
script.insert("id".to_string(), json!(id.0));
} else {
for (_, value) in object {
Box::pin(reduce_app(db, value, app)).await?;
}
}
}
Value::Array(array) => {
for value in array {
Box::pin(reduce_app(db, value, app)).await?;
}
}
_ => {}
}
Ok(())
}
fn skip_creating_new_lock(language: &ScriptLang, content: &str) -> bool {
if language == &ScriptLang::Bun || language == &ScriptLang::Bunnative {
let anns = windmill_common::worker::TypeScriptAnnotations::parse(&content);
@@ -1388,11 +1459,12 @@ pub async fn handle_app_dependency_job(
.clone()
.ok_or_else(|| Error::InternalErr("App Dependency requires script hash".to_owned()))?
.0;
let value = sqlx::query_scalar!("SELECT value FROM app_version WHERE id = $1", id)
let record = sqlx::query!("SELECT app_id, value FROM app_version WHERE id = $1", id)
.fetch_optional(db)
.await?;
.await?
.map(|record| (record.app_id, record.value));
if let Some(value) = value {
if let Some((app_id, value)) = record {
let value = lock_modules_app(
value,
job,
@@ -1409,6 +1481,21 @@ pub async fn handle_app_dependency_job(
)
.await?;
// Compute a lite version of the app value (w/ `inlineScript.{lock,code}`).
let mut value_lite = value.clone();
reduce_app(db, &mut value_lite, app_id).await?;
if let Value::Object(object) = &mut value_lite {
object.insert("version".to_string(), json!(id));
}
sqlx::query!(
"INSERT INTO app_version_lite (id, value) VALUES ($1, $2)
ON CONFLICT (id) DO UPDATE SET value = EXCLUDED.value",
id,
sqlx::types::Json(to_raw_value(&value_lite)) as sqlx::types::Json<Box<RawValue>>,
)
.execute(db)
.await?;
// Re-check cancelation to ensure we don't accidentially override an app.
if sqlx::query_scalar!("SELECT canceled FROM queue WHERE id = $1", job.id)
.fetch_optional(db)
frontend/src/lib/components/apps/components/helpers/RunnableComponent.svelte
+9 -2
View File
@@ -377,11 +377,14 @@
: runnable
if (inlineScript) {
if (inlineScript.id !== undefined) {
requestBody['id'] = inlineScript.id
}
requestBody['raw_code'] = {
content: inlineScript.content,
content: inlineScript.id === undefined ? inlineScript.content : '',
language: inlineScript.language ?? '',
path: inlineScript.path,
lock: inlineScript.lock,
lock: inlineScript.id === undefined ? inlineScript.lock : undefined,
cache_ttl: inlineScript.cache_ttl
}
}
@@ -390,6 +393,10 @@
requestBody['path'] = runType !== 'hubscript' ? `${runType}/${path}` : `script/${path}`
}
if ($app.version !== undefined) {
requestBody['version'] = $app.version
}
const uuid = await AppService.executeComponent({
workspace,
path: defaultIfEmptyString($appPath, `u/${$userStore?.username ?? 'unknown'}/newapp`),
frontend/src/lib/components/apps/types.ts
+2
View File
@@ -117,6 +117,7 @@ export type InlineScript = {
cache_ttl?: number
refreshOn?: { id: string; key: string }[]
suggestedRefreshOn?: { id: string; key: string }[]
id?: number
}
export type AppCssItemName = 'viewer' | 'grid' | AppComponent['type']
@@ -163,6 +164,7 @@ export type App = {
theme: AppTheme | undefined
hideLegacyTopBar?: boolean | undefined
mobileViewOnSmallerScreens?: boolean | undefined
version?: number
}
export type ConnectingInput = {
frontend/src/routes/(root)/(logged)/apps/get/[...path]/+page.svelte
+1 -1
View File
@@ -17,7 +17,7 @@
let can_write = false
async function loadApp() {
app = await AppService.getAppByPath({ workspace: $workspaceStore!, path: $page.params.path })
app = await AppService.getAppLiteByPath({ workspace: $workspaceStore!, path: $page.params.path })
can_write = canWrite(app?.path, app?.extra_perms!, $userStore)
}