* feat: replace LISTEN/NOTIFY with polling-based event system Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * test: add exhaustive tests for polling-based notify events Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * fix: add missing triggers and fix tests for polling-based events - Add variable/resource cache invalidation triggers to migration - Fix flow test to UPDATE flow table instead of INSERT into flow_version - Improve test isolation with unique channel names per test - All 26 tests now pass Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * test: add multi-server polling tests for cross-server event propagation Add 4 tests simulating independent server instances with separate DB connections and polling state: - test_two_servers_both_receive_trigger_event: both servers see same event - test_two_servers_cross_trigger_visibility: each triggers a change, both see both - test_server_catches_up_after_being_offline: server catches up on missed events - test_two_servers_incremental_polling: multi-round polling with cursor advancement Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * feat: add LISTEN_NEW_EVENTS_INTERVAL_SEC env var and e2e server test - Make poll interval configurable via LISTEN_NEW_EVENTS_INTERVAL_SEC (defaults to 30s) - Make migration idempotent with IF NOT EXISTS - Replace mock multi-server tests with actual e2e test that starts two windmill server processes on ports 19100/19200 with 1s poll interval, triggers a DB change, and verifies both servers log the event Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * test: ignore notify_events tests in CI These tests require a running database, like other integration tests in the codebase. Run with --ignored flag locally. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * test: only ignore slow e2e test, not fast DB tests Only test_two_server_processes_both_receive_event is slow (~10s, starts two server processes). The other 26 tests run in <0.2s. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * fix: ignore all notify_events tests in CI All tests depend on the notify_event table from the polling-based events migration, which is not applied in CI. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * fix: use sqlx::test for notify_events tests so they work in CI Convert all 26 fast tests from #[tokio::test] + manual get_db() to #[sqlx::test(fixtures("base"))], which creates temporary databases with all migrations applied. This ensures the notify_event table exists in CI without manual setup. Only the slow e2e multi-server test retains #[tokio::test] + #[ignore]. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * chore: reduce default polling interval from 30s to 10s Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * fix: address review feedback on polling-based events - Remove redundant notify_event_id_idx index (id is already PRIMARY KEY) - Add LIMIT 1000 to poll_notify_events to bound memory per poll cycle - Fix potential UTF-8 panic in token log truncation using str::get - Remove var/resource cache triggers that were re-enabled by mistake (they were intentionally dropped in migration 20250902085504) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>
136 lines
4.3 KiB
PL/PgSQL
136 lines
4.3 KiB
PL/PgSQL
-- Create notify_event table for polling-based event system
|
|
-- This replaces PostgreSQL LISTEN/NOTIFY with a table-based approach
|
|
|
|
CREATE TABLE IF NOT EXISTS notify_event (
|
|
id BIGSERIAL PRIMARY KEY,
|
|
channel TEXT NOT NULL,
|
|
payload TEXT NOT NULL,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS notify_event_created_at_idx ON notify_event (created_at);
|
|
|
|
-- Drop redundant index if it exists (id is already the PRIMARY KEY)
|
|
DROP INDEX IF EXISTS notify_event_id_idx;
|
|
|
|
-- Update notify_config_change function
|
|
CREATE OR REPLACE FUNCTION notify_config_change()
|
|
RETURNS TRIGGER AS $$
|
|
BEGIN
|
|
INSERT INTO notify_event (channel, payload) VALUES ('notify_config_change', NEW.name::text);
|
|
RETURN NEW;
|
|
END;
|
|
$$ LANGUAGE plpgsql;
|
|
|
|
-- Update notify_global_setting_change function
|
|
CREATE OR REPLACE FUNCTION notify_global_setting_change()
|
|
RETURNS TRIGGER AS $$
|
|
BEGIN
|
|
INSERT INTO notify_event (channel, payload) VALUES ('notify_global_setting_change', NEW.name::text);
|
|
RETURN NEW;
|
|
END;
|
|
$$ LANGUAGE plpgsql;
|
|
|
|
-- Update notify_global_setting_delete function
|
|
CREATE OR REPLACE FUNCTION notify_global_setting_delete()
|
|
RETURNS TRIGGER AS $$
|
|
BEGIN
|
|
INSERT INTO notify_event (channel, payload) VALUES ('notify_global_setting_change', OLD.name::text);
|
|
RETURN OLD;
|
|
END;
|
|
$$ LANGUAGE plpgsql;
|
|
|
|
-- Update notify_webhook_change function
|
|
CREATE OR REPLACE FUNCTION notify_webhook_change()
|
|
RETURNS TRIGGER AS $$
|
|
BEGIN
|
|
INSERT INTO notify_event (channel, payload) VALUES ('notify_webhook_change', NEW.workspace_id);
|
|
RETURN NEW;
|
|
END;
|
|
$$ LANGUAGE plpgsql;
|
|
|
|
-- Update notify_workspace_envs_change function
|
|
CREATE OR REPLACE FUNCTION notify_workspace_envs_change()
|
|
RETURNS TRIGGER AS $$
|
|
BEGIN
|
|
INSERT INTO notify_event (channel, payload) VALUES ('notify_workspace_envs_change', COALESCE(NEW.workspace_id, OLD.workspace_id));
|
|
RETURN COALESCE(NEW, OLD);
|
|
END;
|
|
$$ LANGUAGE plpgsql;
|
|
|
|
-- Update notify_workspace_premium_change function
|
|
CREATE OR REPLACE FUNCTION notify_workspace_premium_change()
|
|
RETURNS TRIGGER AS $$
|
|
BEGIN
|
|
INSERT INTO notify_event (channel, payload) VALUES ('notify_workspace_premium_change', NEW.id);
|
|
RETURN NEW;
|
|
END;
|
|
$$ LANGUAGE plpgsql;
|
|
|
|
-- Update notify_team_plan_status_change function
|
|
CREATE OR REPLACE FUNCTION notify_team_plan_status_change()
|
|
RETURNS TRIGGER AS $$
|
|
BEGIN
|
|
INSERT INTO notify_event (channel, payload) VALUES ('notify_workspace_premium_change', NEW.workspace_id);
|
|
RETURN NEW;
|
|
END;
|
|
$$ LANGUAGE plpgsql;
|
|
|
|
-- Update notify_runnable_version_change function
|
|
CREATE OR REPLACE FUNCTION notify_runnable_version_change()
|
|
RETURNS TRIGGER AS $$
|
|
DECLARE
|
|
source_type TEXT;
|
|
kind TEXT;
|
|
BEGIN
|
|
source_type := TG_ARGV[0];
|
|
|
|
IF source_type = 'script' THEN
|
|
kind := NEW.kind;
|
|
ELSE
|
|
kind := 'flow';
|
|
END IF;
|
|
|
|
INSERT INTO notify_event (channel, payload) VALUES ('notify_runnable_version_change', NEW.workspace_id || ':' || source_type || ':' || NEW.path || ':' || kind);
|
|
RETURN NEW;
|
|
END;
|
|
$$ LANGUAGE plpgsql;
|
|
|
|
-- Update notify_http_trigger_change function
|
|
CREATE OR REPLACE FUNCTION notify_http_trigger_change()
|
|
RETURNS TRIGGER AS $$
|
|
BEGIN
|
|
INSERT INTO notify_event (channel, payload) VALUES ('notify_http_trigger_change', COALESCE(NEW.workspace_id, OLD.workspace_id) || ':' || COALESCE(NEW.path, OLD.path));
|
|
RETURN COALESCE(NEW, OLD);
|
|
END;
|
|
$$ LANGUAGE plpgsql;
|
|
|
|
-- Update notify_token_invalidation function
|
|
CREATE OR REPLACE FUNCTION notify_token_invalidation()
|
|
RETURNS TRIGGER AS $$
|
|
BEGIN
|
|
IF OLD.label = 'session' AND OLD.email IS NOT NULL THEN
|
|
INSERT INTO notify_event (channel, payload) VALUES ('notify_token_invalidation', OLD.token);
|
|
END IF;
|
|
RETURN OLD;
|
|
END;
|
|
$$ LANGUAGE plpgsql;
|
|
|
|
-- Update notify_workspace_key_change function
|
|
CREATE OR REPLACE FUNCTION notify_workspace_key_change()
|
|
RETURNS TRIGGER AS $$
|
|
BEGIN
|
|
IF TG_OP = 'DELETE' THEN
|
|
INSERT INTO notify_event (channel, payload) VALUES ('notify_workspace_key_change', OLD.workspace_id);
|
|
RETURN OLD;
|
|
ELSE
|
|
INSERT INTO notify_event (channel, payload) VALUES ('notify_workspace_key_change', NEW.workspace_id);
|
|
RETURN NEW;
|
|
END IF;
|
|
END;
|
|
$$ LANGUAGE plpgsql;
|
|
|
|
-- NOTE: var_cache_invalidation / resource_cache_invalidation triggers were
|
|
-- intentionally dropped in migration 20250902085504. We do NOT re-create them
|
|
-- here to keep this migration scoped to the LISTEN/NOTIFY → polling swap only.
|