Validate REPLICA IDENTITY FULL for delete-tracking streams at startup

AGENTS.md

@@ -93,8 +93,9 @@ TOML, secrets kept out of the file (see `docs/examples/config.toml`).
   (`mechanism`, `username`, `password_env`).
 - `[observability]` (optional; absent = off): `address`/`port` for a Prometheus
   `/metrics` plus `/healthz` and `/readyz` HTTP server.
-- Postgres needs `wal_level = logical` and `REPLICA IDENTITY FULL` on tracked
-  tables. Outboxx auto-creates the slot and publication.
+- Postgres needs `wal_level = logical`, plus `REPLICA IDENTITY FULL` on tables
+  whose stream tracks DELETE (validated at startup; UPDATE emits only the new
+  row, so it doesn't need it). Outboxx auto-creates the slot and publication.
 
 ## Conventions
 

README.md

@@ -91,7 +91,9 @@ GRANT USAGE, CREATE ON SCHEMA public TO outboxx_user;
 -- Grant table access (SELECT needed for logical replication)
 GRANT SELECT ON TABLE my_table TO outboxx_user;
 
--- Enable REPLICA IDENTITY FULL (required for capturing complete row data in UPDATE/DELETE)
+-- Enable REPLICA IDENTITY FULL so DELETE events carry the full row.
+-- Required (and validated at startup) for tables whose stream tracks DELETE.
+-- UPDATE emits only the new row, so it does not need this.
 ALTER TABLE my_table REPLICA IDENTITY FULL;
 ```
 

src/config/config.zig

@@ -110,6 +110,15 @@ pub const Stream = struct {
     source: StreamSource,
     flow: StreamFlow,
     sink: StreamSink,
+
+    /// Whether this stream captures DELETE. Config validation restricts operations
+    /// to the lowercase set, so an exact match is enough.
+    pub fn hasDeleteOperation(self: Stream) bool {
+        for (self.source.operations) |op| {
+            if (std.mem.eql(u8, op, "delete")) return true;
+        }
+        return false;
+    }
 };
 
 pub const TableFilter = struct {

src/config/config_test.zig

@@ -81,6 +81,24 @@ test "createTestDefault" {
     try testing.expectEqualStrings("localhost:9092", cfg.sink.kafka.?.brokers[0]);
 }
 
+test "Stream.hasDeleteOperation reflects the configured operations" {
+    const Stream = config.Stream;
+    const base: Stream = .{
+        .name = "s",
+        .source = .{ .resource = "users", .operations = &.{} },
+        .flow = .{ .format = "json" },
+        .sink = .{ .destination = "t", .routing_key = null },
+    };
+
+    var insert_update = base;
+    insert_update.source.operations = &.{ "insert", "update" };
+    try testing.expect(!insert_update.hasDeleteOperation());
+
+    var with_delete = base;
+    with_delete.source.operations = &.{ "insert", "delete" };
+    try testing.expect(with_delete.hasDeleteOperation());
+}
+
 test "supported adapter types are implemented" {
     try testing.expectEqual(@as(usize, 1), config.SupportedValues.SOURCE_TYPES.len);
     try testing.expectEqualStrings("postgres", config.SupportedValues.SOURCE_TYPES[0]);

src/main.zig

@@ -290,6 +290,15 @@ fn validatePostgres(allocator: std.mem.Allocator, cfg: Config, conninfo: []const
             printStatus("ERROR: Table validation failed for '{s}': {}\n", .{ stream.source.resource, err });
             return err;
         };
+
+        // Only a stream that tracks DELETE needs REPLICA IDENTITY FULL, so the
+        // deleted row carries all columns. Schema is fixed to "public" for now.
+        if (stream.hasDeleteOperation()) {
+            validator.checkReplicaIdentity("public", stream.source.resource) catch |err| {
+                printStatus("ERROR: Replica identity validation failed for '{s}': {}\n", .{ stream.source.resource, err });
+                return err;
+            };
+        }
     }
 
     printStatus("PostgreSQL validation completed successfully!\n", .{});

src/source/postgres/validator.zig

@@ -125,4 +125,46 @@ pub const PostgresValidator = struct {
 
         print("PostgreSQL validation: Table '{s}.{s}' exists ✓\n", .{ schema, table_name });
     }
+
+    /// Require REPLICA IDENTITY FULL on a table whose stream tracks DELETE, so the
+    /// deleted row carries all columns. Any other identity (default/index/nothing)
+    /// drops the non-key columns from the DELETE old row, breaking the documented
+    /// format. Call only for delete-tracking streams: FULL is irrelevant otherwise
+    /// and only inflates UPDATE WAL.
+    pub fn checkReplicaIdentity(self: *Self, schema: []const u8, table_name: []const u8) ValidationError!void {
+        const query = try std.fmt.allocPrintSentinel(self.allocator, "SELECT c.relreplident FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace WHERE n.nspname = '{s}' AND c.relname = '{s}';", .{ schema, table_name }, 0);
+        defer self.allocator.free(query);
+
+        const result = try self.executeQuery(query.ptr);
+        defer c.PQclear(result);
+
+        // checkTableExists runs first, so an empty result only happens on a race
+        // (the table was dropped between the two queries).
+        if (c.PQntuples(result) == 0) {
+            std.log.warn("PostgreSQL validation: Table '{s}.{s}' not found while checking replica identity", .{ schema, table_name });
+            return ValidationError.TableNotFound;
+        }
+
+        const identity = std.mem.span(c.PQgetvalue(result, 0, 0));
+
+        if (identity.len == 0 or identity[0] != 'f') {
+            std.log.warn("PostgreSQL validation: Table '{s}.{s}' has REPLICA IDENTITY {s}, but this stream tracks DELETE and needs the full old row", .{ schema, table_name, replicaIdentityName(identity) });
+            std.log.warn("Fix: ALTER TABLE {s}.{s} REPLICA IDENTITY FULL", .{ schema, table_name });
+            return ValidationError.InvalidReplicaIdentity;
+        }
+
+        print("PostgreSQL validation: Table '{s}.{s}' REPLICA IDENTITY FULL ✓\n", .{ schema, table_name });
+    }
 };
+
+// Human-readable name for a pg_class.relreplident value, for the error message.
+fn replicaIdentityName(identity: []const u8) []const u8 {
+    if (identity.len == 0) return "unknown";
+    return switch (identity[0]) {
+        'd' => "default (primary key only)",
+        'i' => "index",
+        'n' => "nothing",
+        'f' => "full",
+        else => "unknown",
+    };
+}

src/source/postgres/validator_test.zig

@@ -81,6 +81,32 @@ test "PostgresValidator: invalid schema should error" {
     try testing.expectError(error.TableNotFound, result);
 }
 
+test "PostgresValidator: replica identity FULL passes" {
+    const allocator = testing.allocator;
+    var validator = PostgresValidator.init(allocator);
+    defer validator.deinit();
+
+    const conn_str = "host=localhost port=5432 dbname=outboxx_test user=postgres password=password";
+
+    try validator.connect(conn_str);
+    // users is set to REPLICA IDENTITY FULL by the dev init script.
+    try validator.checkReplicaIdentity("public", "users");
+}
+
+test "PostgresValidator: replica identity not FULL should error" {
+    const allocator = testing.allocator;
+    var validator = PostgresValidator.init(allocator);
+    defer validator.deinit();
+
+    const conn_str = "host=localhost port=5432 dbname=outboxx_test user=postgres password=password";
+
+    try validator.connect(conn_str);
+    // system_logs keeps the default replica identity (the init script never
+    // alters it), so a delete-tracking stream on it must be rejected.
+    const result = validator.checkReplicaIdentity("public", "system_logs");
+    try testing.expectError(error.InvalidReplicaIdentity, result);
+}
+
 test "PostgresValidator: invalid connection string should error" {
     const allocator = testing.allocator;
     var validator = PostgresValidator.init(allocator);
@@ -113,6 +139,9 @@ test "PostgresValidator: methods fail when not connected" {
 
     const table_result = validator.checkTableExists("public", "users");
     try testing.expectError(error.ConnectionFailed, table_result);
+
+    const identity_result = validator.checkReplicaIdentity("public", "users");
+    try testing.expectError(error.ConnectionFailed, identity_result);
 }
 
 // Note: Testing invalid PostgreSQL version and wal_level requires mocking or