Detect and report schema (DDL) changes

.gitignore

@@ -2,3 +2,4 @@
 *.exe
 /gateway
 tmp/
+dump.rdb

CLAUDE.md

@@ -59,6 +59,12 @@ Consequences to keep in mind:
   placeholder strings) and listed in `UnchangedToastColumns`. Sinks handle
   this with partial updates (Meilisearch `UpdateDocuments`, RedisJSON
   `JSON.MERGE`).
+- Schema (DDL) changes are detected by diffing pgoutput relation
+  descriptions, which PostgreSQL re-sends before a changed table's next row
+  change. They are **best-effort**, not at-least-once: a reconnect
+  re-describes tables with nothing to diff against, so a missed change cannot
+  be redelivered. Observers must therefore never fail the WAL stream —
+  `ports.SchemaObserver` returns no error by design.
 - Backfill emits `READ` events for existing rows, interleaved with the live
   stream via WAL watermarks (`pg_logical_emit_message`, prefix
   `gtc-backfill`). Chunk rows superseded by live events between the low and
@@ -95,6 +101,7 @@ Consequences to keep in mind:
 | CDC_DLQ_THRESHOLD | 3 | Failure cycles for the same event before it is parked |
 | CDC_DLQ_MAX_ENTRIES | 10000 | DLQ cap; when full, parking fails and the pipeline stalls |
 | CDC_MASK_HMAC_KEY | - | Secret key for the hmac256 mask strategy (startup fails if a transform uses hmac256 without it) |
+| CDC_SCHEMA_EVENTS | true | Publish detected DDL changes to the <prefix>:schema Redis stream (detection/metrics/API always on) |
 | SINK_CONFIG_FILE | - | Path to per-table sink YAML (see config/sinks.example.yaml). Without it, Redis sinks mirror all tables |
 | REDIS_URL | - | Redis connection (enables stream sink if set) |
 | REDIS_STREAM_PREFIX | cdc | Stream key prefix |
@@ -147,6 +154,8 @@ internal/
     transform/         # Declarative per-sink transforms (CEL filters, masking)
     resilience/        # Circuit breaker and retry wrapper for sinks
     dlq/               # Dead-letter queue (Redis store, parking decorator, triage)
+    doctor/            # `gtc doctor` preflight checks
+    schema/            # DDL change history, logging, and Redis notification stream
     server/            # HTTP server: health/readiness/metrics, APIs, dashboard
     metrics/           # Prometheus metrics
 ui/                    # React 19 + Tailwind dashboard; dist/ committed + embedded

README.md

@@ -48,6 +48,11 @@ PostgreSQL ──logical replication──▶ GTC ──▶ Redis Streams   (eve
   `last4`, `redact`, `null`), column dropping, and CEL row filters,
   configured per sink and per table in YAML. The stream can carry full rows while the search index gets
   masked ones — no plugin code, no SMT classes.
+- **Schema-change awareness** — GTC notices when a published table gains,
+  loses or retypes a column, changes replica identity, or is renamed, and
+  reports it on the dashboard, in `/api/schema`, in Prometheus, and on a
+  Redis notification stream — flagging changes that can break consumers. No
+  DDL triggers, no schema-history topic, no superuser.
 - **First-class transactional outbox** — write domain events to an outbox
   table in the same transaction as your business writes, and GTC publishes
   them to per-topic Redis streams with optional delete-after-publish. Five
