feat: add postgres trigger captures (#5165)

* feat: 🚧 add postgres trigger config in back

* feat: add capture section for postgres

* feat: update postgres trigger panel, update capture migration

* fix: ci and casing style

* feat: capture done

* fix: update sqlx

* fix: remove unused

* feat: handle persisting data

* fix: add feature flag

* feat: capture done

* fix: feature flag

* feat: capture done

* nits: add plus into postgres capture label

* feat: trigger done

* Update PostgresTriggerEditorInner.svelte

* Update script_helpers.ts

* nit

* chore: update to meet pr comment requirement

* refactor:

* fix:

* fix: miss closing curly braces

* fix:

* fix: ci

* Update mod.rs

* polishing postgres trigger

* update minor changes on UI

* nits: remove log and update function parameters

* fix: remove wrong name

* nits: change icon

* fix: nit migration

* nits: refacto and remove alert

* nits: change variable name

* update workspaces_export

* fix: ci

---------

Co-authored-by: HugoCasa <hugo@casademont.ch>
Co-authored-by: Guilhem <guilhemlemouel@gmail.com>
This commit is contained in:
dieriba
2025-02-11 18:16:12 +01:00
committed by GitHub
co-authored by HugoCasa Guilhem
parent f44917feab
commit f3e91d714d
59 changed files with 3349 additions and 981 deletions
+91 -91
View File
@@ -336,8 +336,8 @@ pub(crate) async fn tarball_workspace(
{
let scripts = sqlx::query_as::<_, Script>(
"SELECT * FROM script as o WHERE workspace_id = $1 AND archived = false
AND created_at = (select max(created_at) from script where path = o.path AND \
workspace_id = $1)",
AND created_at = (select max(created_at) from script where path = o.path AND \
workspace_id = $1)",
)
.bind(&w_id)
.fetch_all(&mut *tx)
@@ -413,12 +413,12 @@ pub(crate) async fn tarball_workspace(
if !skip_resources.unwrap_or(false) {
let resources = sqlx::query_as!(
Resource,
"SELECT * FROM resource WHERE workspace_id = $1 AND resource_type != 'state' AND resource_type != 'cache'",
&w_id
)
.fetch_all(&mut *tx)
.await?;
Resource,
"SELECT * FROM resource WHERE workspace_id = $1 AND resource_type != 'state' AND resource_type != 'cache'",
&w_id
)
.fetch_all(&mut *tx)
.await?;
for resource in resources {
let resource_str = &to_string_without_metadata(&resource, false, None).unwrap();
@@ -450,14 +450,14 @@ pub(crate) async fn tarball_workspace(
{
let flows = sqlx::query_as::<_, Flow>(
"SELECT flow.workspace_id, flow.path, flow.summary, flow.description, flow.archived, flow.extra_perms, flow.draft_only, flow.dedicated_worker, flow.tag, flow.ws_error_handler_muted, flow.timeout, flow.visible_to_runner_only, flow.on_behalf_of_email, flow_version.schema, flow_version.value, flow_version.created_at as edited_at, flow_version.created_by as edited_by
FROM flow
LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]
WHERE flow.workspace_id = $1 AND flow.archived = false",
)
.bind(&w_id)
.fetch_all(&mut *tx)
.await?;
"SELECT flow.workspace_id, flow.path, flow.summary, flow.description, flow.archived, flow.extra_perms, flow.draft_only, flow.dedicated_worker, flow.tag, flow.ws_error_handler_muted, flow.timeout, flow.visible_to_runner_only, flow.on_behalf_of_email, flow_version.schema, flow_version.value, flow_version.created_at as edited_at, flow_version.created_by as edited_by
FROM flow
LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]
WHERE flow.workspace_id = $1 AND flow.archived = false",
)
.bind(&w_id)
.fetch_all(&mut *tx)
.await?;
for flow in flows {
let flow_str = &to_string_without_metadata(&flow, false, None).unwrap();
@@ -469,14 +469,14 @@ pub(crate) async fn tarball_workspace(
if !skip_variables.unwrap_or(false) {
let variables =
sqlx::query_as::<_, ExportableListableVariable>(if !skip_secrets.unwrap_or(false) {
"SELECT * FROM variable WHERE workspace_id = $1 AND expires_at IS NULL"
} else {
"SELECT * FROM variable WHERE workspace_id = $1 AND is_secret = false AND expires_at IS NULL"
})
.bind(&w_id)
.fetch_all(&mut *tx)
.await?;
sqlx::query_as::<_, ExportableListableVariable>(if !skip_secrets.unwrap_or(false) {
"SELECT * FROM variable WHERE workspace_id = $1 AND expires_at IS NULL"
} else {
"SELECT * FROM variable WHERE workspace_id = $1 AND is_secret = false AND expires_at IS NULL"
})
.bind(&w_id)
.fetch_all(&mut *tx)
.await?;
let mc = build_crypt(&db, &w_id).await?;
@@ -496,14 +496,14 @@ pub(crate) async fn tarball_workspace(
{
let apps = sqlx::query_as::<_, AppWithLastVersion>(
"SELECT app.id, app.path, app.summary, app.versions, app.policy, app.custom_path,
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)]",
)
.bind(&w_id)
.fetch_all(&mut *tx)
.await?;
"SELECT app.id, app.path, app.summary, app.versions, app.policy, app.custom_path,
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)]",
)
.bind(&w_id)
.fetch_all(&mut *tx)
.await?;
for app in apps {
let app_str = &to_string_without_metadata(&app, false, None).unwrap();
@@ -516,7 +516,7 @@ pub(crate) async fn tarball_workspace(
if include_schedules.unwrap_or(false) {
let schedules = sqlx::query_as::<_, Schedule>(
"SELECT * FROM schedule
WHERE workspace_id = $1",
WHERE workspace_id = $1",
)
.bind(&w_id)
.fetch_all(&mut *tx)
@@ -534,13 +534,13 @@ pub(crate) async fn tarball_workspace(
#[cfg(feature = "http_trigger")]
{
let http_triggers = sqlx::query_as!(
crate::http_triggers::HttpTrigger,
"SELECT workspace_id, path, route_path, route_path_key, script_path, is_flow, edited_by, edited_at, email, extra_perms, is_async, requires_auth, http_method as \"http_method: _\", static_asset_config as \"static_asset_config: _\", is_static_website FROM http_trigger
WHERE workspace_id = $1",
&w_id
)
.fetch_all(&mut *tx)
.await?;
crate::http_triggers::HttpTrigger,
"SELECT workspace_id, path, route_path, route_path_key, script_path, is_flow, edited_by, edited_at, email, extra_perms, is_async, requires_auth, http_method as \"http_method: _\", static_asset_config as \"static_asset_config: _\", is_static_website FROM http_trigger
WHERE workspace_id = $1",
&w_id
)
.fetch_all(&mut *tx)
.await?;
for trigger in http_triggers {
let trigger_str = &to_string_without_metadata(&trigger, false, None).unwrap();
@@ -553,13 +553,13 @@ pub(crate) async fn tarball_workspace(
#[cfg(feature = "websocket")]
{
let websocket_triggers = sqlx::query_as!(
crate::websocket_triggers::WebsocketTrigger,
"SELECT workspace_id, path, url, script_path, is_flow, edited_by, email, edited_at, server_id, last_server_ping, extra_perms, error, enabled, filters as \"filters: _\", initial_messages as \"initial_messages: _\", url_runnable_args as \"url_runnable_args: _\", can_return_message FROM websocket_trigger
WHERE workspace_id = $1",
&w_id
)
.fetch_all(&mut *tx)
.await?;
crate::websocket_triggers::WebsocketTrigger,
"SELECT workspace_id, path, url, script_path, is_flow, edited_by, email, edited_at, server_id, last_server_ping, extra_perms, error, enabled, filters as \"filters: _\", initial_messages as \"initial_messages: _\", url_runnable_args as \"url_runnable_args: _\", can_return_message FROM websocket_trigger
WHERE workspace_id = $1",
&w_id
)
.fetch_all(&mut *tx)
.await?;
for trigger in websocket_triggers {
let trigger_str = &to_string_without_metadata(&trigger, false, None).unwrap();
@@ -577,7 +577,7 @@ pub(crate) async fn tarball_workspace(
let kafka_triggers = sqlx::query_as!(
crate::kafka_triggers_ee::KafkaTrigger,
"SELECT * FROM kafka_trigger
WHERE workspace_id = $1",
WHERE workspace_id = $1",
&w_id
)
.fetch_all(&mut *tx)
@@ -599,7 +599,7 @@ pub(crate) async fn tarball_workspace(
let nats_triggers = sqlx::query_as!(
crate::nats_triggers_ee::NatsTrigger,
"SELECT * FROM nats_trigger
WHERE workspace_id = $1",
WHERE workspace_id = $1",
&w_id
)
.fetch_all(&mut *tx)
@@ -619,7 +619,7 @@ pub(crate) async fn tarball_workspace(
let postgres_triggers = sqlx::query_as!(
crate::postgres_triggers::PostgresTrigger,
"SELECT * FROM postgres_trigger
WHERE workspace_id = $1",
WHERE workspace_id = $1",
&w_id
)
.fetch_all(&mut *tx)
@@ -640,7 +640,7 @@ pub(crate) async fn tarball_workspace(
if include_users.unwrap_or(false) {
let users = sqlx::query!(
"SELECT * FROM usr
WHERE workspace_id = $1",
WHERE workspace_id = $1",
&w_id
)
.fetch_all(&mut *tx)
@@ -668,16 +668,16 @@ pub(crate) async fn tarball_workspace(
if include_groups.unwrap_or(false) {
let groups = sqlx::query!(
r#"SELECT g_.workspace_id, name, summary, extra_perms, array_agg(u2g.usr) filter (where u2g.usr is not null) as members
FROM usr u
JOIN usr_to_group u2g ON u2g.usr = u.username AND u2g.workspace_id = u.workspace_id
RIGHT JOIN group_ g_ ON g_.workspace_id = u.workspace_id AND g_.name = u2g.group_
WHERE g_.workspace_id = $1 AND g_.name != 'all'
GROUP BY g_.workspace_id, name, summary, extra_perms"#,
&w_id
)
.fetch_all(&mut *tx)
.await?;
r#"SELECT g_.workspace_id, name, summary, extra_perms, array_agg(u2g.usr) filter (where u2g.usr is not null) as members
FROM usr u
JOIN usr_to_group u2g ON u2g.usr = u.username AND u2g.workspace_id = u.workspace_id
RIGHT JOIN group_ g_ ON g_.workspace_id = u.workspace_id AND g_.name = u2g.group_
WHERE g_.workspace_id = $1 AND g_.name != 'all'
GROUP BY g_.workspace_id, name, summary, extra_perms"#,
&w_id
)
.fetch_all(&mut *tx)
.await?;
for group in groups {
let extra_perms: HashMap<String, bool> = serde_json::from_value(group.extra_perms)
@@ -728,36 +728,36 @@ pub(crate) async fn tarball_workspace(
if include_settings.unwrap_or(false) {
let settings = sqlx::query_as!(
SimplifiedSettings,
r#"SELECT
-- slack_team_id,
-- slack_name,
-- slack_command_script,
-- CASE WHEN slack_email = 'missing@email.xyz' THEN NULL ELSE slack_email END AS slack_email,
auto_invite_domain IS NOT NULL AS "auto_invite_enabled!",
CASE WHEN auto_invite_operator IS TRUE THEN 'operator' ELSE 'developer' END AS "auto_invite_as!",
CASE WHEN auto_add IS TRUE THEN 'add' ELSE 'invite' END AS "auto_invite_mode!",
webhook,
deploy_to,
error_handler,
ai_resource,
ai_models,
code_completion_model,
error_handler_extra_args,
error_handler_muted_on_cancel,
large_file_storage,
git_sync,
default_app,
default_scripts,
workspace.name,
mute_critical_alerts,
color,
operator_settings
FROM workspace_settings
LEFT JOIN workspace ON workspace.id = workspace_settings.workspace_id
WHERE workspace_id = $1"#,
&w_id
).fetch_one(&mut *tx).await?;
SimplifiedSettings,
r#"SELECT
-- slack_team_id,
-- slack_name,
-- slack_command_script,
-- CASE WHEN slack_email = 'missing@email.xyz' THEN NULL ELSE slack_email END AS slack_email,
auto_invite_domain IS NOT NULL AS "auto_invite_enabled!",
CASE WHEN auto_invite_operator IS TRUE THEN 'operator' ELSE 'developer' END AS "auto_invite_as!",
CASE WHEN auto_add IS TRUE THEN 'add' ELSE 'invite' END AS "auto_invite_mode!",
webhook,
deploy_to,
error_handler,
ai_resource,
ai_models,
code_completion_model,
error_handler_extra_args,
error_handler_muted_on_cancel,
large_file_storage,
git_sync,
default_app,
default_scripts,
workspace.name,
mute_critical_alerts,
color,
operator_settings
FROM workspace_settings
LEFT JOIN workspace ON workspace.id = workspace_settings.workspace_id
WHERE workspace_id = $1"#,
&w_id
).fetch_one(&mut *tx).await?;
let settings_str = serde_json::to_value(settings)
.map(|v| serde_json::to_string_pretty(&v).ok())