drop user created publications

internal/db/reset/reset.go

@@ -220,6 +220,9 @@ func resetRemote(ctx context.Context, version string, config pgconn.Config, fsys
 	if err := migration.DropUserSchemas(ctx, conn); err != nil {
 		return err
 	}
+	if err := migration.DropUserPublication(ctx, conn); err != nil {
+		return err
+	}
 	return apply.MigrateAndSeed(ctx, version, conn, fsys)
 }
 

internal/db/reset/reset_test.go

@@ -418,7 +418,11 @@ func TestResetRemote(t *testing.T) {
 			Query("DROP SCHEMA IF EXISTS private CASCADE").
 			Reply("DROP SCHEMA").
 			Query(migration.DropObjects).
-			Reply("INSERT 0")
+			Reply("INSERT 0").
+			Query(migration.ListPublications, migration.ManagedPublications).
+			Reply("SELECT pubname", []interface{}{"pub1"}).
+			Query("DROP PUBLICATION IF EXISTS pub1").
+			Reply("DROP PUBLICATION")
 		helper.MockMigrationHistory(conn).
 			Query(migration.INSERT_MIGRATION_VERSION, "0", "schema", nil).
 			Reply("INSERT 0 1")
@@ -444,7 +448,12 @@ func TestResetRemote(t *testing.T) {
 			Query("DROP SCHEMA IF EXISTS private CASCADE").
 			Reply("DROP SCHEMA").
 			Query(migration.DropObjects).
-			Reply("INSERT 0")
+			Reply("INSERT 0").
+			Query(migration.ListPublications, migration.ManagedPublications).
+			Reply("SELECT pubname", []interface{}{"pub1"}).
+			Query("DROP PUBLICATION IF EXISTS pub1").
+			Reply("DROP PUBLICATION")
+
 		helper.MockMigrationHistory(conn).
 			Query(migration.INSERT_MIGRATION_VERSION, "0", "schema", nil).
 			Reply("INSERT 0 1")

pkg/migration/drop.go

@@ -15,6 +15,8 @@ var (
 	DropObjects string
 	//go:embed queries/list.sql
 	ListSchemas string
+	//go:embed queries/publications.sql
+	ListPublications string
 
 	// Initialised by postgres image and owned by postgres role
 	ManagedSchemas = []string{
@@ -29,6 +31,8 @@ var (
 		`supabase\_migrations`,
 		"vault",
 	}
+
+	ManagedPublications = []string{"supabase_realtime"}
 )
 
 func DropUserSchemas(ctx context.Context, conn *pgx.Conn) error {
@@ -65,3 +69,30 @@ func ListUserSchemas(ctx context.Context, conn *pgx.Conn, exclude ...string) ([]
 	// TODO: show detail and hint from pgconn.PgError
 	return pgxv5.CollectStrings(rows)
 }
+
+func DropUserPublication(ctx context.Context, conn *pgx.Conn) error {
+	excludes := ManagedPublications
+	userPublications, err := ListUserPublications(ctx, conn, excludes...)
+	if err != nil {
+		return err
+	}
+
+	// Drop all user defined publications
+	migration := MigrationFile{}
+	for _, publication := range userPublications {
+		sql := fmt.Sprintf("DROP PUBLICATION IF EXISTS %s", publication)
+		migration.Statements = append(migration.Statements, sql)
+	}
+	return migration.ExecBatch(ctx, conn)
+}
+
+func ListUserPublications(ctx context.Context, conn *pgx.Conn, exclude ...string) ([]string, error) {
+	// Execute the query, passing the exclude slice as a single argument
+	rows, err := conn.Query(ctx, ListPublications, exclude)
+	if err != nil {
+		return nil, fmt.Errorf("failed to list publications: %w", err)
+	}
+
+	// Collect results into a slice of strings
+	return pgxv5.CollectStrings(rows)
+}

pkg/migration/queries/publications.sql

@@ -0,0 +1,2 @@
+-- List user defined publications
+select pubname from pg_publication where pubname not like any($1)