feat: add unique id (#2483)

* feat: add unique id

* fix: sqlx prepare

* feat: add disable option

* fix: cron schedule

backend/Cargo.lock

@@ -7828,6 +7828,8 @@ dependencies = [
  "anyhow",
  "axum",
  "chrono",
+ "cron",
+ "git-version",
  "hex",
  "hmac",
  "hyper",

backend/migrations/20231019154138_add_uid.down.sql

@@ -0,0 +1,2 @@
+-- Add down migration script here
+DELETE FROM global_settings WHERE name = 'uid';

backend/migrations/20231019154138_add_uid.up.sql

@@ -0,0 +1,2 @@
+-- Add up migration script here
+INSERT INTO global_settings (name, value, updated_at) VALUES ('uid', to_jsonb(gen_random_uuid()), now()) ON CONFLICT DO NOTHING;

backend/migrations/20231022165924_update_hub_sync_stats.down.sql

@@ -0,0 +1 @@
+-- Add down migration script here

backend/migrations/20231022165924_update_hub_sync_stats.up.sql

@@ -0,0 +1,15 @@
+-- Add up migration script here
+UPDATE script SET content = 'import wmill from "https://deno.land/x/wmill@v1.189.0/main.ts";
+export async function main() {
+  await run(
+    "workspace", "add", "__automation", "admins", Deno.env.get("BASE_INTERNAL_URL") + "/", "--token", Deno.env.get("WM_TOKEN"));
+
+  await run("hub", "pull");
+}
+
+async function run(...cmd: string[]) {
+  console.log("Running \"" + cmd.join('' '') + "\"");
+  await wmill.parse(cmd);
+}', summary = 'Synchronize Hub Resource types with instance',
+description = 'Basic administrative script to sync latest resource types from hub to share to every workspace. Recommended to run at least once. On a schedule by default.'
+WHERE hash = -28028598712388162 AND workspace_id = 'admins';

backend/src/main.rs

@@ -20,12 +20,14 @@ use tokio::{
     fs::{metadata, DirBuilder},
     sync::RwLock,
 };
+use windmill_api::HTTP_CLIENT;
 use windmill_common::{
     global_settings::{
-        BASE_URL_SETTING, CUSTOM_TAGS_SETTING, ENV_SETTINGS, EXTRA_PIP_INDEX_URL_SETTING,
-        LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING,
-        REQUEST_SIZE_LIMIT_SETTING, RETENTION_PERIOD_SECS_SETTING,
+        BASE_URL_SETTING, CUSTOM_TAGS_SETTING, DISABLE_STATS_SETTING, ENV_SETTINGS,
+        EXTRA_PIP_INDEX_URL_SETTING, LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING,
+        OAUTH_SETTING, REQUEST_SIZE_LIMIT_SETTING, RETENTION_PERIOD_SECS_SETTING,
     },
+    stats::schedule_stats,
     utils::rd_string,
     worker::{reload_custom_tags_setting, WORKER_GROUP},
     DB, METRICS_ADDR,
@@ -334,7 +336,8 @@ Windmill Community Edition {GIT_VERSION}
                                                     if let Err(e) = tx.send(()) {
                                                         tracing::error!(error = %e, "Could not send killpill to server");
                                                     }
-                                                }
+                                                },
+                                                DISABLE_STATS_SETTING => {},
                                                 a @_ => {
                                                     tracing::info!("Unrecognized Global Setting Change Payload: {:?}", a);
                                                 }
@@ -378,6 +381,11 @@ Windmill Community Edition {GIT_VERSION}
             Ok(()) as anyhow::Result<()>
         };
 
