Implement SDK BuildPipelineSyncStages methods (#5584)

* Define types for BuildPipelineSyncStages

Signed-off-by: Shinnosuke Sawada-Dazai <shin@warashi.dev>

* Implement BuildPipelineSyncStages method

Signed-off-by: Shinnosuke Sawada-Dazai <shin@warashi.dev>

* Add config handlings

Signed-off-by: Shinnosuke Sawada-Dazai <shin@warashi.dev>

* Add ManualOperationNone constant to represent absence of manual operation

Signed-off-by: Shinnosuke Sawada-Dazai <shin@warashi.dev>

* Update BuildPipelineSyncStages to use config in deployment logic

Signed-off-by: Shinnosuke Sawada-Dazai <shin@warashi.dev>

---------

Signed-off-by: Shinnosuke Sawada-Dazai <shin@warashi.dev>

Author: Warashi

pkg/app/pipedv1/plugin/example/plugin.go

@@ -36,8 +36,8 @@ func (p *plugin) Version() string {
 }
 
 // BuildPipelineSyncStages implements sdk.PipelineSyncPlugin.
-func (p *plugin) BuildPipelineSyncStages(context.Context, *config, *sdk.Client, sdk.TODO) (sdk.TODO, error) {
-	return sdk.TODO{}, nil
+func (p *plugin) BuildPipelineSyncStages(context.Context, *config, *sdk.Client, *sdk.BuildPipelineSyncStagesRequest) (*sdk.BuildPipelineSyncStagesResponse, error) {
+	return &sdk.BuildPipelineSyncStagesResponse{}, nil
 }
 
 // ExecuteStage implements sdk.PipelineSyncPlugin.

pkg/plugin/sdk/deployment.go

@@ -18,13 +18,15 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"time"
 
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
 	config "github.com/pipe-cd/pipecd/pkg/configv1"
+	"github.com/pipe-cd/pipecd/pkg/model"
 	"github.com/pipe-cd/pipecd/pkg/plugin/api/v1alpha1/deployment"
 	"github.com/pipe-cd/pipecd/pkg/plugin/logpersister"
 	"github.com/pipe-cd/pipecd/pkg/plugin/pipedapi"
@@ -71,7 +73,7 @@ type PipelineSyncPlugin[Config, DeployTargetConfig any] interface {
 	// FetchDefinedStages returns the list of stages that the plugin can execute.
 	FetchDefinedStages() []string
 	// BuildPipelineSyncStages builds the stages that will be executed by the plugin.
-	BuildPipelineSyncStages(context.Context, *Config, *Client, TODO) (TODO, error)
+	BuildPipelineSyncStages(context.Context, *Config, *Client, *BuildPipelineSyncStagesRequest) (*BuildPipelineSyncStagesResponse, error)
 	// ExecuteStage executes the given stage.
 	ExecuteStage(context.Context, *Config, []*DeployTarget[DeployTargetConfig], *Client, logpersister.StageLogPersister, TODO) (TODO, error)
 }
@@ -156,7 +158,7 @@ func (s *DeploymentPluginServiceServer[Config, DeployTargetConfig]) DetermineStr
 	return nil, status.Errorf(codes.Unimplemented, "method DetermineStrategy not implemented")
 }
 func (s *DeploymentPluginServiceServer[Config, DeployTargetConfig]) BuildPipelineSyncStages(ctx context.Context, request *deployment.BuildPipelineSyncStagesRequest) (*deployment.BuildPipelineSyncStagesResponse, error) {
-	return nil, status.Errorf(codes.Unimplemented, "method BuildPipelineSyncStages not implemented")
+	return buildPipelineSyncStages(ctx, s.base, &s.config, nil, request) // TODO: pass the real client
 }
 func (s *DeploymentPluginServiceServer[Config, DeployTargetConfig]) BuildQuickSyncStages(context.Context, *deployment.BuildQuickSyncStagesRequest) (*deployment.BuildQuickSyncStagesResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method BuildQuickSyncStages not implemented")
@@ -213,11 +215,144 @@ func (s *PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) DetermineS
 	return &deployment.DetermineStrategyResponse{Unsupported: true}, nil
 }
 func (s *PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) BuildPipelineSyncStages(ctx context.Context, request *deployment.BuildPipelineSyncStagesRequest) (*deployment.BuildPipelineSyncStagesResponse, error) {
-	return nil, status.Errorf(codes.Unimplemented, "method BuildPipelineSyncStages not implemented")
+	return buildPipelineSyncStages(ctx, s.base, &s.config, nil, request) // TODO: pass the real client
 }
 func (s *PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) BuildQuickSyncStages(context.Context, *deployment.BuildQuickSyncStagesRequest) (*deployment.BuildQuickSyncStagesResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method BuildQuickSyncStages not implemented")
 }
 func (s *PipelineSyncPluginServiceServer[Config, DeployTargetConfig]) ExecuteStage(context.Context, *deployment.ExecuteStageRequest) (*deployment.ExecuteStageResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method ExecuteStage not implemented")
 }
+
+// buildPipelineSyncStages builds the stages that will be executed by the plugin.
+func buildPipelineSyncStages[Config, DeployTargetConfig any](ctx context.Context, plugin PipelineSyncPlugin[Config, DeployTargetConfig], config *Config, client *Client, request *deployment.BuildPipelineSyncStagesRequest) (*deployment.BuildPipelineSyncStagesResponse, error) {
+	resp, err := plugin.BuildPipelineSyncStages(ctx, config, client, newPipelineSyncStagesRequest(request))
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to build pipeline sync stages: %v", err)
+	}
+	return newPipelineSyncStagesResponse(plugin, time.Now(), request, resp)
+}
+
+// ManualOperation represents the manual operation that the user can perform.
+type ManualOperation int
+
+const (
+	// ManualOperationNone indicates that there is no manual operation.
+	ManualOperationNone ManualOperation = iota
+	// ManualOperationSkip indicates that the manual operation is to skip the stage.
+	ManualOperationSkip
+	// ManualOperationApprove indicates that the manual operation is to approve the stage.
+	ManualOperationApprove
+)
+
+// toModelEnum converts the ManualOperation to the model.ManualOperation.
+func (o ManualOperation) toModelEnum() model.ManualOperation {
+	switch o {
+	case ManualOperationNone:
+		return model.ManualOperation_MANUAL_OPERATION_NONE
+	case ManualOperationSkip:
+		return model.ManualOperation_MANUAL_OPERATION_SKIP
+	case ManualOperationApprove:
+		return model.ManualOperation_MANUAL_OPERATION_APPROVE
+	default:
+		return model.ManualOperation_MANUAL_OPERATION_UNKNOWN
+	}
+}
+
+// newPipelineSyncStagesRequest converts the request to the internal representation.
+func newPipelineSyncStagesRequest(request *deployment.BuildPipelineSyncStagesRequest) *BuildPipelineSyncStagesRequest {
+	stages := make([]StageConfig, 0, len(request.Stages))
+	for _, s := range request.GetStages() {
+		stages = append(stages, StageConfig{
+			Index:  int(s.GetIndex()),
+			Name:   s.GetName(),
+			Config: s.GetConfig(),
+		})
+	}
+	return &BuildPipelineSyncStagesRequest{
+		Rollback: request.GetRollback(),
+		Stages:   stages,
+	}
+}
+
+// newPipelineSyncStagesResponse converts the response to the external representation.
+func newPipelineSyncStagesResponse(plugin Plugin, now time.Time, request *deployment.BuildPipelineSyncStagesRequest, response *BuildPipelineSyncStagesResponse) (*deployment.BuildPipelineSyncStagesResponse, error) {
+	// Convert the request stages to a map for easier access.
+	requestStages := make(map[int]*deployment.BuildPipelineSyncStagesRequest_StageConfig, len(request.GetStages()))
+	for _, s := range request.GetStages() {
+		requestStages[int(s.GetIndex())] = s
+	}
+
+	stages := make([]*model.PipelineStage, 0, len(response.Stages))
+	for _, s := range response.Stages {
+		// Find the corresponding stage in the request.
+		requestStage, ok := requestStages[s.Index]
+		if !ok {
+			return nil, status.Errorf(codes.Internal, "missing stage with index %d in the request, it's unexpected behavior of the plugin", s.Index)
+		}
+		id := requestStage.GetId()
+		if id == "" {
+			id = fmt.Sprintf("%s-stage-%d", plugin.Name(), s.Index)
+		}
+		stages = append(stages, &model.PipelineStage{
+			Id:                 id,
+			Name:               s.Name,
+			Desc:               requestStage.GetDesc(),
+			Index:              int32(s.Index),
+			Status:             model.StageStatus_STAGE_NOT_STARTED_YET,
+			StatusReason:       "", // TODO: set the reason
+			Metadata:           s.Metadata,
+			Rollback:           s.Rollback,
+			CreatedAt:          now.Unix(),
+			UpdatedAt:          now.Unix(),
+			AvailableOperation: s.AvailableOperation.toModelEnum(),
+		})
+	}
+	return &deployment.BuildPipelineSyncStagesResponse{
+		Stages: stages,
+	}, nil
+}
+
+// BuildPipelineSyncStagesRequest is the request to build pipeline sync stages.
+// Rollback indicates whether the stages for rollback are requested.
+type BuildPipelineSyncStagesRequest struct {
+	// Rollback indicates whether the stages for rollback are requested.
+	Rollback bool
+	// Stages contains the stage names and their configurations.
+	Stages []StageConfig
+}
+
+// StageConfig represents the configuration of a stage.
+type StageConfig struct {
+	// Index is the order of the stage in the pipeline.
+	Index int
+	// Name is the name of the stage.
+	// It must be one of the stages returned by FetchDefinedStages.
+	Name string
+	// Config is the configuration of the stage.
+	// It should be marshaled to JSON bytes.
+	// The plugin should unmarshal it to the appropriate struct.
+	Config []byte
+}
+
+// BuildPipelineSyncStagesResponse is the response of the request to build pipeline sync stages.
+type BuildPipelineSyncStagesResponse struct {
+	Stages []PipelineStage
+}
+
+// PipelineStage represents a stage in the pipeline.
+type PipelineStage struct {
+	// Index is the order of the stage in the pipeline.
+	// The value must be one of the index of the stage in the request.
+	// The rollback stage should have the same index as the original stage.
+	Index int
+	// Name is the name of the stage.
+	// It must be one of the stages returned by FetchDefinedStages.
+	Name string
+	// Rollback indicates whether the stage is for rollback.
+	Rollback bool
+	// Metadata contains the metadata of the stage.
+	Metadata map[string]string
+	// AvailableOperation indicates the manual operation that the user can perform.
+	AvailableOperation ManualOperation
+}