Fix Postgres CDC issues (table mappings, error handling, state, CDC scaling, etc) (#415)

Author: tpanetti

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/hashicorp/terraform-plugin-docs v0.24.0
 	github.com/hashicorp/terraform-plugin-framework v1.16.1
 	github.com/hashicorp/terraform-plugin-framework-validators v0.19.0
+	github.com/hashicorp/terraform-plugin-go v0.29.0
 	github.com/hashicorp/terraform-plugin-log v0.10.0
 	github.com/stretchr/testify v1.10.0
 	k8s.io/apimachinery v0.34.1
@@ -45,7 +46,6 @@ require (
 	github.com/hashicorp/hc-install v0.9.2 // indirect
 	github.com/hashicorp/terraform-exec v0.24.0 // indirect
 	github.com/hashicorp/terraform-json v0.27.2 // indirect
-	github.com/hashicorp/terraform-plugin-go v0.29.0 // indirect
 	github.com/hashicorp/terraform-registry-address v0.4.0 // indirect
 	github.com/hashicorp/terraform-svchost v0.1.1 // indirect
 	github.com/hashicorp/yamux v0.1.2 // indirect

pkg/internal/api/clickpipe.go

@@ -402,3 +402,79 @@ func (c *ClientImpl) UpdateClickPipeSettings(ctx context.Context, serviceId stri
 
 	return settingsResponse.Result, nil
 }
+
+type ClickPipeCdcScaling struct {
+	ReplicaCpuMillicores int64   `json:"replicaCpuMillicores"`
+	ReplicaMemoryGb      float64 `json:"replicaMemoryGb"`
+}
+
+type ClickPipeCdcScalingRequest struct {
+	ReplicaCpuMillicores int64   `json:"replicaCpuMillicores"`
+	ReplicaMemoryGb      float64 `json:"replicaMemoryGb"`
+}
+
+func (c *ClientImpl) GetClickPipeCdcScaling(ctx context.Context, serviceId string) (*ClickPipeCdcScaling, error) {
+	req, err := http.NewRequest(http.MethodGet, c.getServicePath(serviceId, "/clickpipesCdcScaling"), nil)
+	if err != nil {
+		return nil, err
+	}
+	body, err := c.doRequest(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+
+	scalingResponse := ResponseWithResult[ClickPipeCdcScaling]{}
+	if err := json.Unmarshal(body, &scalingResponse); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal CDC scaling: %w", err)
+	}
+
+	return &scalingResponse.Result, nil
+}
+
+func (c *ClientImpl) UpdateClickPipeCdcScaling(ctx context.Context, serviceId string, request ClickPipeCdcScalingRequest) (*ClickPipeCdcScaling, error) {
+	var payload bytes.Buffer
+	if err := json.NewEncoder(&payload).Encode(request); err != nil {
+		return nil, fmt.Errorf("failed to encode CDC scaling: %w", err)
+	}
+
+	req, err := http.NewRequest(http.MethodPatch, c.getServicePath(serviceId, "/clickpipesCdcScaling"), &payload)
+	if err != nil {
+		return nil, err
+	}
+
+	body, err := c.doRequest(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+
+	scalingResponse := ResponseWithResult[ClickPipeCdcScaling]{}
+	if err := json.Unmarshal(body, &scalingResponse); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal CDC scaling: %w", err)
+	}
+
+	return &scalingResponse.Result, nil
+}
+
+func (c *ClientImpl) WaitForClickPipeCdcScaling(ctx context.Context, serviceId string, expectedCpuMillicores int64, expectedMemoryGb float64, maxElapsedTime time.Duration) (scaling *ClickPipeCdcScaling, err error) {
+	checkScaling := func() error {
+		scaling, err = c.GetClickPipeCdcScaling(ctx, serviceId)
+		if err != nil {
+			return err
+		}
+
+		// Check if the scaling values match the expected values
+		if scaling.ReplicaCpuMillicores == expectedCpuMillicores && scaling.ReplicaMemoryGb == expectedMemoryGb {
+			return nil
+		}
+
+		return fmt.Errorf("CDC scaling not yet applied: current cpu=%d (expected %d), memory=%.1f (expected %.1f)",
+			scaling.ReplicaCpuMillicores, expectedCpuMillicores, scaling.ReplicaMemoryGb, expectedMemoryGb)
+	}
+
+	if maxElapsedTime < 5*time.Second {
+		maxElapsedTime = 5 * time.Second
+	}
+
+	err = backoff.Retry(checkScaling, backoff.NewExponentialBackOff(backoff.WithMaxElapsedTime(maxElapsedTime), backoff.WithMaxInterval(maxElapsedTime/5)))
+	return
+}

pkg/internal/api/clickpipe_models.go

@@ -144,12 +144,12 @@ type ClickPipeKinesisSource struct {
 }
 
 type ClickPipePostgresSource struct {
-	Host                  string                          `json:"host"`
-	Port                  int                             `json:"port"`
-	Database              string                          `json:"database"`
+	Host                  string                          `json:"host,omitempty"`
+	Port                  int                             `json:"port,omitempty"`
+	Database              string                          `json:"database,omitempty"`
 	Credentials           *ClickPipeSourceCredentials     `json:"credentials,omitempty"`
-	Settings              ClickPipePostgresSettings       `json:"settings"`
-	Mappings              []ClickPipePostgresTableMapping `json:"tableMappings"`
+	Settings              *ClickPipePostgresSettings      `json:"settings,omitempty"`
+	Mappings              []ClickPipePostgresTableMapping `json:"tableMappings,omitempty"`
 	TableMappingsToRemove []ClickPipePostgresTableMapping `json:"tableMappingsToRemove,omitempty"`
 	TableMappingsToAdd    []ClickPipePostgresTableMapping `json:"tableMappingsToAdd,omitempty"`
 }
@@ -158,7 +158,7 @@ type ClickPipePostgresSettings struct {
 	SyncIntervalSeconds            *int    `json:"syncIntervalSeconds,omitempty"`
 	PullBatchSize                  *int    `json:"pullBatchSize,omitempty"`
 	PublicationName                *string `json:"publicationName,omitempty"`
-	ReplicationMode                string  `json:"replicationMode"`
+	ReplicationMode                string  `json:"replicationMode,omitempty"`
 	ReplicationSlotName            *string `json:"replicationSlotName,omitempty"`
 	AllowNullableColumns           *bool   `json:"allowNullableColumns,omitempty"`
 	InitialLoadParallelism         *int    `json:"initialLoadParallelism,omitempty"`