pgmq-migration 0.3.0.0 → 0.4.0.0

raw patch · 17 files changed

+2884/−498 lines, 17 filesdep +aesondep +containersdep +directorydep −hasql-migrationdep −transformersdep ~bytestringdep ~hasql-transactiondep ~textPVP ok

version bump matches the API change (PVP)

Dependencies added: aeson, containers, directory, pg-migrate, pg-migrate-embed, pg-migrate-import-hasql-migration

Dependencies removed: hasql-migration, transformers

Dependency ranges changed: bytestring, hasql-transaction, text

API changes (from Hackage documentation)

Files

• CHANGELOG.md +60/−0
• migrations/0001-install-v1.11.0.sql +2076/−0
• migrations/0002-schema-management-comment.sql +2/−0
• migrations/manifest +2/−0
• pgmq-migration.cabal +28/−23
• src/Pgmq/Migration.hs +6/−143
• src/Pgmq/Migration/History/HasqlMigration.hs +120/−0
• src/Pgmq/Migration/Internal/Definition.hs +29/−0
• src/Pgmq/Migration/Migrations.hs +0/−75
• src/Pgmq/Migration/Migrations/V1_10_0_to_V1_10_1.hs +0/−21
• src/Pgmq/Migration/Migrations/V1_10_1_to_V1_11_0.hs +0/−21
• src/Pgmq/Migration/Migrations/V1_11_0.hs +0/−29
• src/Pgmq/Migration/SchemaContract.hs +216/−0
• src/Pgmq/Migration/Sessions.hs +0/−44
• src/Pgmq/Migration/Statements.hs +0/−29
• src/Pgmq/Migration/Transactions.hs +0/−26
• test/Main.hs +345/−87

@@ -1,5 +1,65 @@ # Changelog for pgmq-migration +## 0.4.0.0 -- 2026-07-14++### Breaking Changes++* Replace the `hasql-migration` command/session API with the native+  `pgmqMigrations :: Either DefinitionError MigrationComponent` API. `Pgmq.Migration` now+  exports only `pgmqMigrations`, `MigrationComponent`, and `DefinitionError`; consumers+  compose and run a `pg-migrate` plan instead of calling a runner in this package.+* Remove the migration operations `migrate`, `upgrade`, and `validate`, the migration+  metadata `getMigrations`, `version`, `migrations`, and `upgradeMigrations`, and the+  `hasql-migration` re-exports `MigrationCommand`, `MigrationError`, and+  `SchemaMigration`.+* Remove the following exposed modules. Their contents were either predecessor migration+  command lists or the `hasql-migration` runner plumbing, both of which the native+  component replaces:+  * `Pgmq.Migration.Migrations`+  * `Pgmq.Migration.Migrations.V1_10_0_to_V1_10_1`+  * `Pgmq.Migration.Migrations.V1_10_1_to_V1_11_0`+  * `Pgmq.Migration.Migrations.V1_11_0`+  * `Pgmq.Migration.Sessions`+  * `Pgmq.Migration.Statements`+  * `Pgmq.Migration.Transactions`+* An existing ledger written by a previous release of this package must be imported+  through the direct or the explicitly opted-in equivalent-history route (see below)+  before the native runner takes over. The native runner does not read the old+  `public.schema_migrations` table on its own.+* Require the `pg-migrate` 1.1 family (`pg-migrate`, `pg-migrate-embed`, and+  `pg-migrate-import-hasql-migration`), up from 1.0. That release reshapes types this+  package's callers handle directly: `HistoryImportReport` becomes a multi-field record,+  `CleanupFailed` carries a primary error plus a `NonEmpty CleanupIssue`, and `SqlError`+  and `HistoryValidationError` gain constructors that exhaustive matches must cover.++### New Features++* Add the exposed module `Pgmq.Migration.History.HasqlMigration`, which maps a+  predecessor `hasql-migration` ledger onto the native baseline. It exports+  `AlternativeHistoryPolicy`, `pgmqHasqlMigrationMappings`, and+  `pgmqHasqlMigrationSourceConfig`.+  * `DirectFullInstallHistory` imports a `pgmq_v1.11.0` full-install ledger, verified by+    reproducing the exact base64 MD5 recorded in `public.schema_migrations`.+  * `EquivalentTwoStepUpgradeHistory` imports a `v1.10.0 -> v1.10.1 -> v1.11.0` upgrade+    ledger. It is never selected implicitly, and is additionally guarded by the read-only+    PGMQ 1.11 schema contract.+* Add the exposed module `Pgmq.Migration.SchemaContract`, exporting+  `pgmqV1_11StateValidator` and `pgmqV1_11StateEvidenceKey`. The validator checks the PGMQ+  1.11 schemas, tables, columns, constraints, types, and functions that pgmq-hs depends on+  without modifying database state.+* Append `0002-schema-management-comment` as an observable native-runner canary after the+  imported historical baseline. It is non-destructive: it only sets a `COMMENT ON SCHEMA+  pgmq`, so the first native-only upgrade is provable after either import route.++### Other Changes++* Load `Database.PostgreSQL.Migrate.Embed.RecompilePlugin` in the manifest-embedding+  module. GHC 9.12 offers Template Haskell no way to depend on the migrations directory+  itself, so adding or removing a SQL file could otherwise reuse a stale object file and+  skip strict manifest membership validation. This forces the module to recompile whenever+  GHC runs over the package. Note that a build where *only* SQL files changed can still be+  short-circuited by `cabal`'s own up-to-date check before GHC is invoked.+ ## 0.3.0.0 -- 2026-05-31  * Version bump only — coordinated release with pgmq-effectful 0.3.0.0.

