From c887272673174abaa54740f90bd7e56f2caeafc3 Mon Sep 17 00:00:00 2001 From: Kevin Berry Date: Sun, 6 Sep 2026 13:04:17 -0500 Subject: [PATCH 1/2] feat(table): add optional PG NOTIFY trigger for outbox tables - Add NotifyTrigger class creating a pg_notify function and statement-level AFTER INSERT trigger on the outbox table - Opt-in with notify: true option on pgt_outbox_setup - Channel defaults to _notifications, overridable via notify_channel option - Follows pgt_outbox_* naming convention for generated objects - Add --notify and --notify-channel CLI flags to outboxify - Add tests for default off, notify on, custom channel, and insert --- exe/outboxify | 10 ++- lib/sequel/pgt_outbox/notify_trigger.rb | 67 ++++++++++++++++ lib/sequel/pgt_outbox/table.rb | 17 ++++ test/sequel/test_pgt_outbox.rb | 101 ++++++++++++++++++++++++ 4 files changed, 194 insertions(+), 1 deletion(-) create mode 100644 lib/sequel/pgt_outbox/notify_trigger.rb diff --git a/exe/outboxify b/exe/outboxify index 8484dd9..b532d46 100755 --- a/exe/outboxify +++ b/exe/outboxify @@ -4,9 +4,15 @@ require 'sequel/pgt_outbox' require 'optparse' +notify = false +notify_channel = nil + opts = OptionParser.new do |o| o.banner = 'Usage: outboxify [options] ' + o.on('--notify') { notify = true } + o.on('--notify-channel CHANNEL') { |c| notify_channel = c } end +opts.parse! db_uri = ARGV.shift if db_uri.nil? @@ -25,5 +31,7 @@ if table.nil? exit 2 end -function = DB.pgt_outbox_setup(table) +options = { notify: } +options[:notify_channel] = notify_channel if notify_channel +function = DB.pgt_outbox_setup(table, **options) DB.pgt_outbox_events(table, function) diff --git a/lib/sequel/pgt_outbox/notify_trigger.rb b/lib/sequel/pgt_outbox/notify_trigger.rb new file mode 100644 index 0000000..ad49f89 --- /dev/null +++ b/lib/sequel/pgt_outbox/notify_trigger.rb @@ -0,0 +1,67 @@ +# frozen_string_literal: true + +require_relative '../pgt_outbox' + +module Rubyists + module PgtOutbox + # A PostgreSQL trigger that sends a NOTIFY on a channel when rows are inserted into the outbox + class NotifyTrigger + include PgtOutbox + + DEFAULT_OPTS = { language: :plpgsql, returns: :trigger, replace: true }.freeze + TRIGGER_DEFAULT_OPTS = { after: true, each_statement: true, replace: true }.freeze + + attr_reader(*%i[outbox db channel function_name trigger_name opts]) + + def self.create!(outbox, channel:, function_name: nil, trigger_name: nil, opts: {}) + new(outbox, channel:, function_name:, trigger_name:, opts:).create! + end + + def initialize(outbox, channel:, function_name: nil, trigger_name: nil, opts: {}) + @outbox = outbox + @db = outbox.db + @channel = channel + @function_name = function_name || "pgt_outbox_notify_#{mangled_table_name(db, outbox.name)}" + @trigger_name = trigger_name || "pgt_outbox_notify_after_insert_#{mangled_table_name(db, outbox.name)}" + @opts = opts + end + + def create! + create_function! + create_trigger! + self + end + + private + + def create_function! + db.create_function(function_name, function_sql, function_opts) + end + + def function_sql + <<~SQL + BEGIN + PERFORM pg_catalog.pg_notify('#{channel}', ''); + RETURN NULL; + END; + SQL + end + + def function_opts + @function_opts ||= DEFAULT_OPTS.merge(opts.fetch(:function_opts, {})) + end + + def create_trigger! + db.create_trigger( + outbox.name, + trigger_name, + function_name, + after: true, + each_statement: true, + events: [:insert], + replace: true + ) + end + end + end +end diff --git a/lib/sequel/pgt_outbox/table.rb b/lib/sequel/pgt_outbox/table.rb index 830ed2d..b70d130 100644 --- a/lib/sequel/pgt_outbox/table.rb +++ b/lib/sequel/pgt_outbox/table.rb @@ -28,6 +28,7 @@ def create! string_columns! jsonb_columns! indexes! + notify_trigger! self end @@ -99,6 +100,14 @@ def uuid_function @uuid_function ||= opts.fetch(:uuid_function, :uuid_generate_v4) end + def notify? + @notify ||= opts.fetch(:notify, false) + end + + def notify_channel + @notify_channel ||= opts.fetch(:notify_channel, "#{name}_notifications") + end + def function @function ||= Function.create!(self, opts:) end @@ -156,6 +165,14 @@ def indexes! db.add_index name, Sequel.desc(attempted_column) self end + + def notify_trigger! + return unless notify? + + require_relative 'notify_trigger' + NotifyTrigger.create!(self, channel: notify_channel, opts: opts.fetch(:notify_opts, {})) + self + end end end end diff --git a/test/sequel/test_pgt_outbox.rb b/test/sequel/test_pgt_outbox.rb index fa42bc4..752014b 100755 --- a/test/sequel/test_pgt_outbox.rb +++ b/test/sequel/test_pgt_outbox.rb @@ -230,4 +230,105 @@ def depth_sql(depth) end end +if DB.server_version >= 90_400 + describe 'PG NOTIFY Trigger' do # rubocop:disable Metrics/BlockLength + def get_trigger(table_name, trigger_name) + DB[ + "SELECT tgname FROM pg_trigger " \ + "JOIN pg_class ON pg_class.oid = pg_trigger.tgrelid " \ + "WHERE pg_class.relname = ? AND pg_trigger.tgname = ?", + table_name, trigger_name + ].first + end + + def get_function(func_name) + DB[ + "SELECT proname FROM pg_proc WHERE proname = ?", + func_name + ].first + end + + after do + DB.drop_table(:accounts, :accounts_outbox) + begin + DB.drop_function(:spgt_outbox_events) + rescue Sequel::DatabaseError + # function may not exist + end + begin + DB.drop_function('pgt_outbox_notify_accounts_outbox') + rescue Sequel::DatabaseError + # function may not exist + end + end + + it 'should not create notify trigger by default' do + DB.create_table!(:accounts) do + integer :id + String :s + end + DB.pgt_outbox_setup(:accounts, function_name: :spgt_outbox_events) + + trigger = get_trigger('accounts_outbox', 'pgt_outbox_notify_after_insert_accounts_outbox') + + _(trigger).must_be_nil + end + + it 'should create notify function and trigger when notify: true' do + DB.create_table!(:accounts) do + integer :id + String :s + end + DB.pgt_outbox_setup(:accounts, notify: true, function_name: :spgt_outbox_events) + + func = get_function('pgt_outbox_notify_accounts_outbox') + + _(func).wont_be :nil? + + trigger = get_trigger('accounts_outbox', 'pgt_outbox_notify_after_insert_accounts_outbox') + + _(trigger).wont_be :nil? + end + + it 'should use custom channel name when provided' do + DB.create_table!(:accounts) do + integer :id + String :s + end + DB.pgt_outbox_setup(:accounts, notify: true, notify_channel: 'my_custom_channel', + function_name: :spgt_outbox_events) + + func = get_function('pgt_outbox_notify_accounts_outbox') + + _(func).wont_be :nil? + + # Verify the function body contains the custom channel + func_body = DB[ + 'SELECT proname, prosrc FROM pg_proc WHERE proname = ?', + 'pgt_outbox_notify_accounts_outbox' + ].first[:prosrc] + + _(func_body).must_include 'my_custom_channel' + end + + it 'should send notification on insert into outbox' do + DB.create_table!(:accounts) do + integer :id + String :s + end + DB.pgt_outbox_setup(:accounts, notify: true, function_name: :spgt_outbox_events) + DB.pgt_outbox_events(:accounts, 'spgt_outbox_events') + + # Insert a row which should trigger the notify + DB[:accounts].insert(id: 1, s: 'test') + + # Verify outbox event was created + outbox_event = DB[:accounts_outbox].first + + _(outbox_event).wont_be :nil? + _(outbox_event[:event_type]).must_equal 'accounts_created' + end + end +end + # vim: ft=ruby sts=2 sw=2 ts=2 et From 1ac9815062a2687f8a68858833e928ed1ceade35 Mon Sep 17 00:00:00 2001 From: Kevin Berry Date: Sun, 6 Sep 2026 14:04:52 -0500 Subject: [PATCH 2/2] style: fix rubocop offenses and update rubocop plugin syntax --- .rubocop.yml | 2 +- lib/sequel/pgt_outbox/function.rb | 1 + test/sequel/test_pgt_outbox.rb | 34 ++++++++++++++++++++++++++----- 3 files changed, 31 insertions(+), 6 deletions(-) diff --git a/.rubocop.yml b/.rubocop.yml index 3ad8418..5623b76 100644 --- a/.rubocop.yml +++ b/.rubocop.yml @@ -1,5 +1,5 @@ --- -require: +plugins: - rubocop-rake - rubocop-minitest diff --git a/lib/sequel/pgt_outbox/function.rb b/lib/sequel/pgt_outbox/function.rb index f93caeb..addea9f 100644 --- a/lib/sequel/pgt_outbox/function.rb +++ b/lib/sequel/pgt_outbox/function.rb @@ -8,6 +8,7 @@ module PgtOutbox # The Outbox Function class Function include PgtOutbox + DEFAULT_OPTS = { language: :plpgsql, returns: :trigger, replace: true }.freeze attr_reader(*%i[outbox db opts]) diff --git a/test/sequel/test_pgt_outbox.rb b/test/sequel/test_pgt_outbox.rb index 752014b..a249096 100755 --- a/test/sequel/test_pgt_outbox.rb +++ b/test/sequel/test_pgt_outbox.rb @@ -60,8 +60,10 @@ def depth_sql(depth) ds = DB[:accounts] ds.insert(id: 1, s: 'string') + _(ds.all).must_equal [{ id: 1, s: 'string' }] h = @logs.first + _(h.delete(:created).to_i).must_be_close_to(10, DB.get(Sequel::CURRENT_TIMESTAMP).to_i) _(h.delete(:updated).to_i).must_be_close_to(10, DB.get(Sequel::CURRENT_TIMESTAMP).to_i) _(h).must_equal(id: 1, @@ -75,8 +77,10 @@ def depth_sql(depth) metadata: nil) ds.where(id: 1).update(s: 'string2') + _(ds.all).must_equal [{ id: 1, s: 'string2' }] h = @logs.first + _(h.delete(:created).to_i).must_be_close_to(10, DB.get(Sequel::CURRENT_TIMESTAMP).to_i) _(h.delete(:updated).to_i).must_be_close_to(10, DB.get(Sequel::CURRENT_TIMESTAMP).to_i) _(h).must_equal(id: 2, @@ -90,8 +94,10 @@ def depth_sql(depth) metadata: nil) ds.delete + _(ds.all).must_equal [] h = @logs.first + _(h.delete(:created).to_i).must_be_close_to(10, DB.get(Sequel::CURRENT_TIMESTAMP).to_i) _(h.delete(:updated).to_i).must_be_close_to(10, DB.get(Sequel::CURRENT_TIMESTAMP).to_i) _(h).must_equal(id: 3, @@ -130,21 +136,27 @@ def depth_sql(depth) ds = DB[:accounts] ds.insert(id: 1, s: 'string') + _(ds.all).must_equal [{ id: 1, s: 'string' }] h = @logs.first + _(h.delete(:created).to_i).must_be_close_to(10, DB.get(Sequel::CURRENT_TIMESTAMP).to_i) _(h.delete(:updated).to_i).must_be_close_to(10, DB.get(Sequel::CURRENT_TIMESTAMP).to_i) id = h.delete(:id) + _(id).must_match(/\A\h{8}-\h{4}-\h{4}-\h{4}-\h{12}\z/) _(h).must_equal(attempts: 0, attempted: nil, completed: nil, event_type: 'accounts_created', last_error: nil, data_before: nil, data_after: { 's' => 'string', 'id' => 1 }, metadata: nil) ds.where(id: 1).update(s: 'string2') + _(ds.all).must_equal [{ id: 1, s: 'string2' }] h = @logs.first + _(h.delete(:created).to_i).must_be_close_to(10, DB.get(Sequel::CURRENT_TIMESTAMP).to_i) _(h.delete(:updated).to_i).must_be_close_to(10, DB.get(Sequel::CURRENT_TIMESTAMP).to_i) id = h.delete(:id) + _(id).must_match(/\A\h{8}-\h{4}-\h{4}-\h{4}-\h{12}\z/) _(h).must_equal(attempts: 0, attempted: nil, @@ -156,11 +168,14 @@ def depth_sql(depth) metadata: nil) ds.delete + _(ds.all).must_equal [] h = @logs.first + _(h.delete(:created).to_i).must_be_close_to(10, DB.get(Sequel::CURRENT_TIMESTAMP).to_i) _(h.delete(:updated).to_i).must_be_close_to(10, DB.get(Sequel::CURRENT_TIMESTAMP).to_i) id = h.delete(:id) + _(id).must_match(/\A\h{8}-\h{4}-\h{4}-\h{4}-\h{12}\z/) _(h).must_equal(attempts: 0, attempted: nil, completed: nil, event_type: 'accounts_deleted', last_error: nil, data_before: { 's' => 'string2', 'id' => 1 }, data_after: nil, metadata: nil) @@ -192,21 +207,27 @@ def depth_sql(depth) ds = DB[:accounts] ds.insert(id: 1, s: 'string') + _(ds.all).must_equal [{ id: 1, s: 'string' }] h = @logs.first + _(h.delete(:created).to_i).must_be_close_to(10, DB.get(Sequel::CURRENT_TIMESTAMP).to_i) _(h.delete(:updated).to_i).must_be_close_to(10, DB.get(Sequel::CURRENT_TIMESTAMP).to_i) id = h.delete(:id) + _(id).must_match(/\A\h{8}-\h{4}-\h{4}-\h{4}-\h{12}\z/) _(h).must_equal(attempts: 0, attempted: nil, completed: false, event_type: 'accounts_created', last_error: nil, data_before: nil, data_after: { 's' => 'string', 'id' => 1 }, metadata: nil) ds.where(id: 1).update(s: 'string2') + _(ds.all).must_equal [{ id: 1, s: 'string2' }] h = @logs.first + _(h.delete(:created).to_i).must_be_close_to(10, DB.get(Sequel::CURRENT_TIMESTAMP).to_i) _(h.delete(:updated).to_i).must_be_close_to(10, DB.get(Sequel::CURRENT_TIMESTAMP).to_i) id = h.delete(:id) + _(id).must_match(/\A\h{8}-\h{4}-\h{4}-\h{4}-\h{12}\z/) _(h).must_equal(attempts: 0, attempted: nil, @@ -218,11 +239,14 @@ def depth_sql(depth) metadata: nil) ds.delete + _(ds.all).must_equal [] h = @logs.first + _(h.delete(:created).to_i).must_be_close_to(10, DB.get(Sequel::CURRENT_TIMESTAMP).to_i) _(h.delete(:updated).to_i).must_be_close_to(10, DB.get(Sequel::CURRENT_TIMESTAMP).to_i) id = h.delete(:id) + _(id).must_match(/\A\h{8}-\h{4}-\h{4}-\h{4}-\h{12}\z/) _(h).must_equal(attempts: 0, attempted: nil, completed: false, event_type: 'accounts_deleted', last_error: nil, data_before: { 's' => 'string2', 'id' => 1 }, data_after: nil, metadata: nil) @@ -234,16 +258,16 @@ def depth_sql(depth) describe 'PG NOTIFY Trigger' do # rubocop:disable Metrics/BlockLength def get_trigger(table_name, trigger_name) DB[ - "SELECT tgname FROM pg_trigger " \ - "JOIN pg_class ON pg_class.oid = pg_trigger.tgrelid " \ - "WHERE pg_class.relname = ? AND pg_trigger.tgname = ?", + 'SELECT tgname FROM pg_trigger ' \ + 'JOIN pg_class ON pg_class.oid = pg_trigger.tgrelid ' \ + 'WHERE pg_class.relname = ? AND pg_trigger.tgname = ?', table_name, trigger_name ].first end def get_function(func_name) DB[ - "SELECT proname FROM pg_proc WHERE proname = ?", + 'SELECT proname FROM pg_proc WHERE proname = ?', func_name ].first end @@ -296,7 +320,7 @@ def get_function(func_name) String :s end DB.pgt_outbox_setup(:accounts, notify: true, notify_channel: 'my_custom_channel', - function_name: :spgt_outbox_events) + function_name: :spgt_outbox_events) func = get_function('pgt_outbox_notify_accounts_outbox')