diff --git a/exe/outboxify b/exe/outboxify index 14fc1d2..2ad9e40 100755 --- a/exe/outboxify +++ b/exe/outboxify @@ -5,10 +5,14 @@ require 'sequel/pgt_outbox' require 'optparse' autovacuum = true +notify = false +notify_channel = nil opts = OptionParser.new do |o| o.banner = 'Usage: outboxify [options] ' o.on('--no-autovacuum') { autovacuum = false } + o.on('--notify') { notify = true } + o.on('--notify-channel CHANNEL') { |c| notify_channel = c } end opts.parse! @@ -29,6 +33,7 @@ if table.nil? exit 2 end -options = { autovacuum: } +options = { autovacuum:, 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 ea5c967..4988f25 100644 --- a/lib/sequel/pgt_outbox/table.rb +++ b/lib/sequel/pgt_outbox/table.rb @@ -19,7 +19,7 @@ def initialize(table, db, opts: {}) @db = db end - def create! + def create! # rubocop:disable Metrics/MethodLength db.run 'CREATE EXTENSION IF NOT EXISTS "uuid-ossp"' if uuid_primary_key? create_table! integer_columns! @@ -29,6 +29,7 @@ def create! jsonb_columns! indexes! autovacuum_settings! + notify_trigger! self end @@ -124,6 +125,14 @@ def autovacuum_vacuum_cost_delay opts.fetch(:autovacuum_vacuum_cost_delay, 0) 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 @@ -195,6 +204,14 @@ def autovacuum_settings! db.run "ALTER TABLE #{quoted_name} SET (#{settings.map { |k, v| "#{k} = #{v}" }.join(", ")})" 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 279daeb..9d07897 100755 --- a/test/sequel/test_pgt_outbox.rb +++ b/test/sequel/test_pgt_outbox.rb @@ -315,6 +315,105 @@ def get_reloptions(table_name) _(opts).must_include 'autovacuum_vacuum_cost_delay=20' end end + + 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