[SDK] Implement ListStageCommands() in SDK (#5639)

* Simplify ListStageCommands()

Signed-off-by: t-kikuc <[email protected]>

* Impl ListStageCommands()

Signed-off-by: t-kikuc <[email protected]>

* Fix cases and add comments

Signed-off-by: t-kikuc <[email protected]>

---------

Signed-off-by: t-kikuc <[email protected]>

Author: t-kikuc

pkg/plugin/pipedservice/service.pb.go

@@ -48,7 +48,6 @@ service PluginService {
     rpc GetDeploymentSharedMetadata(GetDeploymentSharedMetadataRequest) returns (GetDeploymentSharedMetadataResponse) {}
 
     // ListStageCommands lists unhandled commands of the given stage and type.
-    // Currently, supported types are only APPROVE_STAGE and SKIP_STAGE.
     rpc ListStageCommands(ListStageCommandsRequest) returns (ListStageCommandsResponse) {}
 }
 
@@ -188,7 +187,6 @@ message GetDeploymentSharedMetadataResponse {
 message ListStageCommandsRequest {
     string deployment_id = 1 [(validate.rules).string.min_len = 1];
     string stage_id = 2 [(validate.rules).string.min_len = 1];
-    model.Command.Type type = 3 [(validate.rules).enum.defined_only = true];
 }
 
 message ListStageCommandsResponse {

pkg/plugin/pipedservice/service.pb.validate.go

@@ -16,12 +16,20 @@ package sdk
 
 import (
 	"context"
+	"iter"
+	"slices"
+	"time"
 
+	"github.com/pipe-cd/pipecd/pkg/model"
 	"github.com/pipe-cd/pipecd/pkg/plugin/pipedapi"
 	"github.com/pipe-cd/pipecd/pkg/plugin/pipedservice"
 	"github.com/pipe-cd/pipecd/pkg/plugin/toolregistry"
 )
 
+const (
+	listStageCommandsInterval = 5 * time.Second
+)
+
 // Client is a toolkit for interacting with the piped service.
 // It provides methods to call the piped service APIs.
 // It's a wrapper around the raw piped service client.
@@ -164,3 +172,48 @@ func (c *Client) LogPersister() StageLogPersister {
 func (c *Client) ToolRegistry() *toolregistry.ToolRegistry {
 	return c.toolRegistry
 }
+
+// ListStageCommands returns the list of stage commands of the given command types.
+func (c Client) ListStageCommands(ctx context.Context, commandTypes ...model.Command_Type) iter.Seq2[*StageCommand, error] {
+	return func(yield func(*StageCommand, error) bool) {
+		returned := map[string]struct{}{}
+
+		for {
+			resp, err := c.base.ListStageCommands(ctx, &pipedservice.ListStageCommandsRequest{
+				DeploymentId: c.deploymentID,
+				StageId:      c.stageID,
+			})
+			if err != nil {
+				if !yield(nil, err) {
+					return
+				}
+				continue
+			}
+
+			for _, command := range resp.Commands {
+				if !slices.Contains(commandTypes, command.Type) {
+					continue
+				}
+
+				if _, ok := returned[command.Id]; ok {
+					continue
+				}
+				returned[command.Id] = struct{}{}
+
+				stageCommand, err := newStageCommand(command)
+				if err != nil {
+					if !yield(nil, err) {
+						return
+					}
+					continue
+				}
+
+				if !yield(&stageCommand, nil) {
+					return
+				}
+			}
+
+			time.Sleep(listStageCommandsInterval)
+		}
+	}
+}

pkg/plugin/pipedservice/service.proto

@@ -688,3 +688,35 @@ func (o StageStatus) toModelEnum() model.StageStatus {
 		return model.StageStatus_STAGE_FAILURE
 	}
 }
+
+// StageCommand represents a command for a stage.
+type StageCommand struct {
+	Commander string
+	Type      CommandType
+}
+
+// CommandType represents the type of the command.
+type CommandType int32
+
+const (
+	CommandTypeApproveStage CommandType = iota
+	CommandTypeSkipStage
+)
+
+// newStageCommand converts the model.Command to the internal representation.
+func newStageCommand(c *model.Command) (StageCommand, error) {
+	switch c.Type {
+	case model.Command_APPROVE_STAGE:
+		return StageCommand{
+			Commander: c.GetCommander(),
+			Type:      CommandTypeApproveStage,
+		}, nil
+	case model.Command_SKIP_STAGE:
+		return StageCommand{
+			Commander: c.GetCommander(),
+			Type:      CommandTypeSkipStage,
+		}, nil
+	default:
+		return StageCommand{}, fmt.Errorf("invalid command type: %d", c.Type)
+	}
+}