@@ -0,0 +1,2076 @@+------------------------------------------------------------+-- Schema, tables, records, privileges, indexes, etc+------------------------------------------------------------+-- When installed as an extension, we don't need to create the `pgmq` schema+-- because it is automatically created by postgres due to being declared in+-- the extension control file+DO+$$+BEGIN+    IF (SELECT NOT EXISTS( SELECT 1 FROM pg_extension WHERE extname = 'pgmq')) THEN+      CREATE SCHEMA IF NOT EXISTS pgmq;+    END IF;+END+$$;++-- Table where queues and metadata about them is stored+CREATE TABLE IF NOT EXISTS pgmq.meta (+    queue_name VARCHAR UNIQUE NOT NULL,+    is_partitioned BOOLEAN NOT NULL,+    is_unlogged BOOLEAN NOT NULL,+    created_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL+);++-- Table to track notification throttling for queues+CREATE UNLOGGED TABLE IF NOT EXISTS pgmq.notify_insert_throttle (+    queue_name           VARCHAR UNIQUE NOT NULL -- Queue name (without 'q_' prefix)+       CONSTRAINT notify_insert_throttle_meta_queue_name_fk+            REFERENCES pgmq.meta (queue_name)+            ON DELETE CASCADE,+    throttle_interval_ms INTEGER NOT NULL DEFAULT 0, -- Min milliseconds between notifications (0 = no throttling)+    last_notified_at     TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT to_timestamp(0) -- Timestamp of last sent notification+);++CREATE INDEX IF NOT EXISTS idx_notify_throttle_active+    ON pgmq.notify_insert_throttle (queue_name, last_notified_at)+    WHERE throttle_interval_ms > 0;++CREATE TABLE IF NOT EXISTS pgmq.topic_bindings+(+    pattern        text NOT NULL, -- Wildcard pattern for routing key matching (* = one segment, # = zero or more segments)+    queue_name     text NOT NULL  -- Name of the queue that receives messages when pattern matches+        CONSTRAINT topic_bindings_meta_queue_name_fk+            REFERENCES pgmq.meta (queue_name)+            ON DELETE CASCADE,+    bound_at       TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, -- Timestamp when the binding was created+    compiled_regex text GENERATED ALWAYS AS (+        -- Pre-compile the pattern to regex for faster matching+        -- This avoids runtime compilation on every send_topic call+        '^' ||+        replace(+                replace(+                        regexp_replace(pattern, '([.+?{}()|\[\]\\^$])', '\\\1', 'g'),+                        '*', '[^.]+'+                ),+                '#', '.*'+        ) || '$'+        ) STORED,                 -- Computed column: stores the compiled regex pattern+    CONSTRAINT topic_bindings_unique_pattern_queue UNIQUE (pattern, queue_name)+);++-- Create covering index for better performance when scanning patterns+-- Includes queue_name and compiled_regex to allow index-only scans (no table access needed)+CREATE INDEX IF NOT EXISTS idx_topic_bindings_covering ON pgmq.topic_bindings (pattern) INCLUDE (queue_name, compiled_regex);++-- Allow pgmq.meta to be dumped by `pg_dump` when pgmq is installed as an extension+DO+$$+BEGIN+    IF EXISTS(SELECT 1 FROM pg_extension WHERE extname = 'pgmq') THEN+        PERFORM pg_catalog.pg_extension_config_dump('pgmq.meta', '');+        PERFORM pg_catalog.pg_extension_config_dump('pgmq.notify_insert_throttle', '');+        PERFORM pg_catalog.pg_extension_config_dump('pgmq.topic_bindings', '');+    END IF;+END+$$;++-- Grant permission to pg_monitor to all tables and sequences+GRANT USAGE ON SCHEMA pgmq TO pg_monitor;+GRANT SELECT ON ALL TABLES IN SCHEMA pgmq TO pg_monitor;+GRANT SELECT ON ALL SEQUENCES IN SCHEMA pgmq TO pg_monitor;+ALTER DEFAULT PRIVILEGES IN SCHEMA pgmq GRANT SELECT ON TABLES TO pg_monitor;+ALTER DEFAULT PRIVILEGES IN SCHEMA pgmq GRANT SELECT ON SEQUENCES TO pg_monitor;++-- This type has the shape of a message in a queue, and is often returned by+-- pgmq functions that return messages+CREATE TYPE pgmq.message_record AS (+    msg_id BIGINT,+    read_ct INTEGER,+    enqueued_at TIMESTAMP WITH TIME ZONE,+    last_read_at TIMESTAMP WITH TIME ZONE,+    vt TIMESTAMP WITH TIME ZONE,+    message JSONB,+    headers JSONB+);++CREATE TYPE pgmq.queue_record AS (+    queue_name VARCHAR,+    is_partitioned BOOLEAN,+    is_unlogged BOOLEAN,+    created_at TIMESTAMP WITH TIME ZONE+);++------------------------------------------------------------+-- Functions+------------------------------------------------------------++-- prevents race conditions during queue creation by acquiring a transaction-level advisory lock+-- uses a transaction advisory lock maintain the lock until transaction commit+-- a race condition would still exist if lock was released before commit+CREATE FUNCTION pgmq.acquire_queue_lock(queue_name TEXT)+RETURNS void AS $$+BEGIN+  PERFORM pg_advisory_xact_lock(hashtext('pgmq.queue_' || queue_name));+END;+$$ LANGUAGE plpgsql;++-- read_grouped_round_robin+-- reads messages while preserving FIFO within groups and interleaving across groups (layered round-robin)+CREATE FUNCTION pgmq.read_grouped_rr(+    queue_name TEXT,+    vt INTEGER,+    qty INTEGER+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+    sql TEXT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    sql := FORMAT(+        $QUERY$+        WITH fifo_groups AS (+            -- Determine the absolute head (oldest) message id per FIFO group, regardless of visibility+            SELECT+                COALESCE(headers->>'x-pgmq-group', '_default_fifo_group') AS fifo_key,+                MIN(msg_id) AS head_msg_id+            FROM pgmq.%1$I+            GROUP BY COALESCE(headers->>'x-pgmq-group', '_default_fifo_group')+        ),+        eligible_groups AS (+            -- Only groups whose head message is currently visible+            -- Acquire a transaction-level advisory lock per group to prevent concurrent selection+            SELECT+                g.fifo_key,+                g.head_msg_id,+                ROW_NUMBER() OVER (ORDER BY g.head_msg_id) AS group_priority+            FROM fifo_groups g+            JOIN pgmq.%2$I h ON h.msg_id = g.head_msg_id+            WHERE h.vt <= clock_timestamp()+              AND pg_try_advisory_xact_lock(pg_catalog.hashtextextended(g.fifo_key, 0))+        ),+        available_messages AS (+            -- All currently visible messages starting at the head for each eligible group+            SELECT+                m.msg_id,+                eg.group_priority,+                ROW_NUMBER() OVER (+                    PARTITION BY eg.fifo_key+                    ORDER BY m.msg_id+                ) AS msg_rank_in_group+            FROM pgmq.%3$I m+            JOIN eligible_groups eg+              ON COALESCE(m.headers->>'x-pgmq-group', '_default_fifo_group') = eg.fifo_key+            WHERE m.vt <= clock_timestamp()+              AND m.msg_id >= eg.head_msg_id+        ),+        ordered_messages AS (+            -- Layered round-robin: take rank 1 of all groups by group_priority, then rank 2, etc.+            -- Assign selection order before locking+            SELECT msg_id, ROW_NUMBER() OVER (ORDER BY msg_rank_in_group, group_priority) as selection_order+            FROM available_messages+        ),+        selected_messages AS (+            -- Lock the messages in the correct order, preserving selection_order+            SELECT om.msg_id, om.selection_order+            FROM ordered_messages om+            JOIN pgmq.%4$I m ON m.msg_id = om.msg_id+            WHERE om.selection_order <= $1+            ORDER BY om.selection_order+            FOR UPDATE OF m SKIP LOCKED+        ),+        updated_messages AS (+            UPDATE pgmq.%5$I m+            SET+                vt = clock_timestamp() + %6$L,+                read_ct = read_ct + 1,+                last_read_at = clock_timestamp()+            FROM selected_messages sm+            WHERE m.msg_id = sm.msg_id+              AND m.vt <= clock_timestamp() -- final guard to avoid duplicate reads under races+            RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.last_read_at, m.vt, m.message, m.headers, sm.selection_order+        )+        SELECT msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers+        FROM updated_messages+        ORDER BY selection_order;+        $QUERY$,+        qtable, qtable, qtable, qtable, qtable, make_interval(secs => vt)+    );+    RETURN QUERY EXECUTE sql USING qty;+END;+$$ LANGUAGE plpgsql;++-- read_grouped_rr_with_poll+-- reads messages using round-robin layering across groups, with polling support+CREATE FUNCTION pgmq.read_grouped_rr_with_poll(+    queue_name TEXT,+    vt INTEGER,+    qty INTEGER,+    max_poll_seconds INTEGER DEFAULT 5,+    poll_interval_ms INTEGER DEFAULT 100+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+    r pgmq.message_record;+    stop_at TIMESTAMP;+BEGIN+    stop_at := clock_timestamp() + make_interval(secs => max_poll_seconds);+    LOOP+      IF (SELECT clock_timestamp() >= stop_at) THEN+        RETURN;+      END IF;++      FOR r IN+        SELECT * FROM pgmq.read_grouped_rr(queue_name, vt, qty)+      LOOP+        RETURN NEXT r;+      END LOOP;+      IF FOUND THEN+        RETURN;+      ELSE+        PERFORM pg_sleep(poll_interval_ms::numeric / 1000);+      END IF;+    END LOOP;+END;+$$ LANGUAGE plpgsql;++-- a helper to format table names and check for invalid characters+CREATE FUNCTION pgmq.format_table_name(queue_name text, prefix text)+RETURNS TEXT AS $$+BEGIN+    IF queue_name ~ '\$|;|--|'''+    THEN+        RAISE EXCEPTION 'queue name contains invalid characters: $, ;, --, or \''';+    END IF;+    RETURN lower(prefix || '_' || queue_name);+END;+$$ LANGUAGE plpgsql;++-- read+-- reads a number of messages from a queue, setting a visibility timeout on them+CREATE FUNCTION pgmq.read(+    queue_name TEXT,+    vt INTEGER,+    qty INTEGER,+    conditional JSONB DEFAULT '{}'+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+    sql TEXT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    sql := FORMAT(+        $QUERY$+        WITH cte AS+        (+            SELECT msg_id+            FROM pgmq.%I+            WHERE vt <= clock_timestamp() AND CASE+                WHEN %L != '{}'::jsonb THEN (message @> %2$L)::integer+                ELSE 1+            END = 1+            ORDER BY msg_id ASC+            LIMIT $1+            FOR UPDATE SKIP LOCKED+        )+        UPDATE pgmq.%I m+        SET+            last_read_at = clock_timestamp(),+            vt = clock_timestamp() + %L,+            read_ct = read_ct + 1+        FROM cte+        WHERE m.msg_id = cte.msg_id+        RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.last_read_at, m.vt, m.message, m.headers;+        $QUERY$,+        qtable, conditional, qtable, make_interval(secs => vt)+    );+    RETURN QUERY EXECUTE sql USING qty;+END;+$$ LANGUAGE plpgsql;++-- read_grouped+-- reads messages with AWS SQS FIFO-style batch retrieval behavior+-- attempts to return as many messages as possible from the same message group+CREATE FUNCTION pgmq.read_grouped(+    queue_name TEXT,+    vt INTEGER,+    qty INTEGER+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+    sql TEXT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    sql := FORMAT(+        $QUERY$+        WITH fifo_groups AS (+            -- Find the minimum msg_id for each FIFO group that's ready to be processed+            SELECT+                COALESCE(headers->>'x-pgmq-group', '_default_fifo_group') as fifo_key,+                MIN(msg_id) as min_msg_id+            FROM pgmq.%I+            WHERE vt <= clock_timestamp()+            GROUP BY COALESCE(headers->>'x-pgmq-group', '_default_fifo_group')+        ),+        locked_groups AS (+            -- Lock the first available message in each FIFO group+            SELECT+                m.msg_id,+                fg.fifo_key+            FROM pgmq.%I m+            INNER JOIN fifo_groups fg ON+                COALESCE(m.headers->>'x-pgmq-group', '_default_fifo_group') = fg.fifo_key+                AND m.msg_id = fg.min_msg_id+            WHERE m.vt <= clock_timestamp()+            ORDER BY m.msg_id ASC+            FOR UPDATE SKIP LOCKED+        ),+        group_priorities AS (+            -- Assign priority to groups based on their oldest message+            SELECT+                fifo_key,+                msg_id as min_msg_id,+                ROW_NUMBER() OVER (ORDER BY msg_id) as group_priority+            FROM locked_groups+        ),+        filtered_groups as (+            SELECT * FROM group_priorities gp+            WHERE NOT EXISTS (+                -- Ensure no earlier message in this group is currently being processed+                SELECT 1+                FROM pgmq.%I m2+                WHERE COALESCE(m2.headers->>'x-pgmq-group', '_default_fifo_group') = gp.fifo_key+                AND m2.vt > clock_timestamp()+                AND m2.msg_id < gp.min_msg_id+            )+        ),+        available_messages as (+            SELECT gp.fifo_key, t.msg_id,gp.group_priority,+                ROW_NUMBER() OVER (PARTITION BY gp.fifo_key ORDER BY t.msg_id) as msg_rank_in_group+            FROM filtered_groups gp+            CROSS JOIN LATERAL (+                SELECT *+                FROM pgmq.%I t+                WHERE COALESCE(t.headers->>'x-pgmq-group', '_default_fifo_group') = gp.fifo_key+                AND t.vt <= clock_timestamp()+                ORDER BY msg_id+                LIMIT $1  -- tip to limit query impact, we know we need at most qty in each group+            ) t+            ORDER BY gp.group_priority+        ),+        batch_selection AS (+            -- Select messages to fill batch, prioritizing earliest group+            SELECT+                msg_id,+                ROW_NUMBER() OVER (ORDER BY group_priority, msg_rank_in_group) as overall_rank+            FROM available_messages+        ),+        selected_messages AS (+            -- Limit to requested quantity+            SELECT msg_id+            FROM batch_selection+            WHERE overall_rank <= $1+            ORDER BY msg_id+            FOR UPDATE SKIP LOCKED+        )+        UPDATE pgmq.%I m+        SET+            vt = clock_timestamp() + %L,+            read_ct = read_ct + 1,+            last_read_at = clock_timestamp()+        FROM selected_messages sm+        WHERE m.msg_id = sm.msg_id+        RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.last_read_at, m.vt, m.message, m.headers;+        $QUERY$,+        qtable, qtable, qtable, qtable, qtable, make_interval(secs => vt)+    );+    RETURN QUERY EXECUTE sql USING qty;+END;+$$ LANGUAGE plpgsql;++-- read_grouped_with_poll+-- reads messages with AWS SQS FIFO-style batch retrieval behavior, with polling support+CREATE FUNCTION pgmq.read_grouped_with_poll(+    queue_name TEXT,+    vt INTEGER,+    qty INTEGER,+    max_poll_seconds INTEGER DEFAULT 5,+    poll_interval_ms INTEGER DEFAULT 100+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+    r pgmq.message_record;+    stop_at TIMESTAMP;+BEGIN+    stop_at := clock_timestamp() + make_interval(secs => max_poll_seconds);+    LOOP+      IF (SELECT clock_timestamp() >= stop_at) THEN+        RETURN;+      END IF;++      FOR r IN+        SELECT * FROM pgmq.read_grouped(queue_name, vt, qty)+      LOOP+        RETURN NEXT r;+      END LOOP;+      IF FOUND THEN+        RETURN;+      ELSE+        PERFORM pg_sleep(poll_interval_ms::numeric / 1000);+      END IF;+    END LOOP;+END;+$$ LANGUAGE plpgsql;++---- read_with_poll+---- reads a number of messages from a queue, setting a visibility timeout on them+CREATE FUNCTION pgmq.read_with_poll(+    queue_name TEXT,+    vt INTEGER,+    qty INTEGER,+    max_poll_seconds INTEGER DEFAULT 5,+    poll_interval_ms INTEGER DEFAULT 100,+    conditional JSONB DEFAULT '{}'+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+    r pgmq.message_record;+    stop_at TIMESTAMP;+    sql TEXT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    stop_at := clock_timestamp() + make_interval(secs => max_poll_seconds);+    LOOP+      IF (SELECT clock_timestamp() >= stop_at) THEN+        RETURN;+      END IF;++      sql := FORMAT(+          $QUERY$+          WITH cte AS+          (+              SELECT msg_id+              FROM pgmq.%I+              WHERE vt <= clock_timestamp() AND CASE+                  WHEN %L != '{}'::jsonb THEN (message @> %2$L)::integer+                  ELSE 1+              END = 1+              ORDER BY msg_id ASC+              LIMIT $1+              FOR UPDATE SKIP LOCKED+          )+          UPDATE pgmq.%I m+          SET+              last_read_at = clock_timestamp(),+              vt = clock_timestamp() + %L,+              read_ct = read_ct + 1+          FROM cte+          WHERE m.msg_id = cte.msg_id+          RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.last_read_at, m.vt, m.message, m.headers;+          $QUERY$,+          qtable, conditional, qtable, make_interval(secs => vt)+      );++      FOR r IN+        EXECUTE sql USING qty+      LOOP+        RETURN NEXT r;+      END LOOP;+      IF FOUND THEN+        RETURN;+      ELSE+        PERFORM pg_sleep(poll_interval_ms::numeric / 1000);+      END IF;+    END LOOP;+END;+$$ LANGUAGE plpgsql;++---- archive+---- removes a message from the queue, and sends it to the archive, where its+---- saved permanently.+CREATE FUNCTION pgmq.archive(+    queue_name TEXT,+    msg_id BIGINT+)+RETURNS BOOLEAN AS $$+DECLARE+    sql TEXT;+    result BIGINT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+    atable TEXT := pgmq.format_table_name(queue_name, 'a');+BEGIN+    sql := FORMAT(+        $QUERY$+        WITH archived AS (+            DELETE FROM pgmq.%I+            WHERE msg_id = $1+            RETURNING msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers+        )+        INSERT INTO pgmq.%I (msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers)+        SELECT msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers+        FROM archived+        RETURNING msg_id;+        $QUERY$,+        qtable, atable+    );+    EXECUTE sql USING msg_id INTO result;+    RETURN NOT (result IS NULL);+END;+$$ LANGUAGE plpgsql;++---- archive+---- removes an array of message ids from the queue, and sends it to the archive,+---- where these messages will be saved permanently.+CREATE FUNCTION pgmq.archive(+    queue_name TEXT,+    msg_ids BIGINT[]+)+RETURNS SETOF BIGINT AS $$+DECLARE+    sql TEXT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+    atable TEXT := pgmq.format_table_name(queue_name, 'a');+BEGIN+    sql := FORMAT(+        $QUERY$+        WITH archived AS (+            DELETE FROM pgmq.%I+            WHERE msg_id = ANY($1)+            RETURNING msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers+        )+        INSERT INTO pgmq.%I (msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers)+        SELECT msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers+        FROM archived+        RETURNING msg_id;+        $QUERY$,+        qtable, atable+    );+    RETURN QUERY EXECUTE sql USING msg_ids;+END;+$$ LANGUAGE plpgsql;++---- delete+---- deletes a message id from the queue permanently+CREATE FUNCTION pgmq.delete(+    queue_name TEXT,+    msg_id BIGINT+)+RETURNS BOOLEAN AS $$+DECLARE+    sql TEXT;+    result BIGINT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    sql := FORMAT(+        $QUERY$+        DELETE FROM pgmq.%I+        WHERE msg_id = $1+        RETURNING msg_id+        $QUERY$,+        qtable+    );+    EXECUTE sql USING msg_id INTO result;+    RETURN NOT (result IS NULL);+END;+$$ LANGUAGE plpgsql;++---- delete+---- deletes an array of message ids from the queue permanently+CREATE FUNCTION pgmq.delete(+    queue_name TEXT,+    msg_ids BIGINT[]+)+RETURNS SETOF BIGINT AS $$+DECLARE+    sql TEXT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    sql := FORMAT(+        $QUERY$+        DELETE FROM pgmq.%I+        WHERE msg_id = ANY($1)+        RETURNING msg_id+        $QUERY$,+        qtable+    );+    RETURN QUERY EXECUTE sql USING msg_ids;+END;+$$ LANGUAGE plpgsql;++-- send: actual implementation+CREATE FUNCTION pgmq.send(+    queue_name TEXT,+    msg JSONB,+    headers JSONB,+    delay TIMESTAMP WITH TIME ZONE+) RETURNS SETOF BIGINT AS $$+DECLARE+    sql TEXT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    sql := FORMAT(+            $QUERY$+        INSERT INTO pgmq.%I (vt, message, headers)+        VALUES ($2, $1, $3)+        RETURNING msg_id;+        $QUERY$,+            qtable+           );+    RETURN QUERY EXECUTE sql USING msg, delay, headers;+END;+$$ LANGUAGE plpgsql;++-- send: 2 args, no delay or headers+CREATE FUNCTION pgmq.send(+    queue_name TEXT,+    msg JSONB+) RETURNS SETOF BIGINT AS $$+    SELECT * FROM pgmq.send(queue_name, msg, NULL, clock_timestamp());+$$ LANGUAGE sql;++-- send: 3 args with headers+CREATE FUNCTION pgmq.send(+    queue_name TEXT,+    msg JSONB,+    headers JSONB+) RETURNS SETOF BIGINT AS $$+    SELECT * FROM pgmq.send(queue_name, msg, headers, clock_timestamp());+$$ LANGUAGE sql;++-- send: 3 args with integer delay+CREATE FUNCTION pgmq.send(+    queue_name TEXT,+    msg JSONB,+    delay INTEGER+) RETURNS SETOF BIGINT AS $$+    SELECT * FROM pgmq.send(queue_name, msg, NULL, clock_timestamp() + make_interval(secs => delay));+$$ LANGUAGE sql;++-- send: 3 args with timestamp+CREATE FUNCTION pgmq.send(+    queue_name TEXT,+    msg JSONB,+    delay TIMESTAMP WITH TIME ZONE+) RETURNS SETOF BIGINT AS $$+    SELECT * FROM pgmq.send(queue_name, msg, NULL, delay);+$$ LANGUAGE sql;++-- send: 4 args with integer delay+CREATE FUNCTION pgmq.send(+    queue_name TEXT,+    msg JSONB,+    headers JSONB,+    delay INTEGER+) RETURNS SETOF BIGINT AS $$+    SELECT * FROM pgmq.send(queue_name, msg, headers, clock_timestamp() + make_interval(secs => delay));+$$ LANGUAGE sql;++-- _validate_batch_params: Private function to validate batch parameters+CREATE FUNCTION pgmq._validate_batch_params(+    msgs JSONB[],+    headers JSONB[]+) RETURNS void AS $$+BEGIN+    -- Validate that msgs is not NULL or empty+    IF msgs IS NULL OR array_length(msgs, 1) IS NULL THEN+        RAISE EXCEPTION 'msgs cannot be NULL or empty';+    END IF;++    -- Validate that headers array length matches msgs array length if headers is provided+    -- Note: array_length returns NULL for empty arrays, so we use COALESCE to treat empty arrays as length 0+    IF headers IS NOT NULL AND COALESCE(array_length(headers, 1), 0) != COALESCE(array_length(msgs, 1), 0) THEN+        RAISE EXCEPTION 'headers array length (%) must match msgs array length (%)',+            COALESCE(array_length(headers, 1), 0), COALESCE(array_length(msgs, 1), 0);+    END IF;+END;+$$ LANGUAGE plpgsql;++-- _send_batch: Private function that performs the actual batch insert without validation+CREATE FUNCTION pgmq._send_batch(+    queue_name TEXT,+    msgs JSONB[],+    headers JSONB[],+    delay TIMESTAMP WITH TIME ZONE+) RETURNS SETOF BIGINT AS $$+DECLARE+    sql TEXT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    sql := FORMAT(+            $QUERY$+        INSERT INTO pgmq.%I (vt, message, headers)+        SELECT $2, unnest($1), unnest(coalesce($3, ARRAY[]::jsonb[]))+        RETURNING msg_id;+        $QUERY$,+            qtable+           );+    RETURN QUERY EXECUTE sql USING msgs, delay, headers;+END;+$$ LANGUAGE plpgsql;++-- send_batch: Public function with validation+CREATE FUNCTION pgmq.send_batch(+    queue_name TEXT,+    msgs JSONB[],+    headers JSONB[],+    delay TIMESTAMP WITH TIME ZONE+) RETURNS SETOF BIGINT AS $$+BEGIN+    PERFORM pgmq._validate_batch_params(msgs, headers);+    RETURN QUERY SELECT * FROM pgmq._send_batch(queue_name, msgs, headers, delay);+END;+$$ LANGUAGE plpgsql;++-- send batch: 2 args+CREATE FUNCTION pgmq.send_batch(+    queue_name TEXT,+    msgs JSONB[]+) RETURNS SETOF BIGINT AS $$+    SELECT * FROM pgmq.send_batch(queue_name, msgs, NULL, clock_timestamp());+$$ LANGUAGE sql;++-- send batch: 3 args with headers+CREATE FUNCTION pgmq.send_batch(+    queue_name TEXT,+    msgs JSONB[],+    headers JSONB[]+) RETURNS SETOF BIGINT AS $$+    SELECT * FROM pgmq.send_batch(queue_name, msgs, headers, clock_timestamp());+$$ LANGUAGE sql;++-- send batch: 3 args with integer delay+CREATE FUNCTION pgmq.send_batch(+    queue_name TEXT,+    msgs JSONB[],+    delay INTEGER+) RETURNS SETOF BIGINT AS $$+    SELECT * FROM pgmq.send_batch(queue_name, msgs, NULL, clock_timestamp() + make_interval(secs => delay));+$$ LANGUAGE sql;++-- send batch: 3 args with timestamp+CREATE FUNCTION pgmq.send_batch(+    queue_name TEXT,+    msgs JSONB[],+    delay TIMESTAMP WITH TIME ZONE+) RETURNS SETOF BIGINT AS $$+    SELECT * FROM pgmq.send_batch(queue_name, msgs, NULL, delay);+$$ LANGUAGE sql;++-- send_batch: 4 args with integer delay+CREATE FUNCTION pgmq.send_batch(+    queue_name TEXT,+    msgs JSONB[],+    headers JSONB[],+    delay INTEGER+) RETURNS SETOF BIGINT AS $$+    SELECT * FROM pgmq.send_batch(queue_name, msgs, headers, clock_timestamp() + make_interval(secs => delay));+$$ LANGUAGE sql;++-- returned by pgmq.metrics() and pgmq.metrics_all+CREATE TYPE pgmq.metrics_result AS (+    queue_name text,+    queue_length bigint,+    newest_msg_age_sec int,+    oldest_msg_age_sec int,+    total_messages bigint,+    scrape_time timestamp with time zone,+    queue_visible_length bigint+);++-- get metrics for a single queue+CREATE FUNCTION pgmq.metrics(queue_name TEXT)+RETURNS pgmq.metrics_result AS $$+DECLARE+    result_row pgmq.metrics_result;+    query TEXT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    query := FORMAT(+        $QUERY$+        WITH q_summary AS (+            SELECT+                count(*) as queue_length,+                count(CASE WHEN vt <= NOW() THEN 1 END) as queue_visible_length,+                EXTRACT(epoch FROM (NOW() - max(enqueued_at)))::int as newest_msg_age_sec,+                EXTRACT(epoch FROM (NOW() - min(enqueued_at)))::int as oldest_msg_age_sec,+                NOW() as scrape_time+            FROM pgmq.%I+        ),+        all_metrics AS (+            SELECT CASE+                WHEN is_called THEN last_value ELSE 0+                END as total_messages+            FROM pgmq.%I+        )+        SELECT+            %L as queue_name,+            q_summary.queue_length,+            q_summary.newest_msg_age_sec,+            q_summary.oldest_msg_age_sec,+            all_metrics.total_messages,+            q_summary.scrape_time,+            q_summary.queue_visible_length+        FROM q_summary, all_metrics+        $QUERY$,+        qtable, qtable || '_msg_id_seq', queue_name+    );+    EXECUTE query INTO result_row;+    RETURN result_row;+END;+$$ LANGUAGE plpgsql;++-- get metrics for all queues+CREATE FUNCTION pgmq."metrics_all"()+RETURNS SETOF pgmq.metrics_result AS $$+DECLARE+    row_name RECORD;+    result_row pgmq.metrics_result;+BEGIN+    FOR row_name IN SELECT queue_name FROM pgmq.meta LOOP+        result_row := pgmq.metrics(row_name.queue_name);+        RETURN NEXT result_row;+    END LOOP;+END;+$$ LANGUAGE plpgsql;++-- list queues+CREATE FUNCTION pgmq."list_queues"()+RETURNS SETOF pgmq.queue_record AS $$+BEGIN+  RETURN QUERY SELECT * FROM pgmq.meta;+END+$$ LANGUAGE plpgsql;++-- purge queue, deleting all entries in it.+CREATE OR REPLACE FUNCTION pgmq."purge_queue"(queue_name TEXT)+RETURNS BIGINT AS $$+DECLARE+  deleted_count INTEGER;+  qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+  -- Get the row count before truncating+  EXECUTE format('SELECT count(*) FROM pgmq.%I', qtable) INTO deleted_count;++  -- Use TRUNCATE for better performance on large tables+  EXECUTE format('TRUNCATE TABLE pgmq.%I', qtable);++  -- Return the number of purged rows+  RETURN deleted_count;+END+$$ LANGUAGE plpgsql;++-- unassign archive, so it can be kept when a queue is deleted+CREATE FUNCTION pgmq."detach_archive"(queue_name TEXT)+RETURNS VOID AS $$+DECLARE+  atable TEXT := pgmq.format_table_name(queue_name, 'a');+BEGIN+  RAISE WARNING 'detach_archive(queue_name) is deprecated and is a no-op. It will be removed in PGMQ v2.0. Archive tables are no longer member objects.';+END+$$ LANGUAGE plpgsql;++-- pop: implementation+CREATE FUNCTION pgmq.pop(queue_name TEXT, qty INTEGER DEFAULT 1)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+    sql TEXT;+    result pgmq.message_record;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    sql := FORMAT(+        $QUERY$+        WITH cte AS+            (+                SELECT msg_id+                FROM pgmq.%I+                WHERE vt <= clock_timestamp()+                ORDER BY msg_id ASC+                LIMIT $1+                FOR UPDATE SKIP LOCKED+            )+        DELETE from pgmq.%I+        WHERE msg_id IN (select msg_id from cte)+        RETURNING msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers;+        $QUERY$,+        qtable, qtable+    );+    RETURN QUERY EXECUTE sql USING qty;+END;+$$ LANGUAGE plpgsql;++-- Sets timestamp vt of a message, returns it+CREATE FUNCTION pgmq.set_vt(queue_name TEXT, msg_id BIGINT, vt TIMESTAMP WITH TIME ZONE)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+    sql TEXT;+    result pgmq.message_record;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    sql := FORMAT(+        $QUERY$+        UPDATE pgmq.%I+        SET vt = $1+        WHERE msg_id = $2+        RETURNING msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers;+        $QUERY$, +        qtable+    );+    RETURN QUERY EXECUTE sql USING vt, msg_id;+END;+$$ LANGUAGE plpgsql;++-- Sets integer vt of a message, returns it+CREATE FUNCTION pgmq.set_vt(queue_name TEXT, msg_id BIGINT, vt INTEGER)+RETURNS SETOF pgmq.message_record AS $$+    SELECT * FROM pgmq.set_vt(queue_name, msg_id, clock_timestamp() + make_interval(secs => vt));+$$ LANGUAGE sql;++-- Sets timestamp vt of multiple messages, returns them+CREATE FUNCTION pgmq.set_vt(+    queue_name TEXT,+    msg_ids BIGINT[],+    vt TIMESTAMP WITH TIME ZONE+)+RETURNS SETOF pgmq.message_record AS $$+DECLARE+    sql TEXT;+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+BEGIN+    sql := FORMAT(+        $QUERY$+        UPDATE pgmq.%I+        SET vt = $1+        WHERE msg_id = ANY($2)+        RETURNING msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers;+        $QUERY$,+        qtable+    );+    RETURN QUERY EXECUTE sql USING vt, msg_ids;+END;+$$ LANGUAGE plpgsql;++-- Sets integer vt of multiple messages, returns them+CREATE FUNCTION pgmq.set_vt(+    queue_name TEXT,+    msg_ids BIGINT[],+    vt INTEGER+)+RETURNS SETOF pgmq.message_record AS $$+    SELECT * FROM pgmq.set_vt(queue_name, msg_ids, clock_timestamp() + make_interval(secs => vt));+$$ LANGUAGE sql;++CREATE FUNCTION pgmq._get_pg_partman_schema()+RETURNS TEXT AS $$+  SELECT+    extnamespace::regnamespace::text+  FROM+    pg_extension+  WHERE+    extname = 'pg_partman';+$$ LANGUAGE SQL;++CREATE FUNCTION pgmq.drop_queue(queue_name TEXT, partitioned BOOLEAN)+RETURNS BOOLEAN AS $$+DECLARE+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+    fq_qtable TEXT := 'pgmq.' || qtable;+    atable TEXT := pgmq.format_table_name(queue_name, 'a');+    fq_atable TEXT := 'pgmq.' || atable;+BEGIN+    RAISE WARNING 'drop_queue(queue_name, partitioned) is deprecated and will be removed in PGMQ v2.0. Use drop_queue(queue_name) instead';++    PERFORM pgmq.drop_queue(queue_name);++    RETURN TRUE;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq.drop_queue(queue_name TEXT)+RETURNS BOOLEAN AS $$+DECLARE+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+    qtable_seq TEXT := qtable || '_msg_id_seq';+    fq_qtable TEXT := 'pgmq.' || qtable;+    atable TEXT := pgmq.format_table_name(queue_name, 'a');+    fq_atable TEXT := 'pgmq.' || atable;+    partitioned BOOLEAN;+BEGIN+    PERFORM pgmq.acquire_queue_lock(queue_name);+    EXECUTE FORMAT(+        $QUERY$+        SELECT is_partitioned FROM pgmq.meta WHERE queue_name = %L+        $QUERY$,+        queue_name+    ) INTO partitioned;++    -- check if the queue exists+    IF NOT EXISTS (+        SELECT 1+        FROM information_schema.tables+        WHERE table_name = qtable and table_schema = 'pgmq'+    ) THEN+        RAISE NOTICE 'pgmq queue `%` does not exist', queue_name;+        RETURN FALSE;+    END IF;++    EXECUTE FORMAT(+        $QUERY$+        DROP TABLE IF EXISTS pgmq.%I+        $QUERY$,+        qtable+    );++    EXECUTE FORMAT(+        $QUERY$+        DROP TABLE IF EXISTS pgmq.%I+        $QUERY$,+        atable+    );++     IF EXISTS (+          SELECT 1+          FROM information_schema.tables+          WHERE table_name = 'meta' and table_schema = 'pgmq'+     ) THEN+        EXECUTE FORMAT(+            $QUERY$+            DELETE FROM pgmq.meta WHERE queue_name = %L+            $QUERY$,+            queue_name+        );+     END IF;++     IF partitioned THEN+        EXECUTE FORMAT(+          $QUERY$+          DELETE FROM %I.part_config where parent_table in (%L, %L)+          $QUERY$,+          pgmq._get_pg_partman_schema(), fq_qtable, fq_atable+        );+     END IF;++    RETURN TRUE;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq.validate_queue_name(queue_name TEXT)+RETURNS void AS $$+BEGIN+  IF length(queue_name) > 47 THEN+    -- complete table identifier must be <= 63+    -- https://www.postgresql.org/docs/17/sql-syntax-lexical.html#SQL-SYNTAX-IDENTIFIERS+    -- e.g. template_pgmq_q_my_queue is an identifier for my_queue when partitioned+    -- template_pgmq_q_ (16) + <a max length queue name> (47) = 63 +    RAISE EXCEPTION 'queue name is too long, maximum length is 47 characters';+  END IF;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq._belongs_to_pgmq(table_name TEXT)+RETURNS BOOLEAN AS $$+DECLARE+    sql TEXT;+    result BOOLEAN;+BEGIN+  SELECT EXISTS (+    SELECT 1+    FROM pg_depend+    WHERE refobjid = (SELECT oid FROM pg_extension WHERE extname = 'pgmq')+    AND objid = (+        SELECT oid+        FROM pg_class+        WHERE relname = table_name+    )+  ) INTO result;+  RETURN result;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq.create_non_partitioned(queue_name TEXT)+RETURNS void AS $$+DECLARE+  qtable TEXT := pgmq.format_table_name(queue_name, 'q');+  qtable_seq TEXT := qtable || '_msg_id_seq';+  atable TEXT := pgmq.format_table_name(queue_name, 'a');+BEGIN+  PERFORM pgmq.validate_queue_name(queue_name);+  PERFORM pgmq.acquire_queue_lock(queue_name);++  EXECUTE FORMAT(+    $QUERY$+    CREATE TABLE IF NOT EXISTS pgmq.%I (+        msg_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,+        read_ct INT DEFAULT 0 NOT NULL,+        enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+        last_read_at TIMESTAMP WITH TIME ZONE,+        vt TIMESTAMP WITH TIME ZONE NOT NULL,+        message JSONB,+        headers JSONB+    )+    $QUERY$,+    qtable+  );++  EXECUTE FORMAT(+    $QUERY$+    CREATE TABLE IF NOT EXISTS pgmq.%I (+      msg_id BIGINT PRIMARY KEY,+      read_ct INT DEFAULT 0 NOT NULL,+      enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+      last_read_at TIMESTAMP WITH TIME ZONE,+      archived_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+      vt TIMESTAMP WITH TIME ZONE NOT NULL,+      message JSONB,+      headers JSONB+    );+    $QUERY$,+    atable+  );++  EXECUTE FORMAT(+    $QUERY$+    CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (vt ASC);+    $QUERY$,+    qtable || '_vt_idx', qtable+  );++  EXECUTE FORMAT(+    $QUERY$+    CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (archived_at);+    $QUERY$,+    'archived_at_idx_' || queue_name, atable+  );++  EXECUTE FORMAT(+    $QUERY$+    INSERT INTO pgmq.meta (queue_name, is_partitioned, is_unlogged)+    VALUES (%L, false, false)+    ON CONFLICT+    DO NOTHING;+    $QUERY$,+    queue_name+  );++END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq.create_unlogged(queue_name TEXT)+RETURNS void AS $$+DECLARE+  qtable TEXT := pgmq.format_table_name(queue_name, 'q');+  qtable_seq TEXT := qtable || '_msg_id_seq';+  atable TEXT := pgmq.format_table_name(queue_name, 'a');+BEGIN+  PERFORM pgmq.validate_queue_name(queue_name);+  PERFORM pgmq.acquire_queue_lock(queue_name);++  EXECUTE FORMAT(+    $QUERY$+    CREATE UNLOGGED TABLE IF NOT EXISTS pgmq.%I (+        msg_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,+        read_ct INT DEFAULT 0 NOT NULL,+        enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+        last_read_at TIMESTAMP WITH TIME ZONE,+        vt TIMESTAMP WITH TIME ZONE NOT NULL,+        message JSONB,+        headers JSONB+    )+    $QUERY$,+    qtable+  );++  EXECUTE FORMAT(+    $QUERY$+    CREATE TABLE IF NOT EXISTS pgmq.%I (+      msg_id BIGINT PRIMARY KEY,+      read_ct INT DEFAULT 0 NOT NULL,+      enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+      last_read_at TIMESTAMP WITH TIME ZONE,+      archived_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+      vt TIMESTAMP WITH TIME ZONE NOT NULL,+      message JSONB,+      headers JSONB+    );+    $QUERY$,+    atable+  );++  EXECUTE FORMAT(+    $QUERY$+    CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (vt ASC);+    $QUERY$,+    qtable || '_vt_idx', qtable+  );++  EXECUTE FORMAT(+    $QUERY$+    CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (archived_at);+    $QUERY$,+    'archived_at_idx_' || queue_name, atable+  );++  EXECUTE FORMAT(+    $QUERY$+    INSERT INTO pgmq.meta (queue_name, is_partitioned, is_unlogged)+    VALUES (%L, false, true)+    ON CONFLICT+    DO NOTHING;+    $QUERY$,+    queue_name+  );+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq._get_partition_col(partition_interval TEXT)+RETURNS TEXT AS $$+DECLARE+  num INTEGER;+BEGIN+    BEGIN+        num := partition_interval::INTEGER;+        RETURN 'msg_id';+    EXCEPTION+        WHEN others THEN+            RETURN 'enqueued_at';+    END;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq._extension_exists(extension_name TEXT)+    RETURNS BOOLEAN+    LANGUAGE SQL+AS $$+SELECT EXISTS (+    SELECT 1+    FROM pg_extension+    WHERE extname = extension_name+)+$$;++CREATE FUNCTION pgmq._ensure_pg_partman_installed()+RETURNS void AS $$+BEGIN+  IF NOT pgmq._extension_exists('pg_partman') THEN+    RAISE EXCEPTION 'pg_partman is required for partitioned queues';+  END IF;+END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq._get_pg_partman_major_version()+RETURNS INT+LANGUAGE SQL+AS $$+  SELECT split_part(extversion, '.', 1)::INT+  FROM pg_extension+  WHERE extname = 'pg_partman'+$$;++CREATE FUNCTION pgmq.create_partitioned(+  queue_name TEXT,+  partition_interval TEXT DEFAULT '10000',+  retention_interval TEXT DEFAULT '100000'+)+RETURNS void AS $$+DECLARE+  partition_col TEXT;+  a_partition_col TEXT;+  qtable TEXT := pgmq.format_table_name(queue_name, 'q');+  qtable_seq TEXT := qtable || '_msg_id_seq';+  atable TEXT := pgmq.format_table_name(queue_name, 'a');+  fq_qtable TEXT := 'pgmq.' || qtable;+  fq_atable TEXT := 'pgmq.' || atable;+BEGIN+  PERFORM pgmq.validate_queue_name(queue_name);+  PERFORM pgmq.acquire_queue_lock(queue_name);+  PERFORM pgmq._ensure_pg_partman_installed();+  SELECT pgmq._get_partition_col(partition_interval) INTO partition_col;++  EXECUTE FORMAT(+    $QUERY$+    CREATE TABLE IF NOT EXISTS pgmq.%I (+        msg_id BIGINT GENERATED ALWAYS AS IDENTITY,+        read_ct INT DEFAULT 0 NOT NULL,+        enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+        last_read_at TIMESTAMP WITH TIME ZONE,+        vt TIMESTAMP WITH TIME ZONE NOT NULL,+        message JSONB,+        headers JSONB+    ) PARTITION BY RANGE (%I)+    $QUERY$,+    qtable, partition_col+  );++  -- https://github.com/pgpartman/pg_partman/blob/master/doc/pg_partman.md+  -- p_parent_table - the existing parent table. MUST be schema qualified, even if in public schema.+  EXECUTE FORMAT(+    $QUERY$+    SELECT %I.create_parent(+      p_parent_table := %L,+      p_control := %L,+      p_interval := %L,+      p_type := case+        when pgmq._get_pg_partman_major_version() = 5 then 'range'+        else 'native'+      end+    )+    $QUERY$,+    pgmq._get_pg_partman_schema(),+    fq_qtable,+    partition_col,+    partition_interval+  );++  EXECUTE FORMAT(+    $QUERY$+    CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (%I);+    $QUERY$,+    qtable || '_part_idx', qtable, partition_col+  );++  EXECUTE FORMAT(+    $QUERY$+    UPDATE %I.part_config+    SET+        retention = %L,+        retention_keep_table = false,+        retention_keep_index = true,+        automatic_maintenance = 'on'+    WHERE parent_table = %L;+    $QUERY$,+    pgmq._get_pg_partman_schema(),+    retention_interval,+    'pgmq.' || qtable+  );++  EXECUTE FORMAT(+    $QUERY$+    INSERT INTO pgmq.meta (queue_name, is_partitioned, is_unlogged)+    VALUES (%L, true, false)+    ON CONFLICT+    DO NOTHING;+    $QUERY$,+    queue_name+  );++  IF partition_col = 'enqueued_at' THEN+    a_partition_col := 'archived_at';+  ELSE+    a_partition_col := partition_col;+  END IF;++  EXECUTE FORMAT(+    $QUERY$+    CREATE TABLE IF NOT EXISTS pgmq.%I (+      msg_id BIGINT NOT NULL,+      read_ct INT DEFAULT 0 NOT NULL,+      enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+      last_read_at TIMESTAMP WITH TIME ZONE,+      archived_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,+      vt TIMESTAMP WITH TIME ZONE NOT NULL,+      message JSONB,+      headers JSONB+    ) PARTITION BY RANGE (%I);+    $QUERY$,+    atable, a_partition_col+  );++  -- https://github.com/pgpartman/pg_partman/blob/master/doc/pg_partman.md+  -- p_parent_table - the existing parent table. MUST be schema qualified, even if in public schema.+  EXECUTE FORMAT(+    $QUERY$+    SELECT %I.create_parent(+      p_parent_table := %L,+      p_control := %L,+      p_interval := %L,+      p_type := case+        when pgmq._get_pg_partman_major_version() = 5 then 'range'+        else 'native'+      end+    )+    $QUERY$,+    pgmq._get_pg_partman_schema(),+    fq_atable,+    a_partition_col,+    partition_interval+  );++  EXECUTE FORMAT(+    $QUERY$+    UPDATE %I.part_config+    SET+        retention = %L,+        retention_keep_table = false,+        retention_keep_index = true,+        automatic_maintenance = 'on'+    WHERE parent_table = %L;+    $QUERY$,+    pgmq._get_pg_partman_schema(),+    retention_interval,+    'pgmq.' || atable+  );++  EXECUTE FORMAT(+    $QUERY$+    CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (archived_at);+    $QUERY$,+    'archived_at_idx_' || queue_name, atable+  );++END;+$$ LANGUAGE plpgsql;++CREATE FUNCTION pgmq.create(queue_name TEXT)+RETURNS void AS $$+BEGIN+    PERFORM pgmq.create_non_partitioned(queue_name);+END;+$$ LANGUAGE plpgsql;++-- _create_fifo_index_if_not_exists+-- internal function to create GIN index on headers for better FIFO performance+CREATE OR REPLACE FUNCTION pgmq._create_fifo_index_if_not_exists(queue_name TEXT)+RETURNS void AS $$+DECLARE+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');+    index_name TEXT := qtable || '_fifo_idx';+BEGIN+    -- Create GIN index on headers for efficient FIFO key lookups+    EXECUTE FORMAT(+        $QUERY$+        CREATE INDEX IF NOT EXISTS %I ON pgmq.%I USING GIN (headers);+        $QUERY$,+        index_name, qtable+    );+END;+$$ LANGUAGE plpgsql;++-- create_fifo_index+-- creates a GIN index on the headers column to improve FIFO read performance+CREATE FUNCTION pgmq.create_fifo_index(queue_name TEXT)+RETURNS void AS $$+BEGIN+    PERFORM pgmq._create_fifo_index_if_not_exists(queue_name);+END;+$$ LANGUAGE plpgsql;++-- create_fifo_indexes_all+-- creates FIFO indexes on all existing queues+CREATE FUNCTION pgmq.create_fifo_indexes_all()+RETURNS void AS $$+DECLARE+    queue_record RECORD;+BEGIN+    FOR queue_record IN SELECT queue_name FROM pgmq.meta LOOP+        PERFORM pgmq.create_fifo_index(queue_record.queue_name);+    END LOOP;+END;+$$ LANGUAGE plpgsql;++CREATE OR REPLACE FUNCTION pgmq.convert_archive_partitioned(+  table_name TEXT,+  partition_interval TEXT DEFAULT '10000',+  retention_interval TEXT DEFAULT '100000',+  leading_partition INT DEFAULT 10+)+RETURNS void AS $$+DECLARE+  a_table_name TEXT := pgmq.format_table_name(table_name, 'a');+  a_table_name_old TEXT := pgmq.format_table_name(table_name, 'a') || '_old';+  qualified_a_table_name TEXT := format('pgmq.%I', a_table_name);+  partition_col TEXT;+  a_partition_col TEXT;+BEGIN++  PERFORM c.relkind+    FROM pg_class c+    JOIN pg_namespace n ON n.oid = c.relnamespace+    WHERE c.relname = a_table_name+    AND c.relkind = 'p';++  IF FOUND THEN+    RAISE NOTICE 'Table %s is already partitioned', a_table_name;+    RETURN;+  END IF;++  PERFORM c.relkind+    FROM pg_class c+    JOIN pg_namespace n ON n.oid = c.relnamespace+    WHERE c.relname = a_table_name+    AND c.relkind = 'r';++  IF NOT FOUND THEN+    RAISE NOTICE 'Table %s does not exists', a_table_name;+    RETURN;+  END IF;++  SELECT pgmq._get_partition_col(partition_interval) INTO partition_col;++  -- For archive tables, use archived_at for time-based partitioning+  IF partition_col = 'enqueued_at' THEN+    a_partition_col := 'archived_at';+  ELSE+    a_partition_col := partition_col;+  END IF;++  EXECUTE 'ALTER TABLE ' || qualified_a_table_name || ' RENAME TO ' || a_table_name_old;++  -- When partitioning by time (archived_at), we need to exclude constraints and indexes+  -- because the existing PRIMARY KEY on msg_id alone is incompatible with partitioning by archived_at.+  -- When partitioning by msg_id, we can keep all constraints including PRIMARY KEY.+  IF a_partition_col = 'archived_at' THEN+    EXECUTE format( 'CREATE TABLE pgmq.%I (LIKE pgmq.%I including defaults including generated including storage including comments) PARTITION BY RANGE (%I)', a_table_name, a_table_name_old, a_partition_col );+  ELSE+    EXECUTE format( 'CREATE TABLE pgmq.%I (LIKE pgmq.%I including all) PARTITION BY RANGE (%I)', a_table_name, a_table_name_old, a_partition_col );+  END IF;++  EXECUTE 'ALTER INDEX pgmq.archived_at_idx_' || table_name || ' RENAME TO archived_at_idx_' || table_name || '_old';+  EXECUTE 'CREATE INDEX archived_at_idx_'|| table_name || ' ON ' || qualified_a_table_name ||'(archived_at)';++  -- https://github.com/pgpartman/pg_partman/blob/master/doc/pg_partman.md+  -- p_parent_table - the existing parent table. MUST be schema qualified, even if in public schema.+  EXECUTE FORMAT(+    $QUERY$+    SELECT %I.create_parent(+      p_parent_table := %L,+      p_control := %L,+      p_interval := %L,+      p_type := case+        when pgmq._get_pg_partman_major_version() = 5 then 'range'+        else 'native'+      end+    )+    $QUERY$,+    pgmq._get_pg_partman_schema(),+    qualified_a_table_name,+    a_partition_col,+    partition_interval+  );++  EXECUTE FORMAT(+    $QUERY$+    UPDATE %I.part_config+    SET+      retention = %L,+      retention_keep_table = false,+      retention_keep_index = false,+      infinite_time_partitions = true+    WHERE+      parent_table = %L;+    $QUERY$,+    pgmq._get_pg_partman_schema(),+    retention_interval,+    qualified_a_table_name+  );+END;+$$ LANGUAGE plpgsql;++CREATE OR REPLACE FUNCTION pgmq.notify_queue_listeners()+RETURNS TRIGGER AS $$+DECLARE+  queue_name_extracted TEXT; -- Queue name extracted from trigger table name+  updated_count        INTEGER; -- Number of rows updated (0 or 1)+BEGIN+  queue_name_extracted := substring(TG_TABLE_NAME from 3);++  UPDATE pgmq.notify_insert_throttle+  SET last_notified_at = clock_timestamp()+  WHERE queue_name = queue_name_extracted+    AND (+      throttle_interval_ms = 0 -- No throttling configured+          OR clock_timestamp() - last_notified_at >=+             (throttle_interval_ms * INTERVAL '1 millisecond') -- Throttle interval has elapsed+    );++  -- Check how many rows were updated (will be 0 or 1)+  GET DIAGNOSTICS updated_count = ROW_COUNT;++  IF updated_count > 0 THEN+    PERFORM PG_NOTIFY('pgmq.' || TG_TABLE_NAME || '.' || TG_OP, NULL);+  END IF;++RETURN NEW;+END;+$$ LANGUAGE plpgsql;++CREATE OR REPLACE FUNCTION pgmq.enable_notify_insert(queue_name TEXT, throttle_interval_ms INTEGER DEFAULT 250)+RETURNS void AS $$+DECLARE+  qtable TEXT := pgmq.format_table_name(queue_name, 'q');+  v_queue_name TEXT := queue_name;+  v_throttle_interval_ms INTEGER := throttle_interval_ms;+BEGIN+  -- Validate that throttle_interval_ms is non-negative+  IF v_throttle_interval_ms < 0 THEN+    RAISE EXCEPTION 'throttle_interval_ms must be non-negative';+  END IF;++  -- Validate that the queue table exists+  IF NOT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'pgmq' AND table_name = qtable) THEN+    RAISE EXCEPTION 'Queue "%" does not exist. Create it first using pgmq.create()', v_queue_name;+  END IF;++  PERFORM pgmq.disable_notify_insert(v_queue_name);++  INSERT INTO pgmq.notify_insert_throttle (queue_name, throttle_interval_ms)+  VALUES (v_queue_name, v_throttle_interval_ms)+  ON CONFLICT ON CONSTRAINT notify_insert_throttle_queue_name_key DO UPDATE+      SET throttle_interval_ms = EXCLUDED.throttle_interval_ms,+          last_notified_at = to_timestamp(0);++  EXECUTE FORMAT(+    $QUERY$+    CREATE CONSTRAINT TRIGGER trigger_notify_queue_insert_listeners+    AFTER INSERT ON pgmq.%I+    DEFERRABLE FOR EACH ROW+    EXECUTE PROCEDURE pgmq.notify_queue_listeners()+    $QUERY$,+    qtable+  );+END;+$$ LANGUAGE plpgsql;++CREATE OR REPLACE FUNCTION pgmq.disable_notify_insert(queue_name TEXT)+RETURNS void AS $$+DECLARE+  qtable TEXT := pgmq.format_table_name(queue_name, 'q');+  v_queue_name TEXT := queue_name;+BEGIN+  EXECUTE FORMAT(+    $QUERY$+    DROP TRIGGER IF EXISTS trigger_notify_queue_insert_listeners ON pgmq.%I;+    $QUERY$,+    qtable+  );++  DELETE FROM pgmq.notify_insert_throttle nit WHERE nit.queue_name = v_queue_name;+END;+$$ LANGUAGE plpgsql;++CREATE OR REPLACE FUNCTION pgmq.list_notify_insert_throttles()+    RETURNS TABLE+            (+                queue_name           text,+                throttle_interval_ms integer,+                last_notified_at     TIMESTAMP WITH TIME ZONE+            )+    LANGUAGE sql+    STABLE+AS+$$+    SELECT queue_name, throttle_interval_ms, last_notified_at+    FROM pgmq.notify_insert_throttle+    ORDER BY queue_name;+$$;++CREATE OR REPLACE FUNCTION pgmq.update_notify_insert(queue_name text, throttle_interval_ms integer)+    RETURNS void+    LANGUAGE plpgsql+AS+$$+BEGIN+    IF throttle_interval_ms < 0 THEN+        RAISE EXCEPTION 'throttle_interval_ms must be non-negative, got: %', throttle_interval_ms;+    END IF;++    IF NOT EXISTS (SELECT 1 FROM pgmq.meta WHERE meta.queue_name = update_notify_insert.queue_name) THEN+        RAISE EXCEPTION 'Queue "%" does not exist. Create the queue first using pgmq.create()', queue_name;+    END IF;++    IF NOT EXISTS (SELECT 1 FROM pgmq.notify_insert_throttle WHERE notify_insert_throttle.queue_name = update_notify_insert.queue_name) THEN+        RAISE EXCEPTION 'Queue "%" does not have notify_insert enabled. Enable it first using pgmq.enable_notify_insert()', queue_name;+    END IF;++    UPDATE pgmq.notify_insert_throttle+    SET throttle_interval_ms = update_notify_insert.throttle_interval_ms,+        last_notified_at = to_timestamp(0)+    WHERE notify_insert_throttle.queue_name = update_notify_insert.queue_name;+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.validate_routing_key(routing_key text)+    RETURNS boolean+    LANGUAGE plpgsql+    IMMUTABLE+AS+$$+BEGIN+    -- Valid routing key examples:+    --   "logs.error"+    --   "app.user-service.auth"+    --   "system_events.db.connection_failed"+    --+    -- Invalid routing key examples:+    --   ""                     - empty+    --   ".logs.error"          - starts with dot+    --   "logs.error."          - ends with dot+    --   "logs..error"          - consecutive dots+    --   "logs.error!"          - invalid character+    --   "logs error"           - space not allowed+    --   "logs.*"               - wildcards not allowed in routing keys++    IF routing_key IS NULL OR routing_key = '' THEN+        RAISE EXCEPTION 'routing_key cannot be NULL or empty';+    END IF;++    IF length(routing_key) > 255 THEN+        RAISE EXCEPTION 'routing_key length cannot exceed 255 characters, got % characters', length(routing_key);+    END IF;++    IF routing_key !~ '^[a-zA-Z0-9._-]+$' THEN+        RAISE EXCEPTION 'routing_key contains invalid characters. Only alphanumeric, dots, hyphens, and underscores are allowed. Got: %', routing_key;+    END IF;++    IF routing_key ~ '^\.' THEN+        RAISE EXCEPTION 'routing_key cannot start with a dot. Got: %', routing_key;+    END IF;++    IF routing_key ~ '\.$' THEN+        RAISE EXCEPTION 'routing_key cannot end with a dot. Got: %', routing_key;+    END IF;++    IF routing_key ~ '\.\.' THEN+        RAISE EXCEPTION 'routing_key cannot contain consecutive dots. Got: %', routing_key;+    END IF;++    RETURN true;+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.validate_topic_pattern(pattern text)+    RETURNS boolean+    LANGUAGE plpgsql+    IMMUTABLE+AS+$$+BEGIN+    -- Valid pattern examples:+    --   "logs.*"           - matches one segment after logs. (e.g., logs.error, logs.info)+    --   "logs.#"           - matches one or more segments after logs. (e.g., logs.error, logs.api.error)+    --   "*.error"          - matches one segment before .error (e.g., app.error, db.error)+    --   "#.error"          - matches one or more segments before .error (e.g., app.error, x.y.error)+    --   "app.*.#"          - mixed wildcards (one segment then one or more)+    --   "#"                - catch-all pattern, matches any routing key+    --+    -- Invalid pattern examples:+    --   ".logs.*"          - starts with dot+    --   "logs.*."          - ends with dot+    --   "logs..error"      - consecutive dots+    --   "logs.**"          - consecutive stars+    --   "logs.##"          - consecutive hashes+    --   "logs.*#"          - adjacent wildcards+    --   "logs.error!"      - invalid character++    IF pattern IS NULL OR pattern = '' THEN+        RAISE EXCEPTION 'pattern cannot be NULL or empty';+    END IF;++    IF length(pattern) > 255 THEN+        RAISE EXCEPTION 'pattern length cannot exceed 255 characters, got % characters', length(pattern);+    END IF;++    IF pattern !~ '^[a-zA-Z0-9._\-*#]+$' THEN+        RAISE EXCEPTION 'pattern contains invalid characters. Only alphanumeric, dots, hyphens, underscores, *, and # are allowed. Got: %', pattern;+    END IF;++    IF pattern ~ '^\.' THEN+        RAISE EXCEPTION 'pattern cannot start with a dot. Got: %', pattern;+    END IF;++    IF pattern ~ '\.$' THEN+        RAISE EXCEPTION 'pattern cannot end with a dot. Got: %', pattern;+    END IF;++    IF pattern ~ '\.\.' THEN+        RAISE EXCEPTION 'pattern cannot contain consecutive dots. Got: %', pattern;+    END IF;++    IF pattern ~ '\*\*' THEN+        RAISE EXCEPTION 'pattern cannot contain consecutive stars (**). Use # for multi-segment matching. Got: %', pattern;+    END IF;++    IF pattern ~ '##' THEN+        RAISE EXCEPTION 'pattern cannot contain consecutive hashes (##). A single # already matches zero or more segments. Got: %', pattern;+    END IF;++    IF pattern ~ '\*#' OR pattern ~ '#\*' THEN+        RAISE EXCEPTION 'pattern cannot contain adjacent wildcards (*# or #*). Separate wildcards with dots. Got: %', pattern;+    END IF;++    RETURN true;+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.bind_topic(pattern text, queue_name text)+    RETURNS void+    LANGUAGE plpgsql+AS+$$+BEGIN+    PERFORM pgmq.validate_topic_pattern(pattern);+    IF queue_name IS NULL OR queue_name = '' THEN+        RAISE EXCEPTION 'queue_name cannot be NULL or empty';+    END IF;++    IF NOT EXISTS (SELECT 1 FROM pgmq.meta WHERE meta.queue_name = bind_topic.queue_name) THEN+        RAISE EXCEPTION 'Queue "%" does not exist. Create the queue first using pgmq.create()', queue_name;+    END IF;++    INSERT INTO pgmq.topic_bindings (pattern, queue_name)+    VALUES (pattern, queue_name)+    ON CONFLICT ON CONSTRAINT topic_bindings_unique_pattern_queue DO NOTHING;+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.unbind_topic(pattern text, queue_name text)+    RETURNS boolean+    LANGUAGE plpgsql+AS+$$+DECLARE+    rows_deleted integer;+BEGIN+    IF pattern IS NULL OR pattern = '' THEN+        RAISE EXCEPTION 'pattern cannot be NULL or empty';+    END IF;++    IF queue_name IS NULL OR queue_name = '' THEN+        RAISE EXCEPTION 'queue_name cannot be NULL or empty';+    END IF;++    DELETE+    FROM pgmq.topic_bindings+    WHERE topic_bindings.pattern = unbind_topic.pattern+      AND topic_bindings.queue_name = unbind_topic.queue_name;++    GET DIAGNOSTICS rows_deleted = ROW_COUNT;++    IF rows_deleted > 0 THEN+        RETURN true;+    ELSE+        RETURN false;+    END IF;+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.test_routing(routing_key text)+    RETURNS TABLE+            (+                pattern        text,+                queue_name     text,+                compiled_regex text+            )+    LANGUAGE plpgsql+    STABLE+AS+$$+BEGIN+    PERFORM pgmq.validate_routing_key(routing_key);+    RETURN QUERY+        SELECT b.pattern,+               b.queue_name,+               b.compiled_regex+        FROM pgmq.topic_bindings b+        WHERE routing_key ~ b.compiled_regex+        ORDER BY b.pattern;+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.send_topic(routing_key text, msg jsonb, headers jsonb, delay integer)+    RETURNS integer+    LANGUAGE plpgsql+    VOLATILE+AS+$$+DECLARE+    b             RECORD;+    matched_count integer := 0;+BEGIN+    PERFORM pgmq.validate_routing_key(routing_key);++    IF msg IS NULL THEN+        RAISE EXCEPTION 'msg cannot be NULL';+    END IF;++    IF delay < 0 THEN+        RAISE EXCEPTION 'delay cannot be negative, got: %', delay;+    END IF;++    -- Filter matching patterns in SQL for better performance (uses index)+    -- Any failure will rollback the entire transaction+    FOR b IN+        SELECT DISTINCT tb.queue_name+        FROM pgmq.topic_bindings tb+        WHERE routing_key ~ tb.compiled_regex+        ORDER BY tb.queue_name -- Deterministic ordering, deduplicated by queue_name+        LOOP+            PERFORM pgmq.send(b.queue_name, msg, headers, delay);+            matched_count := matched_count + 1;+        END LOOP;++    RETURN matched_count;+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.send_topic(routing_key text, msg jsonb)+    RETURNS integer+    LANGUAGE plpgsql+    VOLATILE+AS+$$+BEGIN+    RETURN pgmq.send_topic(routing_key, msg, NULL, 0);+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.send_topic(routing_key text, msg jsonb, delay integer)+    RETURNS integer+    LANGUAGE plpgsql+    VOLATILE+AS+$$+BEGIN+    RETURN pgmq.send_topic(routing_key, msg, NULL, delay);+END;+$$;++CREATE OR REPLACE FUNCTION pgmq.list_topic_bindings()+    RETURNS TABLE+            (+                pattern        text,+                queue_name     text,+                bound_at       TIMESTAMP WITH TIME ZONE,+                compiled_regex text+            )+    LANGUAGE sql+    STABLE+AS+$$+    SELECT pattern, queue_name, bound_at, compiled_regex+    FROM pgmq.topic_bindings+    ORDER BY bound_at DESC, pattern, queue_name;+$$;++CREATE OR REPLACE FUNCTION pgmq.list_topic_bindings(queue_name text)+    RETURNS TABLE+            (+                pattern        text,+                queue_name     text,+                bound_at       TIMESTAMP WITH TIME ZONE,+                compiled_regex text+            )+    LANGUAGE sql+    STABLE+AS+$$+    SELECT pattern, tb.queue_name, bound_at, compiled_regex+    FROM pgmq.topic_bindings tb+    WHERE tb.queue_name = list_topic_bindings.queue_name+    ORDER BY bound_at DESC, pattern;+$$;++-- send_batch_topic: Base implementation with TIMESTAMP WITH TIME ZONE delay+CREATE OR REPLACE FUNCTION pgmq.send_batch_topic(+    routing_key text,+    msgs jsonb[],+    headers jsonb[],+    delay TIMESTAMP WITH TIME ZONE+)+    RETURNS TABLE(queue_name text, msg_id bigint)+    LANGUAGE plpgsql+    VOLATILE+AS+$$+DECLARE+    b RECORD;+BEGIN+    PERFORM pgmq.validate_routing_key(routing_key);++    -- Validate batch parameters once (not per queue)+    PERFORM pgmq._validate_batch_params(msgs, headers);++    -- Filter matching patterns in SQL for better performance (uses index)+    -- Any failure will rollback the entire transaction+    FOR b IN+        SELECT DISTINCT tb.queue_name+        FROM pgmq.topic_bindings tb+        WHERE routing_key ~ tb.compiled_regex+        ORDER BY tb.queue_name -- Deterministic ordering, deduplicated by queue_name+        LOOP+            -- Use private _send_batch to avoid redundant validation+            RETURN QUERY+            SELECT b.queue_name, batch_result.msg_id+            FROM pgmq._send_batch(b.queue_name, msgs, headers, delay) AS batch_result(msg_id);+        END LOOP;++    RETURN;+END;+$$;++-- send_batch_topic: 2 args (routing_key, msgs)+CREATE OR REPLACE FUNCTION pgmq.send_batch_topic(+    routing_key text,+    msgs jsonb[]+)+    RETURNS TABLE(queue_name text, msg_id bigint)+    LANGUAGE sql+    VOLATILE+AS+$$+    SELECT * FROM pgmq.send_batch_topic(routing_key, msgs, NULL, clock_timestamp());+$$;++-- send_batch_topic: 3 args with headers+CREATE OR REPLACE FUNCTION pgmq.send_batch_topic(+    routing_key text,+    msgs jsonb[],+    headers jsonb[]+)+    RETURNS TABLE(queue_name text, msg_id bigint)+    LANGUAGE sql+    VOLATILE+AS+$$+    SELECT * FROM pgmq.send_batch_topic(routing_key, msgs, headers, clock_timestamp());+$$;++-- send_batch_topic: 3 args with integer delay+CREATE OR REPLACE FUNCTION pgmq.send_batch_topic(+    routing_key text,+    msgs jsonb[],+    delay integer+)+    RETURNS TABLE(queue_name text, msg_id bigint)+    LANGUAGE sql+    VOLATILE+AS+$$+    SELECT * FROM pgmq.send_batch_topic(routing_key, msgs, NULL, clock_timestamp() + make_interval(secs => delay));+$$;++-- send_batch_topic: 3 args with timestamp delay+CREATE OR REPLACE FUNCTION pgmq.send_batch_topic(+    routing_key text,+    msgs jsonb[],+    delay TIMESTAMP WITH TIME ZONE+)+    RETURNS TABLE(queue_name text, msg_id bigint)+    LANGUAGE sql+    VOLATILE+AS+$$+    SELECT * FROM pgmq.send_batch_topic(routing_key, msgs, NULL, delay);+$$;++-- send_batch_topic: 4 args with integer delay+CREATE OR REPLACE FUNCTION pgmq.send_batch_topic(+    routing_key text,+    msgs jsonb[],+    headers jsonb[],+    delay integer+)+    RETURNS TABLE(queue_name text, msg_id bigint)+    LANGUAGE sql+    VOLATILE+AS+$$+    SELECT * FROM pgmq.send_batch_topic(routing_key, msgs, headers, clock_timestamp() + make_interval(secs => delay));+$$;

