Decouple FlowConnectionConfigs from API (#3605)

In #3589 we want to stop using a field in a way that's not error-prone
in the rest of the code (so, removing) but also doesn't break API
backcompat (so, not removing).

Trying to split FlowConnectionConfigs into two contracts that are made
sure to be in sync with a little codegen. Generated code has small TODOs
that would be removed by #3589 if we go this route.

Validated that renaming the type of a workflow arg works well with
history replay, only the payload is important.

Author: ilidemi

.github/actions/genprotos/action.yml

@@ -9,11 +9,10 @@ runs:
       uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4
       with:
         path: |
-          ./flow/generated/protos
-          ./flow/cmd/typed_handler.go
+          ./flow/generated
           ./nexus/pt/src/gen
           ./ui/grpc_generated
-        key: ${{ runner.os }}-build-genprotos-${{ hashFiles('buf.gen.yaml', './protos/peers.proto', './protos/flow.proto', './protos/route.proto', './flow/cmd/gen-grpc-wrapper/main.go') }}
+        key: ${{ runner.os }}-build-genprotos-${{ hashFiles('buf.gen.yaml', './protos/peers.proto', './protos/flow.proto', './protos/route.proto', './flow/cmd/codegen/*.go') }}
 
     - if: steps.cache.outputs.cache-hit != 'true'
       uses: bufbuild/buf-action@8f4a1456a0ab6a1eb80ba68e53832e6fcfacc16c # v1

.gitignore

@@ -13,7 +13,7 @@ flow/peer-flow
 go.work
 go.work.sum
 
-# generated protobuf files
+# generated protobuf files and go generate
 ui/grpc_generated
 flow/generated
 nexus/pt/src/gen

flow/.gitignore

@@ -10,6 +10,3 @@ vendor/
 
 # goenv local version. See https://github.com/syndbg/goenv/blob/master/COMMANDS.md#goenv-local for more info.
 .go-version
-
-# generated files
-cmd/typed_handler.go

flow/.golangci.yml

@@ -60,6 +60,8 @@ linters:
       settings:
         hugeParam:
           sizeThreshold: 1024
+        ruleguard:
+          rules: cmd/codegen/ruleguard.rules.go
     gosec:
       excludes:
         - G115

flow/activities/flowable.go

@@ -283,7 +283,7 @@ func (a *FlowableActivity) CreateNormalizedTable(
 
 func (a *FlowableActivity) SyncFlow(
 	ctx context.Context,
-	config *protos.FlowConnectionConfigs,
+	config *protos.FlowConnectionConfigsCore,
 	options *protos.SyncFlowOptions,
 ) error {
 	var currentSyncFlowNum atomic.Int32
@@ -414,7 +414,7 @@ func (a *FlowableActivity) SyncFlow(
 
 func (a *FlowableActivity) syncRecords(
 	ctx context.Context,
-	config *protos.FlowConnectionConfigs,
+	config *protos.FlowConnectionConfigsCore,
 	options *protos.SyncFlowOptions,
 	srcConn connectors.CDCPullConnector,
 	normRequests *concurrency.LastChan,
@@ -461,7 +461,7 @@ func (a *FlowableActivity) syncRecords(
 
 func (a *FlowableActivity) syncPg(
 	ctx context.Context,
-	config *protos.FlowConnectionConfigs,
+	config *protos.FlowConnectionConfigsCore,
 	options *protos.SyncFlowOptions,
 	srcConn connectors.CDCPullPgConnector,
 	normRequests *concurrency.LastChan,
@@ -905,14 +905,14 @@ func (a *FlowableActivity) ScheduledTasks(ctx context.Context) error {
 }
 
 type flowInformation struct {
-	config     *protos.FlowConnectionConfigs
+	config     *protos.FlowConnectionConfigsCore
 	updatedAt  time.Time
 	workflowID string
 }
 
 type metricsFlowMetadata struct {
 	updatedAt           time.Time
-	config              *protos.FlowConnectionConfigs
+	config              *protos.FlowConnectionConfigsCore
 	name                string
 	workflowID          string
 	sourcePeerName      string
@@ -1129,7 +1129,7 @@ func (a *FlowableActivity) getFlowsForMetrics(ctx context.Context) ([]metricsFlo
 
 	infos, err := pgx.CollectRows(rows, func(row pgx.CollectableRow) (metricsFlowMetadata, error) {
 		f := metricsFlowMetadata{
-			config: &protos.FlowConnectionConfigs{},
+			config: &protos.FlowConnectionConfigsCore{},
 		}
 		var configProto []byte
 		if err := rows.Scan(
@@ -1191,7 +1191,7 @@ func (a *FlowableActivity) RecordSlotSizes(ctx context.Context) error {
 			return flowInformation{}, err
 		}
 
-		var config protos.FlowConnectionConfigs
+		var config protos.FlowConnectionConfigsCore
 		if err := proto.Unmarshal(configProto, &config); err != nil {
 			return flowInformation{}, err
 		}
@@ -1543,7 +1543,7 @@ func (a *FlowableActivity) ReplicateXminPartition(ctx context.Context,
 	}
 }
 
-func (a *FlowableActivity) AddTablesToPublication(ctx context.Context, cfg *protos.FlowConnectionConfigs,
+func (a *FlowableActivity) AddTablesToPublication(ctx context.Context, cfg *protos.FlowConnectionConfigsCore,
 	additionalTableMappings []*protos.TableMapping,
 ) error {
 	ctx = context.WithValue(ctx, shared.FlowNameKey, cfg.FlowJobName)
@@ -1571,7 +1571,7 @@ func (a *FlowableActivity) AddTablesToPublication(ctx context.Context, cfg *prot
 
 func (a *FlowableActivity) RemoveTablesFromPublication(
 	ctx context.Context,
-	cfg *protos.FlowConnectionConfigs,
+	cfg *protos.FlowConnectionConfigsCore,
 	removedTablesMapping []*protos.TableMapping,
 ) error {
 	ctx = context.WithValue(ctx, shared.FlowNameKey, cfg.FlowJobName)
@@ -1599,7 +1599,7 @@ func (a *FlowableActivity) RemoveTablesFromPublication(
 
 func (a *FlowableActivity) RemoveTablesFromRawTable(
 	ctx context.Context,
-	cfg *protos.FlowConnectionConfigs,
+	cfg *protos.FlowConnectionConfigsCore,
 	tablesToRemove []*protos.TableMapping,
 ) error {
 	ctx = context.WithValue(ctx, shared.FlowNameKey, cfg.FlowJobName)
@@ -1647,7 +1647,7 @@ func (a *FlowableActivity) RemoveTablesFromRawTable(
 
 func (a *FlowableActivity) RemoveTablesFromCatalog(
 	ctx context.Context,
-	cfg *protos.FlowConnectionConfigs,
+	cfg *protos.FlowConnectionConfigsCore,
 	tablesToRemove []*protos.TableMapping,
 ) error {
 	removedTables := make([]string, 0, len(tablesToRemove))
@@ -1768,7 +1768,7 @@ func (a *FlowableActivity) GetFlowMetadata(
 	}, nil
 }
 
-func (a *FlowableActivity) UpdateCDCConfigInCatalogActivity(ctx context.Context, cfg *protos.FlowConnectionConfigs) error {
+func (a *FlowableActivity) UpdateCDCConfigInCatalogActivity(ctx context.Context, cfg *protos.FlowConnectionConfigsCore) error {
 	return internal.UpdateCDCConfigInCatalog(ctx, a.CatalogPool, internal.LoggerFromCtx(ctx), cfg)
 }
 

flow/activities/flowable_core.go

@@ -79,7 +79,7 @@ func (a *FlowableActivity) getTableNameSchemaMapping(ctx context.Context, flowNa
 
 func (a *FlowableActivity) applySchemaDeltas(
 	ctx context.Context,
-	config *protos.FlowConnectionConfigs,
+	config *protos.FlowConnectionConfigsCore,
 	options *protos.SyncFlowOptions,
 	schemaDeltas []*protos.TableSchemaDelta,
 ) error {
@@ -111,7 +111,7 @@ func (a *FlowableActivity) applySchemaDeltas(
 func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncConnectorCore, Items model.Items](
 	ctx context.Context,
 	a *FlowableActivity,
-	config *protos.FlowConnectionConfigs,
+	config *protos.FlowConnectionConfigsCore,
 	options *protos.SyncFlowOptions,
 	srcConn TPull,
 	normRequests *concurrency.LastChan,
@@ -624,7 +624,7 @@ func (a *FlowableActivity) maintainReplConn(
 
 func (a *FlowableActivity) startNormalize(
 	ctx context.Context,
-	config *protos.FlowConnectionConfigs,
+	config *protos.FlowConnectionConfigsCore,
 	batchID int64,
 	normalizeResponses *concurrency.LastChan,
 ) error {
@@ -683,7 +683,7 @@ func (a *FlowableActivity) startNormalize(
 func (a *FlowableActivity) normalizeLoop(
 	ctx context.Context,
 	logger log.Logger,
-	config *protos.FlowConnectionConfigs,
+	config *protos.FlowConnectionConfigsCore,
 	syncDone <-chan struct{},
 	normalizeRequests *concurrency.LastChan,
 	normalizeResponses *concurrency.LastChan,

flow/activities/snapshot_activity.go

@@ -172,7 +172,7 @@ func (a *SnapshotActivity) GetPeerType(ctx context.Context, name string) (protos
 
 func (a *SnapshotActivity) GetDefaultPartitionKeyForTables(
 	ctx context.Context,
-	input *protos.FlowConnectionConfigs,
+	input *protos.FlowConnectionConfigsCore,
 ) (*protos.GetDefaultPartitionKeyForTablesOutput, error) {
 	connector, err := connectors.GetByNameAs[connectors.QRepPullConnectorCore](ctx, nil, a.CatalogPool, input.SourceName)
 	if err != nil {

flow/cmd/api.go

@@ -23,6 +23,7 @@ import (
 	"google.golang.org/grpc/health/grpc_health_v1"
 	"google.golang.org/grpc/reflection"
 
+	"github.com/PeerDB-io/peerdb/flow/generated/grpc_handler"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
 	"github.com/PeerDB-io/peerdb/flow/middleware"
@@ -259,7 +260,7 @@ func APIMain(ctx context.Context, args *APIServerParams) error {
 		return fmt.Errorf("unable to start scheduler workflow: %w", err)
 	}
 
-	protos.RegisterFlowServiceServer(grpcServer, NewFlowServiceAdapter(flowHandler))
+	protos.RegisterFlowServiceServer(grpcServer, grpc_handler.NewFlowServiceAdapter(flowHandler))
 	grpc_health_v1.RegisterHealthServer(grpcServer, health.NewServer())
 	reflection.Register(grpcServer)
 

flow/cmd/api_error.go

@@ -8,15 +8,13 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/protoadapt"
+
+	"github.com/PeerDB-io/peerdb/flow/generated/grpc_handler"
 )
 
 // APIError is a strongly-typed error that must be a gRPC status error.
 // All handler methods should return this type instead of the generic error interface.
-type APIError interface {
-	error
-	GRPCStatus() *status.Status
-	Code() codes.Code
-}
+type APIError = grpc_handler.APIError
 
 type apiError struct {
 	status *status.Status