diff --git a/README.md b/README.md index e2b39d1..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) - Define triggers that automatically capture table changes +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/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..4f61185 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,77 +99,21 @@ 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 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 -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; -$$; -``` +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/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..9211bf3 --- /dev/null +++ b/docs/guides/subscription-setup.md @@ -0,0 +1,52 @@ +# 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. 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. + +## 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`. + +The pgstream runtime role does not need access to this schema or ownership of application tables. + +## 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. + +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) for its behavior. + +## Customize + +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. +- 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..4dc07b2 --- /dev/null +++ b/docs/guides/upgrade-subscriptions.md @@ -0,0 +1,53 @@ +# 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. Future subscription changes run under the new package owner, which must be able 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. + +`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, 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 9370e63..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) - Define triggers that automatically capture table changes +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 f3116a3..fb6c55f 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 @@ -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/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..48b79de --- /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 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 +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..c85a16e --- /dev/null +++ b/extensions/subscriptions/README.md @@ -0,0 +1,70 @@ +# 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. +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. + +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. + +## Optional reconciliation helper + +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. + +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 + +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: 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. + +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..400420d --- /dev/null +++ b/extensions/subscriptions/examples/set_subscriptions.sql @@ -0,0 +1,72 @@ +-- 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 new file mode 100644 index 0000000..2c1697a --- /dev/null +++ b/extensions/subscriptions/migrations/0001_create_pgstream_subscriptions.sql @@ -0,0 +1,253 @@ +-- 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(); 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..ef4760d --- /dev/null +++ b/extensions/subscriptions/upgrades/from-pgstream-0.1.sql @@ -0,0 +1,201 @@ +-- 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 +$_$; + +COMMIT; 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..0ea6c44 --- /dev/null +++ b/tests/subscription_extension_tests.rs @@ -0,0 +1,478 @@ +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 = + 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"); + +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#" + 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); + + 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")] +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; + + 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; + + apply_legacy_core_migrations(&database).await; + 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 = migrate_pgstream(&database.config) + .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 + ); + 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")] +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; + 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) + .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; + + apply_legacy_core_migrations(&database).await; + + 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 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) + .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"); + + 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); +} 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",