Module: Pgbus::PgmqSchema

Defined in:
lib/pgbus/pgmq_schema.rb

Overview

Manages embedded PGMQ SQL schema for extension-free installations.

Supports three modes:

:auto      - Try extension first, fall back to embedded SQL
:extension - Require the pgmq PostgreSQL extension
:embedded  - Use vendored SQL (no extension needed)

Vendored SQL files live in lib/pgbus/pgmq_schema/pgmq_vVERSION.sql and are exact copies of the upstream pgmq-extension/sql/pgmq.sql at each release.

Defined Under Namespace

Classes: VersionNotFoundError

Constant Summary collapse

SCHEMA_DIR = 
File.expand_path("pgmq_schema", __dir__).freeze

Class Method Summary collapse

Class Method Details

.available_versions ⇒ Object

Returns sorted list of all vendored PGMQ versions.



25
26
27
28
29
 
# File 'lib/pgbus/pgmq_schema.rb', line 25

def available_versions
  Dir.glob(File.join(SCHEMA_DIR, "pgmq_v*.sql"))
     .map { |f| File.basename(f).match(/pgmq_v(.+)\.sql/)[1] }
     .sort_by { |v| Gem::Version.new(v) }
end

.drop_pgmq_functions_sql ⇒ Object

SQL to drop all pgmq functions/types (for clean upgrade). Uses CASCADE so dependent objects are also dropped.



97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
 
# File 'lib/pgbus/pgmq_schema.rb', line 97

def drop_pgmq_functions_sql
  <<~SQL
    DO $$
    DECLARE
      r RECORD;
    BEGIN
      -- Drop all functions in pgmq schema
      FOR r IN
        SELECT pg_catalog.pg_get_functiondef(p.oid) AS funcdef,
               n.nspname || '.' || p.proname || '(' ||
                 pg_catalog.pg_get_function_identity_arguments(p.oid) || ')' AS func_sig
        FROM pg_catalog.pg_proc p
        JOIN pg_catalog.pg_namespace n ON n.oid = p.pronamespace
        WHERE n.nspname = 'pgmq'
      LOOP
        EXECUTE 'DROP FUNCTION IF EXISTS ' || r.func_sig || ' CASCADE';
      END LOOP;

      -- Drop custom composite types in pgmq schema (e.g. message_record,
      -- queue_record, metrics_result). Standalone `CREATE TYPE ... AS (...)`
      -- types get a pg_class shell with relkind = 'c'; only skip row-types
      -- backed by an actual relation (table/view/etc.), which must be
      -- dropped via DROP TABLE rather than DROP TYPE.
      FOR r IN
        SELECT n.nspname || '.' || t.typname AS type_name
        FROM pg_catalog.pg_type t
        JOIN pg_catalog.pg_namespace n ON n.oid = t.typnamespace
        WHERE n.nspname = 'pgmq'
          AND t.typtype = 'c'
          AND NOT EXISTS (
            SELECT 1 FROM pg_catalog.pg_class c
            WHERE c.reltype = t.oid
              AND c.relkind <> 'c'
          )
      LOOP
        EXECUTE 'DROP TYPE IF EXISTS ' || r.type_name || ' CASCADE';
      END LOOP;
    END $$;
  SQL
end

.install_sql(version = latest_version) ⇒ String

Returns the SQL to install PGMQ schema without the extension. Strips the extension-only pg_dump config blocks since they're irrelevant when not installed as an extension.

Parameters:

  • version (String) (defaults to: latest_version) —

    defaults to latest

Returns:

  • (String) —

    SQL



54
55
56
57
 
# File 'lib/pgbus/pgmq_schema.rb', line 54

def install_sql(version = latest_version)
  sql = sql_for_version(version)
  strip_extension_only_blocks(sql)
end

.latest_version ⇒ Object

Returns the latest vendored PGMQ version string.



20
21
22
 
# File 'lib/pgbus/pgmq_schema.rb', line 20

def latest_version
  available_versions.last
end

.reinstall_notify_triggers_sql ⇒ Object

