From c77fb3d61fa29c8c9862bc4ea4754541de93ce44 Mon Sep 17 00:00:00 2001 From: psteinroe Date: Tue, 11 Aug 2026 08:14:27 +0000 Subject: [PATCH 1/3] refactor(subscriptions): extract user-managed SQL package --- README.md | 2 +- docs/concepts/event-structure.md | 8 +- docs/concepts/extensions.md | 8 +- docs/concepts/how-it-works.md | 4 +- docs/concepts/subscriptions.md | 105 ++--- docs/getting-started.md | 22 +- docs/guides/subscription-setup.md | 62 +++ docs/guides/upgrade-subscriptions.md | 54 +++ docs/index.md | 2 +- docs/reference/event-format.md | 4 +- docs/reference/subscriptions-table.md | 42 +- docs/sinks/index.md | 2 +- extensions/README.md | 10 + extensions/subscriptions/README.md | 73 ++++ .../examples/set_subscriptions.sql | 20 + .../0001_create_pgstream_subscriptions.sql | 328 ++++++++++++++ extensions/subscriptions/uninstall.sql | 11 + .../upgrades/from-pgstream-0.1.sql | 276 ++++++++++++ implementation-notes.md | 23 + .../1786434463000_extract_subscriptions.sql | 33 ++ src/test_utils/database.rs | 17 + tests/subscription_extension_tests.rs | 400 ++++++++++++++++++ tests/subscriptions_tests.rs | 43 +- tests/transaction_lsn_tests.rs | 6 +- zensical.toml | 4 + 25 files changed, 1432 insertions(+), 127 deletions(-) create mode 100644 docs/guides/subscription-setup.md create mode 100644 docs/guides/upgrade-subscriptions.md create mode 100644 extensions/README.md create mode 100644 extensions/subscriptions/README.md create mode 100644 extensions/subscriptions/examples/set_subscriptions.sql create mode 100644 extensions/subscriptions/migrations/0001_create_pgstream_subscriptions.sql create mode 100644 extensions/subscriptions/uninstall.sql create mode 100644 extensions/subscriptions/upgrades/from-pgstream-0.1.sql create mode 100644 implementation-notes.md create mode 100644 migrations/1786434463000_extract_subscriptions.sql create mode 100644 tests/subscription_extension_tests.rs diff --git a/README.md b/README.md index e2b39d1..0439b95 100644 --- a/README.md +++ b/README.md @@ -28,7 +28,7 @@ Events are inserted into the `pgstream.events` table and streamed via logical re **Two ways to create events:** -1. **Subscriptions** (optional) - Define triggers that automatically capture table changes +1. **Subscriptions** (optional) - Install the user-managed SQL package in `extensions/subscriptions` to capture table changes with triggers 2. **Manual inserts** - Insert directly into `pgstream.events` from your application or database functions ## Trade-offs diff --git a/docs/concepts/event-structure.md b/docs/concepts/event-structure.md index fd9afda..52cf2ad 100644 --- a/docs/concepts/event-structure.md +++ b/docs/concepts/event-structure.md @@ -32,10 +32,10 @@ The event body sent to your sink. Contains the row data and trigger context. ### Selecting Columns -By default, all columns are included. Use `column_names` to select specific columns: +Use `column_names` to select the columns included in each event: ```sql -INSERT INTO pgstream.subscriptions (key, stream_id, operation, schema_name, table_name, column_names) +INSERT INTO pgstream_subscriptions.subscriptions (key, stream_id, operation, schema_name, table_name, column_names) VALUES ('user-created', 1, 'INSERT', 'public', 'users', ARRAY['id', 'email']); ``` @@ -44,8 +44,8 @@ VALUES ('user-created', 1, 'INSERT', 'public', 'users', ARRAY['id', 'email']); Routing configuration read by sinks. Controls where and how events are delivered. ```sql -INSERT INTO pgstream.subscriptions (key, stream_id, operation, schema_name, table_name, metadata) -VALUES ('user-created', 1, 'INSERT', 'public', 'users', '{"topic": "users", "priority": "high"}'); +INSERT INTO pgstream_subscriptions.subscriptions (key, stream_id, operation, schema_name, table_name, column_names, metadata) +VALUES ('user-created', 1, 'INSERT', 'public', 'users', ARRAY['id', 'email'], '{"topic": "users", "priority": "high"}'); ``` Each sink reads specific metadata fields: diff --git a/docs/concepts/extensions.md b/docs/concepts/extensions.md index 20ba646..470d52c 100644 --- a/docs/concepts/extensions.md +++ b/docs/concepts/extensions.md @@ -7,9 +7,9 @@ Add computed values to payload or metadata using SQL expressions. Add fields to the event body: ```sql -INSERT INTO pgstream.subscriptions (key, stream_id, operation, schema_name, table_name, payload_extensions) +INSERT INTO pgstream_subscriptions.subscriptions (key, stream_id, operation, schema_name, table_name, column_names, payload_extensions) VALUES ( - 'order-created', 1, 'INSERT', 'public', 'orders', + 'order-created', 1, 'INSERT', 'public', 'orders', array['id', 'total', 'created_at'], '[ {"json_path": "total_formatted", "expression": "''$'' || new.total::text"}, {"json_path": "order_date", "expression": "new.created_at::date::text"} @@ -32,9 +32,9 @@ Result: Compute routing values from row data: ```sql -INSERT INTO pgstream.subscriptions (key, stream_id, operation, schema_name, table_name, metadata_extensions) +INSERT INTO pgstream_subscriptions.subscriptions (key, stream_id, operation, schema_name, table_name, column_names, metadata_extensions) VALUES ( - 'order-created', 1, 'INSERT', 'public', 'orders', + 'order-created', 1, 'INSERT', 'public', 'orders', array['id', 'region'], '[ {"json_path": "topic", "expression": "''orders-'' || new.region"} ]' diff --git a/docs/concepts/how-it-works.md b/docs/concepts/how-it-works.md index f34bdba..929e43f 100644 --- a/docs/concepts/how-it-works.md +++ b/docs/concepts/how-it-works.md @@ -51,11 +51,11 @@ But we only care about the events we subscribed to, and do not want to replicate ## Subscriptions (Optional) -When you insert a subscription, Postgres Stream creates a trigger on the target table: +Subscriptions come from a [user-installed SQL package](../guides/subscription-setup.md). When you insert a subscription, that package creates a trigger on the target table; the pgstream daemon does not manage application-table triggers: ```sql -- Auto-generated when you insert into subscriptions table -create or replace function pgstream._publish_after_insert_on_users() +create or replace function pgstream_subscriptions._publish_after_insert_on_users() returns trigger as $$ declare v_jsonb_output jsonb := '[]'::jsonb; diff --git a/docs/concepts/subscriptions.md b/docs/concepts/subscriptions.md index 4649572..92de308 100644 --- a/docs/concepts/subscriptions.md +++ b/docs/concepts/subscriptions.md @@ -4,12 +4,12 @@ Define which table changes to capture and how they should be formatted. ## Overview -Subscriptions are stored in the `pgstream.subscriptions` table. When you insert, update, or delete a subscription, Postgres Stream automatically creates or updates the corresponding database triggers. +Subscriptions are provided by an [optional, user-installed SQL package](../guides/subscription-setup.md) and stored in `pgstream_subscriptions.subscriptions`. When you insert, update, or delete a subscription, the package drops and recreates the corresponding database triggers. The pgstream daemon does not manage these triggers. ## Creating a Subscription ```sql -insert into pgstream.subscriptions ( +insert into pgstream_subscriptions.subscriptions ( key, stream_id, operation, @@ -40,7 +40,7 @@ insert into pgstream.subscriptions ( | `schema_name` | text | Yes | Database schema (usually `public`) | | `table_name` | text | Yes | Target table name | | `when_clause` | text | No | SQL expression to filter events | -| `column_names` | text[] | No | Columns to include in payload (null = all) | +| `column_names` | text[] | Yes | Columns to include in the payload | | `payload_extensions` | jsonb | No | Computed fields to add to payload | | `metadata` | jsonb | No | Static routing metadata | | `metadata_extensions` | jsonb | No | Dynamic routing metadata | @@ -51,23 +51,23 @@ Use `when_clause` to capture only specific events: ```sql -- Only capture high-value orders -insert into pgstream.subscriptions (key, stream_id, operation, schema_name, table_name, when_clause) -values ('high-value-orders', 1, 'INSERT', 'public', 'orders', 'new.total > 1000'); +insert into pgstream_subscriptions.subscriptions (key, stream_id, operation, schema_name, table_name, when_clause, column_names) +values ('high-value-orders', 1, 'INSERT', 'public', 'orders', 'new.total > 1000', array['id', 'total']); -- Only capture status changes -insert into pgstream.subscriptions (key, stream_id, operation, schema_name, table_name, when_clause) -values ('status-changed', 1, 'UPDATE', 'public', 'orders', 'old.status IS DISTINCT FROM new.status'); +insert into pgstream_subscriptions.subscriptions (key, stream_id, operation, schema_name, table_name, when_clause, column_names) +values ('status-changed', 1, 'UPDATE', 'public', 'orders', 'old.status IS DISTINCT FROM new.status', array['id', 'status']); ``` The `when_clause` is a SQL expression. Use `new` to reference the new row (INSERT/UPDATE) and `old` for the previous row (UPDATE/DELETE). ## Selecting Columns -By default, all columns are included. Use `column_names` to select specific columns: +Use `column_names` to select the columns included in the payload: ```sql -- Only include id, email, and created_at -insert into pgstream.subscriptions (key, stream_id, operation, schema_name, table_name, column_names) +insert into pgstream_subscriptions.subscriptions (key, stream_id, operation, schema_name, table_name, column_names) values ('user-created', 1, 'INSERT', 'public', 'users', array['id', 'email', 'created_at']); ``` @@ -99,78 +99,43 @@ You can have multiple subscriptions on the same table: ```sql -- Capture all user inserts -insert into pgstream.subscriptions (key, stream_id, operation, schema_name, table_name) -values ('all-users', 1, 'INSERT', 'public', 'users'); +insert into pgstream_subscriptions.subscriptions (key, stream_id, operation, schema_name, table_name, column_names) +values ('all-users', 1, 'INSERT', 'public', 'users', array['id', 'email', 'email_verified']); -- Also capture verified users separately -insert into pgstream.subscriptions (key, stream_id, operation, schema_name, table_name, when_clause) -values ('verified-users', 1, 'INSERT', 'public', 'users', 'new.email_verified = true'); +insert into pgstream_subscriptions.subscriptions (key, stream_id, operation, schema_name, table_name, when_clause, column_names) +values ('verified-users', 1, 'INSERT', 'public', 'users', 'new.email_verified = true', array['id', 'email']); ``` Both subscriptions will fire for a verified user, creating two events with different `tg_name` values. -## Avoiding Unnecessary Trigger Recreation +## Reconciling a Stream -Each subscription change recreates the trigger, which can be expensive. Use MERGE to only update when values actually change: +Each changed subscription recreates its target trigger. The package provides `set_subscriptions()` so deployments can supply the complete desired set for one stream without rewriting unchanged rows: ```sql -create or replace function set_subscriptions( - p_stream_id bigint, - p_subscriptions pgstream.subscriptions[] -) -returns void -language plpgsql -security definer -set search_path to '' -as $$ -begin - create temporary table temp_subscriptions as - select * from unnest(p_subscriptions); - - -- Only update if values actually changed (avoids trigger recreation) - merge into pgstream.subscriptions as target - using temp_subscriptions as source - on (target.key = source.key and target.stream_id = p_stream_id) - when matched and ( - target.operation is distinct from source.operation or - target.schema_name is distinct from source.schema_name or - target.table_name is distinct from source.table_name or - target.when_clause is distinct from source.when_clause or - target.column_names is distinct from source.column_names or - target.metadata is distinct from source.metadata or - target.payload_extensions is distinct from source.payload_extensions or - target.metadata_extensions is distinct from source.metadata_extensions - ) then update set - operation = source.operation, - schema_name = source.schema_name, - table_name = source.table_name, - when_clause = source.when_clause, - column_names = source.column_names, - metadata = source.metadata, - payload_extensions = source.payload_extensions, - metadata_extensions = source.metadata_extensions - when not matched then insert ( - key, stream_id, operation, schema_name, table_name, - when_clause, column_names, metadata, payload_extensions, metadata_extensions - ) values ( - source.key, p_stream_id, source.operation, source.schema_name, - source.table_name, source.when_clause, source.column_names, - source.metadata, source.payload_extensions, source.metadata_extensions - ); - - -- Remove subscriptions not in input - delete from pgstream.subscriptions - where stream_id = p_stream_id - and not exists ( - select 1 from temp_subscriptions - where pgstream.subscriptions.key = temp_subscriptions.key - ); - - drop table temp_subscriptions; -end; -$$; +select pgstream_subscriptions.set_subscriptions( + 1, + array[ + row( + null::uuid, + 'user-created', + 1::bigint, + 'INSERT'::pgstream_subscriptions.operation_type, + 'public', + 'users', + null::text, + array['id', 'email']::text[], + null::jsonb, + '[]'::jsonb, + '[]'::jsonb + )::pgstream_subscriptions.subscriptions + ] +); ``` +The array is the complete desired state for stream `1`: missing rows are inserted, changed rows are updated, and omitted rows are deleted. See the [package example](https://github.com/psteinroe/postgres-stream/blob/main/extensions/subscriptions/examples/set_subscriptions.sql). + ## Next Steps - [Event Structure](event-structure.md) - Payload and metadata format diff --git a/docs/getting-started.md b/docs/getting-started.md index 57fb7df..3bdf32a 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -48,17 +48,35 @@ docker run -v $(pwd)/config.yaml:/config.yaml \ Each sink has its own image: `kafka-latest`, `nats-latest`, `sqs-latest`, etc. +## Install the Optional Subscription Package + +Subscriptions are user-managed SQL and are not installed by the pgstream daemon. Copy the provided migration into your application's migration system and run it as the role that owns your subscribed tables: + +```bash +psql "$DATABASE_URL" \ + --set ON_ERROR_STOP=1 \ + --single-transaction \ + --file extensions/subscriptions/migrations/0001_create_pgstream_subscriptions.sql +``` + +See [Subscription Setup](guides/subscription-setup.md) for ownership, grants, and customization. + ## Create a Subscription ```sql -INSERT INTO pgstream.subscriptions (key, stream_id, operation, schema_name, table_name) -VALUES ('user-created', 1, 'INSERT', 'public', 'users'); +INSERT INTO pgstream_subscriptions.subscriptions ( + key, stream_id, operation, schema_name, table_name, column_names +) +VALUES ( + 'user-created', 1, 'INSERT', 'public', 'users', array['id', 'email'] +); ``` Now inserts into `users` are streamed to your webhook. ## Next Steps +- [Subscription Setup](guides/subscription-setup.md) - Install the optional SQL package - [Subscriptions](concepts/subscriptions.md) - Filter events, select columns - [Sinks](sinks/index.md) - Configure your destination - [Configuration Reference](reference/configuration.md) - All options diff --git a/docs/guides/subscription-setup.md b/docs/guides/subscription-setup.md new file mode 100644 index 0000000..9eb0de0 --- /dev/null +++ b/docs/guides/subscription-setup.md @@ -0,0 +1,62 @@ +# Subscription Setup + +Subscriptions are an optional SQL package that you install through your own database migration system. The pgstream daemon creates no triggers on application tables and does not migrate the `pgstream_subscriptions` schema. + +## Ownership model + +Subscription changes drop and recreate target-table triggers. PostgreSQL requires the role dropping a trigger to own its table. Run the package migration as the application migration role that owns subscribed tables, or as a role that is a member of all relevant owner roles. + +This privilege belongs to the database deployment path, not the long-lived pgstream runtime role. + +The package accepts trusted SQL through `when_clause`, `payload_extensions`, and `metadata_extensions`. Only trusted database deployment roles should be able to change subscriptions or execute `set_subscriptions()`. + +## Install + +Run pgstream's core migrations first, then copy [`extensions/subscriptions/migrations/0001_create_pgstream_subscriptions.sql`](https://github.com/psteinroe/postgres-stream/blob/main/extensions/subscriptions/migrations/0001_create_pgstream_subscriptions.sql) into your migration system. Review and change its schema, ownership, or grants if needed. + +To apply the repository file directly: + +```bash +psql "$DATABASE_URL" \ + --set ON_ERROR_STOP=1 \ + --single-transaction \ + --file extensions/subscriptions/migrations/0001_create_pgstream_subscriptions.sql +``` + +The installer needs: + +- Ownership of subscribed tables. +- `USAGE` on the core `pgstream` schema. +- `INSERT` on `pgstream.events`. +- Permission to create and own `pgstream_subscriptions`. + +## Grant subscription deployment + +The installer revokes public execution of `set_subscriptions()`. Grant it only to your trusted migration role: + +```sql +grant usage on schema pgstream_subscriptions to application_migrator; +grant execute on function pgstream_subscriptions.set_subscriptions( + bigint, + pgstream_subscriptions.subscriptions[] +) to application_migrator; +``` + +The pgstream runtime role does not need access to this schema or ownership of application tables. + +## Reconcile subscriptions + +Use `pgstream_subscriptions.set_subscriptions()` to supply the complete desired set for a stream. It inserts missing definitions, updates changed definitions, and deletes omitted definitions. Unchanged rows are not written, avoiding unnecessary trigger recreation. + +See [Subscriptions](../concepts/subscriptions.md) and the [complete SQL example](https://github.com/psteinroe/postgres-stream/blob/main/extensions/subscriptions/examples/set_subscriptions.sql). + +## Customize + +The package is intentionally plain SQL rather than a PostgreSQL `CREATE EXTENSION` package. You may copy and change: + +- The `pgstream_subscriptions` schema name. +- Ownership and grants. +- The `set_subscriptions()` interface. +- Trigger names or generated payload logic. + +Once copied, those migrations are owned and versioned by your application. diff --git a/docs/guides/upgrade-subscriptions.md b/docs/guides/upgrade-subscriptions.md new file mode 100644 index 0000000..ecff150 --- /dev/null +++ b/docs/guides/upgrade-subscriptions.md @@ -0,0 +1,54 @@ +# Upgrade Subscriptions from pgstream 0.1 + +The breaking-change release stops installing subscription SQL through the pgstream daemon. Existing subscription objects must be moved from `pgstream` to the user-managed `pgstream_subscriptions` schema before starting the new version. + +## Before upgrading + +1. Back up the database. +2. Test the migration against a production-like copy. +3. Identify the owner of every subscribed table. +4. Connect as a role that can administer the legacy pgstream objects and is the owner, or a member of the owner role, for every subscribed table. + +The migration preserves subscription rows and target triggers, but future calls to `set_subscriptions()` need the package owner to drop and recreate those triggers. + +## Run the upgrade + +Copy and review [`extensions/subscriptions/upgrades/from-pgstream-0.1.sql`](https://github.com/psteinroe/postgres-stream/blob/main/extensions/subscriptions/upgrades/from-pgstream-0.1.sql), then run it before deploying the new pgstream binary: + +```bash +psql "$DATABASE_URL" \ + --set ON_ERROR_STOP=1 \ + --file extensions/subscriptions/upgrades/from-pgstream-0.1.sql +``` + +The script runs in a transaction and: + +1. Creates `pgstream_subscriptions`. +2. Moves the subscription enum, table, helper functions, and generated functions with `ALTER ... SET SCHEMA`. +3. Transfers ownership to the role running the migration. +4. Replaces the coordinator body so future trigger rebuilds use the new schema. +5. Adds `pgstream_subscriptions.set_subscriptions()`. + +`ALTER ... SET SCHEMA` preserves table rows, object OIDs, and dependencies. Existing subscription UUIDs and target-trigger function references are not recreated. + +## Verify + +```sql +select count(*) from pgstream_subscriptions.subscriptions; + +select to_regclass('pgstream.subscriptions') is null as legacy_removed; + +select namespace.nspname, procedure.proname +from pg_catalog.pg_proc as procedure +join pg_catalog.pg_namespace as namespace + on namespace.oid = procedure.pronamespace +where namespace.nspname = 'pgstream_subscriptions'; +``` + +Insert or update a test row covered by an existing subscription and confirm that an event appears in `pgstream.events`. + +After verification, deploy the new pgstream version. Its core migration accepts the moved installation but rejects non-empty legacy `pgstream.subscriptions` installations with an upgrade hint. + +## Rollback + +The upgrade is atomic before commit. If it fails, PostgreSQL rolls back the schema and ownership moves. After commit, restore the database backup or write a reverse migration appropriate for your customized ownership and schema choices. diff --git a/docs/index.md b/docs/index.md index 9370e63..f5498f1 100644 --- a/docs/index.md +++ b/docs/index.md @@ -25,7 +25,7 @@ Events are inserted into the `pgstream.events` table and streamed via logical re **Two ways to create events:** -1. **Subscriptions** (optional) - Define triggers that automatically capture table changes +1. **Subscriptions** (optional) - Install the user-managed SQL package to capture table changes with triggers 2. **Manual inserts** - Insert directly into `pgstream.events` from your application or database functions ## Trade-offs diff --git a/docs/reference/event-format.md b/docs/reference/event-format.md index f3116a3..468468e 100644 --- a/docs/reference/event-format.md +++ b/docs/reference/event-format.md @@ -73,7 +73,7 @@ Unix timestamp in milliseconds when the event was created. The new row values. Present for `INSERT` and `UPDATE` operations. -Contains only the columns specified in `column_names`, or all columns if not specified. +Contains the columns specified in `column_names`. ### `old` @@ -83,7 +83,7 @@ Contains only the columns specified in `column_names`, or all columns if not spe The previous row values. Present for `UPDATE` and `DELETE` operations. -Contains only the columns specified in `column_names`, or all columns if not specified. +Contains the columns specified in `column_names`. ## Operation-Specific Structure diff --git a/docs/reference/subscriptions-table.md b/docs/reference/subscriptions-table.md index a855178..e7738e9 100644 --- a/docs/reference/subscriptions-table.md +++ b/docs/reference/subscriptions-table.md @@ -1,44 +1,55 @@ # Subscriptions Table Reference -Schema reference for `pgstream.subscriptions`. +Schema reference for `pgstream_subscriptions.subscriptions`. ## Schema ```sql -create table pgstream.subscriptions ( +create table pgstream_subscriptions.subscriptions ( + id uuid primary key default pgstream.portable_uuidv7(), key text not null, - stream_id bigint not null, - operation text not null, + stream_id bigint, + operation pgstream_subscriptions.operation_type not null, schema_name text not null, table_name text not null, when_clause text, - column_names text[], + column_names text[] not null, metadata jsonb, - payload_extensions jsonb, - metadata_extensions jsonb, - primary key (key, stream_id) + payload_extensions jsonb default '[]'::jsonb, + metadata_extensions jsonb default '[]'::jsonb, + unique (stream_id, key, schema_name, table_name, operation) ); ``` ## Columns +### `id` + +| | | +|--|--| +| Type | uuid | +| Required | Generated automatically | +| Primary Key | Yes | + +Stable identifier generated with `pgstream.portable_uuidv7()`. + ### `key` | | | |--|--| | Type | text | | Required | Yes | -| Primary Key | Yes (with stream_id) | +| Unique | With stream, schema, table, and operation | -Unique identifier for the subscription within a stream. Used as `tg_name` in generated events. +Identifier for the subscription. Used as `tg_name` in generated events. ### `stream_id` | | | |--|--| | Type | bigint | -| Required | Yes | -| Primary Key | Yes (with key) | +| Required | Recommended | +| Primary Key | No | Must match the `stream.id` in your config. Links the subscription to a specific Postgres Stream instance. @@ -91,10 +102,9 @@ Examples: | | | |--|--| | Type | text[] | -| Required | No | -| Default | All columns | +| Required | Yes | -Array of column names to include in the event payload. If null, all columns are included. +Array of column names to include in the event payload. Example: ```sql @@ -191,4 +201,4 @@ When you insert, update, or delete a subscription: 2. New database triggers are created/modified on target tables 3. Existing triggers are dropped if no subscriptions remain -This happens automatically via internal triggers on the subscriptions table. +This happens through internal triggers installed by the optional subscription SQL package. The package is owned by your database migration role; the pgstream daemon does not perform this DDL. diff --git a/docs/sinks/index.md b/docs/sinks/index.md index 9743477..77394ec 100644 --- a/docs/sinks/index.md +++ b/docs/sinks/index.md @@ -29,7 +29,7 @@ Most sinks support dynamic routing via event metadata. This lets you route event ```sql -- Route to different topics based on table name -insert into pgstream.subscriptions ( +insert into pgstream_subscriptions.subscriptions ( key, stream_id, operation, schema_name, table_name, column_names, metadata_extensions ) values ( diff --git a/extensions/README.md b/extensions/README.md new file mode 100644 index 0000000..4b85804 --- /dev/null +++ b/extensions/README.md @@ -0,0 +1,10 @@ +# Database extensions + +This directory contains optional, user-installed SQL packages for pgstream. +These are not PostgreSQL `CREATE EXTENSION` packages and are never installed or +migrated by the pgstream daemon. + +Copy the relevant SQL into your application's migration system, review it, and +adapt schema names, ownership, and grants to your environment. + +- [`subscriptions`](subscriptions/README.md) — capture table changes with database triggers diff --git a/extensions/subscriptions/README.md b/extensions/subscriptions/README.md new file mode 100644 index 0000000..5241039 --- /dev/null +++ b/extensions/subscriptions/README.md @@ -0,0 +1,73 @@ +# Subscription SQL package + +This optional SQL package captures changes from application tables and writes +them to the core `pgstream.events` table. It lives in the separate +`pgstream_subscriptions` schema and is managed by your database migrations, not +by the pgstream daemon. + +## Why installation is separate + +Changing subscriptions drops and recreates triggers on their target tables. +PostgreSQL requires the role dropping a trigger to own its table. Install this +package as the application migration role that owns the subscribed tables (or a +role that is a member of their owner roles). The long-lived pgstream runtime +role does not need application-table ownership. + +The package uses `SECURITY DEFINER` functions and accepts trusted SQL expressions +in conditions and extensions. Only grant package administration to trusted +database deployment roles. + +## Install + +The core pgstream migrations must run first because the package writes to +`pgstream.events` and uses `pgstream.portable_uuidv7()`. + +Copy [`migrations/0001_create_pgstream_subscriptions.sql`](migrations/0001_create_pgstream_subscriptions.sql) +into your migration system and adapt it as needed. To apply it directly: + +```shell +psql "$DATABASE_URL" \ + --set ON_ERROR_STOP=1 \ + --single-transaction \ + --file extensions/subscriptions/migrations/0001_create_pgstream_subscriptions.sql +``` + +The installing role needs `USAGE` on `pgstream` and `INSERT` on +`pgstream.events`. It also needs ownership of every target table so future +subscription changes can drop their triggers. + +`set_subscriptions` is not executable by `PUBLIC`. Grant it only to the trusted +role that deploys subscription definitions: + +```sql +grant usage on schema pgstream_subscriptions to application_migrator; +grant execute on function pgstream_subscriptions.set_subscriptions( + bigint, + pgstream_subscriptions.subscriptions[] +) to application_migrator; +``` + +See [`examples/set_subscriptions.sql`](examples/set_subscriptions.sql) for a +complete reconciliation call. + +## Upgrade from pgstream 0.1 + +Before starting the breaking-change pgstream version, copy and run +[`upgrades/from-pgstream-0.1.sql`](upgrades/from-pgstream-0.1.sql) as a role that +can administer both the legacy pgstream objects and all subscribed tables. + +The migration moves objects with `ALTER ... SET SCHEMA`. It does not copy +subscription rows or recreate target triggers: PostgreSQL preserves object OIDs +and dependencies while moving them. It updates ownership and replaces only the +coordinator function body so future subscription changes use the new schema. + +Back up the database and test the migration against a production-like copy +first. The migration aborts when the legacy source is absent or the destination +already exists. + +## Uninstall + +[`uninstall.sql`](uninstall.sql) deletes every subscription first, allowing the +coordinator to remove generated target triggers and functions, and then drops +the package schema. It must run as the package owner with ownership of every +subscribed table. diff --git a/extensions/subscriptions/examples/set_subscriptions.sql b/extensions/subscriptions/examples/set_subscriptions.sql new file mode 100644 index 0000000..5f29a6c --- /dev/null +++ b/extensions/subscriptions/examples/set_subscriptions.sql @@ -0,0 +1,20 @@ +-- This array is the complete desired subscription set for stream 1. Existing +-- definitions omitted from the array are deleted, which removes their triggers. +SELECT pgstream_subscriptions.set_subscriptions( + 1, + ARRAY[ + ROW( + NULL::uuid, + 'user-created', + 1::bigint, + 'INSERT'::pgstream_subscriptions.operation_type, + 'public', + 'users', + 'new.email_verified = true', + ARRAY['id', 'email', 'created_at']::text[], + '{"topic":"users"}'::jsonb, + '[]'::jsonb, + '[]'::jsonb + )::pgstream_subscriptions.subscriptions + ] +); diff --git a/extensions/subscriptions/migrations/0001_create_pgstream_subscriptions.sql b/extensions/subscriptions/migrations/0001_create_pgstream_subscriptions.sql new file mode 100644 index 0000000..a37a26e --- /dev/null +++ b/extensions/subscriptions/migrations/0001_create_pgstream_subscriptions.sql @@ -0,0 +1,328 @@ +-- Optional pgstream subscription package. +-- +-- Copy this migration into your database migration system and adjust schema, +-- ownership, and grants for your environment before applying it. Run it as a +-- role that owns every table that subscriptions may target. + +CREATE SCHEMA pgstream_subscriptions; + +create type pgstream_subscriptions.operation_type as enum ( + 'INSERT', + 'UPDATE', + 'DELETE' +); + +create table if not exists pgstream_subscriptions.subscriptions ( + id uuid primary key default pgstream.portable_uuidv7(), + key text not null, + stream_id bigint, + operation pgstream_subscriptions.operation_type not null, + schema_name text not null, + table_name text not null, + when_clause text, + column_names text[] not null, + metadata jsonb, + payload_extensions jsonb default '[]'::jsonb, + metadata_extensions jsonb default '[]'::jsonb, + unique (stream_id, key, schema_name, table_name, operation) +); + +CREATE OR REPLACE FUNCTION pgstream_subscriptions.build_extensions(p_extensions jsonb) +RETURNS text +LANGUAGE plpgsql +STABLE +AS $$ +declare + rec record; + path_parts text[]; + top_key text; + grouped_paths jsonb := '{}'::jsonb; + result_parts text[]; + nested_items text[]; + nested_expr text; + i int; +begin + -- Group extensions by top-level key + for rec in + select + value->>'json_path' as json_path, + value->>'expression' as expression + from jsonb_array_elements(p_extensions) + order by value->>'json_path' + loop + path_parts := string_to_array(rec.json_path, '.'); + top_key := path_parts[1]; + + -- Add to grouped paths + if not grouped_paths ? top_key then + grouped_paths := jsonb_set(grouped_paths, array[top_key], '[]'::jsonb); + end if; + + grouped_paths := jsonb_set( + grouped_paths, + array[top_key], + (grouped_paths->top_key) || jsonb_build_array( + jsonb_build_object('parts', to_jsonb(path_parts), 'expr', rec.expression) + ) + ); + end loop; + + -- Build result for each top-level key + for rec in select key, value from jsonb_each(grouped_paths) order by key loop + top_key := rec.key; + nested_items := '{}'; + + -- Process all items for this top-level key + for nested_expr, path_parts in + select + item->>'expr', + array(select jsonb_array_elements_text(item->'parts')) + from jsonb_array_elements(rec.value) item + loop + if array_length(path_parts, 1) = 1 then + -- Top-level only, no nesting + result_parts := array_append(result_parts, format('%L, %s', top_key, nested_expr)); + nested_items := null; + exit; + else + -- Build nested structure from second element onward + nested_expr := nested_expr; + for i in reverse array_length(path_parts, 1) .. 2 loop + nested_expr := format('jsonb_build_object(%L, %s)', path_parts[i], nested_expr); + end loop; + nested_items := array_append(nested_items, nested_expr); + end if; + end loop; + + -- Combine nested items if any + if nested_items is not null and array_length(nested_items, 1) > 0 then + if array_length(nested_items, 1) = 1 then + result_parts := array_append(result_parts, format('%L, %s', top_key, nested_items[1])); + else + result_parts := array_append(result_parts, format('%L, (%s)', top_key, array_to_string(nested_items, ' || '))); + end if; + end if; + end loop; + + if array_length(result_parts, 1) > 0 then + return 'jsonb_build_object(' || array_to_string(result_parts, ', ') || ')'; + else + return '''{}''::jsonb'; + end if; +end; +$$; + +CREATE OR REPLACE FUNCTION pgstream_subscriptions.sync_database_trigger() RETURNS trigger + LANGUAGE plpgsql SECURITY DEFINER + AS $_$ +declare + v_table_name text := coalesce(new.table_name, old.table_name); + v_schema_name text := coalesce(new.schema_name, old.schema_name); + v_when_clause text; + v_if_blocks text; + v_op pgstream_subscriptions.operation_type; +begin + foreach v_op in array array['INSERT', 'UPDATE', 'DELETE']::pgstream_subscriptions.operation_type[] loop + execute format( + $sql$drop trigger if exists pgstream_%s on %I.%I;$sql$, + lower(v_op::text), v_schema_name, v_table_name + ); + + execute format( + $sql$drop function if exists pgstream_subscriptions._publish_after_%s_on_%s;$sql$, + lower(v_op::text), v_table_name + ); + + if exists (select 1 from pgstream_subscriptions.subscriptions where table_name = v_table_name and schema_name = v_schema_name and operation = v_op) then + -- if there is at least one subscription for v_op operation without a when_clause or with an empty one, we do not add the when clause at all + v_when_clause := ( + case when exists ( + select 1 + from pgstream_subscriptions.subscriptions + where table_name = v_table_name and schema_name = v_schema_name and operation = v_op and (when_clause is null or when_clause = '') + ) then null + else ( + select string_agg(when_clause, ') or (') + from pgstream_subscriptions.subscriptions + where table_name = v_table_name and schema_name = v_schema_name and operation = v_op and when_clause is not null and when_clause != '' + ) + end + ); + + -- Build if blocks that collect both payload and metadata per subscription + v_if_blocks := ( + select string_agg(format( + $sql$ + if %s then + v_payloads := array_append(v_payloads, jsonb_build_object( + 'tg_name', %L, + 'new', case when tg_op is distinct from 'DELETE' then jsonb_build_object( + %s + ) else null end, + 'old', case when tg_op is distinct from 'INSERT' then jsonb_build_object( + %s + ) else null end + ) || v_base_payload || (%s)); + v_metadatas := array_append(v_metadatas, coalesce(%L::jsonb, '{}'::jsonb) || (%s)); + end if; + $sql$, + coalesce(nullif(subscription.when_clause, ''), 'true'), + subscription.key, + (select string_agg(format($s$%L, new.%I$s$, column_name, column_name), ', ') from unnest(subscription.column_names) as column_name), + (select string_agg(format($s$%L, old.%I$s$, column_name, column_name), ', ') from unnest(subscription.column_names) as column_name), + pgstream_subscriptions.build_extensions(subscription.payload_extensions), + subscription.metadata, + pgstream_subscriptions.build_extensions(subscription.metadata_extensions) + ), e'\n') from pgstream_subscriptions.subscriptions as subscription where table_name = v_table_name and schema_name = v_schema_name and operation = v_op + ); + + execute format( + $sql$ + create or replace function pgstream_subscriptions._publish_after_%s_on_%s () + returns trigger + as $inner$ + declare + v_payloads jsonb[] := '{}'; + v_metadatas jsonb[] := '{}'; + + v_base_payload jsonb := jsonb_build_object( + 'tg_op', tg_op, + 'tg_table_name', tg_table_name, + 'tg_table_schema', tg_table_schema, + 'timestamp', (extract(epoch from now()) * 1000)::bigint + ); + begin + %s + + if array_length(v_payloads, 1) > 0 then + insert into pgstream.events (payload, metadata, stream_id, lsn) + select p, m, %L, pg_current_wal_lsn() + from unnest(v_payloads, v_metadatas) as t(p, m); + end if; + + if tg_op = 'DELETE' then + return old; + end if; + + return new; + end + $inner$ + language plpgsql + set search_path = '' + security definer; + $sql$, + lower(v_op::text), + v_table_name, + v_if_blocks, + (select distinct stream_id from pgstream_subscriptions.subscriptions where table_name = v_table_name and schema_name = v_schema_name and operation = v_op limit 1) + ); + + execute format( + $sql$ + create constraint trigger pgstream_%s + after %s on %I.%I + deferrable initially deferred + for each row + %s + execute procedure pgstream_subscriptions._publish_after_%s_on_%s() + $sql$, + lower(v_op::text), + lower(v_op::text), + v_schema_name, + v_table_name, + case when v_when_clause is not null and length(v_when_clause) > 0 + then 'when ((' || v_when_clause || '))' + else '' + end, + lower(v_op::text), + v_table_name + ); + end if; + end loop; + + if tg_op = 'DELETE' then + return old; + end if; + + return new; +end +$_$; + +CREATE TRIGGER sync_database_trigger +AFTER INSERT OR DELETE OR UPDATE ON pgstream_subscriptions.subscriptions +FOR EACH ROW EXECUTE FUNCTION pgstream_subscriptions.sync_database_trigger(); + +-- Reconcile the complete desired subscription set for one stream. Rows that did +-- not change are left untouched so their database triggers are not rebuilt. +CREATE OR REPLACE FUNCTION pgstream_subscriptions.set_subscriptions( + p_stream_id bigint, + p_subscriptions pgstream_subscriptions.subscriptions[] +) +RETURNS void +LANGUAGE plpgsql +SECURITY DEFINER +SET search_path = '' +AS $$ +begin + merge into pgstream_subscriptions.subscriptions as target + using ( + select desired.* + from unnest(p_subscriptions) as desired + ) as source + on target.key = source.key and target.stream_id = p_stream_id + when matched and ( + target.operation is distinct from source.operation or + target.schema_name is distinct from source.schema_name or + target.table_name is distinct from source.table_name or + target.when_clause is distinct from source.when_clause or + target.column_names is distinct from source.column_names or + target.metadata is distinct from source.metadata or + target.payload_extensions is distinct from source.payload_extensions or + target.metadata_extensions is distinct from source.metadata_extensions + ) then update set + operation = source.operation, + schema_name = source.schema_name, + table_name = source.table_name, + when_clause = source.when_clause, + column_names = source.column_names, + metadata = source.metadata, + payload_extensions = source.payload_extensions, + metadata_extensions = source.metadata_extensions + when not matched then insert ( + key, + stream_id, + operation, + schema_name, + table_name, + when_clause, + column_names, + metadata, + payload_extensions, + metadata_extensions + ) values ( + source.key, + p_stream_id, + source.operation, + source.schema_name, + source.table_name, + source.when_clause, + source.column_names, + source.metadata, + source.payload_extensions, + source.metadata_extensions + ); + + delete from pgstream_subscriptions.subscriptions as existing + where existing.stream_id = p_stream_id + and not exists ( + select 1 + from unnest(p_subscriptions) as desired + where desired.key = existing.key + ); +end; +$$; + +REVOKE ALL ON FUNCTION pgstream_subscriptions.set_subscriptions( + bigint, + pgstream_subscriptions.subscriptions[] +) FROM PUBLIC; diff --git a/extensions/subscriptions/uninstall.sql b/extensions/subscriptions/uninstall.sql new file mode 100644 index 0000000..5b90a1b --- /dev/null +++ b/extensions/subscriptions/uninstall.sql @@ -0,0 +1,11 @@ +-- Remove the optional subscription package and all triggers it manages. +-- Run this as the package owner with ownership of every subscribed table. +BEGIN; + +-- Deleting definitions invokes the package coordinator, which drops the +-- generated target-table triggers and functions before the schema is removed. +DELETE FROM pgstream_subscriptions.subscriptions; + +DROP SCHEMA pgstream_subscriptions CASCADE; + +COMMIT; diff --git a/extensions/subscriptions/upgrades/from-pgstream-0.1.sql b/extensions/subscriptions/upgrades/from-pgstream-0.1.sql new file mode 100644 index 0000000..06e8a24 --- /dev/null +++ b/extensions/subscriptions/upgrades/from-pgstream-0.1.sql @@ -0,0 +1,276 @@ +-- Upgrade subscriptions created by pgstream 0.1.x in place. +-- +-- Copy and review this migration before running it. It must run as a role that +-- can administer both the legacy pgstream objects and every subscribed table. +-- ALTER ... SET SCHEMA preserves subscription rows, object OIDs, and trigger +-- dependencies; target-table triggers are not recreated by this migration. + +BEGIN; + +DO $$ +BEGIN + IF to_regclass('pgstream.subscriptions') IS NULL THEN + RAISE EXCEPTION 'pgstream.subscriptions does not exist; use the fresh-install migration instead'; + END IF; + + IF to_regclass('pgstream_subscriptions.subscriptions') IS NOT NULL THEN + RAISE EXCEPTION 'pgstream_subscriptions.subscriptions already exists'; + END IF; +END; +$$; + +LOCK TABLE pgstream.subscriptions IN ACCESS EXCLUSIVE MODE; + +CREATE SCHEMA pgstream_subscriptions AUTHORIZATION CURRENT_USER; + +-- Move generated functions without recreating them. Target triggers reference +-- these functions by OID, so their dependencies remain valid. +DO $$ +DECLARE + v_function_oid oid; + v_signature text; +BEGIN + FOR v_function_oid IN + SELECT DISTINCT procedure.oid + FROM pg_catalog.pg_trigger AS trigger + JOIN pg_catalog.pg_proc AS procedure ON procedure.oid = trigger.tgfoid + JOIN pg_catalog.pg_namespace AS namespace ON namespace.oid = procedure.pronamespace + WHERE NOT trigger.tgisinternal + AND trigger.tgname IN ('pgstream_insert', 'pgstream_update', 'pgstream_delete') + AND namespace.nspname = 'pgstream' + AND procedure.proname LIKE '\_publish\_after\_%' ESCAPE '\' + LOOP + SELECT v_function_oid::regprocedure::text INTO v_signature; + EXECUTE format('ALTER FUNCTION %s SET SCHEMA pgstream_subscriptions', v_signature); + + SELECT v_function_oid::regprocedure::text INTO v_signature; + EXECUTE format('ALTER FUNCTION %s OWNER TO CURRENT_USER', v_signature); + END LOOP; +END; +$$; + +ALTER TYPE pgstream.operation_type SET SCHEMA pgstream_subscriptions; +ALTER TYPE pgstream_subscriptions.operation_type OWNER TO CURRENT_USER; + +ALTER TABLE pgstream.subscriptions SET SCHEMA pgstream_subscriptions; +ALTER TABLE pgstream_subscriptions.subscriptions OWNER TO CURRENT_USER; + +ALTER FUNCTION pgstream.build_extensions(jsonb) SET SCHEMA pgstream_subscriptions; +ALTER FUNCTION pgstream_subscriptions.build_extensions(jsonb) OWNER TO CURRENT_USER; + +ALTER FUNCTION pgstream.sync_database_trigger() SET SCHEMA pgstream_subscriptions; +ALTER FUNCTION pgstream_subscriptions.sync_database_trigger() OWNER TO CURRENT_USER; + +-- Replace only the coordinator body so future rebuilds use the new schema. +CREATE OR REPLACE FUNCTION pgstream_subscriptions.sync_database_trigger() RETURNS trigger + LANGUAGE plpgsql SECURITY DEFINER + AS $_$ +declare + v_table_name text := coalesce(new.table_name, old.table_name); + v_schema_name text := coalesce(new.schema_name, old.schema_name); + v_when_clause text; + v_if_blocks text; + v_op pgstream_subscriptions.operation_type; +begin + foreach v_op in array array['INSERT', 'UPDATE', 'DELETE']::pgstream_subscriptions.operation_type[] loop + execute format( + $sql$drop trigger if exists pgstream_%s on %I.%I;$sql$, + lower(v_op::text), v_schema_name, v_table_name + ); + + execute format( + $sql$drop function if exists pgstream_subscriptions._publish_after_%s_on_%s;$sql$, + lower(v_op::text), v_table_name + ); + + if exists (select 1 from pgstream_subscriptions.subscriptions where table_name = v_table_name and schema_name = v_schema_name and operation = v_op) then + -- if there is at least one subscription for v_op operation without a when_clause or with an empty one, we do not add the when clause at all + v_when_clause := ( + case when exists ( + select 1 + from pgstream_subscriptions.subscriptions + where table_name = v_table_name and schema_name = v_schema_name and operation = v_op and (when_clause is null or when_clause = '') + ) then null + else ( + select string_agg(when_clause, ') or (') + from pgstream_subscriptions.subscriptions + where table_name = v_table_name and schema_name = v_schema_name and operation = v_op and when_clause is not null and when_clause != '' + ) + end + ); + + -- Build if blocks that collect both payload and metadata per subscription + v_if_blocks := ( + select string_agg(format( + $sql$ + if %s then + v_payloads := array_append(v_payloads, jsonb_build_object( + 'tg_name', %L, + 'new', case when tg_op is distinct from 'DELETE' then jsonb_build_object( + %s + ) else null end, + 'old', case when tg_op is distinct from 'INSERT' then jsonb_build_object( + %s + ) else null end + ) || v_base_payload || (%s)); + v_metadatas := array_append(v_metadatas, coalesce(%L::jsonb, '{}'::jsonb) || (%s)); + end if; + $sql$, + coalesce(nullif(subscription.when_clause, ''), 'true'), + subscription.key, + (select string_agg(format($s$%L, new.%I$s$, column_name, column_name), ', ') from unnest(subscription.column_names) as column_name), + (select string_agg(format($s$%L, old.%I$s$, column_name, column_name), ', ') from unnest(subscription.column_names) as column_name), + pgstream_subscriptions.build_extensions(subscription.payload_extensions), + subscription.metadata, + pgstream_subscriptions.build_extensions(subscription.metadata_extensions) + ), e'\n') from pgstream_subscriptions.subscriptions as subscription where table_name = v_table_name and schema_name = v_schema_name and operation = v_op + ); + + execute format( + $sql$ + create or replace function pgstream_subscriptions._publish_after_%s_on_%s () + returns trigger + as $inner$ + declare + v_payloads jsonb[] := '{}'; + v_metadatas jsonb[] := '{}'; + + v_base_payload jsonb := jsonb_build_object( + 'tg_op', tg_op, + 'tg_table_name', tg_table_name, + 'tg_table_schema', tg_table_schema, + 'timestamp', (extract(epoch from now()) * 1000)::bigint + ); + begin + %s + + if array_length(v_payloads, 1) > 0 then + insert into pgstream.events (payload, metadata, stream_id, lsn) + select p, m, %L, pg_current_wal_lsn() + from unnest(v_payloads, v_metadatas) as t(p, m); + end if; + + if tg_op = 'DELETE' then + return old; + end if; + + return new; + end + $inner$ + language plpgsql + set search_path = '' + security definer; + $sql$, + lower(v_op::text), + v_table_name, + v_if_blocks, + (select distinct stream_id from pgstream_subscriptions.subscriptions where table_name = v_table_name and schema_name = v_schema_name and operation = v_op limit 1) + ); + + execute format( + $sql$ + create constraint trigger pgstream_%s + after %s on %I.%I + deferrable initially deferred + for each row + %s + execute procedure pgstream_subscriptions._publish_after_%s_on_%s() + $sql$, + lower(v_op::text), + lower(v_op::text), + v_schema_name, + v_table_name, + case when v_when_clause is not null and length(v_when_clause) > 0 + then 'when ((' || v_when_clause || '))' + else '' + end, + lower(v_op::text), + v_table_name + ); + end if; + end loop; + + if tg_op = 'DELETE' then + return old; + end if; + + return new; +end +$_$; + +-- Reconcile the complete desired subscription set for one stream. Rows that did +-- not change are left untouched so their database triggers are not rebuilt. +CREATE OR REPLACE FUNCTION pgstream_subscriptions.set_subscriptions( + p_stream_id bigint, + p_subscriptions pgstream_subscriptions.subscriptions[] +) +RETURNS void +LANGUAGE plpgsql +SECURITY DEFINER +SET search_path = '' +AS $$ +begin + merge into pgstream_subscriptions.subscriptions as target + using ( + select desired.* + from unnest(p_subscriptions) as desired + ) as source + on target.key = source.key and target.stream_id = p_stream_id + when matched and ( + target.operation is distinct from source.operation or + target.schema_name is distinct from source.schema_name or + target.table_name is distinct from source.table_name or + target.when_clause is distinct from source.when_clause or + target.column_names is distinct from source.column_names or + target.metadata is distinct from source.metadata or + target.payload_extensions is distinct from source.payload_extensions or + target.metadata_extensions is distinct from source.metadata_extensions + ) then update set + operation = source.operation, + schema_name = source.schema_name, + table_name = source.table_name, + when_clause = source.when_clause, + column_names = source.column_names, + metadata = source.metadata, + payload_extensions = source.payload_extensions, + metadata_extensions = source.metadata_extensions + when not matched then insert ( + key, + stream_id, + operation, + schema_name, + table_name, + when_clause, + column_names, + metadata, + payload_extensions, + metadata_extensions + ) values ( + source.key, + p_stream_id, + source.operation, + source.schema_name, + source.table_name, + source.when_clause, + source.column_names, + source.metadata, + source.payload_extensions, + source.metadata_extensions + ); + + delete from pgstream_subscriptions.subscriptions as existing + where existing.stream_id = p_stream_id + and not exists ( + select 1 + from unnest(p_subscriptions) as desired + where desired.key = existing.key + ); +end; +$$; + +REVOKE ALL ON FUNCTION pgstream_subscriptions.set_subscriptions( + bigint, + pgstream_subscriptions.subscriptions[] +) FROM PUBLIC; + +COMMIT; diff --git a/implementation-notes.md b/implementation-notes.md new file mode 100644 index 0000000..17fc786 --- /dev/null +++ b/implementation-notes.md @@ -0,0 +1,23 @@ +# Implementation Notes + +Running notes on how the subscription extraction interprets or diverges from the agreed plan. + +## Design decisions + +- The optional package lives under `extensions/subscriptions/`; `extensions/README.md` explicitly distinguishes these user-managed SQL packages from PostgreSQL `CREATE EXTENSION` packages. +- Subscription objects use the `pgstream_subscriptions` schema, while emitted events continue to target the core `pgstream.events` table. +- Existing SQLx migration files remain unchanged. A forward core migration removes an empty legacy installation, accepts an already-moved installation, and blocks upgrades that still contain legacy subscription rows. +- The upgrade migration moves existing objects with `ALTER ... SET SCHEMA` so subscription rows, UUIDs, generated function OIDs, and target-trigger dependencies are preserved. + +## Deviations + +- None yet. + +## Tradeoffs + +- The package keeps the existing drop-and-recreate trigger model. This intentionally requires the SQL package owner to own subscribed tables, but removes that requirement from the long-lived pgstream runtime role. +- The package migrations are plain, editable SQL and are not executed or versioned by the Rust daemon. This maximizes compatibility with user migration systems at the cost of centralized automatic upgrades. + +## Open questions + +- None yet. diff --git a/migrations/1786434463000_extract_subscriptions.sql b/migrations/1786434463000_extract_subscriptions.sql new file mode 100644 index 0000000..8162ecc --- /dev/null +++ b/migrations/1786434463000_extract_subscriptions.sql @@ -0,0 +1,33 @@ +-- Subscription management is now an optional, user-installed SQL package under +-- extensions/subscriptions. Existing installations with subscriptions must run +-- that package's upgrade migration before starting this pgstream version. +DO $$ +DECLARE + v_has_subscriptions boolean; +BEGIN + IF to_regclass('pgstream.subscriptions') IS NOT NULL THEN + EXECUTE 'SELECT EXISTS (SELECT 1 FROM pgstream.subscriptions)' + INTO v_has_subscriptions; + + IF v_has_subscriptions THEN + RAISE EXCEPTION USING + MESSAGE = 'legacy pgstream subscriptions must be upgraded before pgstream can start', + DETAIL = 'The pgstream.subscriptions table still contains subscription definitions.', + HINT = 'Run extensions/subscriptions/upgrades/from-pgstream-0.1.sql as the owner of the subscribed tables, then start pgstream again.'; + END IF; + END IF; +END; +$$; + +DO $$ +BEGIN + IF to_regclass('pgstream.subscriptions') IS NOT NULL THEN + DROP TRIGGER IF EXISTS sync_database_trigger ON pgstream.subscriptions; + DROP TABLE pgstream.subscriptions; + END IF; +END; +$$; +DROP FUNCTION IF EXISTS pgstream.sync_database_trigger(); +DROP FUNCTION IF EXISTS pgstream.build_extensions(jsonb); +DROP FUNCTION IF EXISTS pgstream.build_payload_from_extensions(jsonb); +DROP TYPE IF EXISTS pgstream.operation_type; diff --git a/src/test_utils/database.rs b/src/test_utils/database.rs index cfba764..4becd1e 100644 --- a/src/test_utils/database.rs +++ b/src/test_utils/database.rs @@ -61,6 +61,23 @@ impl TestDatabase { } } + /// Creates a migrated database with the optional subscription SQL package. + pub async fn spawn_with_subscriptions() -> Self { + let database = Self::spawn().await; + database.install_subscriptions().await; + database + } + + /// Installs the user-managed subscription SQL package for tests that need it. + pub async fn install_subscriptions(&self) { + sqlx::raw_sql(include_str!( + "../../extensions/subscriptions/migrations/0001_create_pgstream_subscriptions.sql" + )) + .execute(&self.pool) + .await + .expect("Failed to install pgstream subscription SQL package"); + } + /// Optional helper: create today's partition for `pgstream.events`, /// like in your original `setup_database`. pub async fn ensure_today_partition(&self) { diff --git a/tests/subscription_extension_tests.rs b/tests/subscription_extension_tests.rs new file mode 100644 index 0000000..540c0c9 --- /dev/null +++ b/tests/subscription_extension_tests.rs @@ -0,0 +1,400 @@ +use postgres_stream::test_utils::TestDatabase; +use sqlx::Row; + +const LEGACY_INIT: &str = include_str!("../migrations/1765364029646_init.sql"); +const LEGACY_LSN: &str = include_str!("../migrations/1766831075000_add_lsn.sql"); +const LEGACY_METADATA: &str = + include_str!("../migrations/1767795878880_add_metadata_extensions.sql"); +const EXTRACT_SUBSCRIPTIONS: &str = + include_str!("../migrations/1786434463000_extract_subscriptions.sql"); +const INSTALL_SUBSCRIPTIONS: &str = + include_str!("../extensions/subscriptions/migrations/0001_create_pgstream_subscriptions.sql"); +const UPGRADE_SUBSCRIPTIONS: &str = + include_str!("../extensions/subscriptions/upgrades/from-pgstream-0.1.sql"); +const UNINSTALL_SUBSCRIPTIONS: &str = include_str!("../extensions/subscriptions/uninstall.sql"); + +async fn create_users_table(database: &TestDatabase) { + sqlx::query( + r#" + create table public.users ( + id bigint generated always as identity primary key, + email text not null + ) + "#, + ) + .execute(&database.pool) + .await + .expect("Failed to create users table"); +} + +async fn set_user_subscription(database: &TestDatabase) { + sqlx::query( + r#" + select pgstream_subscriptions.set_subscriptions( + 1, + array[ + row( + null::uuid, + 'user-created', + 1::bigint, + 'INSERT'::pgstream_subscriptions.operation_type, + 'public', + 'users', + null::text, + array['id', 'email']::text[], + null::jsonb, + '[]'::jsonb, + '[]'::jsonb + )::pgstream_subscriptions.subscriptions + ] + ) + "#, + ) + .execute(&database.pool) + .await + .expect("Failed to reconcile subscriptions"); +} + +async fn insert_trigger_oid(database: &TestDatabase) -> i64 { + sqlx::query_scalar( + r#" + select trigger.oid::bigint + from pg_catalog.pg_trigger as trigger + join pg_catalog.pg_class as relation on relation.oid = trigger.tgrelid + join pg_catalog.pg_namespace as namespace on namespace.oid = relation.relnamespace + where namespace.nspname = 'public' + and relation.relname = 'users' + and trigger.tgname = 'pgstream_insert' + "#, + ) + .fetch_one(&database.pool) + .await + .expect("Failed to find subscription trigger") +} + +#[tokio::test(flavor = "multi_thread")] +async fn core_migrations_do_not_install_subscriptions() { + let database = TestDatabase::spawn().await; + + let row = sqlx::query( + r#" + select + to_regclass('pgstream.subscriptions')::text as legacy, + to_regclass('pgstream_subscriptions.subscriptions')::text as extracted + "#, + ) + .fetch_one(&database.pool) + .await + .expect("Failed to inspect subscription schemas"); + + assert_eq!(row.get::, _>("legacy"), None); + assert_eq!(row.get::, _>("extracted"), None); +} + +#[tokio::test(flavor = "multi_thread")] +async fn set_subscriptions_reconciles_and_skips_unchanged_definitions() { + let database = TestDatabase::spawn_with_subscriptions().await; + create_users_table(&database).await; + + set_user_subscription(&database).await; + let initial_trigger_oid = insert_trigger_oid(&database).await; + + set_user_subscription(&database).await; + let unchanged_trigger_oid = insert_trigger_oid(&database).await; + assert_eq!(initial_trigger_oid, unchanged_trigger_oid); + + sqlx::query( + r#" + select pgstream_subscriptions.set_subscriptions( + 1, + array[]::pgstream_subscriptions.subscriptions[] + ) + "#, + ) + .execute(&database.pool) + .await + .expect("Failed to remove subscriptions"); + + let trigger_exists: bool = sqlx::query_scalar( + r#" + select exists ( + select 1 + from pg_catalog.pg_trigger as trigger + join pg_catalog.pg_class as relation on relation.oid = trigger.tgrelid + join pg_catalog.pg_namespace as namespace on namespace.oid = relation.relnamespace + where namespace.nspname = 'public' + and relation.relname = 'users' + and trigger.tgname = 'pgstream_insert' + ) + "#, + ) + .fetch_one(&database.pool) + .await + .expect("Failed to inspect removed trigger"); + + assert!(!trigger_exists); +} + +#[tokio::test(flavor = "multi_thread")] +async fn core_migration_blocks_nonempty_legacy_subscriptions() { + let database = TestDatabase::spawn_without_migrations().await; + + for migration in [LEGACY_INIT, LEGACY_LSN, LEGACY_METADATA] { + sqlx::raw_sql(migration) + .execute(&database.pool) + .await + .expect("Failed to install legacy pgstream schema"); + } + create_users_table(&database).await; + + sqlx::query( + r#" + insert into pgstream.subscriptions ( + key, stream_id, operation, schema_name, table_name, column_names + ) values ( + 'user-created', 1, 'INSERT', 'public', 'users', array['id', 'email'] + ) + "#, + ) + .execute(&database.pool) + .await + .expect("Failed to create legacy subscription"); + + let error = sqlx::raw_sql(EXTRACT_SUBSCRIPTIONS) + .execute(&database.pool) + .await + .expect_err("Core migration should block legacy subscriptions"); + assert!( + error + .to_string() + .contains("legacy pgstream subscriptions must be upgraded") + ); + assert_eq!( + sqlx::query_scalar::<_, i64>("select count(*) from pgstream.subscriptions") + .fetch_one(&database.pool) + .await + .expect("Failed to verify preserved legacy subscriptions"), + 1 + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn package_owner_manages_triggers_without_runtime_table_privileges() { + let database = TestDatabase::spawn().await; + database.ensure_today_partition().await; + + sqlx::query("create role app_owner nologin") + .execute(&database.pool) + .await + .expect("Failed to create application owner"); + sqlx::query("create role pgstream_runtime nologin") + .execute(&database.pool) + .await + .expect("Failed to create pgstream runtime role"); + + let database_name = database.config.name.replace('"', "\"\""); + sqlx::query(&format!( + "grant create on database \"{database_name}\" to app_owner" + )) + .execute(&database.pool) + .await + .expect("Failed to grant schema creation"); + sqlx::query("grant create on schema public to app_owner") + .execute(&database.pool) + .await + .expect("Failed to grant application schema creation"); + sqlx::query("grant usage on schema pgstream to app_owner") + .execute(&database.pool) + .await + .expect("Failed to grant core schema usage"); + sqlx::query("grant insert on pgstream.events to app_owner") + .execute(&database.pool) + .await + .expect("Failed to grant event insertion"); + + let mut connection = database.pool.acquire().await.expect("Failed to connect"); + sqlx::query("set role app_owner") + .execute(&mut *connection) + .await + .expect("Failed to assume application owner role"); + sqlx::query( + "create table public.users (id bigint generated always as identity primary key, email text not null)", + ) + .execute(&mut *connection) + .await + .expect("Application owner failed to create target table"); + sqlx::raw_sql(INSTALL_SUBSCRIPTIONS) + .execute(&mut *connection) + .await + .expect("Application owner failed to install subscriptions"); + sqlx::query( + r#" + insert into pgstream_subscriptions.subscriptions ( + key, stream_id, operation, schema_name, table_name, column_names + ) values ( + 'user-created', 1, 'INSERT', 'public', 'users', array['id', 'email'] + ) + "#, + ) + .execute(&mut *connection) + .await + .expect("Application owner failed to create subscription"); + sqlx::query("insert into public.users (email) values ('owner@example.com')") + .execute(&mut *connection) + .await + .expect("Subscription trigger failed under application owner"); + sqlx::query("reset role") + .execute(&mut *connection) + .await + .expect("Failed to reset role"); + drop(connection); + + let runtime_has_trigger: bool = sqlx::query_scalar( + "select has_table_privilege('pgstream_runtime', 'public.users', 'TRIGGER')", + ) + .fetch_one(&database.pool) + .await + .expect("Failed to inspect runtime privileges"); + assert!(!runtime_has_trigger); + + let target_owner: String = sqlx::query_scalar( + "select pg_get_userbyid(relowner) from pg_class where oid = 'public.users'::regclass", + ) + .fetch_one(&database.pool) + .await + .expect("Failed to inspect target owner"); + assert_eq!(target_owner, "app_owner"); + + let event_name: String = sqlx::query_scalar("select payload->>'tg_name' from pgstream.events") + .fetch_one(&database.pool) + .await + .expect("Subscription did not emit an event"); + assert_eq!(event_name, "user-created"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn uninstall_removes_package_and_managed_triggers() { + let database = TestDatabase::spawn_with_subscriptions().await; + create_users_table(&database).await; + set_user_subscription(&database).await; + + sqlx::raw_sql(UNINSTALL_SUBSCRIPTIONS) + .execute(&database.pool) + .await + .expect("Failed to uninstall subscription package"); + + let package_exists: bool = sqlx::query_scalar( + "select exists(select 1 from pg_namespace where nspname = 'pgstream_subscriptions')", + ) + .fetch_one(&database.pool) + .await + .expect("Failed to inspect package schema"); + assert!(!package_exists); + + let trigger_exists: bool = sqlx::query_scalar( + "select exists(select 1 from pg_trigger where tgrelid = 'public.users'::regclass and tgname = 'pgstream_insert')", + ) + .fetch_one(&database.pool) + .await + .expect("Failed to inspect managed trigger"); + assert!(!trigger_exists); +} + +#[tokio::test(flavor = "multi_thread")] +async fn upgrade_moves_subscriptions_without_recreating_rows_or_triggers() { + let database = TestDatabase::spawn_without_migrations().await; + + for migration in [LEGACY_INIT, LEGACY_LSN, LEGACY_METADATA] { + sqlx::raw_sql(migration) + .execute(&database.pool) + .await + .expect("Failed to install legacy pgstream schema"); + } + + database.ensure_today_partition().await; + create_users_table(&database).await; + + let subscription_id: sqlx::types::Uuid = sqlx::query_scalar( + r#" + insert into pgstream.subscriptions ( + key, + stream_id, + operation, + schema_name, + table_name, + column_names, + payload_extensions, + metadata_extensions + ) values ( + 'user-created', + 1, + 'INSERT', + 'public', + 'users', + array['id', 'email'], + '[]', + '[]' + ) + returning id + "#, + ) + .fetch_one(&database.pool) + .await + .expect("Failed to create legacy subscription"); + + let trigger_oid = insert_trigger_oid(&database).await; + let trigger_function_oid: i64 = + sqlx::query_scalar("select tgfoid::bigint from pg_trigger where oid::bigint = $1") + .bind(trigger_oid) + .fetch_one(&database.pool) + .await + .expect("Failed to find legacy trigger function"); + + sqlx::raw_sql(UPGRADE_SUBSCRIPTIONS) + .execute(&database.pool) + .await + .expect("Failed to upgrade subscription package"); + + let moved_subscription_id: sqlx::types::Uuid = + sqlx::query_scalar("select id from pgstream_subscriptions.subscriptions") + .fetch_one(&database.pool) + .await + .expect("Failed to find moved subscription"); + assert_eq!(subscription_id, moved_subscription_id); + assert_eq!(trigger_oid, insert_trigger_oid(&database).await); + + let moved_function = sqlx::query( + r#" + select procedure.oid::bigint as oid, namespace.nspname as schema_name + from pg_catalog.pg_proc as procedure + join pg_catalog.pg_namespace as namespace on namespace.oid = procedure.pronamespace + where procedure.oid::bigint = $1 + "#, + ) + .bind(trigger_function_oid) + .fetch_one(&database.pool) + .await + .expect("Failed to inspect moved trigger function"); + assert_eq!(moved_function.get::("oid"), trigger_function_oid); + assert_eq!( + moved_function.get::("schema_name"), + "pgstream_subscriptions" + ); + + // The core extraction migration must accept an already-upgraded installation. + sqlx::raw_sql(EXTRACT_SUBSCRIPTIONS) + .execute(&database.pool) + .await + .expect("Core extraction migration removed the user-managed package"); + + sqlx::query("insert into public.users (email) values ('upgrade@example.com')") + .execute(&database.pool) + .await + .expect("Moved trigger failed"); + + let event_name: String = sqlx::query_scalar("select payload->>'tg_name' from pgstream.events") + .fetch_one(&database.pool) + .await + .expect("Moved trigger did not emit an event"); + assert_eq!(event_name, "user-created"); +} diff --git a/tests/subscriptions_tests.rs b/tests/subscriptions_tests.rs index 7a1ece5..9a433fa 100644 --- a/tests/subscriptions_tests.rs +++ b/tests/subscriptions_tests.rs @@ -38,9 +38,9 @@ async fn create_subscription( ) -> sqlx::types::Uuid { let row = sqlx::query( r#" - insert into pgstream.subscriptions + insert into pgstream_subscriptions.subscriptions (key, stream_id, operation, schema_name, table_name, when_clause, column_names, payload_extensions, metadata, metadata_extensions) - values ($1, $2, $3::pgstream.operation_type, 'public', 'users', $4, array['id', 'name', 'email', 'age'], $5, $6, $7) + values ($1, $2, $3::pgstream_subscriptions.operation_type, 'public', 'users', $4, array['id', 'name', 'email', 'age'], $5, $6, $7) returning id "#, ) @@ -133,7 +133,7 @@ async fn get_events_with_metadata( #[tokio::test(flavor = "multi_thread")] async fn test_subscription_creates_triggers_and_events() { - let db = TestDatabase::spawn().await; + let db = TestDatabase::spawn_with_subscriptions().await; db.ensure_today_partition().await; // Create test table @@ -170,7 +170,7 @@ async fn test_subscription_creates_triggers_and_events() { #[tokio::test(flavor = "multi_thread")] async fn test_subscription_with_when_clause() { - let db = TestDatabase::spawn().await; + let db = TestDatabase::spawn_with_subscriptions().await; db.ensure_today_partition().await; create_test_table(&db.pool).await; @@ -221,7 +221,7 @@ async fn test_subscription_with_when_clause() { #[tokio::test(flavor = "multi_thread")] async fn test_update_operation() { - let db = TestDatabase::spawn().await; + let db = TestDatabase::spawn_with_subscriptions().await; db.ensure_today_partition().await; create_test_table(&db.pool).await; @@ -266,7 +266,7 @@ async fn test_update_operation() { #[tokio::test(flavor = "multi_thread")] async fn test_delete_operation() { - let db = TestDatabase::spawn().await; + let db = TestDatabase::spawn_with_subscriptions().await; db.ensure_today_partition().await; create_test_table(&db.pool).await; @@ -309,7 +309,7 @@ async fn test_delete_operation() { #[tokio::test(flavor = "multi_thread")] async fn test_payload_extensions() { - let db = TestDatabase::spawn().await; + let db = TestDatabase::spawn_with_subscriptions().await; db.ensure_today_partition().await; create_test_table(&db.pool).await; @@ -340,11 +340,12 @@ async fn test_payload_extensions() { .await; // Verify build_extensions generates valid SQL - let payload_expr: String = sqlx::query_scalar("select pgstream.build_extensions($1)") - .bind(&payload_extensions) - .fetch_one(&db.pool) - .await - .expect("Failed to get build_extensions result"); + let payload_expr: String = + sqlx::query_scalar("select pgstream_subscriptions.build_extensions($1)") + .bind(&payload_extensions) + .fetch_one(&db.pool) + .await + .expect("Failed to get build_extensions result"); // Assert the SQL expression contains expected JSONB operations assert!( @@ -390,7 +391,7 @@ async fn test_payload_extensions() { #[tokio::test(flavor = "multi_thread")] async fn test_multiple_subscriptions_same_stream() { - let db = TestDatabase::spawn().await; + let db = TestDatabase::spawn_with_subscriptions().await; db.ensure_today_partition().await; create_test_table(&db.pool).await; @@ -458,7 +459,7 @@ async fn test_multiple_subscriptions_same_stream() { #[tokio::test(flavor = "multi_thread")] async fn test_subscription_deletion_removes_trigger() { - let db = TestDatabase::spawn().await; + let db = TestDatabase::spawn_with_subscriptions().await; db.ensure_today_partition().await; create_test_table(&db.pool).await; @@ -491,7 +492,7 @@ async fn test_subscription_deletion_removes_trigger() { // Delete the subscription sqlx::query( r#" - delete from pgstream.subscriptions + delete from pgstream_subscriptions.subscriptions where id = $1 "#, ) @@ -518,7 +519,7 @@ async fn test_subscription_deletion_removes_trigger() { #[tokio::test(flavor = "multi_thread")] async fn test_events_have_lsn_captured() { - let db = TestDatabase::spawn().await; + let db = TestDatabase::spawn_with_subscriptions().await; db.ensure_today_partition().await; create_test_table(&db.pool).await; @@ -575,7 +576,7 @@ async fn test_events_have_lsn_captured() { #[tokio::test(flavor = "multi_thread")] async fn test_lsn_can_be_used_for_queries() { - let db = TestDatabase::spawn().await; + let db = TestDatabase::spawn_with_subscriptions().await; db.ensure_today_partition().await; create_test_table(&db.pool).await; @@ -630,7 +631,7 @@ async fn test_lsn_can_be_used_for_queries() { #[tokio::test(flavor = "multi_thread")] async fn test_static_metadata() { - let db = TestDatabase::spawn().await; + let db = TestDatabase::spawn_with_subscriptions().await; db.ensure_today_partition().await; create_test_table(&db.pool).await; @@ -681,7 +682,7 @@ async fn test_static_metadata() { #[tokio::test(flavor = "multi_thread")] async fn test_metadata_extensions() { - let db = TestDatabase::spawn().await; + let db = TestDatabase::spawn_with_subscriptions().await; db.ensure_today_partition().await; create_test_table(&db.pool).await; @@ -730,7 +731,7 @@ async fn test_metadata_extensions() { #[tokio::test(flavor = "multi_thread")] async fn test_merged_static_and_dynamic_metadata() { - let db = TestDatabase::spawn().await; + let db = TestDatabase::spawn_with_subscriptions().await; db.ensure_today_partition().await; create_test_table(&db.pool).await; @@ -793,7 +794,7 @@ async fn test_merged_static_and_dynamic_metadata() { #[tokio::test(flavor = "multi_thread")] async fn test_metadata_extensions_with_nested_paths() { - let db = TestDatabase::spawn().await; + let db = TestDatabase::spawn_with_subscriptions().await; db.ensure_today_partition().await; create_test_table(&db.pool).await; diff --git a/tests/transaction_lsn_tests.rs b/tests/transaction_lsn_tests.rs index cb37fe0..6db08ae 100644 --- a/tests/transaction_lsn_tests.rs +++ b/tests/transaction_lsn_tests.rs @@ -8,7 +8,7 @@ use std::time::Duration; #[tokio::test(flavor = "multi_thread")] async fn test_insert_and_update_share_commit_lsn_distinct_from_row_lsns() { - let db = TestDatabase::spawn().await; + let db = TestDatabase::spawn_with_subscriptions().await; db.ensure_today_partition().await; sqlx::query( @@ -29,10 +29,10 @@ async fn test_insert_and_update_share_commit_lsn_distinct_from_row_lsns() { for (key, operation) in [("user_insert", "INSERT"), ("user_update", "UPDATE")] { sqlx::query( r#" - insert into pgstream.subscriptions + insert into pgstream_subscriptions.subscriptions (key, stream_id, operation, schema_name, table_name, column_names, payload_extensions, metadata_extensions) values - ($1, $2, $3::pgstream.operation_type, 'public', 'users', array['id', 'name'], '[]', '[]') + ($1, $2, $3::pgstream_subscriptions.operation_type, 'public', 'users', array['id', 'name'], '[]', '[]') "#, ) .bind(key) diff --git a/zensical.toml b/zensical.toml index 49bb143..ec67c22 100644 --- a/zensical.toml +++ b/zensical.toml @@ -10,6 +10,10 @@ repo_name = "psteinroe/postgres-stream" nav = [ { "Introduction" = "index.md" }, { "Getting Started" = "getting-started.md" }, + { "Guides" = [ + "guides/subscription-setup.md", + "guides/upgrade-subscriptions.md" + ]}, { "Concepts" = [ "concepts/how-it-works.md", "concepts/subscriptions.md", From 5e9c3a0b59d9c5686c55c9e300607033f4a74162 Mon Sep 17 00:00:00 2001 From: psteinroe Date: Tue, 11 Aug 2026 08:32:30 +0000 Subject: [PATCH 2/3] refactor(subscriptions): keep reconciliation helper optional --- README.md | 2 +- docs/concepts/subscriptions.md | 25 +---- docs/guides/subscription-setup.md | 26 ++---- docs/guides/upgrade-subscriptions.md | 5 +- docs/index.md | 2 +- docs/reference/event-format.md | 4 +- extensions/README.md | 2 +- extensions/subscriptions/README.md | 23 ++--- .../examples/set_subscriptions.sql | 92 +++++++++++++++---- .../0001_create_pgstream_subscriptions.sql | 75 --------------- .../upgrades/from-pgstream-0.1.sql | 75 --------------- implementation-notes.md | 23 ----- tests/subscription_extension_tests.rs | 20 +++- 13 files changed, 118 insertions(+), 256 deletions(-) delete mode 100644 implementation-notes.md diff --git a/README.md b/README.md index 0439b95..f52b897 100644 --- a/README.md +++ b/README.md @@ -28,7 +28,7 @@ Events are inserted into the `pgstream.events` table and streamed via logical re **Two ways to create events:** -1. **Subscriptions** (optional) - Install the user-managed SQL package in `extensions/subscriptions` to capture table changes with triggers +1. **Subscriptions** (optional) - Managed triggers 2. **Manual inserts** - Insert directly into `pgstream.events` from your application or database functions ## Trade-offs diff --git a/docs/concepts/subscriptions.md b/docs/concepts/subscriptions.md index 92de308..4f61185 100644 --- a/docs/concepts/subscriptions.md +++ b/docs/concepts/subscriptions.md @@ -111,30 +111,9 @@ Both subscriptions will fire for a verified user, creating two events with diffe ## Reconciling a Stream -Each changed subscription recreates its target trigger. The package provides `set_subscriptions()` so deployments can supply the complete desired set for one stream without rewriting unchanged rows: +Each changed subscription recreates its target trigger. The package includes an optional [`set_subscriptions()` example](https://github.com/psteinroe/postgres-stream/blob/main/extensions/subscriptions/examples/set_subscriptions.sql) that you can copy into your own migrations; it is not installed by default. -```sql -select pgstream_subscriptions.set_subscriptions( - 1, - array[ - row( - null::uuid, - 'user-created', - 1::bigint, - 'INSERT'::pgstream_subscriptions.operation_type, - 'public', - 'users', - null::text, - array['id', 'email']::text[], - null::jsonb, - '[]'::jsonb, - '[]'::jsonb - )::pgstream_subscriptions.subscriptions - ] -); -``` - -The array is the complete desired state for stream `1`: missing rows are inserted, changed rows are updated, and omitted rows are deleted. See the [package example](https://github.com/psteinroe/postgres-stream/blob/main/extensions/subscriptions/examples/set_subscriptions.sql). +The helper accepts the complete desired state for one stream: missing rows are inserted, changed rows are updated, and omitted rows are deleted. Unchanged rows are not written, avoiding unnecessary trigger recreation. ## Next Steps diff --git a/docs/guides/subscription-setup.md b/docs/guides/subscription-setup.md index 9eb0de0..9211bf3 100644 --- a/docs/guides/subscription-setup.md +++ b/docs/guides/subscription-setup.md @@ -4,11 +4,11 @@ Subscriptions are an optional SQL package that you install through your own data ## Ownership model -Subscription changes drop and recreate target-table triggers. PostgreSQL requires the role dropping a trigger to own its table. Run the package migration as the application migration role that owns subscribed tables, or as a role that is a member of all relevant owner roles. +Subscription changes drop and recreate target-table triggers. Postgres requires the role dropping a trigger to own its table. Run the package migration as the application migration role that owns subscribed tables, or as a role that is a member of all relevant owner roles. This privilege belongs to the database deployment path, not the long-lived pgstream runtime role. -The package accepts trusted SQL through `when_clause`, `payload_extensions`, and `metadata_extensions`. Only trusted database deployment roles should be able to change subscriptions or execute `set_subscriptions()`. +The package accepts trusted SQL through `when_clause`, `payload_extensions`, and `metadata_extensions`. Only trusted database deployment roles should be able to change subscriptions. ## Install @@ -30,29 +30,19 @@ The installer needs: - `INSERT` on `pgstream.events`. - Permission to create and own `pgstream_subscriptions`. -## Grant subscription deployment - -The installer revokes public execution of `set_subscriptions()`. Grant it only to your trusted migration role: - -```sql -grant usage on schema pgstream_subscriptions to application_migrator; -grant execute on function pgstream_subscriptions.set_subscriptions( - bigint, - pgstream_subscriptions.subscriptions[] -) to application_migrator; -``` - The pgstream runtime role does not need access to this schema or ownership of application tables. -## Reconcile subscriptions +## Optional reconciliation helper + +The package does not install `set_subscriptions()`. If you want to reconcile the complete desired subscription set for a stream, copy the optional [`examples/set_subscriptions.sql`](https://github.com/psteinroe/postgres-stream/blob/main/extensions/subscriptions/examples/set_subscriptions.sql) function into your own migrations and adapt it as needed. -Use `pgstream_subscriptions.set_subscriptions()` to supply the complete desired set for a stream. It inserts missing definitions, updates changed definitions, and deletes omitted definitions. Unchanged rows are not written, avoiding unnecessary trigger recreation. +The migration role that creates the function already owns it and can execute it; no additional `EXECUTE` grant is needed. Add grants only when intentionally delegating the helper to another trusted role. -See [Subscriptions](../concepts/subscriptions.md) and the [complete SQL example](https://github.com/psteinroe/postgres-stream/blob/main/extensions/subscriptions/examples/set_subscriptions.sql). +See [Subscriptions](../concepts/subscriptions.md) for its behavior. ## Customize -The package is intentionally plain SQL rather than a PostgreSQL `CREATE EXTENSION` package. You may copy and change: +The package is intentionally plain SQL rather than a Postgres `CREATE EXTENSION` package. You may copy and change: - The `pgstream_subscriptions` schema name. - Ownership and grants. diff --git a/docs/guides/upgrade-subscriptions.md b/docs/guides/upgrade-subscriptions.md index ecff150..4dc07b2 100644 --- a/docs/guides/upgrade-subscriptions.md +++ b/docs/guides/upgrade-subscriptions.md @@ -9,7 +9,7 @@ The breaking-change release stops installing subscription SQL through the pgstre 3. Identify the owner of every subscribed table. 4. Connect as a role that can administer the legacy pgstream objects and is the owner, or a member of the owner role, for every subscribed table. -The migration preserves subscription rows and target triggers, but future calls to `set_subscriptions()` need the package owner to drop and recreate those triggers. +The migration preserves subscription rows and target triggers. Future subscription changes run under the new package owner, which must be able to drop and recreate those triggers. ## Run the upgrade @@ -27,7 +27,6 @@ The script runs in a transaction and: 2. Moves the subscription enum, table, helper functions, and generated functions with `ALTER ... SET SCHEMA`. 3. Transfers ownership to the role running the migration. 4. Replaces the coordinator body so future trigger rebuilds use the new schema. -5. Adds `pgstream_subscriptions.set_subscriptions()`. `ALTER ... SET SCHEMA` preserves table rows, object OIDs, and dependencies. Existing subscription UUIDs and target-trigger function references are not recreated. @@ -51,4 +50,4 @@ After verification, deploy the new pgstream version. Its core migration accepts ## Rollback -The upgrade is atomic before commit. If it fails, PostgreSQL rolls back the schema and ownership moves. After commit, restore the database backup or write a reverse migration appropriate for your customized ownership and schema choices. +The upgrade is atomic before commit. If it fails, Postgres rolls back the schema and ownership moves. After commit, restore the database backup or write a reverse migration appropriate for your customized ownership and schema choices. diff --git a/docs/index.md b/docs/index.md index f5498f1..44e0bb9 100644 --- a/docs/index.md +++ b/docs/index.md @@ -25,7 +25,7 @@ Events are inserted into the `pgstream.events` table and streamed via logical re **Two ways to create events:** -1. **Subscriptions** (optional) - Install the user-managed SQL package to capture table changes with triggers +1. **Subscriptions** (optional) - Managed triggers 2. **Manual inserts** - Insert directly into `pgstream.events` from your application or database functions ## Trade-offs diff --git a/docs/reference/event-format.md b/docs/reference/event-format.md index 468468e..fb6c55f 100644 --- a/docs/reference/event-format.md +++ b/docs/reference/event-format.md @@ -170,9 +170,9 @@ They don't have trigger metadata unless you include it. ## Data Type Mapping -PostgreSQL types are serialized to JSON: +Postgres types are serialized to JSON: -| PostgreSQL | JSON | +| Postgres | JSON | |------------|------| | integer, bigint | number | | numeric, decimal | number | diff --git a/extensions/README.md b/extensions/README.md index 4b85804..48b79de 100644 --- a/extensions/README.md +++ b/extensions/README.md @@ -1,7 +1,7 @@ # Database extensions This directory contains optional, user-installed SQL packages for pgstream. -These are not PostgreSQL `CREATE EXTENSION` packages and are never installed or +These are not Postgres `CREATE EXTENSION` packages and are never installed or migrated by the pgstream daemon. Copy the relevant SQL into your application's migration system, review it, and diff --git a/extensions/subscriptions/README.md b/extensions/subscriptions/README.md index 5241039..c85a16e 100644 --- a/extensions/subscriptions/README.md +++ b/extensions/subscriptions/README.md @@ -8,7 +8,7 @@ by the pgstream daemon. ## Why installation is separate Changing subscriptions drops and recreates triggers on their target tables. -PostgreSQL requires the role dropping a trigger to own its table. Install this +Postgres requires the role dropping a trigger to own its table. Install this package as the application migration role that owns the subscribed tables (or a role that is a member of their owner roles). The long-lived pgstream runtime role does not need application-table ownership. @@ -36,19 +36,16 @@ The installing role needs `USAGE` on `pgstream` and `INSERT` on `pgstream.events`. It also needs ownership of every target table so future subscription changes can drop their triggers. -`set_subscriptions` is not executable by `PUBLIC`. Grant it only to the trusted -role that deploys subscription definitions: +## Optional reconciliation helper -```sql -grant usage on schema pgstream_subscriptions to application_migrator; -grant execute on function pgstream_subscriptions.set_subscriptions( - bigint, - pgstream_subscriptions.subscriptions[] -) to application_migrator; -``` +The package does not install `set_subscriptions()`. The optional +[`examples/set_subscriptions.sql`](examples/set_subscriptions.sql) defines the +helper documented in the subscription guide. Copy it into your own migrations +and adapt its interface or security settings as needed. -See [`examples/set_subscriptions.sql`](examples/set_subscriptions.sql) for a -complete reconciliation call. +The migration role that creates the function owns it and can execute it without +an additional grant. Add grants only if you intentionally delegate execution to +another role. ## Upgrade from pgstream 0.1 @@ -57,7 +54,7 @@ Before starting the breaking-change pgstream version, copy and run can administer both the legacy pgstream objects and all subscribed tables. The migration moves objects with `ALTER ... SET SCHEMA`. It does not copy -subscription rows or recreate target triggers: PostgreSQL preserves object OIDs +subscription rows or recreate target triggers: Postgres preserves object OIDs and dependencies while moving them. It updates ownership and replaces only the coordinator function body so future subscription changes use the new schema. diff --git a/extensions/subscriptions/examples/set_subscriptions.sql b/extensions/subscriptions/examples/set_subscriptions.sql index 5f29a6c..400420d 100644 --- a/extensions/subscriptions/examples/set_subscriptions.sql +++ b/extensions/subscriptions/examples/set_subscriptions.sql @@ -1,20 +1,72 @@ --- This array is the complete desired subscription set for stream 1. Existing --- definitions omitted from the array are deleted, which removes their triggers. -SELECT pgstream_subscriptions.set_subscriptions( - 1, - ARRAY[ - ROW( - NULL::uuid, - 'user-created', - 1::bigint, - 'INSERT'::pgstream_subscriptions.operation_type, - 'public', - 'users', - 'new.email_verified = true', - ARRAY['id', 'email', 'created_at']::text[], - '{"topic":"users"}'::jsonb, - '[]'::jsonb, - '[]'::jsonb - )::pgstream_subscriptions.subscriptions - ] -); +-- Optional reconciliation helper. This function is not installed by the +-- subscription package. Copy it into your own migrations and adapt it as needed. + +-- Reconcile the complete desired subscription set for one stream. Rows that did +-- not change are left untouched so their database triggers are not rebuilt. +CREATE OR REPLACE FUNCTION pgstream_subscriptions.set_subscriptions( + p_stream_id bigint, + p_subscriptions pgstream_subscriptions.subscriptions[] +) +RETURNS void +LANGUAGE plpgsql +SECURITY DEFINER +SET search_path = '' +AS $$ +begin + merge into pgstream_subscriptions.subscriptions as target + using ( + select desired.* + from unnest(p_subscriptions) as desired + ) as source + on target.key = source.key and target.stream_id = p_stream_id + when matched and ( + target.operation is distinct from source.operation or + target.schema_name is distinct from source.schema_name or + target.table_name is distinct from source.table_name or + target.when_clause is distinct from source.when_clause or + target.column_names is distinct from source.column_names or + target.metadata is distinct from source.metadata or + target.payload_extensions is distinct from source.payload_extensions or + target.metadata_extensions is distinct from source.metadata_extensions + ) then update set + operation = source.operation, + schema_name = source.schema_name, + table_name = source.table_name, + when_clause = source.when_clause, + column_names = source.column_names, + metadata = source.metadata, + payload_extensions = source.payload_extensions, + metadata_extensions = source.metadata_extensions + when not matched then insert ( + key, + stream_id, + operation, + schema_name, + table_name, + when_clause, + column_names, + metadata, + payload_extensions, + metadata_extensions + ) values ( + source.key, + p_stream_id, + source.operation, + source.schema_name, + source.table_name, + source.when_clause, + source.column_names, + source.metadata, + source.payload_extensions, + source.metadata_extensions + ); + + delete from pgstream_subscriptions.subscriptions as existing + where existing.stream_id = p_stream_id + and not exists ( + select 1 + from unnest(p_subscriptions) as desired + where desired.key = existing.key + ); +end; +$$; diff --git a/extensions/subscriptions/migrations/0001_create_pgstream_subscriptions.sql b/extensions/subscriptions/migrations/0001_create_pgstream_subscriptions.sql index a37a26e..2c1697a 100644 --- a/extensions/subscriptions/migrations/0001_create_pgstream_subscriptions.sql +++ b/extensions/subscriptions/migrations/0001_create_pgstream_subscriptions.sql @@ -251,78 +251,3 @@ $_$; CREATE TRIGGER sync_database_trigger AFTER INSERT OR DELETE OR UPDATE ON pgstream_subscriptions.subscriptions FOR EACH ROW EXECUTE FUNCTION pgstream_subscriptions.sync_database_trigger(); - --- Reconcile the complete desired subscription set for one stream. Rows that did --- not change are left untouched so their database triggers are not rebuilt. -CREATE OR REPLACE FUNCTION pgstream_subscriptions.set_subscriptions( - p_stream_id bigint, - p_subscriptions pgstream_subscriptions.subscriptions[] -) -RETURNS void -LANGUAGE plpgsql -SECURITY DEFINER -SET search_path = '' -AS $$ -begin - merge into pgstream_subscriptions.subscriptions as target - using ( - select desired.* - from unnest(p_subscriptions) as desired - ) as source - on target.key = source.key and target.stream_id = p_stream_id - when matched and ( - target.operation is distinct from source.operation or - target.schema_name is distinct from source.schema_name or - target.table_name is distinct from source.table_name or - target.when_clause is distinct from source.when_clause or - target.column_names is distinct from source.column_names or - target.metadata is distinct from source.metadata or - target.payload_extensions is distinct from source.payload_extensions or - target.metadata_extensions is distinct from source.metadata_extensions - ) then update set - operation = source.operation, - schema_name = source.schema_name, - table_name = source.table_name, - when_clause = source.when_clause, - column_names = source.column_names, - metadata = source.metadata, - payload_extensions = source.payload_extensions, - metadata_extensions = source.metadata_extensions - when not matched then insert ( - key, - stream_id, - operation, - schema_name, - table_name, - when_clause, - column_names, - metadata, - payload_extensions, - metadata_extensions - ) values ( - source.key, - p_stream_id, - source.operation, - source.schema_name, - source.table_name, - source.when_clause, - source.column_names, - source.metadata, - source.payload_extensions, - source.metadata_extensions - ); - - delete from pgstream_subscriptions.subscriptions as existing - where existing.stream_id = p_stream_id - and not exists ( - select 1 - from unnest(p_subscriptions) as desired - where desired.key = existing.key - ); -end; -$$; - -REVOKE ALL ON FUNCTION pgstream_subscriptions.set_subscriptions( - bigint, - pgstream_subscriptions.subscriptions[] -) FROM PUBLIC; diff --git a/extensions/subscriptions/upgrades/from-pgstream-0.1.sql b/extensions/subscriptions/upgrades/from-pgstream-0.1.sql index 06e8a24..ef4760d 100644 --- a/extensions/subscriptions/upgrades/from-pgstream-0.1.sql +++ b/extensions/subscriptions/upgrades/from-pgstream-0.1.sql @@ -198,79 +198,4 @@ begin end $_$; --- Reconcile the complete desired subscription set for one stream. Rows that did --- not change are left untouched so their database triggers are not rebuilt. -CREATE OR REPLACE FUNCTION pgstream_subscriptions.set_subscriptions( - p_stream_id bigint, - p_subscriptions pgstream_subscriptions.subscriptions[] -) -RETURNS void -LANGUAGE plpgsql -SECURITY DEFINER -SET search_path = '' -AS $$ -begin - merge into pgstream_subscriptions.subscriptions as target - using ( - select desired.* - from unnest(p_subscriptions) as desired - ) as source - on target.key = source.key and target.stream_id = p_stream_id - when matched and ( - target.operation is distinct from source.operation or - target.schema_name is distinct from source.schema_name or - target.table_name is distinct from source.table_name or - target.when_clause is distinct from source.when_clause or - target.column_names is distinct from source.column_names or - target.metadata is distinct from source.metadata or - target.payload_extensions is distinct from source.payload_extensions or - target.metadata_extensions is distinct from source.metadata_extensions - ) then update set - operation = source.operation, - schema_name = source.schema_name, - table_name = source.table_name, - when_clause = source.when_clause, - column_names = source.column_names, - metadata = source.metadata, - payload_extensions = source.payload_extensions, - metadata_extensions = source.metadata_extensions - when not matched then insert ( - key, - stream_id, - operation, - schema_name, - table_name, - when_clause, - column_names, - metadata, - payload_extensions, - metadata_extensions - ) values ( - source.key, - p_stream_id, - source.operation, - source.schema_name, - source.table_name, - source.when_clause, - source.column_names, - source.metadata, - source.payload_extensions, - source.metadata_extensions - ); - - delete from pgstream_subscriptions.subscriptions as existing - where existing.stream_id = p_stream_id - and not exists ( - select 1 - from unnest(p_subscriptions) as desired - where desired.key = existing.key - ); -end; -$$; - -REVOKE ALL ON FUNCTION pgstream_subscriptions.set_subscriptions( - bigint, - pgstream_subscriptions.subscriptions[] -) FROM PUBLIC; - COMMIT; diff --git a/implementation-notes.md b/implementation-notes.md deleted file mode 100644 index 17fc786..0000000 --- a/implementation-notes.md +++ /dev/null @@ -1,23 +0,0 @@ -# Implementation Notes - -Running notes on how the subscription extraction interprets or diverges from the agreed plan. - -## Design decisions - -- The optional package lives under `extensions/subscriptions/`; `extensions/README.md` explicitly distinguishes these user-managed SQL packages from PostgreSQL `CREATE EXTENSION` packages. -- Subscription objects use the `pgstream_subscriptions` schema, while emitted events continue to target the core `pgstream.events` table. -- Existing SQLx migration files remain unchanged. A forward core migration removes an empty legacy installation, accepts an already-moved installation, and blocks upgrades that still contain legacy subscription rows. -- The upgrade migration moves existing objects with `ALTER ... SET SCHEMA` so subscription rows, UUIDs, generated function OIDs, and target-trigger dependencies are preserved. - -## Deviations - -- None yet. - -## Tradeoffs - -- The package keeps the existing drop-and-recreate trigger model. This intentionally requires the SQL package owner to own subscribed tables, but removes that requirement from the long-lived pgstream runtime role. -- The package migrations are plain, editable SQL and are not executed or versioned by the Rust daemon. This maximizes compatibility with user migration systems at the cost of centralized automatic upgrades. - -## Open questions - -- None yet. diff --git a/tests/subscription_extension_tests.rs b/tests/subscription_extension_tests.rs index 540c0c9..7709155 100644 --- a/tests/subscription_extension_tests.rs +++ b/tests/subscription_extension_tests.rs @@ -9,6 +9,8 @@ const EXTRACT_SUBSCRIPTIONS: &str = include_str!("../migrations/1786434463000_extract_subscriptions.sql"); const INSTALL_SUBSCRIPTIONS: &str = include_str!("../extensions/subscriptions/migrations/0001_create_pgstream_subscriptions.sql"); +const SET_SUBSCRIPTIONS_EXAMPLE: &str = + include_str!("../extensions/subscriptions/examples/set_subscriptions.sql"); const UPGRADE_SUBSCRIPTIONS: &str = include_str!("../extensions/subscriptions/upgrades/from-pgstream-0.1.sql"); const UNINSTALL_SUBSCRIPTIONS: &str = include_str!("../extensions/subscriptions/uninstall.sql"); @@ -92,10 +94,22 @@ async fn core_migrations_do_not_install_subscriptions() { } #[tokio::test(flavor = "multi_thread")] -async fn set_subscriptions_reconciles_and_skips_unchanged_definitions() { +async fn optional_set_subscriptions_reconciles_and_skips_unchanged_definitions() { let database = TestDatabase::spawn_with_subscriptions().await; create_users_table(&database).await; + let installed_by_default: bool = sqlx::query_scalar( + "select to_regprocedure('pgstream_subscriptions.set_subscriptions(bigint, pgstream_subscriptions.subscriptions[])') is not null", + ) + .fetch_one(&database.pool) + .await + .expect("Failed to inspect optional helper"); + assert!(!installed_by_default); + + sqlx::raw_sql(SET_SUBSCRIPTIONS_EXAMPLE) + .execute(&database.pool) + .await + .expect("Failed to install optional set_subscriptions example"); set_user_subscription(&database).await; let initial_trigger_oid = insert_trigger_oid(&database).await; @@ -276,6 +290,10 @@ async fn package_owner_manages_triggers_without_runtime_table_privileges() { async fn uninstall_removes_package_and_managed_triggers() { let database = TestDatabase::spawn_with_subscriptions().await; create_users_table(&database).await; + sqlx::raw_sql(SET_SUBSCRIPTIONS_EXAMPLE) + .execute(&database.pool) + .await + .expect("Failed to install optional set_subscriptions example"); set_user_subscription(&database).await; sqlx::raw_sql(UNINSTALL_SUBSCRIPTIONS) From cd9c9a53cb1b9c32b533ecf924662bf2cd3d557c Mon Sep 17 00:00:00 2001 From: psteinroe Date: Tue, 11 Aug 2026 09:13:15 +0000 Subject: [PATCH 3/3] test(migrations): exercise subscription upgrade via SQLx --- tests/subscription_extension_tests.rs | 112 ++++++++++++++++++++------ 1 file changed, 86 insertions(+), 26 deletions(-) diff --git a/tests/subscription_extension_tests.rs b/tests/subscription_extension_tests.rs index 7709155..0ea6c44 100644 --- a/tests/subscription_extension_tests.rs +++ b/tests/subscription_extension_tests.rs @@ -1,12 +1,10 @@ -use postgres_stream::test_utils::TestDatabase; -use sqlx::Row; - -const LEGACY_INIT: &str = include_str!("../migrations/1765364029646_init.sql"); -const LEGACY_LSN: &str = include_str!("../migrations/1766831075000_add_lsn.sql"); -const LEGACY_METADATA: &str = - include_str!("../migrations/1767795878880_add_metadata_extensions.sql"); -const EXTRACT_SUBSCRIPTIONS: &str = - include_str!("../migrations/1786434463000_extract_subscriptions.sql"); +use std::borrow::Cow; + +use postgres_stream::{migrations::migrate_pgstream, test_utils::TestDatabase}; +use sqlx::{Row, migrate::Migrator}; + +const EXTRACT_SUBSCRIPTIONS_VERSION: i64 = 1_786_434_463_000; +static CORE_MIGRATOR: Migrator = sqlx::migrate!("./migrations"); const INSTALL_SUBSCRIPTIONS: &str = include_str!("../extensions/subscriptions/migrations/0001_create_pgstream_subscriptions.sql"); const SET_SUBSCRIPTIONS_EXAMPLE: &str = @@ -15,6 +13,34 @@ const UPGRADE_SUBSCRIPTIONS: &str = include_str!("../extensions/subscriptions/upgrades/from-pgstream-0.1.sql"); const UNINSTALL_SUBSCRIPTIONS: &str = include_str!("../extensions/subscriptions/uninstall.sql"); +async fn apply_legacy_core_migrations(database: &TestDatabase) { + sqlx::query("create schema if not exists pgstream") + .execute(&database.pool) + .await + .expect("Failed to create pgstream schema"); + + let mut connection = database.pool.acquire().await.expect("Failed to connect"); + sqlx::query("set search_path = 'pgstream'") + .execute(&mut *connection) + .await + .expect("Failed to set migration search path"); + + let migrator = Migrator { + migrations: Cow::Owned( + CORE_MIGRATOR + .iter() + .filter(|migration| migration.version < EXTRACT_SUBSCRIPTIONS_VERSION) + .cloned() + .collect(), + ), + ..Migrator::DEFAULT + }; + migrator + .run_direct(&mut *connection) + .await + .expect("Failed to apply legacy core migrations through SQLx"); +} + async fn create_users_table(database: &TestDatabase) { sqlx::query( r#" @@ -91,6 +117,15 @@ async fn core_migrations_do_not_install_subscriptions() { assert_eq!(row.get::, _>("legacy"), None); assert_eq!(row.get::, _>("extracted"), None); + + let extraction_applied: bool = sqlx::query_scalar( + "select exists(select 1 from pgstream._sqlx_migrations where version = $1 and success)", + ) + .bind(EXTRACT_SUBSCRIPTIONS_VERSION) + .fetch_one(&database.pool) + .await + .expect("Failed to inspect migration history"); + assert!(extraction_applied); } #[tokio::test(flavor = "multi_thread")] @@ -153,12 +188,7 @@ async fn optional_set_subscriptions_reconciles_and_skips_unchanged_definitions() async fn core_migration_blocks_nonempty_legacy_subscriptions() { let database = TestDatabase::spawn_without_migrations().await; - for migration in [LEGACY_INIT, LEGACY_LSN, LEGACY_METADATA] { - sqlx::raw_sql(migration) - .execute(&database.pool) - .await - .expect("Failed to install legacy pgstream schema"); - } + apply_legacy_core_migrations(&database).await; create_users_table(&database).await; sqlx::query( @@ -174,8 +204,7 @@ async fn core_migration_blocks_nonempty_legacy_subscriptions() { .await .expect("Failed to create legacy subscription"); - let error = sqlx::raw_sql(EXTRACT_SUBSCRIPTIONS) - .execute(&database.pool) + let error = migrate_pgstream(&database.config) .await .expect_err("Core migration should block legacy subscriptions"); assert!( @@ -190,6 +219,14 @@ async fn core_migration_blocks_nonempty_legacy_subscriptions() { .expect("Failed to verify preserved legacy subscriptions"), 1 ); + let extraction_applied: bool = sqlx::query_scalar( + "select exists(select 1 from pgstream._sqlx_migrations where version = $1)", + ) + .bind(EXTRACT_SUBSCRIPTIONS_VERSION) + .fetch_one(&database.pool) + .await + .expect("Failed to inspect failed migration history"); + assert!(!extraction_applied); } #[tokio::test(flavor = "multi_thread")] @@ -322,12 +359,7 @@ async fn uninstall_removes_package_and_managed_triggers() { async fn upgrade_moves_subscriptions_without_recreating_rows_or_triggers() { let database = TestDatabase::spawn_without_migrations().await; - for migration in [LEGACY_INIT, LEGACY_LSN, LEGACY_METADATA] { - sqlx::raw_sql(migration) - .execute(&database.pool) - .await - .expect("Failed to install legacy pgstream schema"); - } + apply_legacy_core_migrations(&database).await; database.ensure_today_partition().await; create_users_table(&database).await; @@ -399,11 +431,19 @@ async fn upgrade_moves_subscriptions_without_recreating_rows_or_triggers() { "pgstream_subscriptions" ); - // The core extraction migration must accept an already-upgraded installation. - sqlx::raw_sql(EXTRACT_SUBSCRIPTIONS) - .execute(&database.pool) + // The actual SQLx runner must accept an already-upgraded installation and + // record the core extraction migration. + migrate_pgstream(&database.config) .await .expect("Core extraction migration removed the user-managed package"); + let extraction_applied: bool = sqlx::query_scalar( + "select exists(select 1 from pgstream._sqlx_migrations where version = $1 and success)", + ) + .bind(EXTRACT_SUBSCRIPTIONS_VERSION) + .fetch_one(&database.pool) + .await + .expect("Failed to inspect migration history"); + assert!(extraction_applied); sqlx::query("insert into public.users (email) values ('upgrade@example.com')") .execute(&database.pool) @@ -415,4 +455,24 @@ async fn upgrade_moves_subscriptions_without_recreating_rows_or_triggers() { .await .expect("Moved trigger did not emit an event"); assert_eq!(event_name, "user-created"); + + sqlx::query( + "update pgstream_subscriptions.subscriptions set column_names = array['id'] where id = $1", + ) + .bind(subscription_id) + .execute(&database.pool) + .await + .expect("Failed to rebuild a moved subscription trigger"); + let rebuilt_trigger_oid = insert_trigger_oid(&database).await; + assert_ne!(rebuilt_trigger_oid, trigger_oid); + + sqlx::query("insert into public.users (email) values ('rebuilt@example.com')") + .execute(&database.pool) + .await + .expect("Rebuilt trigger failed"); + let event_count: i64 = sqlx::query_scalar("select count(*) from pgstream.events") + .fetch_one(&database.pool) + .await + .expect("Failed to count upgraded subscription events"); + assert_eq!(event_count, 2); }