diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -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.
diff --git a/migrations/0001-install-v1.11.0.sql b/migrations/0001-install-v1.11.0.sql
new file mode 100644
--- /dev/null
+++ b/migrations/0001-install-v1.11.0.sql
@@ -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));
+$$;
diff --git a/migrations/0002-schema-management-comment.sql b/migrations/0002-schema-management-comment.sql
new file mode 100644
--- /dev/null
+++ b/migrations/0002-schema-management-comment.sql
@@ -0,0 +1,2 @@
+COMMENT ON SCHEMA pgmq IS
+  'Managed by pg-migrate component pgmq through 0002-schema-management-comment';
diff --git a/migrations/manifest b/migrations/manifest
new file mode 100644
--- /dev/null
+++ b/migrations/manifest
@@ -0,0 +1,2 @@
+0001-install-v1.11.0.sql
+0002-schema-management-comment.sql
diff --git a/pgmq-migration.cabal b/pgmq-migration.cabal
--- a/pgmq-migration.cabal
+++ b/pgmq-migration.cabal
@@ -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
diff --git a/src/Pgmq/Migration.hs b/src/Pgmq/Migration.hs
--- a/src/Pgmq/Migration.hs
+++ b/src/Pgmq/Migration.hs
@@ -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)
diff --git a/src/Pgmq/Migration/History/HasqlMigration.hs b/src/Pgmq/Migration/History/HasqlMigration.hs
new file mode 100644
--- /dev/null
+++ b/src/Pgmq/Migration/History/HasqlMigration.hs
@@ -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
diff --git a/src/Pgmq/Migration/Internal/Definition.hs b/src/Pgmq/Migration/Internal/Definition.hs
new file mode 100644
--- /dev/null
+++ b/src/Pgmq/Migration/Internal/Definition.hs
@@ -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
diff --git a/src/Pgmq/Migration/Migrations.hs b/src/Pgmq/Migration/Migrations.hs
deleted file mode 100644
--- a/src/Pgmq/Migration/Migrations.hs
+++ /dev/null
@@ -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
diff --git a/src/Pgmq/Migration/Migrations/V1_10_0_to_V1_10_1.hs b/src/Pgmq/Migration/Migrations/V1_10_0_to_V1_10_1.hs
deleted file mode 100644
--- a/src/Pgmq/Migration/Migrations/V1_10_0_to_V1_10_1.hs
+++ /dev/null
@@ -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")
-  ]
diff --git a/src/Pgmq/Migration/Migrations/V1_10_1_to_V1_11_0.hs b/src/Pgmq/Migration/Migrations/V1_10_1_to_V1_11_0.hs
deleted file mode 100644
--- a/src/Pgmq/Migration/Migrations/V1_10_1_to_V1_11_0.hs
+++ /dev/null
@@ -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")
-  ]
diff --git a/src/Pgmq/Migration/Migrations/V1_11_0.hs b/src/Pgmq/Migration/Migrations/V1_11_0.hs
deleted file mode 100644
--- a/src/Pgmq/Migration/Migrations/V1_11_0.hs
+++ /dev/null
@@ -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")
-  ]
diff --git a/src/Pgmq/Migration/SchemaContract.hs b/src/Pgmq/Migration/SchemaContract.hs
new file mode 100644
--- /dev/null
+++ b/src/Pgmq/Migration/SchemaContract.hs
@@ -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
diff --git a/src/Pgmq/Migration/Sessions.hs b/src/Pgmq/Migration/Sessions.hs
deleted file mode 100644
--- a/src/Pgmq/Migration/Sessions.hs
+++ /dev/null
@@ -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
diff --git a/src/Pgmq/Migration/Statements.hs b/src/Pgmq/Migration/Statements.hs
deleted file mode 100644
--- a/src/Pgmq/Migration/Statements.hs
+++ /dev/null
@@ -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
diff --git a/src/Pgmq/Migration/Transactions.hs b/src/Pgmq/Migration/Transactions.hs
deleted file mode 100644
--- a/src/Pgmq/Migration/Transactions.hs
+++ /dev/null
@@ -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
diff --git a/test/Main.hs b/test/Main.hs
--- a/test/Main.hs
+++ b/test/Main.hs
@@ -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)))