fix(flowable): do not gate pg dest logic by pg source (#4483)

Postgres destination mirrors store metadata on the destination PG peer
instead of catalog. We handle this logic for PG to PG mirrors but not
MySQL to PG or Mongo to PG.

This PR extends this logic to fix CDC for those mirrors too, by only
checking if destination is PG regardless of what source is

Author: Amogh-Bharadwaj

flow/activities/flowable_core.go

@@ -201,18 +201,16 @@ func pullAndSyncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDC
 	}
 
 	syncBatchID, err := func() (int64, error) {
-		// special case pg-pg replication, where batch ID is stored on destination instead of catalog
-		if _, isSourcePg := any(srcConn).(*connpostgres.PostgresConnector); isSourcePg {
-			dstPgConn, dstPgClose, err := connectors.GetPostgresConnectorByName(ctx, config.Env, a.CatalogPool, config.DestinationName)
-			if err != nil {
-				if !errors.Is(err, errors.ErrUnsupported) {
-					return 0, fmt.Errorf("failed to get destination connector to get last sync batch ID: %w", err)
-				}
-				// else fallthrough to loading from catalog
-			} else {
-				defer dstPgClose(ctx)
-				return dstPgConn.GetLastSyncBatchID(ctx, flowName)
+		// when destination is PG, batch ID is stored on destination instead of catalog
+		dstPgConn, dstPgClose, err := connectors.GetPostgresConnectorByName(ctx, config.Env, a.CatalogPool, config.DestinationName)
+		if err != nil {
+			if !errors.Is(err, errors.ErrUnsupported) {
+				return 0, fmt.Errorf("failed to get destination connector to get last sync batch ID: %w", err)
 			}
+			// else fallthrough to loading from catalog
+		} else {
+			defer dstPgClose(ctx)
+			return dstPgConn.GetLastSyncBatchID(ctx, flowName)
 		}
 		pgMetadata := connmetadata.NewPostgresMetadataFromCatalog(logger, a.CatalogPool)
 		return pgMetadata.GetLastSyncBatchID(ctx, flowName)