Validate Mirror: PostgreSQL Checks

flow/activities/flowable.go

@@ -642,6 +642,8 @@ func (a *FlowableActivity) ConsolidateQRepPartitions(ctx context.Context, config
 		return err
 	}
 
+	defer connectors.CloseConnector(dstConn)
+
 	shutdown := utils.HeartbeatRoutine(ctx, 2*time.Minute, func() string {
 		return fmt.Sprintf("consolidating partitions for job - %s", config.FlowJobName)
 	})
@@ -665,6 +667,8 @@ func (a *FlowableActivity) CleanupQRepFlow(ctx context.Context, config *protos.Q
 		return err
 	}
 
+	defer dst.Close()
+
 	return dst.CleanupQRepFlow(config)
 }
 

flow/cmd/handler.go

@@ -8,7 +8,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/PeerDB-io/peer-flow/connectors"
 	"github.com/PeerDB-io/peer-flow/generated/protos"
 	"github.com/PeerDB-io/peer-flow/shared"
 	peerflow "github.com/PeerDB-io/peer-flow/workflows"
@@ -122,6 +121,12 @@ func (h *FlowRequestHandler) CreateCDCFlow(
 	ctx context.Context, req *protos.CreateCDCFlowRequest,
 ) (*protos.CreateCDCFlowResponse, error) {
 	cfg := req.ConnectionConfigs
+	_, validateErr := h.ValidateCDCMirror(ctx, req)
+	if validateErr != nil {
+		slog.Error("validate mirror error", slog.Any("error", validateErr))
+		return nil, fmt.Errorf("invalid mirror: %w", validateErr)
+	}
+
 	workflowID := fmt.Sprintf("%s-peerflow-%s", cfg.FlowJobName, uuid.New())
 	workflowOptions := client.StartWorkflowOptions{
 		ID:        workflowID,
@@ -556,50 +561,6 @@ func (h *FlowRequestHandler) handleWorkflowNotClosed(ctx context.Context, workfl
 	return nil
 }
 
-func (h *FlowRequestHandler) ValidatePeer(
-	ctx context.Context,
-	req *protos.ValidatePeerRequest,
-) (*protos.ValidatePeerResponse, error) {
-	if req.Peer == nil {
-		return &protos.ValidatePeerResponse{
-			Status:  protos.ValidatePeerStatus_INVALID,
-			Message: "no peer provided",
-		}, nil
-	}
-
-	if len(req.Peer.Name) == 0 {
-		return &protos.ValidatePeerResponse{
-			Status:  protos.ValidatePeerStatus_INVALID,
-			Message: "no peer name provided",
-		}, nil
-	}
-
-	conn, err := connectors.GetConnector(ctx, req.Peer)
-	if err != nil {
-		return &protos.ValidatePeerResponse{
-			Status: protos.ValidatePeerStatus_INVALID,
-			Message: fmt.Sprintf("peer type is missing or "+
-				"your requested configuration for %s peer %s was invalidated: %s",
-				req.Peer.Type, req.Peer.Name, err),
-		}, nil
-	}
-
-	connErr := conn.ConnectionActive()
-	if connErr != nil {
-		return &protos.ValidatePeerResponse{
-			Status: protos.ValidatePeerStatus_INVALID,
-			Message: fmt.Sprintf("failed to establish active connection to %s peer %s: %v",
-				req.Peer.Type, req.Peer.Name, connErr),
-		}, nil
-	}
-
-	return &protos.ValidatePeerResponse{
-		Status: protos.ValidatePeerStatus_VALID,
-		Message: fmt.Sprintf("%s peer %s is valid",
-			req.Peer.Type, req.Peer.Name),
-	}, nil
-}
-
 func (h *FlowRequestHandler) CreatePeer(
 	ctx context.Context,
 	req *protos.CreatePeerRequest,

flow/cmd/validate_mirror.go

@@ -0,0 +1,54 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+
+	connpostgres "github.com/PeerDB-io/peer-flow/connectors/postgres"
+	"github.com/PeerDB-io/peer-flow/generated/protos"
+)
+
+func (h *FlowRequestHandler) ValidateCDCMirror(
+	ctx context.Context, req *protos.CreateCDCFlowRequest,
+) (*protos.ValidateCDCMirrorResponse, error) {
+	pgPeer, err := connpostgres.NewPostgresConnector(ctx, req.ConnectionConfigs.Source.GetPostgresConfig())
+	if err != nil {
+		return &protos.ValidateCDCMirrorResponse{
+			Ok: false,
+		}, fmt.Errorf("failed to create postgres connector: %v", err)
+	}
+
+	defer pgPeer.Close()
+
+	sourcePeerConfig := req.ConnectionConfigs.Source.GetPostgresConfig()
+	if sourcePeerConfig == nil {
+		slog.Error("/validatecdc source peer config is nil", slog.Any("peer", req.ConnectionConfigs.Source))
+		return nil, fmt.Errorf("source peer config is nil")
+	}
+
+	// Check permissions of postgres peer
+	err = pgPeer.CheckReplicationPermissions(sourcePeerConfig.User)
+	if err != nil {
+		return &protos.ValidateCDCMirrorResponse{
+			Ok: false,
+		}, fmt.Errorf("failed to check replication permissions: %v", err)
+	}
+
+	// Check source tables
+	sourceTables := make([]string, 0, len(req.ConnectionConfigs.TableMappings))
+	for _, tableMapping := range req.ConnectionConfigs.TableMappings {
+		sourceTables = append(sourceTables, tableMapping.SourceTableIdentifier)
+	}
+
+	err = pgPeer.CheckSourceTables(sourceTables, req.ConnectionConfigs.PublicationName)
+	if err != nil {
+		return &protos.ValidateCDCMirrorResponse{
+			Ok: false,
+		}, fmt.Errorf("provided source tables invalidated: %v", err)
+	}
+
+	return &protos.ValidateCDCMirrorResponse{
+		Ok: true,
+	}, nil
+}

flow/cmd/validate_peer.go

@@ -0,0 +1,73 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+
+	"github.com/PeerDB-io/peer-flow/connectors"
+	connpostgres "github.com/PeerDB-io/peer-flow/connectors/postgres"
+	"github.com/PeerDB-io/peer-flow/generated/protos"
+)
+
+func (h *FlowRequestHandler) ValidatePeer(
+	ctx context.Context,
+	req *protos.ValidatePeerRequest,
+) (*protos.ValidatePeerResponse, error) {
+	if req.Peer == nil {
+		return &protos.ValidatePeerResponse{
+			Status:  protos.ValidatePeerStatus_INVALID,
+			Message: "no peer provided",
+		}, nil
+	}
+
+	if len(req.Peer.Name) == 0 {
+		return &protos.ValidatePeerResponse{
+			Status:  protos.ValidatePeerStatus_INVALID,
+			Message: "no peer name provided",
+		}, nil
+	}
+
+	conn, err := connectors.GetConnector(ctx, req.Peer)
+	if err != nil {
+		return &protos.ValidatePeerResponse{
+			Status: protos.ValidatePeerStatus_INVALID,
+			Message: fmt.Sprintf("peer type is missing or "+
+				"your requested configuration for %s peer %s was invalidated: %s",
+				req.Peer.Type, req.Peer.Name, err),
+		}, nil
+	}
+
+	defer conn.Close()
+
+	if req.Peer.Type == protos.DBType_POSTGRES {
+		version, err := conn.(*connpostgres.PostgresConnector).GetPostgresVersion()
+		if err != nil {
+			slog.Error("/peer/validate: pg version check", slog.Any("error", err))
+			return nil, err
+		}
+
+		if version < 12 {
+			return &protos.ValidatePeerResponse{
+				Status: protos.ValidatePeerStatus_INVALID,
+				Message: fmt.Sprintf("%s peer %s must be of version 12 or above. Current version: %d",
+					req.Peer.Type, req.Peer.Name, version),
+			}, nil
+		}
+	}
+
+	connErr := conn.ConnectionActive()
+	if connErr != nil {
+		return &protos.ValidatePeerResponse{
+			Status: protos.ValidatePeerStatus_INVALID,
+			Message: fmt.Sprintf("failed to establish active connection to %s peer %s: %v",
+				req.Peer.Type, req.Peer.Name, connErr),
+		}, nil
+	}
+
+	return &protos.ValidatePeerResponse{
+		Status: protos.ValidatePeerStatus_VALID,
+		Message: fmt.Sprintf("%s peer %s is valid",
+			req.Peer.Type, req.Peer.Name),
+	}, nil
+}

flow/connectors/postgres/client.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"log"
 	"regexp"
+	"strconv"
 	"strings"
 
 	"github.com/PeerDB-io/peer-flow/connectors/utils"
@@ -572,3 +573,105 @@ func (c *PostgresConnector) getCurrentLSN() (pglogrepl.LSN, error) {
 func (c *PostgresConnector) getDefaultPublicationName(jobName string) string {
 	return fmt.Sprintf("peerflow_pub_%s", jobName)
 }
+
+func (c *PostgresConnector) CheckSourceTables(tableNames []string, pubName string) error {
+	if c.pool == nil {
+		return fmt.Errorf("check tables: pool is nil")
+	}
+
+	// Check that we can select from all tables
+	for _, tableName := range tableNames {
+		var row pgx.Row
+		err := c.pool.QueryRow(c.ctx, fmt.Sprintf("SELECT * FROM %s LIMIT 0;", tableName)).Scan(&row)
+		if err != nil && err != pgx.ErrNoRows {
+			return err
+		}
+	}
+
+	// Check if tables belong to publication
+	tableArr := make([]string, 0, len(tableNames))
+	for _, tableName := range tableNames {
+		tableArr = append(tableArr, fmt.Sprintf("'%s'", tableName))
+	}
+
+	tableStr := strings.Join(tableArr, ",")
+
+	if pubName != "" {
+		var pubTableCount int
+		err := c.pool.QueryRow(c.ctx, fmt.Sprintf("select COUNT(DISTINCT(schemaname||'.'||tablename)) from pg_publication_tables "+
+			"where schemaname||'.'||tablename in (%s) and pubname=$1;", tableStr), pubName).Scan(&pubTableCount)
+		if err != nil {
+			return err
+		}
+
+		if pubTableCount != len(tableNames) {
+			return fmt.Errorf("not all tables belong to publication")
+		}
+	}
+
+	return nil
+}
+
+func (c *PostgresConnector) CheckReplicationPermissions(username string) error {
+	if c.pool == nil {
+		return fmt.Errorf("check replication permissions: pool is nil")
+	}
+
+	var replicationRes bool
+	err := c.pool.QueryRow(c.ctx, "SELECT rolreplication FROM pg_roles WHERE rolname = $1;", username).Scan(&replicationRes)
+	if err != nil {
+		return err
+	}
+
+	if !replicationRes {
+		return fmt.Errorf("postgres user does not have replication role")
+	}
+
+	// check wal_level
+	var walLevel string
+	err = c.pool.QueryRow(c.ctx, "SHOW wal_level;").Scan(&walLevel)
+	if err != nil {
+		return err
+	}
+
+	if walLevel != "logical" {
+		return fmt.Errorf("wal_level is not logical")
+	}
+
+	// max_wal_senders must be at least 2
+	var maxWalSendersRes string
+	err = c.pool.QueryRow(c.ctx, "SHOW max_wal_senders;").Scan(&maxWalSendersRes)
+	if err != nil {
+		return err
+	}
+
+	maxWalSenders, err := strconv.Atoi(maxWalSendersRes)
+	if err != nil {
+		return err
+	}
+
+	if maxWalSenders < 2 {
+		return fmt.Errorf("max_wal_senders must be at least 2")
+	}
+
+	return nil
+}
+
+func (c *PostgresConnector) GetPostgresVersion() (int, error) {
+	if c.pool == nil {
+		return -1, fmt.Errorf("version check: pool is nil")
+	}
+
+	var versionRes string
+	err := c.pool.QueryRow(c.ctx, "SHOW server_version_num;").Scan(&versionRes)
+	if err != nil {
+		return -1, err
+	}
+
+	version, err := strconv.Atoi(versionRes)
+	if err != nil {
+		return -1, err
+	}
+
+	return version / 10000, nil
+}

protos/route.proto

@@ -197,6 +197,10 @@ message MirrorStatusResponse {
   peerdb_flow.FlowStatus current_flow_state = 5;
 }
 
+message ValidateCDCMirrorResponse{
+  bool ok = 1;
+}
+
 message FlowStateChangeRequest {
   string flow_job_name = 1;
   peerdb_flow.FlowStatus requested_flow_state = 2;
@@ -224,6 +228,12 @@ service FlowService {
       post: "/v1/peers/validate",
       body: "*"
      };
+  }
+    rpc ValidateCDCMirror(CreateCDCFlowRequest) returns (ValidateCDCMirrorResponse) {
+    option (google.api.http) = {
+      post: "/v1/mirrors/cdc/validate",
+      body: "*"
+     };
   }
   rpc CreatePeer(CreatePeerRequest) returns (CreatePeerResponse) {
     option (google.api.http) = {

ui/app/api/mirrors/cdc/route.ts

@@ -1,8 +1,5 @@
 import { UCreateMirrorResponse } from '@/app/dto/MirrorsDTO';
-import {
-  CreateCDCFlowRequest,
-  CreateCDCFlowResponse,
-} from '@/grpc_generated/route';
+import { CreateCDCFlowRequest } from '@/grpc_generated/route';
 import { GetFlowHttpAddressFromEnv } from '@/rpc/http';
 
 export async function POST(request: Request) {
@@ -15,18 +12,18 @@ export async function POST(request: Request) {
     createCatalogEntry: true,
   };
   try {
-    const createStatus: CreateCDCFlowResponse = await fetch(
-      `${flowServiceAddr}/v1/flows/cdc/create`,
-      {
-        method: 'POST',
-        body: JSON.stringify(req),
-      }
-    ).then((res) => {
+    const createStatus = await fetch(`${flowServiceAddr}/v1/flows/cdc/create`, {
+      method: 'POST',
+      body: JSON.stringify(req),
+    }).then((res) => {
       return res.json();
     });
 
+    if (!createStatus.worflowId) {
+      return new Response(JSON.stringify(createStatus));
+    }
     let response: UCreateMirrorResponse = {
-      created: !!createStatus.worflowId,
+      created: true,
     };
 
     return new Response(JSON.stringify(response));