diff --git a/CHANGELOG.md b/CHANGELOG.md index 1d05cee..9af966f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,12 @@ This package uses **lockstep major versioning** with the core NSchema package: ` As a consequence, breaking changes that are specific to this provider (rather than the core API) are signalled by a **minor version bump** rather than a major one, and called out explicitly in this changelog. +## [5.7.0] - 2026-10-07 + +### Added + +- **Publication support.**, Includes column lists, row filters and `TABLES IN SCHEMA` on PostgreSQL 15 and later. + ## [5.6.1] - 2026-08-13 ### Fixed @@ -162,6 +168,7 @@ First stable release of the PostgreSQL provider for NSchema, tracking the 1.0 re - `SqlType.Citext` and `SqlType.Jsonb` Postgres-specific type helpers on `SqlType`. - SourceLink and symbol packages (`.snupkg`) published alongside the main package for source-level debugging. +[5.7.0]: https://github.com/nschema-org/NSchema.Postgres/compare/v5.6.1...v5.7.0 [5.6.1]: https://github.com/nschema-org/NSchema.Postgres/compare/v5.6.0...v5.6.1 [5.6.0]: https://github.com/nschema-org/NSchema.Postgres/compare/v5.5.0...v5.6.0 [5.5.0]: https://github.com/nschema-org/NSchema.Postgres/compare/v5.4.0...v5.5.0 diff --git a/Directory.Packages.props b/Directory.Packages.props index 2e19636..0889a1d 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -4,19 +4,19 @@ true - - - - + + + + - - + + - - - + + + \ No newline at end of file diff --git a/NSchema.Postgres.slnx b/NSchema.Postgres.slnx index 659c89d..716744c 100644 --- a/NSchema.Postgres.slnx +++ b/NSchema.Postgres.slnx @@ -3,6 +3,7 @@ + diff --git a/dotnet-tools.json b/dotnet-tools.json index 0ee375b..c836883 100644 --- a/dotnet-tools.json +++ b/dotnet-tools.json @@ -3,7 +3,7 @@ "isRoot": true, "tools": { "ritten": { - "version": "0.6.0", + "version": "0.22.0", "commands": [ "ritten" ], diff --git a/src/NSchema.Postgres/Models/PublicationRow.cs b/src/NSchema.Postgres/Models/PublicationRow.cs new file mode 100644 index 0000000..f918fa3 --- /dev/null +++ b/src/NSchema.Postgres/Models/PublicationRow.cs @@ -0,0 +1,10 @@ +namespace NSchema.Postgres.Models; + +internal sealed record PublicationRow( + string Name, + bool AllTables, + bool Insert, + bool Update, + bool Delete, + bool Truncate, + string? Comment); diff --git a/src/NSchema.Postgres/Models/PublishedTableRow.cs b/src/NSchema.Postgres/Models/PublishedTableRow.cs new file mode 100644 index 0000000..9a20b7e --- /dev/null +++ b/src/NSchema.Postgres/Models/PublishedTableRow.cs @@ -0,0 +1,8 @@ +namespace NSchema.Postgres.Models; + +internal sealed record PublishedTableRow( + string Publication, + string Schema, + string Table, + string[]? Columns, + string? Filter); diff --git a/src/NSchema.Postgres/Models/TableRow.cs b/src/NSchema.Postgres/Models/TableRow.cs index 8c46cb4..c6a6fd3 100644 --- a/src/NSchema.Postgres/Models/TableRow.cs +++ b/src/NSchema.Postgres/Models/TableRow.cs @@ -1,3 +1,3 @@ namespace NSchema.Postgres.Models; -internal sealed record TableRow(string Schema, string Name); +internal sealed record TableRow(string Schema, string Name, char ReplicaIdentity = 'd', string? ReplicaIdentityIndex = null); diff --git a/src/NSchema.Postgres/NSchema.Postgres.csproj b/src/NSchema.Postgres/NSchema.Postgres.csproj index 0c3e495..c7e008b 100644 --- a/src/NSchema.Postgres/NSchema.Postgres.csproj +++ b/src/NSchema.Postgres/NSchema.Postgres.csproj @@ -31,7 +31,7 @@ true true snupkg - 5.6.1 + 5.7.0 $(Version.Split('-')[0]) $(Version.Split('-')[0]) true diff --git a/src/NSchema.Postgres/Sql/PostgresDatabaseIntrospector.cs b/src/NSchema.Postgres/Sql/PostgresDatabaseIntrospector.cs index 390ff41..2c56288 100644 --- a/src/NSchema.Postgres/Sql/PostgresDatabaseIntrospector.cs +++ b/src/NSchema.Postgres/Sql/PostgresDatabaseIntrospector.cs @@ -10,6 +10,7 @@ using NSchema.Model.Enums; using NSchema.Model.Extensions; using NSchema.Model.Indexes; +using NSchema.Model.Publications; using NSchema.Model.Routines; using NSchema.Model.Schemas; using NSchema.Model.Sequences; @@ -93,6 +94,9 @@ private async ValueTask> Read(PlanningScope scope, Cancellation // through NSchema comes back looking the same and no longer routing a single row the way it used to. var partitioning = await QueryPartitioning(conn, schemas, cancellationToken); + // Publications are database-global, so the schema list does not narrow them. + var publications = await QueryPublications(conn, cancellationToken); + var database = Build( tables, columns, primaryKeys, foreignKeys, uniqueConstraints, checkConstraints, exclusionConstraints, indexes, triggers, schemaComments, tableComments, columnComments, indexComments, constraintComments, @@ -102,6 +106,7 @@ private async ValueTask> Read(PlanningScope scope, Cancellation functions, functionComments, procedures, procedureComments, aggregates, aggregateComments, extensions ); + database.Publications.AddRange(publications); return partitioning.Count == 0 ? Result.Success(database) @@ -146,6 +151,107 @@ ORDER BY 1 // ── Queries ─────────────────────────────────────────────────────────────── + // Column lists, row filters and TABLES IN SCHEMA arrived in Postgres 15; before it, none of their catalog + // columns exist. + private static async Task> QueryPublications(NpgsqlConnection conn, CancellationToken ct) + { + var modern = conn.PostgreSqlVersion.Major >= 15; + + var publications = new List(); + await using (var cmd = conn.CreateCommand()) + { + cmd.CommandText = """ + SELECT p.pubname, p.puballtables, p.pubinsert, p.pubupdate, p.pubdelete, p.pubtruncate, + obj_description(p.oid, 'pg_publication') + FROM pg_publication p + ORDER BY p.pubname + """; + await using var reader = await cmd.ExecuteReaderAsync(ct); + while (await reader.ReadAsync(ct)) + { + publications.Add(new PublicationRow(reader.GetString(0), reader.GetBoolean(1), reader.GetBoolean(2), + reader.GetBoolean(3), reader.GetBoolean(4), reader.GetBoolean(5), reader.IsDBNull(6) ? null : reader.GetString(6))); + } + } + + var tables = new List(); + await using (var cmd = conn.CreateCommand()) + { + cmd.CommandText = modern + ? """ + SELECT p.pubname, n.nspname, c.relname, + (SELECT array_agg(a.attname ORDER BY a.attnum) + FROM pg_attribute a + WHERE a.attrelid = pr.prrelid AND a.attnum = ANY(pr.prattrs::int2[])), + pg_get_expr(pr.prqual, pr.prrelid) + FROM pg_publication_rel pr + JOIN pg_publication p ON p.oid = pr.prpubid + JOIN pg_class c ON c.oid = pr.prrelid + JOIN pg_namespace n ON n.oid = c.relnamespace + ORDER BY p.pubname, n.nspname, c.relname + """ + : """ + SELECT p.pubname, n.nspname, c.relname, NULL::text[], NULL::text + FROM pg_publication_rel pr + JOIN pg_publication p ON p.oid = pr.prpubid + JOIN pg_class c ON c.oid = pr.prrelid + JOIN pg_namespace n ON n.oid = c.relnamespace + ORDER BY p.pubname, n.nspname, c.relname + """; + await using var reader = await cmd.ExecuteReaderAsync(ct); + while (await reader.ReadAsync(ct)) + { + tables.Add(new PublishedTableRow(reader.GetString(0), reader.GetString(1), reader.GetString(2), + reader.IsDBNull(3) ? null : reader.GetFieldValue(3), reader.IsDBNull(4) ? null : reader.GetString(4))); + } + } + + var schemas = new List<(string Publication, string Schema)>(); + if (modern) + { + await using var cmd = conn.CreateCommand(); + cmd.CommandText = """ + SELECT p.pubname, n.nspname + FROM pg_publication_namespace pn + JOIN pg_publication p ON p.oid = pn.pnpubid + JOIN pg_namespace n ON n.oid = pn.pnnspid + ORDER BY p.pubname, n.nspname + """; + await using var reader = await cmd.ExecuteReaderAsync(ct); + while (await reader.ReadAsync(ct)) + { + schemas.Add((reader.GetString(0), reader.GetString(1))); + } + } + + return [.. publications.Select(p => new Publication + { + Name = p.Name, + AllTables = p.AllTables, + Tables = [.. tables.Where(t => t.Publication == p.Name).Select(t => new PublishedTable( + new ObjectAddress(t.Schema, t.Table), + t.Columns?.Select(c => new SqlIdentifier(c)).ToList(), + t.Filter is { } filter ? new SqlText(filter) : null))], + Schemas = [.. schemas.Where(s => s.Publication == p.Name).Select(s => new SqlIdentifier(s.Schema))], + Operations = (p.Insert ? PublishedOperations.Insert : PublishedOperations.None) + | (p.Update ? PublishedOperations.Update : PublishedOperations.None) + | (p.Delete ? PublishedOperations.Delete : PublishedOperations.None) + | (p.Truncate ? PublishedOperations.Truncate : PublishedOperations.None), + Comment = p.Comment, + })]; + } + + // Postgres's default ('d', the primary key) is the engine's default, which the model leaves unstated. An index + // identity whose index has since been dropped identifies nothing, as Postgres treats it. + internal static ReplicaIdentity? MapReplicaIdentity(TableRow row) => row.ReplicaIdentity switch + { + 'f' => ReplicaIdentity.Full, + 'n' => ReplicaIdentity.Nothing, + 'i' when row.ReplicaIdentityIndex is { } index => ReplicaIdentity.UsingIndex(index), + 'i' => ReplicaIdentity.Nothing, + _ => null, + }; + private static void AddSchemasParameter(NpgsqlCommand cmd, string[]? schemas) { var parameter = cmd.Parameters.Add("schemas", NpgsqlDbType.Array | NpgsqlDbType.Text); @@ -156,22 +262,27 @@ private static async Task> QueryTables(NpgsqlConnection conn, str { var rows = new List(); await using var cmd = conn.CreateCommand(); + // The replica identity rides along: relreplident, and the index when it names one. cmd.CommandText = """ - SELECT table_schema, table_name - FROM information_schema.tables - WHERE table_type = 'BASE TABLE' - AND (@schemas::text[] IS NULL OR table_schema = ANY(@schemas)) - AND table_schema NOT IN ('pg_catalog', 'information_schema') - AND table_schema NOT LIKE 'pg\_toast%' ESCAPE '\' - AND table_schema NOT LIKE 'pg\_temp%' ESCAPE '\' - ORDER BY table_schema, table_name + SELECT t.table_schema, t.table_name, c.relreplident, ri.relname + FROM information_schema.tables t + JOIN pg_namespace n ON n.nspname = t.table_schema + JOIN pg_class c ON c.relnamespace = n.oid AND c.relname = t.table_name + LEFT JOIN pg_index i ON i.indrelid = c.oid AND i.indisreplident + LEFT JOIN pg_class ri ON ri.oid = i.indexrelid + WHERE t.table_type = 'BASE TABLE' + AND (@schemas::text[] IS NULL OR t.table_schema = ANY(@schemas)) + AND t.table_schema NOT IN ('pg_catalog', 'information_schema') + AND t.table_schema NOT LIKE 'pg\_toast%' ESCAPE '\' + AND t.table_schema NOT LIKE 'pg\_temp%' ESCAPE '\' + ORDER BY t.table_schema, t.table_name """; AddSchemasParameter(cmd, schemas); await using var reader = await cmd.ExecuteReaderAsync(ct); while (await reader.ReadAsync(ct)) { - rows.Add(new TableRow(reader.GetString(0), reader.GetString(1))); + rows.Add(new TableRow(reader.GetString(0), reader.GetString(1), reader.GetChar(2), reader.IsDBNull(3) ? null : reader.GetString(3))); } return rows; @@ -1832,6 +1943,7 @@ List allTableGrants Indexes = [.. idxs], Grants = [.. grants], Triggers = [.. triggers], + ReplicaIdentity = MapReplicaIdentity(tableRow), }; } diff --git a/src/NSchema.Postgres/Sql/PostgresSqlDialect.Publications.cs b/src/NSchema.Postgres/Sql/PostgresSqlDialect.Publications.cs new file mode 100644 index 0000000..e0c3314 --- /dev/null +++ b/src/NSchema.Postgres/Sql/PostgresSqlDialect.Publications.cs @@ -0,0 +1,84 @@ +using NSchema.Model.Publications; +using NSchema.Plan.Domain; +using NSchema.Plan.Domain.Publications; + +namespace NSchema.Postgres.Sql; + +internal sealed partial class PostgresSqlDialect +{ + // ── Publications ────────────────────────────────────────────────────────── + + // Tables first, each run opened once: the only spelling Postgres before 15 accepts, and valid after it. + protected override Result> CreatePublication(CreatePublication action) + { + var publication = action.Publication; + var targets = new List(); + if (publication.Tables.Count > 0) + { + targets.Add("TABLE " + string.Join(", ", publication.Tables.Select(PublishedTable))); + } + if (publication.Schemas.Count > 0) + { + targets.Add("TABLES IN SCHEMA " + string.Join(", ", publication.Schemas.Select(Quote))); + } + + var sql = $"CREATE PUBLICATION {Quote(publication.Name)}"; + if (publication.AllTables) + { + sql += " FOR ALL TABLES"; + } + else if (targets.Count > 0) + { + sql += " FOR " + string.Join(", ", targets); + } + if (publication.Operations != PublishedOperations.All) + { + sql += $" WITH (publish = '{Publish(publication.Operations)}')"; + } + + return Statement(sql); + } + + protected override Result> DropPublication(DropPublication action) => + Statement($"DROP PUBLICATION {Quote(action.PublicationName)}"); + + protected override Result> RenamePublication(RenamePublication action) => + Statement($"ALTER PUBLICATION {Quote(action.OldName)} RENAME TO {Quote(action.NewName)}"); + + protected override Result> AddPublicationTable(AddPublicationTable action) => + Statement($"ALTER PUBLICATION {Quote(action.PublicationName)} ADD TABLE {PublishedTable(action.Table)}"); + + protected override Result> DropPublicationTable(DropPublicationTable action) => + Statement($"ALTER PUBLICATION {Quote(action.PublicationName)} DROP TABLE {Qualify(action.Table)}"); + + protected override Result> AddPublicationSchema(AddPublicationSchema action) => + Statement($"ALTER PUBLICATION {Quote(action.PublicationName)} ADD TABLES IN SCHEMA {Quote(action.Schema)}"); + + protected override Result> DropPublicationSchema(DropPublicationSchema action) => + Statement($"ALTER PUBLICATION {Quote(action.PublicationName)} DROP TABLES IN SCHEMA {Quote(action.Schema)}"); + + protected override Result> SetPublicationOperations(SetPublicationOperations action) => + Statement($"ALTER PUBLICATION {Quote(action.PublicationName)} SET (publish = '{Publish(action.NewOperations)}')"); + + protected override Result> SetPublicationComment(SetPublicationComment action) => + Comment($"PUBLICATION {Quote(action.PublicationName)}", action.NewComment); + + private string PublishedTable(PublishedTable table) + { + var sql = Qualify(table.Table); + if (table.Columns is { } columns) + { + sql += $" ({ColumnList(columns)})"; + } + if (table.Filter is { } filter) + { + sql += $" WHERE ({filter.Value})"; + } + return sql; + } + + private static string Publish(PublishedOperations operations) => string.Join(", ", + new[] { PublishedOperations.Insert, PublishedOperations.Update, PublishedOperations.Delete, PublishedOperations.Truncate } + .Where(o => operations.HasFlag(o)) + .Select(o => o.ToString().ToLowerInvariant())); +} diff --git a/src/NSchema.Postgres/Sql/PostgresSqlDialect.Tables.cs b/src/NSchema.Postgres/Sql/PostgresSqlDialect.Tables.cs index 9ea09d5..ea9cb27 100644 --- a/src/NSchema.Postgres/Sql/PostgresSqlDialect.Tables.cs +++ b/src/NSchema.Postgres/Sql/PostgresSqlDialect.Tables.cs @@ -1,3 +1,4 @@ +using NSchema.Model.Tables; using NSchema.Plan.Domain; using NSchema.Plan.Domain.Tables; @@ -25,4 +26,14 @@ protected override Result> CreateTable(CreateTable a protected override Result> SetTableComment(SetTableComment action) => Comment($"TABLE {Qualify(action.Table)}", action.NewComment); + + // No declared identity is the engine's default, which Postgres spells DEFAULT (the primary key). + protected override Result> SetReplicaIdentity(SetReplicaIdentity action) => + Statement($"ALTER TABLE {Qualify(action.Table)} REPLICA IDENTITY " + action.NewIdentity switch + { + null => "DEFAULT", + { Kind: ReplicaIdentityKind.Full } => "FULL", + { Kind: ReplicaIdentityKind.Index, Index: { } index } => $"USING INDEX {Quote(index)}", + _ => "NOTHING", + }); } diff --git a/tests/NSchema.Postgres.Tests/PostgresPluginEndToEndTests.cs b/tests/NSchema.Postgres.Tests/PostgresPluginEndToEndTests.cs index 140dbba..1e21385 100644 --- a/tests/NSchema.Postgres.Tests/PostgresPluginEndToEndTests.cs +++ b/tests/NSchema.Postgres.Tests/PostgresPluginEndToEndTests.cs @@ -142,6 +142,46 @@ SCRIPT backfill RUN ON ADD COLUMN {_schema}.widgets.status AS $$ } /// Builds an app wired only through the plugin manifest, plus the ephemeral state planning requires. + [Fact] + public async Task Apply_Publication_ThenRefreshAndPlanAgain_ShowsNoChanges() + { + // Arrange — Postgres stores the column list in column order and rewrites the filter, so a clean second + // plan proves both are absorbed rather than reported as drift. + var publication = $"pub_{Guid.NewGuid():N}"; + await File.WriteAllTextAsync(Path.Combine(_projectDir, "schema.sql"), $""" + CREATE SCHEMA {_schema}; + + CREATE TABLE {_schema}.orders ( + id bigint NOT NULL, + status text NOT NULL, + total numeric(12,2), + CONSTRAINT orders_pkey PRIMARY KEY (id) + ) REPLICA IDENTITY FULL; + + CREATE PUBLICATION {publication} + FOR TABLE {_schema}.orders (total, status, id) WHERE (status <> 'draft') + PUBLISH (INSERT, UPDATE); + """, TestContext.Current.CancellationToken); + + try + { + using var app = BuildApp(); + + // Act + var first = await Plan(app); + (await app.Operations.Apply(new ApplyArguments { Plan = first }, TestContext.Current.CancellationToken)).IsSuccess.ShouldBeTrue(); + var second = await Plan(app); + + // Assert + second.Diff.IsEmpty.ShouldBeTrue(); + (await Scalar($"SELECT relreplident::text FROM pg_class WHERE oid = '{_schema}.orders'::regclass")).ShouldBe("f"); + } + finally + { + await Exec($"""DROP PUBLICATION IF EXISTS "{publication}" """); + } + } + private NSchemaApplication BuildApp() { var builder = NSchemaApplication.CreateBuilder(); diff --git a/tests/NSchema.Postgres.Tests/Sql/PostgresSqlDialectSnapshotTests.cs b/tests/NSchema.Postgres.Tests/Sql/PostgresSqlDialectSnapshotTests.cs index 6d5d9d1..b316348 100644 --- a/tests/NSchema.Postgres.Tests/Sql/PostgresSqlDialectSnapshotTests.cs +++ b/tests/NSchema.Postgres.Tests/Sql/PostgresSqlDialectSnapshotTests.cs @@ -5,6 +5,7 @@ using NSchema.Model.Domains; using NSchema.Model.Enums; using NSchema.Model.Extensions; +using NSchema.Model.Publications; using NSchema.Model.Indexes; using NSchema.Model.Routines; using NSchema.Model.Scripts; @@ -19,6 +20,7 @@ using NSchema.Plan.Domain.Domains; using NSchema.Plan.Domain.Enums; using NSchema.Plan.Domain.Extensions; +using NSchema.Plan.Domain.Publications; using NSchema.Plan.Domain.Indexes; using NSchema.Plan.Domain.Routines; using NSchema.Plan.Domain.Schemas; @@ -373,6 +375,40 @@ public Task SequenceOperations() => VerifyActions( new SetSequenceComment(new ObjectAddress("public", "invoice_id"), "Invoice numbers", null), new DropSequence(new ObjectAddress("public", "invoice_id"))); + // ── Publications ───────────────────────────────────────────────────────── + + [Fact] + public Task PublicationOperations() => VerifyActions( + new CreatePublication(new Publication { Name = "empty" }), + new CreatePublication(new Publication { Name = "everything", AllTables = true, Operations = PublishedOperations.Insert }), + new CreatePublication(new Publication + { + Name = "audit_feed", + Tables = [new PublishedTable(new ObjectAddress("sales", "orders"), Filter: "status <> 'draft'")], + Schemas = ["audit", "archive"], + }), + new CreatePublication(new Publication + { + Name = "orders_feed", + Tables = + [ + new PublishedTable(new ObjectAddress("sales", "orders"), ["id", "status"], "status <> 'draft'"), + new PublishedTable(new ObjectAddress("sales", "order_lines")), + ], + Operations = PublishedOperations.None, + }), + new DropPublication("orders_feed"), + new RenamePublication("orders_feed", "order_changes"), + new AddPublicationTable("orders_feed", new PublishedTable(new ObjectAddress("sales", "orders"), ["id"])), + new DropPublicationTable("orders_feed", new ObjectAddress("sales", "orders")), + new AddPublicationSchema("orders_feed", "audit"), + new DropPublicationSchema("orders_feed", "audit"), + new SetPublicationOperations("orders_feed", PublishedOperations.All, PublishedOperations.Insert | PublishedOperations.Delete), + new SetPublicationComment("orders_feed", null, "Order changes"), + new SetReplicaIdentity(new ObjectAddress("sales", "orders"), null, ReplicaIdentity.Full), + new SetReplicaIdentity(new ObjectAddress("sales", "orders"), null, ReplicaIdentity.UsingIndex("ux_orders")), + new SetReplicaIdentity(new ObjectAddress("sales", "orders"), ReplicaIdentity.Full, null)); + // ── Extensions ──────────────────────────────────────────────────────────── [Fact] diff --git a/tests/NSchema.Postgres.Tests/Sql/PostgresSqlDialectTests.cs b/tests/NSchema.Postgres.Tests/Sql/PostgresSqlDialectTests.cs index 4618b42..ac6f8f3 100644 --- a/tests/NSchema.Postgres.Tests/Sql/PostgresSqlDialectTests.cs +++ b/tests/NSchema.Postgres.Tests/Sql/PostgresSqlDialectTests.cs @@ -6,6 +6,7 @@ using NSchema.Model.Domains; using NSchema.Model.Enums; using NSchema.Model.Extensions; +using NSchema.Model.Publications; using NSchema.Model.Indexes; using NSchema.Model.Routines; using NSchema.Model.Scripts; @@ -20,6 +21,7 @@ using NSchema.Plan.Domain.Domains; using NSchema.Plan.Domain.Enums; using NSchema.Plan.Domain.Extensions; +using NSchema.Plan.Domain.Publications; using NSchema.Plan.Domain.Indexes; using NSchema.Plan.Domain.Routines; using NSchema.Plan.Domain.Schemas; @@ -891,6 +893,127 @@ public async Task CreateExtension_ThenIntrospect_ReportsExtension() database.Extensions.ShouldNotContain(e => e.Name == "plpgsql"); } + // ── Publications ───────────────────────────────────────────────────────── + + [Fact] + public async Task CreatePublication_ThenIntrospect_RoundTrips() + { + // Arrange — publications are database-global, so each test names its own and drops it. + await Exec($"""CREATE TABLE "{_schema}".orders (id int PRIMARY KEY, status text, total numeric)"""); + await Exec($"""CREATE TABLE "{_schema}".lines (id int PRIMARY KEY)"""); + var name = $"pub_{Guid.NewGuid():N}"; + var publication = new Publication + { + Name = name, + Tables = + [ + new PublishedTable(new ObjectAddress(_schema, "orders"), ["total", "id", "status"], "status <> 'draft'"), + new PublishedTable(new ObjectAddress(_schema, "lines")), + ], + Operations = PublishedOperations.Insert | PublishedOperations.Update, + }; + + try + { + // Act + await Run(new CreatePublication(publication)); + await Run(new SetPublicationComment(name, null, "order changes")); + + // Assert — the column list comes back in column order and the filter in Postgres's own spelling, which + // is what the recorded declaration exists to absorb. + var introspected = (await Introspect()).Publications.Single(p => p.Name == name); + introspected.Operations.ShouldBe(PublishedOperations.Insert | PublishedOperations.Update); + introspected.Comment.ShouldBe("order changes"); + var orders = introspected.Tables.Single(t => t.Table.Name == "orders"); + orders.SameColumns(publication.Tables[0]).ShouldBeTrue(); + orders.Filter!.Value.ShouldContain("draft"); + introspected.Tables.Single(t => t.Table.Name == "lines").Columns.ShouldBeNull(); + } + finally + { + await Exec($"""DROP PUBLICATION IF EXISTS "{name}" """); + } + } + + [Fact] + public async Task AlterPublication_TablesSchemasAndOperations() + { + // Arrange + await Exec($"""CREATE TABLE "{_schema}".orders (id int PRIMARY KEY, status text)"""); + var name = $"pub_{Guid.NewGuid():N}"; + await Run(new CreatePublication(new Publication { Name = name })); + + try + { + // Act + // Postgres refuses a column list beside TABLES IN SCHEMA, so this entry publishes every column. + await Run(new AddPublicationTable(name, new PublishedTable(new ObjectAddress(_schema, "orders"), Filter: "id > 0"))); + await Run(new AddPublicationSchema(name, "public")); + await Run(new SetPublicationOperations(name, PublishedOperations.All, PublishedOperations.Delete)); + var renamed = $"{name}_x"; + await Run(new RenamePublication(name, renamed)); + name = renamed; + + // Assert + var introspected = (await Introspect()).Publications.Single(p => p.Name == name); + introspected.Tables.ShouldHaveSingleItem().Filter.ShouldNotBeNull(); + introspected.Schemas.ShouldBe([new SqlIdentifier("public")]); + introspected.Operations.ShouldBe(PublishedOperations.Delete); + + // Act — and back out. + await Run(new DropPublicationTable(name, new ObjectAddress(_schema, "orders"))); + await Run(new DropPublicationSchema(name, "public")); + + // Assert + var emptied = (await Introspect()).Publications.Single(p => p.Name == name); + emptied.Tables.ShouldBeEmpty(); + emptied.Schemas.ShouldBeEmpty(); + } + finally + { + await Exec($"""DROP PUBLICATION IF EXISTS "{name}" """); + } + } + + [Fact] + public async Task DropPublication_RemovesIt() + { + // Arrange + var name = $"pub_{Guid.NewGuid():N}"; + await Run(new CreatePublication(new Publication { Name = name, AllTables = true })); + + // Act + await Run(new DropPublication(name)); + + // Assert + (await Introspect()).Publications.ShouldNotContain(p => p.Name == name); + } + + [Theory] + [InlineData(ReplicaIdentityKind.Full)] + [InlineData(ReplicaIdentityKind.Nothing)] + [InlineData(ReplicaIdentityKind.Index)] + public async Task SetReplicaIdentity_ThenIntrospect_RoundTrips(ReplicaIdentityKind kind) + { + // Arrange + await Exec($"""CREATE TABLE "{_schema}".events (id int NOT NULL, kind text)"""); + await Exec($"""CREATE UNIQUE INDEX ux_events ON "{_schema}".events (id)"""); + var identity = new ReplicaIdentity(kind, kind == ReplicaIdentityKind.Index ? "ux_events" : null); + var table = new ObjectAddress(_schema, "events"); + + // Act + await Run(new SetReplicaIdentity(table, null, identity)); + + // Assert + (await Introspect()).Schemas[0].Tables.Single(t => t.Name == "events").ReplicaIdentity.ShouldBe(identity); + + // Act — the engine's default is no identity at all in the model. + await Run(new SetReplicaIdentity(table, identity, null)); + + // Assert + (await Introspect()).Schemas[0].Tables.Single(t => t.Name == "events").ReplicaIdentity.ShouldBeNull(); + } + // ── Composite types ────────────────────────────────────────────────────── [Fact] diff --git a/tests/NSchema.Postgres.Tests/Sql/Snapshots/PostgresSqlDialectSnapshotTests.PublicationOperations.verified.txt b/tests/NSchema.Postgres.Tests/Sql/Snapshots/PostgresSqlDialectSnapshotTests.PublicationOperations.verified.txt new file mode 100644 index 0000000..44022b6 --- /dev/null +++ b/tests/NSchema.Postgres.Tests/Sql/Snapshots/PostgresSqlDialectSnapshotTests.PublicationOperations.verified.txt @@ -0,0 +1,62 @@ +[ + { + Sql: CREATE PUBLICATION "empty", + RunOutsideTransaction: false + }, + { + Sql: CREATE PUBLICATION "everything" FOR ALL TABLES WITH (publish = 'insert'), + RunOutsideTransaction: false + }, + { + Sql: CREATE PUBLICATION "audit_feed" FOR TABLE "sales"."orders" WHERE (status <> 'draft'), TABLES IN SCHEMA "audit", "archive", + RunOutsideTransaction: false + }, + { + Sql: CREATE PUBLICATION "orders_feed" FOR TABLE "sales"."orders" ("id", "status") WHERE (status <> 'draft'), "sales"."order_lines" WITH (publish = ''), + RunOutsideTransaction: false + }, + { + Sql: DROP PUBLICATION "orders_feed", + RunOutsideTransaction: false + }, + { + Sql: ALTER PUBLICATION "orders_feed" RENAME TO "order_changes", + RunOutsideTransaction: false + }, + { + Sql: ALTER PUBLICATION "orders_feed" ADD TABLE "sales"."orders" ("id"), + RunOutsideTransaction: false + }, + { + Sql: ALTER PUBLICATION "orders_feed" DROP TABLE "sales"."orders", + RunOutsideTransaction: false + }, + { + Sql: ALTER PUBLICATION "orders_feed" ADD TABLES IN SCHEMA "audit", + RunOutsideTransaction: false + }, + { + Sql: ALTER PUBLICATION "orders_feed" DROP TABLES IN SCHEMA "audit", + RunOutsideTransaction: false + }, + { + Sql: ALTER PUBLICATION "orders_feed" SET (publish = 'insert, delete'), + RunOutsideTransaction: false + }, + { + Sql: COMMENT ON PUBLICATION "orders_feed" IS $comment$Order changes$comment$, + RunOutsideTransaction: false + }, + { + Sql: ALTER TABLE "sales"."orders" REPLICA IDENTITY FULL, + RunOutsideTransaction: false + }, + { + Sql: ALTER TABLE "sales"."orders" REPLICA IDENTITY USING INDEX "ux_orders", + RunOutsideTransaction: false + }, + { + Sql: ALTER TABLE "sales"."orders" REPLICA IDENTITY DEFAULT, + RunOutsideTransaction: false + } +] \ No newline at end of file