SQL to re-install the per-queue NOTIFY insert triggers that drop_pgmq_functions_sql cascades away (issue #360). Dropping pgmq.notify_queue_listeners() with CASCADE also drops the trigger_notify_queue_insert_listeners trigger from every queue table; install_sql re-creates the function but nothing re-creates the triggers, so NOTIFY-gated wakeups silently die fleet-wide until each queue happens to be re-ensured by a process restart.

The dropped state is fully recoverable: pgmq.notify_insert_throttle is a TABLE (preserved by the upgrade — its rows record exactly which queues had notify enabled and at what throttle, FK-bound to pgmq.meta so it can't reference a dropped queue), and pgmq.enable_notify_insert is idempotent. Replaying it per recorded row restores every trigger at its original interval. No-ops when the throttle table or the enable function is absent (a vendored version without the notify feature).



153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
 
# File 'lib/pgbus/pgmq_schema.rb', line 153

def reinstall_notify_triggers_sql
  <<~SQL
    DO $$
    DECLARE
      r RECORD;
    BEGIN
      IF to_regclass('pgmq.notify_insert_throttle') IS NULL THEN
        RETURN;
      END IF;
      IF to_regprocedure('pgmq.enable_notify_insert(text, integer)') IS NULL THEN
        RETURN;
      END IF;

      -- The FOR loop iterates a snapshot, so enable_notify_insert's
      -- internal DELETE + re-INSERT of the same throttle row is safe.
      FOR r IN
        SELECT queue_name, throttle_interval_ms
        FROM pgmq.notify_insert_throttle
      LOOP
        PERFORM pgmq.enable_notify_insert(r.queue_name, r.throttle_interval_ms);
      END LOOP;
    END $$;
  SQL
end

.sql_for_version(version) ⇒ Object

Returns the raw SQL content for a given version.



44
45
46
 
# File 'lib/pgbus/pgmq_schema.rb', line 44

def sql_for_version(version)
  File.read(sql_path(version))
end

.sql_path(version) ⇒ String

Returns the filesystem path to the vendored SQL file for a given version.

Parameters:

  • version (String) —

    e.g. "1.11.0"

Returns:

  • (String) —

    absolute path

Raises:



36
37
38
39
40
41
 
# File 'lib/pgbus/pgmq_schema.rb', line 36

def sql_path(version)
  path = File.join(SCHEMA_DIR, "pgmq_v#{version}.sql")
  raise VersionNotFoundError, "No vendored PGMQ SQL for version #{version}" unless File.exist?(path)

  path
end

.version_tracking_extension_sql(version = latest_version) ⇒ String

Returns SQL to record an extension-based installation in the version tracking table.

Parameters:

  • version (String) (defaults to: latest_version)

Returns:

  • (String) —

    SQL



81
82
83
84
85
86
87
88
89
90
91
92
93
 
# File 'lib/pgbus/pgmq_schema.rb', line 81

def version_tracking_extension_sql(version = latest_version)
  <<~SQL
    CREATE TABLE IF NOT EXISTS pgbus_pgmq_schema_versions (
      id SERIAL PRIMARY KEY,
      version VARCHAR NOT NULL,
      installed_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,
      install_method VARCHAR NOT NULL DEFAULT 'embedded'
    );

    INSERT INTO pgbus_pgmq_schema_versions (version, install_method)
    VALUES ('#{version}', 'extension');
  SQL
end

.version_tracking_sql(version = latest_version) ⇒ String

Returns SQL to create the version tracking table and record an installation.

Parameters:

  • version (String) (defaults to: latest_version) —

    defaults to latest

Returns:

  • (String) —

    SQL



63
64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'lib/pgbus/pgmq_schema.rb', line 63

def version_tracking_sql(version = latest_version)
  <<~SQL
    CREATE TABLE IF NOT EXISTS pgbus_pgmq_schema_versions (
      id SERIAL PRIMARY KEY,
      version VARCHAR NOT NULL,
      installed_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,
      install_method VARCHAR NOT NULL DEFAULT 'embedded'
    );

    INSERT INTO pgbus_pgmq_schema_versions (version, install_method)
    VALUES ('#{version}', 'embedded');
  SQL
end