fix(postgresqlpublication): Add forgotten management of publication ownership

Author: oxyno-zeta

internal/controller/postgresql/postgres/create-publication-builder.go

@@ -10,6 +10,7 @@ type CreatePublicationBuilder struct {
 	tablesPart string
 	allTables  string
 	withPart   string
+	owner      string
 	tables     []string
 	schemaList []string
 }
@@ -79,6 +80,12 @@ func (b *CreatePublicationBuilder) SetForAllTables() *CreatePublicationBuilder {
 	return b
 }
 
+func (b *CreatePublicationBuilder) SetOwner(n string) *CreatePublicationBuilder {
+	b.owner = n
+
+	return b
+}
+
 func (b *CreatePublicationBuilder) SetName(n string) *CreatePublicationBuilder {
 	b.name = n
 

internal/controller/postgresql/postgres/postgres.go

@@ -67,6 +67,7 @@ type PG interface { //nolint:interfacebloat // This is needed
 	GetPublication(ctx context.Context, dbname, name string) (*PublicationResult, error)
 	CreatePublication(ctx context.Context, dbname string, builder *CreatePublicationBuilder) error
 	UpdatePublication(ctx context.Context, dbname, publicationName string, builder *UpdatePublicationBuilder) error
+	ChangePublicationOwner(ctx context.Context, dbname string, publicationName string, owner string) error
 	DropReplicationSlot(ctx context.Context, name string) error
 	CreateReplicationSlot(ctx context.Context, dbname, name, plugin string) error
 	GetReplicationSlot(ctx context.Context, name string) (*ReplicationSlotResult, error)

internal/controller/postgresql/postgres/publication.go

@@ -12,9 +12,10 @@ const (
 	CreatePublicationSQLTemplate                = `CREATE PUBLICATION "%s" %s %s`
 	DropPublicationSQLTemplate                  = `DROP PUBLICATION "%s"`
 	AlterPublicationRenameSQLTemplate           = `ALTER PUBLICATION "%s" RENAME TO "%s"`
+	AlterPublicationChangeOwnerSQLTemplate      = `ALTER PUBLICATION "%s" OWNER TO "%s"`
 	AlterPublicationGeneralOperationSQLTemplate = `ALTER PUBLICATION "%s" SET %s`
 	GetPublicationSQLTemplate                   = `SELECT
-  puballtables, pubinsert, pubupdate, pubdelete, pubtruncate, pubviaroot
+  pg_catalog.pg_get_userbyid(pubowner), puballtables, pubinsert, pubupdate, pubdelete, pubtruncate, pubviaroot
 FROM pg_catalog.pg_publication
 WHERE pubname = '%s';`
 	GetReplicationSlotSQLTemplate    = `SELECT slot_name,plugin,database FROM pg_replication_slots WHERE slot_name = '%s'`
@@ -23,6 +24,7 @@ WHERE pubname = '%s';`
 )
 
 type PublicationResult struct {
+	Owner              string
 	AllTables          bool
 	Insert             bool
 	Update             bool
@@ -178,6 +180,23 @@ func (c *pg) UpdatePublication(ctx context.Context, dbname, publicationName stri
 	return nil
 }
 
+func (c *pg) ChangePublicationOwner(ctx context.Context, dbname string, publicationName string, owner string) error {
+	// Connect to db
+	err := c.connect(dbname)
+	if err != nil {
+		return err
+	}
+
+	// Change owner
+	_, err = c.db.ExecContext(ctx, fmt.Sprintf(AlterPublicationChangeOwnerSQLTemplate, publicationName, owner))
+	if err != nil {
+		return err
+	}
+
+	// Default
+	return nil
+}
+
 func (c *pg) CreatePublication(ctx context.Context, dbname string, builder *CreatePublicationBuilder) error {
 	// Connect to db
 	err := c.connect(dbname)
@@ -193,6 +212,12 @@ func (c *pg) CreatePublication(ctx context.Context, dbname string, builder *Crea
 		return err
 	}
 
+	// Change owner
+	err = c.ChangePublicationOwner(ctx, dbname, builder.name, builder.owner)
+	if err != nil {
+		return err
+	}
+
 	// Default
 	return nil
 }
@@ -217,7 +242,7 @@ func (c *pg) GetPublication(ctx context.Context, dbname, name string) (*Publicat
 
 	for rows.Next() {
 		// Scan
-		err = rows.Scan(&res.AllTables, &res.Insert, &res.Update, &res.Delete, &res.Truncate, &res.PublicationViaRoot)
+		err = rows.Scan(&res.Owner, &res.AllTables, &res.Insert, &res.Update, &res.Delete, &res.Truncate, &res.PublicationViaRoot)
 		// Check error
 		if err != nil {
 			return nil, err

internal/controller/postgresql/postgresqlpublication_controller.go

@@ -258,6 +258,18 @@ func (r *PostgresqlPublicationReconciler) mainReconcile(
 				return r.manageError(ctx, reqLogger, instance, originalPatch, err)
 			}
 		}
+
+		// Check if owner are aligned
+		if pgDB.Status.Roles.Owner != pubRes.Owner {
+			reqLogger.Info("Owner aren't aligned, update need to be done")
+
+			// Change owner
+			err = pg.ChangePublicationOwner(ctx, pgDB.Status.Database, nameToSearch, pgDB.Status.Roles.Owner)
+			// Check error
+			if err != nil {
+				return r.manageError(ctx, reqLogger, instance, originalPatch, err)
+			}
+		}
 	}
 
 	// Get replication slot
@@ -384,6 +396,9 @@ func (*PostgresqlPublicationReconciler) manageCreate(
 		builder = builder.AddTable(table.TableName, table.Columns, table.AdditionalWhere)
 	})
 
+	// Manage owner
+	builder = builder.SetOwner(pgDB.Status.Roles.Owner)
+
 	// Create publication
 	err := pg.CreatePublication(ctx, pgDB.Status.Database, builder)
 	// Check error