@@ -62,8 +67,9 @@ PostgreSQL ──logical replication──▶ GTC ──▶ Redis Streams   (eve
   silently dropped.
 - **Embedded live dashboard** — a React dashboard served from the binary at
   `/` (PocketBase-style, no separate deployment): live throughput, WAL lag,
-  sink health and breaker states, per-table activity, backfill progress, and
-  DLQ triage with one-click retry/discard. Light and dark themes.
+  sink health and breaker states, per-table activity, schema-change history,
+  backfill progress, and DLQ triage with one-click retry/discard. Light and
+  dark themes.
 - **Observability** — Prometheus metrics (events, per-sink latency, errors,
   retries, breaker state, WAL lag in bytes), health and readiness endpoints,
   structured JSON logs.
@@ -224,6 +230,7 @@ lives in an optional YAML file.
 | `CDC_DLQ_THRESHOLD` | `3` | Failure cycles for the same event before it is parked |
 | `CDC_DLQ_MAX_ENTRIES` | `10000` | DLQ cap; when full, parking fails and the pipeline stalls |
 | `CDC_MASK_HMAC_KEY` | – | Secret key for the `hmac256` mask strategy (required when a transform uses it) |
+| `CDC_SCHEMA_EVENTS` | `true` | Publish detected DDL changes to the `<prefix>:schema` Redis stream (detection, metrics and `/api/schema` are always on) |
 | `SINK_CONFIG_FILE` | – | Path to the per-table sink YAML (below) |
 | `REDIS_URL` | – | Enables the Redis Stream sink |
 | `REDIS_STREAM_PREFIX` | `cdc` | Stream key prefix |
@@ -349,6 +356,55 @@ Behavior worth knowing:
 - Delivery is at-least-once like everything else; consumers dedupe by the
   `id` field.
 
+## Schema changes (DDL)
+
+PostgreSQL's logical decoding does not stream DDL statements — there is no
+`ALTER TABLE` event to subscribe to. What it *does* do is re-describe a
+published table whenever its shape has changed, immediately before that
+table's next row change. GTC diffs those descriptions and reports the
+**effect** of the DDL:
+
+| Detected | Kind | Breaking? |
+|----------|------|-----------|
+| New column | `column_added` | no — additive |
+| Column removed | `column_dropped` | yes |
+| Column type altered | `column_type_changed` | yes |
+| Primary key / identity index replaced | `key_changed` | yes |
+| `REPLICA IDENTITY` altered | `replica_identity_changed` | yes |
+| Table renamed or moved schema | `table_renamed` | yes |
+
+Every change is counted in `cdc_schema_changes_total{schema,table,kind}`,
+logged (breaking ones at WARN, so existing log alerting picks them up), kept
+in a rolling in-memory history behind `GET /api/schema`, and shown on the
+dashboard's **Schema** page. With `REDIS_URL` set, it is also published to
+the `<REDIS_STREAM_PREFIX>:schema` stream so consumers can react
+programmatically:
+
+```bash
+redis-cli XRANGE cdc:schema - +
+# fields: table, lsn, breaking, payload (full JSON diff)
+```
+
+Set `CDC_SCHEMA_EVENTS=false` to keep detection, metrics and the API while
+skipping the Redis stream.
+
+None of this needs DDL triggers, a schema-history topic, or superuser — it
+falls out of the replication stream you are already reading. Two honest
+limitations follow from that:
+
+- **Detection is tied to the next row change.** A table altered at 02:00 and
+  first written at 09:00 reports its change at 09:00. The LSN on the change
+  is where GTC *detected* it, not where the DDL committed.
+- **Several statements can collapse into one change.** If a column is added
+  and another dropped between two row changes, they arrive as a single
+  combined change — the net effect, not a statement log.
+
+Unlike row events, schema notifications are **best-effort**: a change cannot
+be re-derived on redelivery (a reconnect re-describes the table with nothing
+to diff against), so a failed Redis publish is counted in
+`cdc_schema_publish_errors_total` and logged rather than stalling the
+pipeline. Row delivery keeps its at-least-once guarantee regardless.
+
 ## Backfill and replay
 
 GTC syncs *existing* data, not just new changes. When the replication slot is
@@ -453,6 +509,7 @@ networks.
 |----------|---------|
 | `GET /` | Embedded dashboard |
 | `GET /api/stats` | Dashboard's JSON snapshot (uptime, lag, sinks, tables, backfill, DLQ) |
+| `GET /api/schema` | Recently detected schema (DDL) changes, newest first |
 | `GET /health` | Liveness — process is up |
 | `GET /readiness` | Readiness — live replication stream **and** all sinks healthy |
 | `GET /metrics` | Prometheus metrics |

cmd/gateway/modules.go

@@ -19,6 +19,7 @@ import (
 	"github.com/emoss08/gtc/internal/infrastructure/config"
 	"github.com/emoss08/gtc/internal/infrastructure/dlq"
 	"github.com/emoss08/gtc/internal/infrastructure/resilience"
+	"github.com/emoss08/gtc/internal/infrastructure/schema"
 	"github.com/emoss08/gtc/internal/infrastructure/server"
 	"github.com/emoss08/gtc/internal/infrastructure/transform"
 	"github.com/emoss08/gtc/ui"
@@ -34,6 +35,8 @@ func Module() fx.Option {
 			config.LoadSinksConfig,
 			newBackfillCoordinator,
 			newDLQManager,
+			newSchemaRecorder,
+			newSchemaObserver,
 			newWALReader,
 			func(r *wal.Reader) ports.WALReader { return r },
 			newSinks,
@@ -107,9 +110,46 @@ func newDLQManager(cfg *config.Config, logger *slog.Logger) (*dlq.Manager, error
 	}), nil
 }
 
+// newSchemaRecorder keeps the in-memory DDL history that /api/schema and the
+// dashboard read.
+func newSchemaRecorder(logger *slog.Logger) *schema.Recorder {
+	return schema.NewRecorder(logger)
+}
+
+// newSchemaObserver fans schema changes to the recorder and, when Redis is
+// configured, to the "<prefix>:schema" notification stream.
+func newSchemaObserver(
+	cfg *config.Config,
+	recorder *schema.Recorder,
+	lc fx.Lifecycle,
+	logger *slog.Logger,
+) (ports.SchemaObserver, error) {
+	observers := schema.Observers{recorder}
+
+	if rc := redissink.LoadConfig(); rc.Enabled && cfg.SchemaEvents {
+		publisher, err := schema.NewRedisPublisher(schema.RedisPublisherParams{
+			URL:    rc.URL,
+			Prefix: rc.StreamPrefix,
+			MaxLen: rc.MaxStreamLen,
+			Logger: logger,
+		})
+		if err != nil {
+			return nil, err
+		}
+		logger.Info("publishing schema changes", slog.String("stream", publisher.Stream()))
+		observers = append(observers, publisher)
+		lc.Append(fx.Hook{
+			OnStop: func(context.Context) error { return publisher.Close() },
+		})
+	}
+
+	return observers, nil
+}
+
 func newWALReader(
 	cfg *config.Config,
 	coordinator *backfill.Coordinator,
+	schemaObserver ports.SchemaObserver,
 	logger *slog.Logger,
 ) *wal.Reader {
 	walCfg := wal.Config{
@@ -121,6 +161,7 @@ func newWALReader(
 		AutoCreatePub:     cfg.AutoCreatePub,
 		SlotRetryInterval: cfg.SlotRetryInterval,
 		SlotRetryTimeout:  cfg.SlotRetryTimeout,
+		SchemaObserver:    schemaObserver,
 	}
 
 	if coordinator != nil {
@@ -326,6 +367,7 @@ func newHTTPServer(
 	dlqManager *dlq.Manager,
 	reader *wal.Reader,
 	health *server.HealthStatus,
+	schemaRecorder *schema.Recorder,
 	logger *slog.Logger,
 ) *server.Server {
 	var backfillManager ports.BackfillManager
@@ -343,6 +385,7 @@ func newHTTPServer(
 		Backfill: backfillManager,
 		DLQ:      dlqPort,
 		WAL:      reader,
+		Schema:   schemaRecorder,
 		Info: server.InstanceInfo{
 			Version:     version,
 			SlotName:    cfg.SlotName,

internal/adapters/primary/wal/decoder.go

@@ -23,6 +23,9 @@ type txnState struct {
 	inTxn      bool
 	xid        uint32
 	commitTime time.Time
+	// finalLSN is the transaction's commit position, used to place messages
+	// that PostgreSQL sends without a WAL position of their own.
+	finalLSN pglogrepl.LSN
 }
 
 func NewDecoder() *Decoder {
@@ -34,6 +37,9 @@ func NewDecoder() *Decoder {
 
 type DecodeResult struct {
 	Events []domain.CDCEvent
+	// SchemaChanges are DDL changes detected by diffing this position's
+	// relation descriptions against the previously seen ones.
+	SchemaChanges []domain.SchemaChange
 	// Message is a logical decoding message (pg_logical_emit_message)
 	// found at this stream position; used for backfill watermarks.
 	Message *LogicalMessage
@@ -109,6 +115,7 @@ func (d *Decoder) decodeWALData(walData []byte, lsn pglogrepl.LSN) (*DecodeResul
 		d.txn.inTxn = true
 		d.txn.xid = msg.Xid
 		d.txn.commitTime = msg.CommitTime
+		d.txn.finalLSN = msg.FinalLSN
 
 	case *pglogrepl.CommitMessage:
 		d.txn.inTxn = false
@@ -117,6 +124,14 @@ func (d *Decoder) decodeWALData(walData []byte, lsn pglogrepl.LSN) (*DecodeResul
 		result.AckLSN = msg.TransactionEndLSN
 
 	case *pglogrepl.RelationMessageV2:
+		// PostgreSQL re-sends a relation description whenever a published
+		// table's shape changed, just before the next row change for it.
+		// Diffing against the previous one is how GTC detects DDL.
+		if prev, ok := d.relations[msg.RelationID]; ok {
+			if change := d.diffRelation(prev, msg, lsn); change != nil {
+				result.SchemaChanges = append(result.SchemaChanges, *change)
+			}
+		}
 		d.relations[msg.RelationID] = msg
 
 	case *pglogrepl.InsertMessageV2:
@@ -175,6 +190,134 @@ func (d *Decoder) decodeWALData(walData []byte, lsn pglogrepl.LSN) (*DecodeResul
 	return result, nil
 }
 
+// diffRelation reports what changed between two descriptions of the same
+// table, or nil when they are identical (PostgreSQL re-sends an unchanged
+// relation after a reconnect, and that is not a DDL change).
+func (d *Decoder) diffRelation(
+	prev, cur *pglogrepl.RelationMessageV2,
+	lsn pglogrepl.LSN,
+) *domain.SchemaChange {
+	// PostgreSQL sends relation descriptions without a WAL position of their
+	// own. Fall back to the commit position of the transaction they arrived
+	// in, so the change can still be ordered against row events.
+	if lsn == 0 {
+		lsn = d.txn.finalLSN
+	}
+	change := domain.SchemaChange{
+		Schema:             cur.Namespace,
+		Table:              cur.RelationName,
+		RelationID:         cur.RelationID,
+		KeyColumns:         keyColumns(cur),
+		PreviousKeyColumns: keyColumns(prev),
+		ReplicaIdentity:    string(cur.ReplicaIdentity),
+		LSN:                lsn.String(),
+		TransactionID:      d.txn.xid,
+		Timestamp:          d.txn.commitTime,
+	}
+
+	if prev.Namespace != cur.Namespace || prev.RelationName != cur.RelationName {
+		change.PreviousSchema = prev.Namespace
+		change.PreviousTable = prev.RelationName
+		change.Kinds = append(change.Kinds, domain.SchemaChangeTableRenamed)
+	}
+
+	prevCols := make(map[string]*pglogrepl.RelationMessageColumn, len(prev.Columns))
+	for i := range prev.Columns {
+		prevCols[prev.Columns[i].Name] = prev.Columns[i]
+	}
+	curCols := make(map[string]*pglogrepl.RelationMessageColumn, len(cur.Columns))
+	for i := range cur.Columns {
+		curCols[cur.Columns[i].Name] = cur.Columns[i]
+	}
+
+	// Iterate the relation's own column order so output is deterministic.
+	for i := range cur.Columns {
+		col := cur.Columns[i]
+		before, existed := prevCols[col.Name]
+		if !existed {
+			change.AddedColumns = append(change.AddedColumns, d.columnDef(col))
+			continue
+		}
+		if before.DataType != col.DataType {
+			change.ChangedColumns = append(change.ChangedColumns, domain.ColumnTypeChange{
+				Name: col.Name,
+				From: d.columnDef(before),
+				To:   d.columnDef(col),
+			})
+		}
+	}
+	for i := range prev.Columns {
+		col := prev.Columns[i]
+		if _, stillThere := curCols[col.Name]; !stillThere {
+			change.DroppedColumns = append(change.DroppedColumns, d.columnDef(col))
+		}
+	}
+
+	if len(change.AddedColumns) > 0 {
+		change.Kinds = append(change.Kinds, domain.SchemaChangeColumnAdded)
+	}
+	if len(change.DroppedColumns) > 0 {
+		change.Kinds = append(change.Kinds, domain.SchemaChangeColumnDropped)
+	}
+	if len(change.ChangedColumns) > 0 {
+		change.Kinds = append(change.Kinds, domain.SchemaChangeColumnTypeChanged)
+	}
+	if prev.ReplicaIdentity != cur.ReplicaIdentity {
+		change.PreviousReplicaIdentity = string(prev.ReplicaIdentity)
+		change.Kinds = append(change.Kinds, domain.SchemaChangeReplicaIdentity)
+	} else if !equalStrings(change.PreviousKeyColumns, change.KeyColumns) {
+		// The identity setting is unchanged but its columns are not: the
+		// primary key or the identity index itself was altered.
+		change.Kinds = append(change.Kinds, domain.SchemaChangeKeyChanged)
+	}
+
+	if len(change.Kinds) == 0 {
+		return nil
+	}
+	return &change
+}
+
+func (d *Decoder) columnDef(col *pglogrepl.RelationMessageColumn) domain.ColumnDef {
+	return domain.ColumnDef{
+		Name:      col.Name,
+		Type:      d.typeName(col.DataType),
+		TypeOID:   col.DataType,
+		PartOfKey: col.Flags&1 == 1,
+	}
+}
+
+// typeName resolves a type OID to its PostgreSQL name, falling back to the
+// raw OID for types the client does not know (extensions, user-defined types).
+func (d *Decoder) typeName(oid uint32) string {
+	if dt, ok := d.typeMap.TypeForOID(oid); ok {
+		return dt.Name
+	}
+	return fmt.Sprintf("oid:%d", oid)
+}
+
+// keyColumns returns the columns PostgreSQL marks as the replica identity.
+func keyColumns(rel *pglogrepl.RelationMessageV2) []string {
+	var keys []string
+	for i := range rel.Columns {
+		if rel.Columns[i].Flags&1 == 1 {
+			keys = append(keys, rel.Columns[i].Name)
+		}
+	}
+	return keys
+}
+
+func equalStrings(a, b []string) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := range a {
+		if a[i] != b[i] {
+			return false
+		}
+	}
+	return true
+}
+
 func (d *Decoder) relation(relationID uint32) (*pglogrepl.RelationMessageV2, error) {
 	rel, ok := d.relations[relationID]
 	if !ok {