+        if mode == Mode::Server || mode == Mode::Standalone {
+            let instance_name = rd_string(8);
+            schedule_stats(&db, instance_name, &HTTP_CLIENT).await;
+        }
+
         futures::try_join!(shutdown_signal, server_f, metrics_f, workers_f, monitor_f)?;
     } else {
         tracing::info!("Nothing to do, exiting.");

backend/windmill-api/src/apps.rs

@@ -547,12 +547,12 @@ async fn create_app(
     Ok((StatusCode::CREATED, app.path))
 }
 
-async fn list_hub_apps(ApiAuthed { email, .. }: ApiAuthed) -> impl IntoResponse {
+async fn list_hub_apps(Extension(db): Extension<DB>) -> impl IntoResponse {
     let (status_code, headers, response) = query_elems_from_hub(
         &HTTP_CLIENT,
         "https://hub.windmill.dev/searchUiData?approved=true",
-        &email,
         None,
+        &db,
     )
     .await?;
     Ok::<_, Error>((
@@ -563,15 +563,15 @@ async fn list_hub_apps(ApiAuthed { email, .. }: ApiAuthed) -> impl IntoResponse
 }
 
 pub async fn get_hub_app_by_id(
-    ApiAuthed { email, .. }: ApiAuthed,
     Path(id): Path<i32>,
+    Extension(db): Extension<DB>,
 ) -> JsonResult<serde_json::Value> {
     let value = http_get_from_hub(
         &HTTP_CLIENT,
         &format!("https://hub.windmill.dev/apps/{id}/json"),
-        &email,
         false,
         None,
+        &db,
     )
     .await?
     .json()

backend/windmill-api/src/embeddings.rs

@@ -228,9 +228,9 @@ impl EmbeddingsDb {
         let response = http_get_from_hub(
             &HTTP_CLIENT,
             "https://hub.windmill.dev/scripts/embeddings",
-            "todo@windmill.dev",
             false,
             None,
+            pg_db,
         )
         .await?;
         let hub_scripts = response.json::<Vec<HubScript>>().await?;
@@ -257,9 +257,9 @@ impl EmbeddingsDb {
         let response = http_get_from_hub(
             &HTTP_CLIENT,
             "https://hub.windmill.dev/resource_types/embeddings",
-            "todo@windmill.dev",
             false,
             None,
+            pg_db,
         )
         .await?;
         let hub_resource_types = response.json::<Vec<HubResourceType>>().await?;

backend/windmill-api/src/flows.rs

@@ -163,12 +163,12 @@ async fn list_flows(
     Ok(Json(rows))
 }
 
-async fn list_hub_flows(ApiAuthed { email, .. }: ApiAuthed) -> impl IntoResponse {
+async fn list_hub_flows(Extension(db): Extension<DB>) -> impl IntoResponse {
     let (status_code, headers, response) = query_elems_from_hub(
         &HTTP_CLIENT,
         "https://hub.windmill.dev/searchFlowData?approved=true",
-        &email,
         None,
+        &db,
     )
     .await?;
     Ok::<_, Error>((
@@ -197,15 +197,15 @@ async fn list_paths(
 }
 
 pub async fn get_hub_flow_by_id(
-    ApiAuthed { email, .. }: ApiAuthed,
     Path(id): Path<i32>,
+    Extension(db): Extension<DB>,
 ) -> JsonResult<serde_json::Value> {
     let value = http_get_from_hub(
         &HTTP_CLIENT,
         &format!("https://hub.windmill.dev/flows/{id}/json"),
-        &email,
         false,
         None,
+        &db,
     )
     .await?
     .json()

backend/windmill-api/src/integration.rs

@@ -1,5 +1,7 @@
-use crate::{db::ApiAuthed, HTTP_CLIENT};
-use axum::{body::StreamBody, extract::Query, response::IntoResponse, routing::get, Router};
+use crate::{db::DB, HTTP_CLIENT};
+use axum::{
+    body::StreamBody, extract::Query, response::IntoResponse, routing::get, Extension, Router,
+};
 use windmill_common::{error::Error, utils::query_elems_from_hub};
 
 pub fn global_service() -> Router {
@@ -11,8 +13,8 @@ struct ListHubIntegrationsQuery {
     kind: Option<String>,
 }
 async fn list_hub_integrations(
-    ApiAuthed { email, .. }: ApiAuthed,
     Query(query): Query<ListHubIntegrationsQuery>,
+    Extension(db): Extension<DB>,
 ) -> impl IntoResponse {
     let mut query_params = vec![];
 
@@ -23,8 +25,8 @@ async fn list_hub_integrations(
     let (status_code, headers, response) = query_elems_from_hub(
         &HTTP_CLIENT,
         "https://hub.windmill.dev/integrations/list",
-        &email,
         Some(query_params),
+        &db,
     )
     .await?;
     Ok::<_, Error>((

backend/windmill-api/src/scripts.rs

@@ -259,8 +259,8 @@ struct TopHubScriptsQuery {
 }
 
 async fn get_top_hub_scripts(
-    ApiAuthed { email, .. }: ApiAuthed,
     Query(query): Query<TopHubScriptsQuery>,
+    Extension(db): Extension<DB>,
 ) -> impl IntoResponse {
     let mut query_params = vec![];
     if let Some(query_limit) = query.limit {
@@ -276,8 +276,8 @@ async fn get_top_hub_scripts(
     let (status_code, headers, response) = query_elems_from_hub(
         &HTTP_CLIENT,
         "https://hub.windmill.dev/scripts/top",
-        &email,
         Some(query_params),
+        &db,
     )
     .await?;
     Ok::<_, Error>((
@@ -638,18 +638,18 @@ async fn create_script(
 }
 
 pub async fn get_hub_script_by_path(
-    authed: ApiAuthed,
     Path(path): Path<StripPath>,
+    Extension(db): Extension<DB>,
 ) -> Result<String> {
-    windmill_common::scripts::get_hub_script_by_path(&authed.email, path, &HTTP_CLIENT).await
+    windmill_common::scripts::get_hub_script_by_path(path, &HTTP_CLIENT, &db).await
 }
 
 pub async fn get_full_hub_script_by_path(
-    ApiAuthed { email, .. }: ApiAuthed,
     Path(path): Path<StripPath>,
+    Extension(db): Extension<DB>,
 ) -> JsonResult<HubScript> {
     Ok(Json(
-        windmill_common::scripts::get_full_hub_script_by_path(&email, path, &HTTP_CLIENT).await?,
+        windmill_common::scripts::get_full_hub_script_by_path(path, &HTTP_CLIENT, &db).await?,
     ))
 }
 

backend/windmill-common/Cargo.toml

@@ -46,3 +46,5 @@ lazy_static.workspace = true
 tracing-flame = { version = "^0", optional = true }
 itertools.workspace = true
 regex.workspace = true
+git-version.workspace = true
+cron.workspace = true

backend/windmill-common/src/global_settings.rs

@@ -7,6 +7,8 @@ pub const REQUEST_SIZE_LIMIT_SETTING: &str = "request_size_limit_mb";
 pub const LICENSE_KEY_SETTING: &str = "license_key";
 pub const NPM_CONFIG_REGISTRY_SETTING: &str = "npm_config_registry";
 pub const EXTRA_PIP_INDEX_URL_SETTING: &str = "pip_extra_index_url";
+pub const UNIQUE_ID_SETTING: &str = "uid";
+pub const DISABLE_STATS_SETTING: &str = "disable_stats";
 
 pub const ENV_SETTINGS: [&str; 54] = [
     "DISABLE_NSJAIL",

backend/windmill-common/src/lib.rs

@@ -25,6 +25,7 @@ pub mod oauth2;
 pub mod schedule;
 pub mod scripts;
 pub mod server;
+pub mod stats;
 pub mod users;
 pub mod utils;
 pub mod variables;

backend/windmill-common/src/scripts.rs

@@ -11,6 +11,11 @@ use std::{
     hash::{Hash, Hasher},
 };
 
+use crate::{
+    error::{to_anyhow, Error},
+    utils::http_get_from_hub,
+    DB,
+};
 use serde::de::Error as _;
 use serde::{ser::SerializeSeq, Deserialize, Deserializer, Serialize};
 use serde_json::to_string_pretty;
@@ -263,15 +268,10 @@ pub fn to_hex_string(i: &i64) -> String {
 
 #[cfg(feature = "reqwest")]
 pub async fn get_hub_script_by_path(
-    email: &str,
     path: StripPath,
     http_client: &reqwest::Client,
+    db: &DB,
 ) -> crate::error::Result<String> {
-    use crate::{
-        error::{to_anyhow, Error},
-        utils::http_get_from_hub,
-    };
-
     let path = path
         .to_path()
         .strip_prefix("hub/")
@@ -280,9 +280,9 @@ pub async fn get_hub_script_by_path(
     let content = http_get_from_hub(
         http_client,
         &format!("https://hub.windmill.dev/raw/{path}.ts"),
-        email,
         true,
         None,
+        db,
     )
     .await?
     .text()
@@ -293,15 +293,10 @@ pub async fn get_hub_script_by_path(
 
 #[cfg(feature = "reqwest")]
 pub async fn get_full_hub_script_by_path(
-    email: &str,
     path: StripPath,
     http_client: &reqwest::Client,
+    db: &DB,
 ) -> crate::error::Result<HubScript> {
-    use crate::{
-        error::{to_anyhow, Error},
-        utils::http_get_from_hub,
-    };
-
     let path = path
         .to_path()
         .strip_prefix("hub/")
@@ -310,9 +305,9 @@ pub async fn get_full_hub_script_by_path(
     let value = http_get_from_hub(
         http_client,
         &format!("https://hub.windmill.dev/raw2/{path}"),
-        email,
         true,
         None,
+        db,
     )
     .await?
     .json::<HubScript>()

backend/windmill-common/src/stats.rs

@@ -0,0 +1,110 @@
+use std::str::FromStr;
+
+use crate::{
+    error::{to_anyhow, Result},
+    global_settings::{DISABLE_STATS_SETTING, UNIQUE_ID_SETTING},
+    utils::GIT_VERSION,
+    DB,
+};
+
+use chrono::Utc;
+use cron::Schedule;
+
+pub async fn get_disable_stats_setting(db: &DB) -> bool {
+    let q = sqlx::query!(
+        "SELECT value FROM global_settings WHERE name = $1",
+        DISABLE_STATS_SETTING
+    )
+    .fetch_optional(db)
+    .await;
+
+    if let Ok(q) = q {
+        if let Some(q) = q {
+            if let Ok(v) = serde_json::from_value::<bool>(q.value.clone()) {
+                return v;
+            } else {
+                tracing::error!(
+                    "Could not parse DISABLE_STATS_SETTING found: {:#?}",
+                    &q.value
+                );
+            }
+        }
+    };
+
+    false
+}
+
+pub async fn schedule_stats(db: &DB, instance_name: String, http_client: &reqwest::Client) -> () {
+    let http_client = http_client.clone();
+    let db = db.clone();
+    tokio::spawn(async move {
+        loop {
+            let disabled = get_disable_stats_setting(&db).await;
+            if !disabled {
+                tracing::info!("Sending stats");
+                let result = send_stats(&instance_name, &http_client, &db).await;
+                if result.is_err() {
+                    tracing::error!("Error sending stats: {}", result.err().unwrap());
+                } else {
+                    tracing::info!("Stats sent");
+                }
+            }
+
+            let s = "0 0 */24 * * * *"; // Every 24 hours
+            let s = Schedule::from_str(&s);
+            if s.is_err() {
+                tracing::error!("Invalid schedule for stats");
+                return;
+            }
+            let s = s.unwrap();
+
+            let next_time = s.upcoming(Utc).next();
+            if next_time.is_none() {
+                tracing::error!("Invalid schedule for stats");
+                return;
+            }
+            let next_time = next_time.unwrap();
+            let duration_to_next = next_time - Utc::now();
+
+            tokio::time::sleep(tokio::time::Duration::from_millis(
+                duration_to_next.num_milliseconds() as u64,
+            ))
+            .await;
+        }
+    });
+}
+
+pub async fn send_stats(
+    instance_name: &String,
+    http_client: &reqwest::Client,
+    db: &DB,
+) -> Result<()> {
+    let uid = sqlx::query_scalar!(
+        "SELECT value FROM global_settings WHERE name = $1",
+        UNIQUE_ID_SETTING
+    )
+    .fetch_one(db)
+    .await?;
+
+    let uid = serde_json::from_value::<String>(uid).map_err(to_anyhow)?;
+
+    let payload = serde_json::json!({
+        "uid": uid,
+        "version": GIT_VERSION,
+        "instance_name": instance_name,
+    });
+
+    let request = http_client
+        .post("https://hub.windmill.dev/stats")
+        .body(serde_json::to_string(&payload).map_err(to_anyhow)?)
+        .header("content-type", "application/json");
+
+    request
+        .send()
+        .await
+        .map_err(to_anyhow)?
+        .error_for_status()
+        .map_err(to_anyhow)?;
+
+    Ok(())
+}

backend/windmill-common/src/utils.rs

@@ -6,15 +6,22 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
-use crate::error::{Error, Result};
+use crate::error::{to_anyhow, Error, Result};
+use crate::global_settings::UNIQUE_ID_SETTING;
+use crate::DB;
+use git_version::git_version;
 use hyper::{HeaderMap, StatusCode};
 use rand::{distributions::Alphanumeric, thread_rng, Rng};
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
+use sqlx::{Pool, Postgres};
 
 pub const MAX_PER_PAGE: usize = 10000;
 pub const DEFAULT_PER_PAGE: usize = 1000;
 
+pub const GIT_VERSION: &str =
+    git_version!(args = ["--tag", "--always"], fallback = "unknown-version");
+
 #[derive(Deserialize)]
 pub struct Pagination {
     pub page: Option<usize>,
@@ -78,10 +85,10 @@ pub fn not_found_if_none<T, U: AsRef<str>>(opt: Option<T>, kind: &str, name: U)
 pub async fn query_elems_from_hub(
     http_client: &reqwest::Client,
     url: &str,
-    email: &str,
     query_params: Option<Vec<(&str, String)>>,
+    db: &DB,
 ) -> Result<(StatusCode, HeaderMap, reqwest::Response)> {
-    let response = http_get_from_hub(http_client, url, email, false, query_params).await?;
+    let response = http_get_from_hub(http_client, url, false, query_params, db).await?;
 
     let status = response.status();
 
@@ -92,21 +99,34 @@ pub async fn query_elems_from_hub(
 pub async fn http_get_from_hub(
     http_client: &reqwest::Client,
     url: &str,
-    email: &str,
     plain: bool,
     query_params: Option<Vec<(&str, String)>>,
+    db: &Pool<Postgres>,
 ) -> Result<reqwest::Response> {
-    let mut request = http_client
-        .get(url)
-        .header(
-            "Accept",
-            if plain {
-                "text/plain"
-            } else {
-                "application/json"
-            },
-        )
-        .header("X-email", email);
+    let uid = sqlx::query_scalar!(
+        "SELECT value FROM global_settings WHERE name = $1",
+        UNIQUE_ID_SETTING
+    )
+    .fetch_optional(db)
+    .await?
+    .map(|v| serde_json::from_value::<String>(v));
+
+    let mut request = http_client.get(url).header(
+        "Accept",
+        if plain {
+            "text/plain"
+        } else {
+            "application/json"
+        },
+    );
+
+    if let Some(uid) = uid {
+        if let Ok(uid) = uid {
+            request = request.header("X-uid", uid);
+        } else {
+            tracing::info!("Invalid uid in global settings: {}", uid.err().unwrap())
+        }
+    }
 
     if let Some(query_params) = query_params {
         for (key, value) in query_params {
@@ -114,7 +134,7 @@ pub async fn http_get_from_hub(
         }
     }
 
-    let response = request.send().await.map_err(crate::error::to_anyhow)?;
+    let response = request.send().await.map_err(to_anyhow)?;
 
     Ok(response)
 }

backend/windmill-worker/src/worker.rs

@@ -1491,9 +1491,9 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
             &job,
             logs.to_string(),
             mem_peak.to_owned(),
-            serde_json::from_str(result.get()).unwrap_or_else(
-                |_| json!({"message": format!("Non serializable error: {}", result.get())}),
-            ),
+            serde_json::from_str(result.get()).unwrap_or_else(|_| {
+                json!({ "message": format!("Non serializable error: {}", result.get()) })
+            }),
             metrics.clone(),
             rsmq.clone(),
         )
@@ -2057,7 +2057,7 @@ async fn handle_code_execution_job(
             let cache_path = format!("{HUB_CACHE_DIR}/{version}");
             let script;
             if tokio::fs::metadata(&cache_path).await.is_err() {
-                script =  get_full_hub_script_by_path(&job.email, StripPath(script_path.clone()), &HTTP_CLIENT).await?;
+                script = get_full_hub_script_by_path(StripPath(script_path.clone()), &HTTP_CLIENT, db).await?;
                 write_file(HUB_CACHE_DIR, &version, &serde_json::to_string(&script).map_err(to_anyhow)?).await?;
                 tracing::info!("wrote hub script {script_path} to cache");
             } else {

cli/deps.ts

@@ -1,6 +1,6 @@
 // windmill
-export { setClient } from "https://deno.land/x/windmill@v1.95.1/mod.ts";
-export * from "https://deno.land/x/windmill@v1.95.1/windmill-api/index.ts";
+export { setClient } from "https://deno.land/x/windmill@v1.188.1/mod.ts";
+export * from "https://deno.land/x/windmill@v1.188.1/windmill-api/index.ts";
 export { SEP } from "https://deno.land/std@0.201.0/path/separator.ts";
 // cliffy
 export { Command } from "https://deno.land/x/cliffy@v1.0.0-rc.2/command/mod.ts";

cli/hub.ts

@@ -1,5 +1,5 @@
 // deno-lint-ignore-file no-explicit-any
-import { Command, ResourceService, log } from "./deps.ts";
+import { Command, ResourceService, SettingService, log } from "./deps.ts";
 import { requireLogin, resolveWorkspace } from "./context.ts";
 import { pushResourceType } from "./resource-type.ts";
 import { GlobalOptions } from "./types.ts";
@@ -16,6 +16,20 @@ async function pull(opts: GlobalOptions) {
   }
 
   const userInfo = await requireLogin(opts);
+
+  const uid = await SettingService.getGlobal({
+    key: "uid",
+  });
+
+  const headers = {
+    Accept: "application/json",
+    "X-email": userInfo.email,
+  };
+
+  if (uid) {
+    headers["X-uid"] = uid;
+  }
+
   const list: {
     id: number;
     name: string;
@@ -27,10 +41,7 @@ async function pull(opts: GlobalOptions) {
     created_at: Date;
     comments: never[];
   }[] = await fetch("https://hub.windmill.dev/resource_types/list", {
-    headers: {
-      Accept: "application/json",
-      "X-email": userInfo.email,
-    },
+    headers,
   })
     .then((r) => r.json())
     .then((list: { id: number; name: string }[]) =>

frontend/src/lib/components/InstanceSettings.svelte

@@ -127,6 +127,14 @@
 				fieldType: 'boolean',
 				storage: 'config'
 			}
+		],
+		Telemetry: [
+			{
+				label: 'Disable telemetry',
+				key: 'disable_stats',
+				fieldType: 'boolean',
+				storage: 'setting'
+			}
 		]
 	}