@@ -0,0 +1,2 @@+COMMENT ON SCHEMA pgmq IS+  'Managed by pg-migrate component pgmq through 0002-schema-management-comment';

@@ -0,0 +1,2 @@+0001-install-v1.11.0.sql+0002-schema-management-comment.sql

@@ -1,10 +1,10 @@ cabal-version:      3.4 name:               pgmq-migration-version:            0.3.0.0+version:            0.4.0.0 synopsis:           PGMQ schema migrations without PostgreSQL extension description:   Installs the PGMQ schema into PostgreSQL without requiring the pgmq extension.-  Uses hasql-migration for tracking applied migrations.+  Exposes a native pg-migrate component and verified predecessor-history imports.  homepage:           https://github.com/shinzui/pgmq-hs license:            MIT@@ -15,6 +15,8 @@ build-type:         Simple extra-doc-files:    CHANGELOG.md extra-source-files:+  migrations/*.sql+  migrations/manifest   vendor/pgmq/pgmq-extension/sql/pgmq--1.10.0--1.10.1.sql   vendor/pgmq/pgmq-extension/sql/pgmq--1.10.1--1.11.0.sql   vendor/pgmq/pgmq-extension/sql/pgmq.sql@@ -30,27 +32,26 @@   import:             warnings   exposed-modules:     Pgmq.Migration-    Pgmq.Migration.Migrations-    Pgmq.Migration.Migrations.V1_10_0_to_V1_10_1-    Pgmq.Migration.Migrations.V1_10_1_to_V1_11_0-    Pgmq.Migration.Migrations.V1_11_0-    Pgmq.Migration.Sessions-    Pgmq.Migration.Statements-    Pgmq.Migration.Transactions+    Pgmq.Migration.History.HasqlMigration+    Pgmq.Migration.SchemaContract +  other-modules:      Pgmq.Migration.Internal.Definition   default-extensions:     ImportQualifiedPost     OverloadedStrings    build-depends:-    , base               >=4.18    && <5-    , bytestring         ^>=0.12-    , file-embed         ^>=0.0.16-    , hasql              ^>=1.10-    , hasql-migration    ^>=0.3.1-    , hasql-transaction  ^>=1.2-    , text               ^>=2.1-    , transformers       ^>=0.6+    , aeson                              ^>=2.2+    , base                               >=4.18     && <5+    , bytestring                         ^>=0.12+    , containers                         ^>=0.7+    , file-embed                         ^>=0.0.16+    , hasql                              ^>=1.10+    , hasql-transaction                  ^>=1.2+    , pg-migrate                         ^>=1.1.0.0+    , pg-migrate-embed                   ^>=1.1.0.0+    , pg-migrate-import-hasql-migration  ^>=1.1.0.0+    , text                               ^>=2.1    hs-source-dirs:     src   default-language:   GHC2024@@ -62,11 +63,15 @@   hs-source-dirs:   test   main-is:          Main.hs   build-depends:-    , base               >=4.18  && <5-    , ephemeral-pg       >=0.2.1+    , base                               >=4.18  && <5+    , bytestring+    , containers+    , directory+    , ephemeral-pg                       >=0.2.1     , hasql-    , hasql-migration-    , hasql-transaction+    , pg-migrate+    , pg-migrate-import-hasql-migration     , pgmq-migration-    , tasty              ^>=1.5-    , tasty-hunit        ^>=0.10+    , tasty                              ^>=1.5+    , tasty-hunit                        ^>=0.10+    , text

