-- Auto-generated when you insert into subscriptions tablecreate or replacefunctionasync_triggers._publish_after_insert_on_users() returns trigger as $$ declare v_jsonb_output jsonb :='[]'::jsonb; v_base_payload jsonb := jsonb_build_object( 'tg_op', tg_op, 'tg_table_name', tg_table_name, 'tg_table_schema', tg_table_schema, 'timestamp', (extract(epoch from now()) *1000)::bigint ); begin-- Check subscription "user-signup" condition if new.email_verified= true then v_jsonb_output := v_jsonb_output || (jsonb_build_object( 'tg_name', 'user-signup', 'new', jsonb_build_object('id', new.id, 'email', new.email), 'old', null ) || v_base_payload ||'{}'::jsonb); -- payload_extensions would go here end if; -- Write to events table if any subscriptions matched if jsonb_array_length(v_jsonb_output) >0 then insert intoasync_triggers.events (payload, stream_id) select elem, 1from jsonb_array_elements(v_jsonb_output) as t(elem); end if; return new; end; $$ language plpgsql set search_path ='' security definer; -- Trigger that calls the function create constraint trigger async_triggers_insert after insert onpublic.users deferrable initially deferred for each row when ((new.email_verified= true)) execute procedure async_triggers._publish_after_insert_on_users();

-- Helper function that only recreates triggers when subscription actually changescreate or replacefunctionset_subscriptions( p_stream_id bigint, p_subscriptions async_triggers.subscriptions[] ) returns void language plpgsql security definer set search_path to ''as $$ begin create temporary table temp_subscriptions asselect*from unnest(p_subscriptions); -- Only update if values actually changed (avoids trigger recreation) merge into async_triggers.subscriptionsas target using temp_subscriptions as source on (target.key=source.keyandtarget.stream_id= p_stream_id) when matched and ( target.operation is distinct fromsource.operationortarget.schema_name is distinct fromsource.schema_nameortarget.table_name is distinct fromsource.table_nameortarget.when_clause is distinct fromsource.when_clauseortarget.column_names is distinct fromsource.column_namesortarget.metadata is distinct fromsource.metadataortarget.payload_extensions is distinct fromsource.payload_extensions ) then updateset operation =source.operation, schema_name =source.schema_name, table_name =source.table_name, when_clause =source.when_clause, column_names =source.column_names, metadata =source.metadata, payload_extensions =source.payload_extensions when not matched then insert ( key, stream_id, operation, schema_name, table_name, when_clause, column_names, metadata, payload_extensions ) values ( source.key, p_stream_id, source.operation, source.schema_name, source.table_name, source.when_clause, source.column_names, source.metadata, source.payload_extensions ); -- Remove subscriptions not in inputdeletefromasync_triggers.subscriptionswhere stream_id = p_stream_id and not exists ( select1from temp_subscriptions whereasync_triggers.subscriptions.key =temp_subscriptions.key ); droptable temp_subscriptions; end; $$;