feat(table): add optional PG NOTIFY trigger for outbox tables

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] <db uri> <table>'
   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)

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

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

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