From 3a62b907b44b53bc1758a89a7658dc325a212273 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 14 Jun 2026 19:14:48 +0000 Subject: [PATCH] Fix websocket env_id: enrich the generic notify trigger The API's websocket feed listens on the generic `db_notifications` channel, fed by `notify_trigger()`. The earlier env_id work enriched `notify_specific_trigger_name()` instead, which only feeds the per-table channels used by backend services -- so websocket events reaching the frontend carried no env_id, and per-env subscription/permission filtering dropped every environment-scoped event (live updates were broken). - New migration enriches the generic `notify_trigger()` with the changed row's environment, resolved per table (deployments/tasks/resources via cluster; secrets/permissions via their env_id column; envs via their own id). Other tables carry a null env_id. - Websocket subscription matching now treats a no-env subscription as a wildcard ("every permitted event of this table"), so globally loaded collections such as envs and env_user_permissions -- whose events now carry an env_id -- still reach subscribers, gated by permissions. Verified against PostgreSQL that db_notifications payloads carry the correct env_id for every environment-scoped table, including DELETE. --- api/src/routes/v2/ws.rs | 20 +++- .../down.sql | 45 +++++++++ .../up.sql | 95 +++++++++++++++++++ 3 files changed, 157 insertions(+), 3 deletions(-) create mode 100644 db/migrations/2026-06-14-190000_ws-env-id-generic-trigger/down.sql create mode 100644 db/migrations/2026-06-14-190000_ws-env-id-generic-trigger/up.sql diff --git a/api/src/routes/v2/ws.rs b/api/src/routes/v2/ws.rs index 6f02483..212e20a 100644 --- a/api/src/routes/v2/ws.rs +++ b/api/src/routes/v2/ws.rs @@ -44,8 +44,10 @@ pub enum ClientMessage { /// to the (collection, environment) pairs it has subscribed to. struct DbEventsWs { scope: AccessScope, - /// Active subscriptions as (table, env) pairs. `None` env means a global - /// collection. An event is forwarded only when its (table, env_id) is here. + /// Active subscriptions as (table, env) pairs. A `None` env is a wildcard: + /// every permitted event of that table (used by globally loaded + /// collections). A `Some(env)` subscription receives only that + /// environment's events. See [`Self::is_subscribed`]. subscriptions: HashSet<(DbTable, Option)>, } @@ -74,6 +76,18 @@ impl DbEventsWs { } impl DbEventsWs { + /// Whether the client is subscribed to this event. A wildcard subscription + /// (no env) for a table receives every event of that table the identity is + /// permitted to see -- used by globally loaded collections such as `envs` + /// and `env_user_permissions`, whose rows belong to different environments. + /// An env-scoped subscription receives only that environment's events. + fn is_subscribed(&self, event: &DbEvent) -> bool { + self.subscriptions.contains(&(event.table, None)) + || event + .env_id + .is_some_and(|env_id| self.subscriptions.contains(&(event.table, Some(env_id)))) + } + fn handle_client_message(&mut self, text: &str) { match serde_json::from_str::(text) { Ok(ClientMessage::Subscribe { table, env_id }) => { @@ -111,7 +125,7 @@ impl StreamHandler> for DbEventsWs { return; } // ...and only those the client is currently subscribed to. - if !self.subscriptions.contains(&(event.table, event.env_id)) { + if !self.is_subscribed(&event) { return; } match serde_json::to_string(&event) { diff --git a/db/migrations/2026-06-14-190000_ws-env-id-generic-trigger/down.sql b/db/migrations/2026-06-14-190000_ws-env-id-generic-trigger/down.sql new file mode 100644 index 0000000..68ff9cb --- /dev/null +++ b/db/migrations/2026-06-14-190000_ws-env-id-generic-trigger/down.sql @@ -0,0 +1,45 @@ +-- Restore the original generic notify function that emits only the row id, +-- without the resolved env_id. + +CREATE OR REPLACE FUNCTION notify_trigger() RETURNS trigger AS $trigger$ +DECLARE + rec RECORD; + payload TEXT; + column_name TEXT; + column_value TEXT; + payload_items TEXT[]; +BEGIN + -- Set record row depending on operation + CASE TG_OP + WHEN 'INSERT', 'UPDATE' THEN + rec := NEW; + WHEN 'DELETE' THEN + rec := OLD; + ELSE + RAISE EXCEPTION 'Unknown TG_OP: "%". Should not occur!', TG_OP; + END CASE; + + -- Get required fields + FOREACH column_name IN ARRAY TG_ARGV LOOP + EXECUTE format('SELECT $1.%I::TEXT', column_name) + INTO column_value + USING rec; + payload_items := array_append(payload_items, '"' || replace(column_name, '"', '\"') || '":"' || replace(column_value, '"', '\"') || '"'); + END LOOP; + + -- Build the payload + payload := '' + || '{' + || '"timestamp":"' || CURRENT_TIMESTAMP || '",' + || '"operation":"' || TG_OP || '",' + || '"schema":"' || TG_TABLE_SCHEMA || '",' + || '"table":"' || TG_TABLE_NAME || '",' + || '"data":{' || array_to_string(payload_items, ',') || '}' + || '}'; + + -- Notify the channel + PERFORM pg_notify('db_notifications', payload); + + RETURN rec; +END; +$trigger$ LANGUAGE plpgsql; diff --git a/db/migrations/2026-06-14-190000_ws-env-id-generic-trigger/up.sql b/db/migrations/2026-06-14-190000_ws-env-id-generic-trigger/up.sql new file mode 100644 index 0000000..562ace7 --- /dev/null +++ b/db/migrations/2026-06-14-190000_ws-env-id-generic-trigger/up.sql @@ -0,0 +1,95 @@ +-- The API's websocket feed listens on the generic `db_notifications` channel, +-- which is fed by `notify_trigger()` (attached to every table). A previous +-- migration added `env_id` to `notify_specific_trigger_name()` instead, which +-- only feeds the per-table channels consumed by the backend services -- so the +-- websocket events reaching the frontend carried no `env_id`, and per-env +-- subscription/permission filtering dropped every env-scoped event. +-- +-- This enriches the generic `notify_trigger()` with the environment of the +-- changed row, resolved per table. Because the trigger uses OLD on DELETE, the +-- environment is available even for deletions. +-- +-- deployments -> k8s_clusters.env_id (via cluster_id) +-- deployment_tasks -> k8s_clusters.env_id (via cluster_id) +-- deployment_resources -> deployments -> k8s_clusters.env_id +-- secrets -> secrets.env_id +-- env_user_permissions -> env_user_permissions.env_id +-- deployment_permissions -> deployment_permissions.env_id +-- envs -> envs.id (the environment itself) +-- All other tables are not environment-scoped and carry a null env_id. + +CREATE OR REPLACE FUNCTION notify_trigger() RETURNS trigger AS $trigger$ +DECLARE + rec RECORD; + payload TEXT; + column_name TEXT; + column_value TEXT; + payload_items TEXT[]; + v_env_id UUID; + env_id_json TEXT; +BEGIN + -- Set record row depending on operation + CASE TG_OP + WHEN 'INSERT', 'UPDATE' THEN + rec := NEW; + WHEN 'DELETE' THEN + rec := OLD; + ELSE + RAISE EXCEPTION 'Unknown TG_OP: "%". Should not occur!', TG_OP; + END CASE; + + -- Resolve the environment of the changed row, where applicable. + v_env_id := NULL; + CASE TG_TABLE_NAME + WHEN 'deployments' THEN + SELECT k.env_id INTO v_env_id FROM k8s_clusters k WHERE k.id = rec.cluster_id; + WHEN 'deployment_tasks' THEN + SELECT k.env_id INTO v_env_id FROM k8s_clusters k WHERE k.id = rec.cluster_id; + WHEN 'deployment_resources' THEN + SELECT k.env_id INTO v_env_id + FROM deployments d + JOIN k8s_clusters k ON k.id = d.cluster_id + WHERE d.id = rec.deployment_id; + WHEN 'secrets' THEN + v_env_id := rec.env_id; + WHEN 'env_user_permissions' THEN + v_env_id := rec.env_id; + WHEN 'deployment_permissions' THEN + v_env_id := rec.env_id; + WHEN 'envs' THEN + v_env_id := rec.id; + ELSE + v_env_id := NULL; + END CASE; + + IF v_env_id IS NULL THEN + env_id_json := 'null'; + ELSE + env_id_json := '"' || v_env_id::TEXT || '"'; + END IF; + + -- Get required fields + FOREACH column_name IN ARRAY TG_ARGV LOOP + EXECUTE format('SELECT $1.%I::TEXT', column_name) + INTO column_value + USING rec; + payload_items := array_append(payload_items, '"' || replace(column_name, '"', '\"') || '":"' || replace(column_value, '"', '\"') || '"'); + END LOOP; + + -- Build the payload + payload := '' + || '{' + || '"timestamp":"' || CURRENT_TIMESTAMP || '",' + || '"operation":"' || TG_OP || '",' + || '"schema":"' || TG_TABLE_SCHEMA || '",' + || '"table":"' || TG_TABLE_NAME || '",' + || '"env_id":' || env_id_json || ',' + || '"data":{' || array_to_string(payload_items, ',') || '}' + || '}'; + + -- Notify the channel + PERFORM pg_notify('db_notifications', payload); + + RETURN rec; +END; +$trigger$ LANGUAGE plpgsql;