Logical Decoding

-- 1) Enable on the source
-- postgresql.conf:
--   wal_level = logical
--   max_replication_slots = 10
--   max_wal_senders = 10

-- 2) Create a publication for the tables you want
CREATE PUBLICATION shop_pub FOR TABLE orders, line_items;

-- Or all tables, or specific ones with column lists
-- CREATE PUBLICATION shop_pub FOR ALL TABLES;
-- CREATE PUBLICATION shop_pub FOR TABLE orders (id, status, total_cents);

-- 3) Pick a replication user
CREATE ROLE repl WITH REPLICATION LOGIN PASSWORD 'strong-secret';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO repl;
GRANT USAGE  ON SCHEMA public TO repl;

-- pg_hba.conf:
-- host replication repl 10.0.0.0/8 scram-sha-256

-- 4) Create a replication slot on the source
SELECT pg_create_logical_replication_slot('shop_slot', 'pgoutput');
-- The slot holds onto WAL until the consumer acknowledges it. If the consumer
-- disappears, WAL grows unbounded -- monitor + drop dead slots.

-- 5) Two common output plugins
-- pgoutput   built-in, binary protocol; used by Postgres logical replication
--            and Debezium 2.x
-- wal2json   JSON output; easy to consume with any language
SELECT pg_create_logical_replication_slot('shop_slot_json', 'wal2json');

-- 6) Peek at changes WITHOUT consuming them (debug)
SELECT * FROM pg_logical_slot_peek_changes('shop_slot_json', NULL, NULL,
  'pretty-print', '1');

-- 7) Consume them (advances the slot's confirmed flush LSN)
SELECT * FROM pg_logical_slot_get_changes('shop_slot_json', NULL, NULL);

-- 8) Postgres-to-Postgres logical replication (no Kafka needed)
-- On the SUBSCRIBER:
CREATE SUBSCRIPTION shop_sub
  CONNECTION 'host=source dbname=shop user=repl password=strong-secret'
  PUBLICATION shop_pub
  WITH (slot_name = 'shop_slot', create_slot = false, copy_data = true);

-- Inspect
SELECT subname, received_lsn, latest_end_lsn FROM pg_stat_subscription;

-- 9) Debezium for production CDC
-- Debezium reads pgoutput, normalises events, ships to Kafka/Redis Streams.
-- docker-compose excerpt:
--
-- debezium:
--   image: debezium/connect:2.5
--   environment:
--     BOOTSTRAP_SERVERS: kafka:9092
--     CONFIG_STORAGE_TOPIC: dbz_configs
--     OFFSET_STORAGE_TOPIC: dbz_offsets
--     STATUS_STORAGE_TOPIC: dbz_status
-- Then POST a connector config:
-- {
--   "name": "shop-orders",
--   "config": {
--     "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
--     "database.hostname": "source",
--     "database.dbname": "shop",
--     "database.user": "repl",
--     "database.password": "strong-secret",
--     "plugin.name": "pgoutput",
--     "slot.name": "debezium",
--     "publication.name": "shop_pub",
--     "topic.prefix": "shop"
--   }
-- }

-- 10) Monitor the slot — UNCONSUMED WAL grows here
SELECT slot_name, active,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) AS lag
FROM pg_replication_slots;

-- If lag balloons (consumer is down), either restart the consumer or
-- DROP the slot to release WAL (and accept that you've missed events).

-- 11) Caveats
-- - Logical decoding does NOT include DDL (schema changes). Handle separately.
-- - TOAST values without REPLICA IDENTITY FULL show as TOAST_PLACEHOLDER on
--   UPDATE; use REPLICA IDENTITY FULL on critical tables if you need old
--   values:  ALTER TABLE orders REPLICA IDENTITY FULL;
-- - Large transactions stream as one block; long-running tx delay all
--   downstream consumers until commit.

-- 12) Use cases
-- - Cache invalidation: stream orders changes -> wipe Redis keys
-- - Materialised search index: orders changes -> Elasticsearch bulk
-- - Audit log: any change -> append to an immutable store
-- - Outbox -> CDC -> message broker: 'transactional outbox' without polling
-- - Cross-DB replication: a transient sync to a new schema during a migration

-- Stream row-level changes to a downstream system.
CREATE PUBLICATION app_pub FOR ALL TABLES;
-- Subscriber
CREATE SUBSCRIPTION app_sub
    CONNECTION 'host=primary dbname=app'
    PUBLICATION app_pub;