refactor(subscriptions): extract user-managed SQL package

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

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:

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"}
   ]'

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;

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
 

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

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.

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.

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

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 |