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

pkg/internal/api/clickpipe.go

@@ -393,3 +393,55 @@ 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
+}

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"`

pkg/internal/api/interface.go

@@ -35,6 +35,8 @@ type Client interface {
 	DeleteClickPipe(ctx context.Context, serviceId string, clickPipeId string) error
 	GetClickPipeSettings(ctx context.Context, serviceId string, clickPipeId string) (map[string]any, error)
 	UpdateClickPipeSettings(ctx context.Context, serviceId string, clickPipeId string, settings map[string]any) (map[string]any, error)
+	GetClickPipeCdcScaling(ctx context.Context, serviceId string) (*ClickPipeCdcScaling, error)
+	UpdateClickPipeCdcScaling(ctx context.Context, serviceId string, request ClickPipeCdcScalingRequest) (*ClickPipeCdcScaling, error)
 
 	GetReversePrivateEndpointPath(serviceId, reversePrivateEndpointId string) string
 	ListReversePrivateEndpoints(ctx context.Context, serviceId string) ([]*ReversePrivateEndpoint, error)