@@ -1,147 +1,10 @@--- | PGMQ schema migration support------ This module provides functions to install the PGMQ schema into PostgreSQL--- without requiring the pgmq extension. It uses hasql-migration for tracking--- applied migrations.------ == Fresh Installation------ For new projects, use 'migrate' to install the complete PGMQ schema:------ @--- import Hasql.Connection (acquire)--- import Hasql.Session (run)--- import Pgmq.Migration (migrate)------ main :: IO ()--- main = do---   Right conn <- acquire "host=localhost dbname=mydb"---   result <- run migrate conn---   case result of---     Right (Right ()) -> putStrLn "Migration successful"---     Right (Left err) -> print err---     Left sessionErr  -> print sessionErr--- @------ == Upgrading Existing Installations------ For projects that previously installed PGMQ via this package (e.g., at v1.10.0),--- use 'upgrade' to apply only the incremental changes:------ @--- import Hasql.Connection (acquire)--- import Hasql.Session (run)--- import Pgmq.Migration (upgrade)------ main :: IO ()--- main = do---   Right conn <- acquire "host=localhost dbname=mydb"---   result <- run upgrade conn---   case result of---     Right (Right ()) -> putStrLn "Upgrade successful"---     Right (Left err) -> print err---     Left sessionErr  -> print sessionErr--- @------ The hasql-migration library tracks which migrations have been applied,--- so 'upgrade' will only apply migrations that haven't run yet.------ == Which Function Should I Use?------ * __New project, fresh database__: Use 'migrate'--- * __Existing project using pgmq-migration__: Use 'upgrade'--- * __Not sure__: Use 'upgrade' - it's safe on fresh databases too---   (it will apply all needed migrations)+-- | Native pg-migrate component for installing PGMQ without the extension. module Pgmq.Migration-  ( -- * Migration Operations-    migrate,-    upgrade,-    validate,-    getMigrations,--    -- * Migration Info-    version,-    migrations,-    upgradeMigrations,--    -- * Re-exports-    MigrationCommand,-    MigrationError (..),-    SchemaMigration (..),+  ( DefinitionError,+    MigrationComponent,+    pgmqMigrations,   ) where -import Control.Monad (foldM)-import Hasql.Migration (MigrationCommand, MigrationError (..), SchemaMigration (..))-import Hasql.Migration qualified as Migration-import Hasql.Session (Session)-import Pgmq.Migration.Migrations qualified as Migrations-import Pgmq.Migration.Sessions qualified as Sessions---- | Current version of the PGMQ schema-version :: String-version = Migrations.version---- | Full migration commands for the current version.------ Use this for fresh installations. Applies the complete schema.-migrations :: [MigrationCommand]-migrations = Migrations.migrations---- | Incremental upgrade migrations.------ Use this to upgrade existing installations that were set up via this package.--- Contains only the delta migrations (e.g., v1.10.0 -> v1.10.1 -> v1.11.0).-upgradeMigrations :: [MigrationCommand]-upgradeMigrations = Migrations.upgradeMigrations---- | Run all PGMQ migrations for a fresh installation.------ This function is idempotent - it will only run migrations that haven't--- been applied yet. Returns 'Left' 'MigrationError' if a migration fails.------ Use this for new projects with a fresh database.-migrate :: Session (Either MigrationError ())-migrate = runMigrations migrations---- | Run upgrade migrations for existing installations.------ This function applies only the incremental migrations needed to upgrade--- from a previous version installed via this package (e.g., v1.10.0 -> v1.10.1 -> v1.11.0).------ It is idempotent and safe to run on any database - migrations that have--- already been applied will be skipped.------ Use this for existing projects that need to upgrade their PGMQ schema.-upgrade :: Session (Either MigrationError ())-upgrade = runMigrations upgradeMigrations---- | Run a list of migrations-runMigrations :: [MigrationCommand] -> Session (Either MigrationError ())-runMigrations cmds = foldM runIfOk (Right ()) cmds-  where-    runIfOk :: Either MigrationError () -> MigrationCommand -> Session (Either MigrationError ())-    runIfOk (Left err) _ = pure (Left err)-    runIfOk (Right ()) cmd = Sessions.runMigrationSession cmd---- | Validate all migrations without applying them------ Checks that previously applied migrations haven't changed.--- Returns a list of validation errors, or an empty list if validation passes.-validate :: Session [MigrationError]-validate = do-  results <- mapM validateOne validationCommands-  pure $ concat results-  where-    validationCommands = map Migration.MigrationValidation migrations--    validateOne :: MigrationCommand -> Session [MigrationError]-    validateOne cmd = do-      result <- Sessions.runMigrationSession cmd-      pure $ case result of-        Left err -> [err]-        Right () -> []---- | Get all applied migrations-getMigrations :: Session [SchemaMigration]-getMigrations = Sessions.getMigrationsSession+import Database.PostgreSQL.Migrate (DefinitionError, MigrationComponent)+import Pgmq.Migration.Internal.Definition (pgmqMigrations)

@@ -0,0 +1,120 @@+{-# LANGUAGE TemplateHaskell #-}++-- | Explicit imports from the predecessor @hasql-migration@ ledger.+module Pgmq.Migration.History.HasqlMigration+  ( AlternativeHistoryPolicy (..),+    pgmqHasqlMigrationMappings,+    pgmqHasqlMigrationSourceConfig,+  )+where++import Data.ByteString (ByteString)+import Data.FileEmbed (embedFile)+import Data.List.NonEmpty (NonEmpty ((:|)))+import Data.Map.Strict qualified as Map+import Data.Text qualified as Text+import Database.PostgreSQL.Migrate+  ( ConnectionProvider,+    EvidenceRequirement (AllOf, Evidence),+    HistoryDefinitionError,+    HistoryMapping,+    MigrationId,+    PayloadRelation (EquivalentState, SamePayload),+    evidenceKey,+    historyMapping,+    migrationId,+  )+import Database.PostgreSQL.Migrate.History.HasqlMigration+  ( HasqlMigrationDefinitionError,+    HasqlMigrationSourceConfig,+    defaultHasqlMigrationTable,+    hasqlMigrationSourceConfig,+  )+import Pgmq.Migration.Internal.Definition (embeddedMigrationEntries)+import Pgmq.Migration.SchemaContract+  ( pgmqV1_11StateEvidenceKey,+    pgmqV1_11StateValidator,+  )++-- | Select exactly one predecessor history shape. Equivalent history is never implicit.+data AlternativeHistoryPolicy+  = DirectFullInstallHistory+  | EquivalentTwoStepUpgradeHistory+  deriving stock (Eq, Show)++-- | Map the selected predecessor history shape to the native PGMQ baseline.+pgmqHasqlMigrationMappings ::+  AlternativeHistoryPolicy ->+  Either HistoryDefinitionError (NonEmpty HistoryMapping)+pgmqHasqlMigrationMappings policy = do+  directKey <- evidenceKey ("hasql-migration:" <> Text.pack directFilename)+  firstUpgradeKey <- evidenceKey ("hasql-migration:" <> Text.pack firstUpgradeFilename)+  secondUpgradeKey <- evidenceKey ("hasql-migration:" <> Text.pack secondUpgradeFilename)+  pure $ case policy of+    DirectFullInstallHistory ->+      historyMapping targetMigration (Evidence directKey) (SamePayload directKey) :| []+    EquivalentTwoStepUpgradeHistory ->+      historyMapping+        targetMigration+        ( AllOf+            ( Evidence firstUpgradeKey+                :| [ Evidence secondUpgradeKey,+                     Evidence pgmqV1_11StateEvidenceKey+                   ]+            )+        )+        EquivalentState+        :| []++-- | Build a strict source-reader configuration for the selected legacy shape.+--+-- The adapter independently reproduces and checks each base64 MD5 stored in+-- @public.schema_migrations@ before the mappings can be imported.+pgmqHasqlMigrationSourceConfig ::+  ConnectionProvider ->+  AlternativeHistoryPolicy ->+  Either HasqlMigrationDefinitionError HasqlMigrationSourceConfig+pgmqHasqlMigrationSourceConfig sourceProvider policy =+  hasqlMigrationSourceConfig+    sourceProvider+    defaultHasqlMigrationTable+    selectedFilenames+    True+    selectedPayloads+    selectedValidators+    "verified pgmq-migration cutover to native pg-migrate history"+  where+    (selectedFilenames, selectedPayloads, selectedValidators) = case policy of+      DirectFullInstallHistory ->+        ( directFilename :| [],+          Map.singleton directFilename directPayload,+          []+        )+      EquivalentTwoStepUpgradeHistory ->+        ( firstUpgradeFilename :| [secondUpgradeFilename],+          Map.fromList+            [ (firstUpgradeFilename, firstUpgradePayload),+              (secondUpgradeFilename, secondUpgradePayload)+            ],+          [pgmqV1_11StateValidator]+        )++directFilename, firstUpgradeFilename, secondUpgradeFilename :: FilePath+directFilename = "pgmq_v1.11.0"+firstUpgradeFilename = "pgmq_v1.10.0_to_v1.10.1"+secondUpgradeFilename = "pgmq_v1.10.1_to_v1.11.0"++directPayload, firstUpgradePayload, secondUpgradePayload :: ByteString+directPayload = case embeddedMigrationEntries of+  (_, payload) :| _ -> payload+firstUpgradePayload =+  $(embedFile "vendor/pgmq/pgmq-extension/sql/pgmq--1.10.0--1.10.1.sql")+secondUpgradePayload =+  $(embedFile "vendor/pgmq/pgmq-extension/sql/pgmq--1.10.1--1.11.0.sql")++targetMigration :: MigrationId+targetMigration =+  requireDefinition (migrationId "pgmq" "0001-install-v1.11.0")++requireDefinition :: (Show error) => Either error value -> value+requireDefinition = either (error . ("invalid static PGMQ history definition: " <>) . show) id

@@ -0,0 +1,29 @@+{-# LANGUAGE TemplateHaskell #-}+{-# OPTIONS_GHC -fplugin=Database.PostgreSQL.Migrate.Embed.RecompilePlugin #-}++-- | GHC 9.12 cannot track the manifest's sibling SQL directory as a Template Haskell+-- dependency, so the plugin above forces this module to be reconsidered on every build.+-- Without it, adding or removing a SQL file leaves stale embedded bytes and skips strict+-- manifest membership validation.+module Pgmq.Migration.Internal.Definition+  ( embeddedMigrationEntries,+    pgmqMigrations,+  )+where++import Data.ByteString (ByteString)+import Data.List.NonEmpty (NonEmpty)+import Database.PostgreSQL.Migrate+  ( DefinitionError,+    MigrationComponent,+    migrationComponentFromEmbeddedSql,+  )+import Database.PostgreSQL.Migrate.Embed (embedMigrationManifest)++embeddedMigrationEntries :: NonEmpty (FilePath, ByteString)+embeddedMigrationEntries =+  $(embedMigrationManifest "migrations/manifest")++pgmqMigrations :: Either DefinitionError MigrationComponent+pgmqMigrations =+  migrationComponentFromEmbeddedSql "pgmq" mempty embeddedMigrationEntries

@@ -1,75 +0,0 @@--- | Aggregation of all PGMQ migrations------ This module provides two migration paths:------ == Fresh Installation------ For new projects, use 'migrations' which installs the latest PGMQ schema directly:------ @--- import Pgmq.Migration.Migrations (migrations)------ -- Apply migrations to a fresh database--- runMigrations conn migrations--- @------ == Upgrading Existing Installations------ For projects that previously installed PGMQ via this package, use 'upgradeMigrations'--- to apply only the incremental changes:------ @--- import Pgmq.Migration.Migrations (upgradeMigrations)------ -- Apply only upgrade migrations (v1.10.0 -> v1.10.1 -> v1.11.0)--- runMigrations conn upgradeMigrations--- @------ The hasql-migration library tracks which migrations have been applied,--- so running 'upgradeMigrations' on a database with v1.10.0 installed will--- only apply the delta migrations needed to reach the current version.------ == Compatibility Note------ Existing deployments that ran the old hand-written upgrade migrations--- (e.g., @pgmq_v1.9.0_to_v1.10.0_upgrade@, @pgmq_v1.10.0_to_v1.11.0_upgrade@)--- have those names recorded in @schema_migrations@. The vendored migrations use--- different names, so @hasql-migration@ will treat them as new, unapplied migrations.--- This is correct — the vendored migrations re-apply upstream function definitions--- using @CREATE OR REPLACE FUNCTION@ and @IF NOT EXISTS@ guards, which are safe--- to re-run.-module Pgmq.Migration.Migrations-  ( version,-    migrations,-    upgradeMigrations,-  )-where--import Hasql.Migration (MigrationCommand (..))-import Pgmq.Migration.Migrations.V1_10_0_to_V1_10_1 qualified as V1_10_0_to_V1_10_1-import Pgmq.Migration.Migrations.V1_10_1_to_V1_11_0 qualified as V1_10_1_to_V1_11_0-import Pgmq.Migration.Migrations.V1_11_0 qualified as V1_11_0---- | Current version of the PGMQ schema-version :: String-version = V1_11_0.version---- | Full migration commands for the current version.------ Use this for fresh installations. Applies the complete v1.11.0 schema.-migrations :: [MigrationCommand]-migrations = V1_11_0.migrations---- | Incremental upgrade migrations.------ Use this to upgrade existing installations that were set up via this package.--- Contains only the delta migrations (v1.10.0 -> v1.10.1 -> v1.11.0).------ The hasql-migration library will skip migrations that have already been applied,--- so this is safe to run on any database regardless of its current version.-upgradeMigrations :: [MigrationCommand]-upgradeMigrations =-  [ MigrationInitialization-  ]-    ++ V1_10_0_to_V1_10_1.migrations-    ++ V1_10_1_to_V1_11_0.migrations

@@ -1,21 +0,0 @@-{-# LANGUAGE TemplateHaskell #-}---- | PGMQ v1.10.0 to v1.10.1 migration------ Embeds the upstream pgmq--1.10.0--1.10.1.sql migration directly.--- Changes: Function signature refinements, read_grouped LATERAL join optimization.-module Pgmq.Migration.Migrations.V1_10_0_to_V1_10_1-  ( migrations,-  )-where--import Data.FileEmbed (embedFile)-import Hasql.Migration (MigrationCommand (..))---- | Migration commands to upgrade from v1.10.0 to v1.10.1-migrations :: [MigrationCommand]-migrations =-  [ MigrationScript-      "pgmq_v1.10.0_to_v1.10.1"-      $(embedFile "vendor/pgmq/pgmq-extension/sql/pgmq--1.10.0--1.10.1.sql")-  ]

@@ -1,21 +0,0 @@-{-# LANGUAGE TemplateHaskell #-}---- | PGMQ v1.10.1 to v1.11.0 migration------ Embeds the upstream pgmq--1.10.1--1.11.0.sql migration directly.--- Changes: Topic routing, batch validation, notification throttle management.-module Pgmq.Migration.Migrations.V1_10_1_to_V1_11_0-  ( migrations,-  )-where--import Data.FileEmbed (embedFile)-import Hasql.Migration (MigrationCommand (..))---- | Migration commands to upgrade from v1.10.1 to v1.11.0-migrations :: [MigrationCommand]-migrations =-  [ MigrationScript-      "pgmq_v1.10.1_to_v1.11.0"-      $(embedFile "vendor/pgmq/pgmq-extension/sql/pgmq--1.10.1--1.11.0.sql")-  ]

@@ -1,29 +0,0 @@-{-# LANGUAGE TemplateHaskell #-}---- | PGMQ v1.11.0 migrations------ Embeds the upstream pgmq.sql directly from the vendored git subtree.--- This is the single source of truth for the full PGMQ schema.-module Pgmq.Migration.Migrations.V1_11_0-  ( version,-    migrations,-  )-where--import Data.FileEmbed (embedFile)-import Hasql.Migration (MigrationCommand (..))---- | Version string for this migration set-version :: String-version = "v1.11.0"---- | All migration commands for v1.11.0------ Uses the upstream pgmq.sql directly. The file is already ordered correctly--- (types before functions, tables before references) and PostgreSQL executes--- it as one transaction.-migrations :: [MigrationCommand]-migrations =-  [ MigrationInitialization,-    MigrationScript "pgmq_v1.11.0" $(embedFile "vendor/pgmq/pgmq-extension/sql/pgmq.sql")-  ]

@@ -0,0 +1,216 @@+{-# LANGUAGE LambdaCase #-}++-- | Read-only validation of the PGMQ 1.11 database objects consumed by pgmq-hs.+module Pgmq.Migration.SchemaContract+  ( pgmqV1_11StateValidator,+    pgmqV1_11StateEvidenceKey,+  )+where++import Data.Aeson qualified as Aeson+import Data.Text (Text)+import Data.Text qualified as Text+import Database.PostgreSQL.Migrate+  ( EvidenceKey,+    StateValidator,+    evidenceKey,+    stateValidationError,+    stateValidator,+  )+import Hasql.Decoders qualified as Decoders+import Hasql.Encoders qualified as Encoders+import Hasql.Statement (Statement)+import Hasql.Statement qualified as Statement+import Hasql.Transaction qualified as Transaction++-- | Stable evidence key required by the equivalent two-step import route.+pgmqV1_11StateEvidenceKey :: EvidenceKey+pgmqV1_11StateEvidenceKey =+  requireDefinition "PGMQ 1.11 state evidence key" (evidenceKey "pgmq_schema_contract_v1.11")++-- | Validate the PGMQ 1.11 objects used by pgmq-hs without changing database state.+pgmqV1_11StateValidator :: StateValidator+pgmqV1_11StateValidator =+  stateValidator pgmqV1_11StateEvidenceKey $ do+    missing <- Transaction.statement () missingContractObjectsStatement+    pure $ case missing of+      [] ->+        Right+          ( Aeson.object+              [ "contract" Aeson..= ("pgmq-1.11" :: Text),+                "checked_objects" Aeson..= length requiredContractObjects+              ]+          )+      _ ->+        Left+          ( requireDefinition+              "PGMQ 1.11 state validation diagnostic"+              ( stateValidationError+                  ("PGMQ 1.11 schema contract is missing: " <> Text.intercalate ", " missing)+              )+          )++data RequiredObject+  = RequiredSchema !Text+  | RequiredTable !Text+  | RequiredColumn !Text !Text !Text !Bool+  | RequiredConstraint !Text !Text !Char+  | RequiredType !Text+  | RequiredFunction !Text++-- Reviewed against vendor/pgmq/pgmq-extension/sql/pgmq.sql at PGMQ v1.11.0.+-- Function identities use PostgreSQL's regprocedure input notation.+requiredContractObjects :: [RequiredObject]+requiredContractObjects =+  [ RequiredSchema "pgmq",+    RequiredTable "meta",+    RequiredColumn "meta" "queue_name" "character varying" True,+    RequiredColumn "meta" "is_partitioned" "boolean" True,+    RequiredColumn "meta" "is_unlogged" "boolean" True,+    RequiredColumn "meta" "created_at" "timestamp with time zone" True,+    RequiredConstraint "meta" "meta_queue_name_key" 'u',+    RequiredTable "topic_bindings",+    RequiredColumn "topic_bindings" "pattern" "text" True,+    RequiredColumn "topic_bindings" "queue_name" "text" True,+    RequiredColumn "topic_bindings" "bound_at" "timestamp with time zone" True,+    RequiredColumn "topic_bindings" "compiled_regex" "text" False,+    RequiredConstraint "topic_bindings" "topic_bindings_meta_queue_name_fk" 'f',+    RequiredConstraint "topic_bindings" "topic_bindings_unique_pattern_queue" 'u',+    RequiredType "message_record",+    RequiredType "queue_record",+    RequiredType "metrics_result"+  ]+    <> fmap RequiredFunction requiredFunctions++requiredFunctions :: [Text]+requiredFunctions =+  [ "pgmq.archive(text,bigint)",+    "pgmq.archive(text,bigint[])",+    "pgmq.bind_topic(text,text)",+    "pgmq.create(text)",+    "pgmq.create_fifo_index(text)",+    "pgmq.create_fifo_indexes_all()",+    "pgmq.create_partitioned(text,text,text)",+    "pgmq.create_unlogged(text)",+    "pgmq.delete(text,bigint)",+    "pgmq.delete(text,bigint[])",+    "pgmq.detach_archive(text)",+    "pgmq.disable_notify_insert(text)",+    "pgmq.drop_queue(text)",+    "pgmq.enable_notify_insert(text,integer)",+    "pgmq.list_notify_insert_throttles()",+    "pgmq.list_queues()",+    "pgmq.list_topic_bindings()",+    "pgmq.list_topic_bindings(text)",+    "pgmq.metrics(text)",+    "pgmq.metrics_all()",+    "pgmq.pop(text,integer)",+    "pgmq.purge_queue(text)",+    "pgmq.read(text,integer,integer,jsonb)",+    "pgmq.read_grouped(text,integer,integer)",+    "pgmq.read_grouped_rr(text,integer,integer)",+    "pgmq.read_grouped_rr_with_poll(text,integer,integer,integer,integer)",+    "pgmq.read_grouped_with_poll(text,integer,integer,integer,integer)",+    "pgmq.read_with_poll(text,integer,integer,integer,integer,jsonb)",+    "pgmq.send(text,jsonb)",+    "pgmq.send(text,jsonb,integer)",+    "pgmq.send(text,jsonb,timestamp with time zone)",+    "pgmq.send(text,jsonb,jsonb)",+    "pgmq.send(text,jsonb,jsonb,integer)",+    "pgmq.send(text,jsonb,jsonb,timestamp with time zone)",+    "pgmq.send_batch(text,jsonb[])",+    "pgmq.send_batch(text,jsonb[],integer)",+    "pgmq.send_batch(text,jsonb[],timestamp with time zone)",+    "pgmq.send_batch(text,jsonb[],jsonb[])",+    "pgmq.send_batch(text,jsonb[],jsonb[],integer)",+    "pgmq.send_batch(text,jsonb[],jsonb[],timestamp with time zone)",+    "pgmq.send_batch_topic(text,jsonb[])",+    "pgmq.send_batch_topic(text,jsonb[],integer)",+    "pgmq.send_batch_topic(text,jsonb[],timestamp with time zone)",+    "pgmq.send_batch_topic(text,jsonb[],jsonb[])",+    "pgmq.send_batch_topic(text,jsonb[],jsonb[],integer)",+    "pgmq.send_batch_topic(text,jsonb[],jsonb[],timestamp with time zone)",+    "pgmq.send_topic(text,jsonb)",+    "pgmq.send_topic(text,jsonb,integer)",+    "pgmq.send_topic(text,jsonb,jsonb,integer)",+    "pgmq.set_vt(text,bigint,integer)",+    "pgmq.set_vt(text,bigint,timestamp with time zone)",+    "pgmq.set_vt(text,bigint[],integer)",+    "pgmq.set_vt(text,bigint[],timestamp with time zone)",+    "pgmq.test_routing(text)",+    "pgmq.unbind_topic(text,text)",+    "pgmq.update_notify_insert(text,integer)",+    "pgmq.validate_routing_key(text)",+    "pgmq.validate_topic_pattern(text)"+  ]++missingContractObjectsStatement :: Statement () [Text]+missingContractObjectsStatement =+  Statement.preparable+    contractQuery+    Encoders.noParams+    (Decoders.rowList (Decoders.column (Decoders.nonNullable Decoders.text)))++contractQuery :: Text+contractQuery =+  "SELECT required.identity\n"+    <> "FROM (VALUES\n  "+    <> Text.intercalate ",\n  " (renderRequiredObject <$> requiredContractObjects)+    <> "\n) AS required(identity, present)\n"+    <> "WHERE NOT required.present\n"+    <> "ORDER BY required.identity"++renderRequiredObject :: RequiredObject -> Text+renderRequiredObject = \case+  RequiredSchema schemaName ->+    row ("schema:" <> schemaName) ("pg_catalog.to_regnamespace(" <> literal schemaName <> ") IS NOT NULL")+  RequiredTable tableName ->+    row+      ("table:pgmq." <> tableName)+      ("pg_catalog.to_regclass(" <> literal ("pgmq." <> tableName) <> ") IS NOT NULL")+  RequiredColumn tableName columnName typeName requiredNotNull ->+    row+      ("column:pgmq." <> tableName <> "." <> columnName)+      ( "EXISTS (SELECT 1 FROM pg_catalog.pg_attribute AS attribute "+          <> "JOIN pg_catalog.pg_class AS relation ON relation.oid = attribute.attrelid "+          <> "JOIN pg_catalog.pg_namespace AS namespace ON namespace.oid = relation.relnamespace "+          <> "WHERE namespace.nspname = 'pgmq' AND relation.relname = "+          <> literal tableName+          <> " AND attribute.attname = "+          <> literal columnName+          <> " AND attribute.attnum > 0 AND NOT attribute.attisdropped "+          <> "AND pg_catalog.format_type(attribute.atttypid, attribute.atttypmod) = "+          <> literal typeName+          <> if requiredNotNull then " AND attribute.attnotnull)" else ")"+      )+  RequiredConstraint tableName constraintName constraintType ->+    row+      ("constraint:pgmq." <> tableName <> "." <> constraintName)+      ( "EXISTS (SELECT 1 FROM pg_catalog.pg_constraint AS constraint_record "+          <> "JOIN pg_catalog.pg_class AS relation ON relation.oid = constraint_record.conrelid "+          <> "JOIN pg_catalog.pg_namespace AS namespace ON namespace.oid = relation.relnamespace "+          <> "WHERE namespace.nspname = 'pgmq' AND relation.relname = "+          <> literal tableName+          <> " AND constraint_record.conname = "+          <> literal constraintName+          <> " AND constraint_record.contype = "+          <> literal (Text.singleton constraintType)+          <> ")"+      )+  RequiredType typeName ->+    row+      ("type:pgmq." <> typeName)+      ("pg_catalog.to_regtype(" <> literal ("pgmq." <> typeName) <> ") IS NOT NULL")+  RequiredFunction functionIdentity ->+    row+      ("function:" <> functionIdentity)+      ("pg_catalog.to_regprocedure(" <> literal functionIdentity <> ") IS NOT NULL")++row :: Text -> Text -> Text+row identity expression = "(" <> literal identity <> ", " <> expression <> ")"++literal :: Text -> Text+literal value = "'" <> Text.replace "'" "''" value <> "'"++requireDefinition :: String -> Either error value -> value+requireDefinition label = either (const (error (label <> " is invalid"))) id

@@ -1,44 +0,0 @@--- | Session-level migration operations-module Pgmq.Migration.Sessions-  ( runMigrationSession,-    getMigrationsSession,-  )-where--import Control.Monad.Trans.Except (ExceptT (..), runExceptT)-import Data.Text (Text)-import Hasql.Migration (MigrationCommand, MigrationError, SchemaMigration)-import Hasql.Session (Session)-import Hasql.Session qualified as Session-import Hasql.Transaction.Sessions qualified as Transaction.Sessions-import Pgmq.Migration.Statements qualified as Statements-import Pgmq.Migration.Transactions qualified as Transactions---- | Run a single migration command in a session--- Returns Left MigrationError on failure, Right () on success-runMigrationSession :: MigrationCommand -> Session (Either MigrationError ())-runMigrationSession cmd = runExceptT $ do-  -- Get the current search path-  searchPath <- ExceptT $ fmap Right $ Session.statement () Statements.getSearchPath-  -- Run the migration in a transaction with proper isolation-  result <- ExceptT $ fmap Right $ runMigrationTransaction searchPath cmd-  -- Convert Maybe MigrationError to Either-  case result of-    Nothing -> pure ()-    Just err -> ExceptT $ pure $ Left err---- | Run migration in a transaction with serializable isolation-runMigrationTransaction :: Text -> MigrationCommand -> Session (Maybe MigrationError)-runMigrationTransaction searchPath cmd =-  Transaction.Sessions.transaction-    Transaction.Sessions.Serializable-    Transaction.Sessions.Write-    (Transactions.runMigrationWithSearchPath searchPath cmd)---- | Get all applied migrations-getMigrationsSession :: Session [SchemaMigration]-getMigrationsSession =-  Transaction.Sessions.transaction-    Transaction.Sessions.ReadCommitted-    Transaction.Sessions.Read-    Transactions.getMigrationsTransaction

@@ -1,29 +0,0 @@-{-# LANGUAGE QuasiQuotes #-}---- | Search path management statements for migrations-module Pgmq.Migration.Statements-  ( getSearchPath,-    setSearchPath,-  )-where--import Data.Text (Text)-import Hasql.Decoders qualified as Decoders-import Hasql.Encoders qualified as Encoders-import Hasql.Statement (Statement, unpreparable)---- | Get the current search path-getSearchPath :: Statement () Text-getSearchPath =-  unpreparable-    "SHOW search_path"-    Encoders.noParams-    (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.text)))---- | Set the search path-setSearchPath :: Statement Text ()-setSearchPath =-  unpreparable-    "SELECT set_config('search_path', $1, true)"-    (Encoders.param (Encoders.nonNullable Encoders.text))-    Decoders.noResult

@@ -1,26 +0,0 @@--- | Transaction-level migration operations-module Pgmq.Migration.Transactions-  ( runMigrationWithSearchPath,-    getMigrationsTransaction,-  )-where--import Control.Monad (void)-import Data.Text (Text)-import Hasql.Migration (MigrationCommand, MigrationError, SchemaMigration)-import Hasql.Migration qualified as Migration-import Hasql.Transaction (Transaction)-import Hasql.Transaction qualified as Transaction-import Pgmq.Migration.Statements qualified as Statements---- | Run a migration command, preserving and restoring the search path-runMigrationWithSearchPath :: Text -> MigrationCommand -> Transaction (Maybe MigrationError)-runMigrationWithSearchPath searchPath cmd = do-  -- Set the search path for this transaction-  void $ Transaction.statement searchPath Statements.setSearchPath-  -- Run the migration-  Migration.runMigration cmd---- | Get all applied migrations-getMigrationsTransaction :: Transaction [SchemaMigration]-getMigrationsTransaction = Migration.getMigrations

@@ -1,12 +1,57 @@+{-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-}  module Main (main) where +import Control.Monad (filterM, forM_)+import Data.ByteString qualified as ByteString+import Data.Foldable (toList)+import Data.List.NonEmpty (NonEmpty (..))+import Data.Map.Strict qualified as Map+import Data.Text (Text)+import Data.Text.Encoding qualified as Text+import Database.PostgreSQL.Migrate+  ( EquivalentHistoryPolicy (AllowEquivalentHistory),+    HistoryImportError (..),+    HistoryImportOutcome (AlreadyImported, Imported),+    HistoryImportReport (HistoryImportReport, importResults),+    HistoryImportResult (importOutcome),+    HistoryValidationError (..),+    ImportOptions,+    MigrationOutcome (AlreadyApplied, AppliedNow),+    MigrationPlan,+    MigrationReport (results),+    MigrationResult (outcome),+    VerificationIssue (PendingMigration),+    VerificationReport (VerificationReport),+    connectionProviderFromSettings,+    defaultImportOptions,+    defaultRunOptions,+    migrationId,+    migrationPlan,+    runMigrationPlan,+    verifyMigrationPlan,+    withEquivalentHistory,+  )+import Database.PostgreSQL.Migrate.History.HasqlMigration+  ( HasqlMigrationImportError (..),+    HasqlMigrationSourceConfig,+    defaultHasqlMigrationTable,+    hasqlMigrationSourceConfig,+    importHasqlMigrationHistory,+  )+import Database.PostgreSQL.Migrate.Internal+  ( ComponentDescription (..),+    PlanDescription (..),+    componentNameText,+    planDescription,+  ) import EphemeralPg   ( connectionSettings,     withCached,   ) import Hasql.Connection qualified as Connection+import Hasql.Connection.Settings qualified as Settings import Hasql.Decoders qualified as Decoders import Hasql.Encoders qualified as Encoders import Hasql.Session (Session)@@ -14,8 +59,14 @@ import Hasql.Statement (preparable) import Hasql.Statement qualified as Statement import Pgmq.Migration qualified as Migration+import Pgmq.Migration.History.HasqlMigration+  ( AlternativeHistoryPolicy (..),+    pgmqHasqlMigrationMappings,+    pgmqHasqlMigrationSourceConfig,+  )+import System.Directory (doesFileExist) import Test.Tasty (TestTree, defaultMain, testGroup)-import Test.Tasty.HUnit (assertBool, assertFailure, testCase, (@?=))+import Test.Tasty.HUnit (assertFailure, testCase, (@?=))  main :: IO () main = do@@ -25,29 +76,60 @@     case connResult of       Left err -> error $ "Failed to connect: " <> show err       Right conn ->-        defaultMain (tests conn)+        defaultMain (tests connSettings conn)   case result of     Left startErr -> error $ "Failed to start temp database: " <> show startErr     Right () -> pure () -tests :: Connection.Connection -> TestTree-tests conn =+tests :: Settings.Settings -> Connection.Connection -> TestTree+tests settings conn =   testGroup     "pgmq-migration"     [ testGroup-        "fresh install"-        [ testCase "migrate on fresh database succeeds" (testMigrateFresh conn),-          testCase "migrate is idempotent" (testMigrateIdempotent conn),-          testCase "getMigrations returns applied migrations" (testGetMigrations conn),-          testCase "version is v1.11.0" testVersion+        "native definition"+        [ testCase "baseline bytes equal vendored pgmq.sql" testNativePayload,+          testCase "component pgmq has two migrations and no dependencies" testNativeComponent         ],       testGroup-        "upgrade"-        [ testCase "upgrade is idempotent" (testUpgradeIdempotent conn),-          testCase "upgrade after migrate succeeds" (testUpgradeAfterMigrate conn)+        "native runner"+        [ testCase "fresh install applies once and is idempotent" (testNativeRunner settings conn)+        ],+      testGroup+        "history import"+        [ testCase "direct row imports without executing the target action" (testDirectHistoryImport settings conn),+          testCase "direct import rejects altered bytes, checksum, and duplicate rows" (testDirectHistoryRejections settings conn),+          testCase "two-step history requires explicit equivalent opt-in" (testEquivalentHistoryImport settings conn),+          testCase "two-step history rejects an incomplete PGMQ contract" (testEquivalentContractRejections settings conn)         ]     ] +testNativePayload :: IO ()+testNativePayload = do+  nativePath <- findFile ["pgmq-migration/migrations/0001-install-v1.11.0.sql", "migrations/0001-install-v1.11.0.sql"]+  vendorPath <- findFile ["vendor/pgmq/pgmq-extension/sql/pgmq.sql", "../vendor/pgmq/pgmq-extension/sql/pgmq.sql"]+  native <- ByteString.readFile nativePath+  vendored <- ByteString.readFile vendorPath+  native @?= vendored++testNativeComponent :: IO ()+testNativeComponent = do+  component <- either (assertFailure . show) pure Migration.pgmqMigrations+  plan <- either (assertFailure . show) pure (migrationPlan (component :| []))+  let PlanDescription components = planDescription plan+  case toList components of+    [ComponentDescription {name, dependencies, migrations}] -> do+      componentNameText name @?= "pgmq"+      dependencies @?= mempty+      length migrations @?= 2+    actual -> assertFailure ("unexpected native PGMQ plan: " <> show actual)++findFile :: [FilePath] -> IO FilePath+findFile candidates = do+  existing <- filterM doesFileExist candidates+  case existing of+    path : _ -> pure path+    [] -> assertFailure ("could not find any of: " <> show candidates) >> pure "."+ -- | Reset the database to a clean state by dropping the pgmq schema -- and migration tracking table resetDb :: Connection.Connection -> IO ()@@ -61,6 +143,7 @@     resetSession = do       Session.statement () dropPgmqSchema       Session.statement () dropMigrationTable+      Session.statement () dropNativeMigrationSchema      dropPgmqSchema :: Statement.Statement () ()     dropPgmqSchema =@@ -76,93 +159,268 @@         Encoders.noParams         Decoders.noResult --- | Run a test with a clean database+    dropNativeMigrationSchema :: Statement.Statement () ()+    dropNativeMigrationSchema =+      preparable+        "DROP SCHEMA IF EXISTS pgmigrate CASCADE"+        Encoders.noParams+        Decoders.noResult+ withCleanDb :: Connection.Connection -> (Connection.Connection -> IO ()) -> IO () withCleanDb conn action = do   resetDb conn   action conn -testMigrateFresh :: Connection.Connection -> IO ()-testMigrateFresh conn = withCleanDb conn $ \c -> do-  migResult <- Connection.use c Migration.migrate-  case migResult of-    Left sessionErr -> assertFailure $ "Session error: " <> show sessionErr-    Right (Left migrationErr) -> assertFailure $ "Migration error: " <> show migrationErr-    Right (Right ()) -> pure ()+testNativeRunner :: Settings.Settings -> Connection.Connection -> IO ()+testNativeRunner settings conn = withCleanDb conn $ \c -> do+  plan <- nativePlan+  first <- runMigrationPlan defaultRunOptions settings plan+  case first of+    Left err -> assertFailure ("fresh native migration failed: " <> show err)+    Right report -> (outcome <$> toList (results report)) @?= [AppliedNow, AppliedNow]+  second <- runMigrationPlan defaultRunOptions settings plan+  case second of+    Left err -> assertFailure ("repeated native migration failed: " <> show err)+    Right report -> (outcome <$> toList (results report)) @?= [AlreadyApplied, AlreadyApplied]+  functionExists c "pgmq.metrics_all()" >>= (@?= True)+  hasCanaryComment c >>= (@?= True) -testMigrateIdempotent :: Connection.Connection -> IO ()-testMigrateIdempotent conn = withCleanDb conn $ \c -> do-  -- Run migration first time-  result1 <- Connection.use c Migration.migrate-  case result1 of-    Left sessionErr -> assertFailure $ "First migration session error: " <> show sessionErr-    Right (Left migrationErr) -> assertFailure $ "First migration error: " <> show migrationErr-    Right (Right ()) -> pure ()+testDirectHistoryImport :: Settings.Settings -> Connection.Connection -> IO ()+testDirectHistoryImport settings conn = withCleanDb conn $ \c -> do+  prepareDirectHistory c+  runSql c "DROP FUNCTION pgmq.metrics_all()" -  -- Run migration second time - should succeed without error-  result2 <- Connection.use c Migration.migrate-  case result2 of-    Left sessionErr -> assertFailure $ "Second migration session error: " <> show sessionErr-    Right (Left migrationErr) -> assertFailure $ "Second migration error: " <> show migrationErr-    Right (Right ()) -> pure ()+  first <- runPolicyImport settings defaultImportOptions DirectFullInstallHistory+  historyOutcomes first @?= [Imported]+  functionExists c "pgmq.metrics_all()" >>= (@?= False) -testGetMigrations :: Connection.Connection -> IO ()-testGetMigrations conn = withCleanDb conn $ \c -> do-  -- Run migrations first-  _ <- Connection.use c Migration.migrate+  second <- runPolicyImport settings defaultImportOptions DirectFullInstallHistory+  historyOutcomes second @?= [AlreadyImported] -  -- Get applied migrations-  migrationsResult <- Connection.use c Migration.getMigrations-  case migrationsResult of-    Left sessionErr -> assertFailure $ "Session error: " <> show sessionErr-    Right appliedMigrations -> do-      -- Should have applied at least the pgmq_v1.11.0 SQL migration-      assertBool "Should have applied migrations" (length appliedMigrations >= 1)+  plan <- nativePlan+  canaryId <- either (assertFailure . show) pure (migrationId "pgmq" "0002-schema-management-comment")+  beforeCanary <- verifyMigrationPlan defaultRunOptions settings plan+  case beforeCanary of+    Left err -> assertFailure ("native verify failed after direct import: " <> show err)+    Right (VerificationReport issues _ _ _) -> issues @?= [PendingMigration canaryId]+  nativeRun <- runMigrationPlan defaultRunOptions settings plan+  case nativeRun of+    Left err -> assertFailure ("native runner failed after direct import: " <> show err)+    Right report -> (outcome <$> toList (results report)) @?= [AlreadyApplied, AppliedNow]+  afterCanary <- verifyMigrationPlan defaultRunOptions settings plan+  case afterCanary of+    Left err -> assertFailure ("native verify failed after direct canary: " <> show err)+    Right (VerificationReport issues _ _ _) -> issues @?= []+  repeated <- runMigrationPlan defaultRunOptions settings plan+  case repeated of+    Left err -> assertFailure ("native rerun failed after direct canary: " <> show err)+    Right report -> (outcome <$> toList (results report)) @?= [AlreadyApplied, AlreadyApplied]+  functionExists c "pgmq.metrics_all()" >>= (@?= False)+  hasCanaryComment c >>= (@?= True) -testVersion :: IO ()-testVersion =-  Migration.version @?= "v1.11.0"+testDirectHistoryRejections :: Settings.Settings -> Connection.Connection -> IO ()+testDirectHistoryRejections settings conn = do+  withCleanDb conn $ \c -> do+    prepareDirectHistory c+    nativePath <- findFile ["pgmq-migration/migrations/0001-install-v1.11.0.sql", "migrations/0001-install-v1.11.0.sql"]+    payload <- (<> "\n-- altered") <$> ByteString.readFile nativePath+    let provider = connectionProviderFromSettings settings+    config <-+      either (assertFailure . show) pure $+        hasqlMigrationSourceConfig+          provider+          defaultHasqlMigrationTable+          (directLegacyFilename :| [])+          True+          (Map.singleton directLegacyFilename payload)+          []+          "test altered direct PGMQ payload"+    runImportWith settings defaultImportOptions DirectFullInstallHistory config+      >>= assertImportError (\case HasqlMigrationChecksumMismatch name _ _ -> name == directLegacyFilename; _ -> False) --- | Run upgrade twice after a fresh install - second run should be a no-op.--- Upgrade migrations assume the pgmq schema already exists, so we install--- the base schema first.-testUpgradeIdempotent :: Connection.Connection -> IO ()-testUpgradeIdempotent conn = withCleanDb conn $ \c -> do-  -- First install the base schema-  migResult <- Connection.use c Migration.migrate-  case migResult of-    Left sessionErr -> assertFailure $ "Migrate session error: " <> show sessionErr-    Right (Left migrationErr) -> assertFailure $ "Migrate error: " <> show migrationErr-    Right (Right ()) -> pure ()+  withCleanDb conn $ \c -> do+    prepareDirectHistory c+    runSql c "UPDATE public.schema_migrations SET checksum = 'altered' WHERE filename = 'pgmq_v1.11.0'"+    runPolicyImportEither settings defaultImportOptions DirectFullInstallHistory+      >>= assertImportError (\case HasqlMigrationChecksumMismatch name "altered" _ -> name == directLegacyFilename; _ -> False) -  -- Run upgrade first time-  upgrade1Result <- Connection.use c Migration.upgrade-  case upgrade1Result of-    Left sessionErr -> assertFailure $ "First upgrade session error: " <> show sessionErr-    Right (Left migrationErr) -> assertFailure $ "First upgrade migration error: " <> show migrationErr-    Right (Right ()) -> pure ()+  withCleanDb conn $ \c -> do+    prepareDirectHistory c+    runSql c "INSERT INTO public.schema_migrations SELECT * FROM public.schema_migrations WHERE filename = 'pgmq_v1.11.0'"+    runPolicyImportEither settings defaultImportOptions DirectFullInstallHistory+      >>= assertImportError (\case HasqlMigrationDuplicateLedgerFilename name -> name == directLegacyFilename; _ -> False) -  -- Run upgrade second time - should succeed-  upgrade2Result <- Connection.use c Migration.upgrade-  case upgrade2Result of-    Left sessionErr -> assertFailure $ "Second upgrade session error: " <> show sessionErr-    Right (Left migrationErr) -> assertFailure $ "Second upgrade migration error: " <> show migrationErr-    Right (Right ()) -> pure ()+testEquivalentHistoryImport :: Settings.Settings -> Connection.Connection -> IO ()+testEquivalentHistoryImport settings conn = withCleanDb conn $ \c -> do+  prepareTwoStepHistory c+  runPolicyImportEither settings defaultImportOptions EquivalentTwoStepUpgradeHistory+    >>= assertImportError+      ( \case+          HasqlMigrationTargetImportFailed (HistoryImportValidationFailed (HistoryEquivalentStateDisallowed _)) -> True+          _ -> False+      ) --- | Run migrate then upgrade - tests that CREATE OR REPLACE FUNCTION--- makes re-applying safe-testUpgradeAfterMigrate :: Connection.Connection -> IO ()-testUpgradeAfterMigrate conn = withCleanDb conn $ \c -> do-  -- First do a fresh install-  migResult <- Connection.use c Migration.migrate-  case migResult of-    Left sessionErr -> assertFailure $ "Migrate session error: " <> show sessionErr-    Right (Left migrationErr) -> assertFailure $ "Migrate error: " <> show migrationErr-    Right (Right ()) -> pure ()+  first <- runPolicyImport settings equivalentImportOptions EquivalentTwoStepUpgradeHistory+  historyOutcomes first @?= [Imported]+  second <- runPolicyImport settings equivalentImportOptions EquivalentTwoStepUpgradeHistory+  historyOutcomes second @?= [AlreadyImported] -  -- Now run upgrade - should succeed (migrations are safe to re-apply)-  upgradeResult <- Connection.use c Migration.upgrade-  case upgradeResult of-    Left sessionErr -> assertFailure $ "Upgrade session error: " <> show sessionErr-    Right (Left migrationErr) -> assertFailure $ "Upgrade migration error: " <> show migrationErr-    Right (Right ()) -> pure ()+  plan <- nativePlan+  canaryId <- either (assertFailure . show) pure (migrationId "pgmq" "0002-schema-management-comment")+  beforeCanary <- verifyMigrationPlan defaultRunOptions settings plan+  case beforeCanary of+    Left err -> assertFailure ("native verify failed after equivalent import: " <> show err)+    Right (VerificationReport issues _ _ _) -> issues @?= [PendingMigration canaryId]+  nativeRun <- runMigrationPlan defaultRunOptions settings plan+  case nativeRun of+    Left err -> assertFailure ("native runner failed after equivalent import: " <> show err)+    Right report -> (outcome <$> toList (results report)) @?= [AlreadyApplied, AppliedNow]+  afterCanary <- verifyMigrationPlan defaultRunOptions settings plan+  case afterCanary of+    Left err -> assertFailure ("native verify failed after equivalent canary: " <> show err)+    Right (VerificationReport issues _ _ _) -> issues @?= []+  repeated <- runMigrationPlan defaultRunOptions settings plan+  case repeated of+    Left err -> assertFailure ("native rerun failed after equivalent canary: " <> show err)+    Right report -> (outcome <$> toList (results report)) @?= [AlreadyApplied, AlreadyApplied]+  hasCanaryComment c >>= (@?= True)++testEquivalentContractRejections :: Settings.Settings -> Connection.Connection -> IO ()+testEquivalentContractRejections settings conn =+  forM_ destructiveChanges $ \sql ->+    withCleanDb conn $ \c -> do+      prepareTwoStepHistory c+      runSql c sql+      runPolicyImportEither settings equivalentImportOptions EquivalentTwoStepUpgradeHistory+        >>= assertImportError+          ( \case+              HasqlMigrationTargetImportFailed (HistoryStateValidationFailed _ _) -> True+              _ -> False+          )+  where+    destructiveChanges =+      [ "DROP FUNCTION pgmq.send_topic(text,jsonb)",+        "DROP TYPE pgmq.metrics_result CASCADE",+        "DROP TABLE pgmq.topic_bindings CASCADE"+      ]++equivalentImportOptions :: ImportOptions+equivalentImportOptions =+  withEquivalentHistory AllowEquivalentHistory defaultImportOptions++directLegacyFilename :: FilePath+directLegacyFilename = "pgmq_v1.11.0"++nativePlan :: IO MigrationPlan+nativePlan = do+  component <- either (assertFailure . show) pure Migration.pgmqMigrations+  either (assertFailure . show) pure (migrationPlan (component :| []))++runPolicyImport ::+  Settings.Settings ->+  ImportOptions ->+  AlternativeHistoryPolicy ->+  IO HistoryImportReport+runPolicyImport settings options policy =+  runPolicyImportEither settings options policy >>= either (assertFailure . show) pure++runPolicyImportEither ::+  Settings.Settings ->+  ImportOptions ->+  AlternativeHistoryPolicy ->+  IO (Either HasqlMigrationImportError HistoryImportReport)+runPolicyImportEither settings options policy = do+  let provider = connectionProviderFromSettings settings+  config <- either (assertFailure . show) pure (pgmqHasqlMigrationSourceConfig provider policy)+  runImportWith settings options policy config++runImportWith ::+  Settings.Settings ->+  ImportOptions ->+  AlternativeHistoryPolicy ->+  HasqlMigrationSourceConfig ->+  IO (Either HasqlMigrationImportError HistoryImportReport)+runImportWith settings options policy config = do+  mappings <- either (assertFailure . show) pure (pgmqHasqlMigrationMappings policy)+  plan <- nativePlan+  let provider = connectionProviderFromSettings settings+  importHasqlMigrationHistory options config provider plan mappings++historyOutcomes :: HistoryImportReport -> [HistoryImportOutcome]+historyOutcomes HistoryImportReport {importResults} = importOutcome <$> toList importResults++assertImportError ::+  (HasqlMigrationImportError -> Bool) ->+  Either HasqlMigrationImportError HistoryImportReport ->+  IO ()+assertImportError predicate actual =+  case actual of+    Left err | predicate err -> pure ()+    Left err -> assertFailure ("unexpected history import error: " <> show err)+    Right report -> assertFailure ("expected history import failure, received: " <> show report)++prepareDirectHistory :: Connection.Connection -> IO ()+prepareDirectHistory connection = do+  installHistoricalSchema connection+  runSql connection legacyLedgerDefinition+  runSql+    connection+    "INSERT INTO public.schema_migrations (filename, checksum) VALUES ('pgmq_v1.11.0', '+qm4gAAF+A+99qM9BxGD0g==')"++prepareTwoStepHistory :: Connection.Connection -> IO ()+prepareTwoStepHistory connection = do+  installHistoricalSchema connection+  runSql connection legacyLedgerDefinition+  runSql+    connection+    ( "INSERT INTO public.schema_migrations (filename, checksum) VALUES "+        <> "('pgmq_v1.10.0_to_v1.10.1', 'C56QJtvtxB2pGcEHR82LFA=='), "+        <> "('pgmq_v1.10.1_to_v1.11.0', 'KMM7gGjkepkD1YA1hUCpEQ==')"+    )++installHistoricalSchema :: Connection.Connection -> IO ()+installHistoricalSchema connection = do+  path <- findFile ["pgmq-migration/migrations/0001-install-v1.11.0.sql", "migrations/0001-install-v1.11.0.sql"]+  payload <- ByteString.readFile path+  runSql connection (Text.decodeUtf8 payload)++legacyLedgerDefinition :: Text+legacyLedgerDefinition =+  "CREATE TABLE public.schema_migrations "+    <> "(filename text NOT NULL, checksum text NOT NULL, "+    <> "executed_at timestamp without time zone NOT NULL DEFAULT now())"++runSql :: Connection.Connection -> Text -> IO ()+runSql connection sql = do+  result <- Connection.use connection (Session.script sql)+  case result of+    Left err -> assertFailure ("SQL fixture failed: " <> show err)+    Right () -> pure ()++functionExists :: Connection.Connection -> Text -> IO Bool+functionExists connection identity = do+  result <- Connection.use connection (Session.statement identity functionExistsStatement)+  case result of+    Left err -> assertFailure ("function inspection failed: " <> show err) >> pure False+    Right exists -> pure exists++functionExistsStatement :: Statement.Statement Text Bool+functionExistsStatement =+  preparable+    "SELECT pg_catalog.to_regprocedure($1) IS NOT NULL"+    (Encoders.param (Encoders.nonNullable Encoders.text))+    (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.bool)))++hasCanaryComment :: Connection.Connection -> IO Bool+hasCanaryComment connection = do+  result <- Connection.use connection (Session.statement () canaryCommentStatement)+  case result of+    Left err -> assertFailure ("schema comment inspection failed: " <> show err) >> pure False+    Right matches -> pure matches++canaryCommentStatement :: Statement.Statement () Bool+canaryCommentStatement =+  preparable+    "SELECT obj_description(to_regnamespace('pgmq'), 'pg_namespace') = 'Managed by pg-migrate component pgmq through 0002-schema-management-comment'"+    Encoders.noParams+    (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.bool)))