refactor: address PR review — simplify streaming, remove apiUrl, extract stepStreamer

- Remove apiUrl from StreamLogs interface/Config — client uses internal baseUrl
- Extract inline closures into stepStreamer struct (step_streamer.go)
- Simplify drain: cancel() + wg.Wait() instead of timeout race
- Use defer streamer.drain() for cleanup safety
- Rename waitForStageWithStepCallback → waitForStage (no callback indirection)
- Replace fmt with log framework in pipeline.go
- Only stream steps with state 'running' (not already-completed ones)
- Regenerate mocks

Signed-off-by: Alex <132889147+alexvcodesphere@users.noreply.github.com>

Author: alexvcodesphere

api/client.go

@@ -13,9 +13,10 @@ import (
 )
 
 type Client struct {
-	ctx  context.Context
-	api  *openapi_client.APIClient
-	time Time
+	ctx     context.Context
+	api     *openapi_client.APIClient
+	time    Time
+	baseUrl *url.URL
 }
 
 type Configuration struct {
@@ -40,9 +41,10 @@ func (c Configuration) GetApiUrl() *url.URL {
 // For use in tests
 func NewClientWithCustomDeps(ctx context.Context, opts Configuration, api *openapi_client.APIClient, time Time) *Client {
 	return &Client{
-		ctx:  context.WithValue(ctx, openapi_client.ContextAccessToken, opts.Token),
-		api:  api,
-		time: time,
+		ctx:     context.WithValue(ctx, openapi_client.ContextAccessToken, opts.Token),
+		api:     api,
+		time:    time,
+		baseUrl: opts.GetApiUrl(),
 	}
 }
 

api/workspace.go

@@ -229,8 +229,8 @@ type logEntry struct {
 // log entries to the provided writer until the context is cancelled or the
 // stream ends. This is used during pipeline execution to provide real-time
 // log output.
-func (c *Client) StreamLogs(ctx context.Context, apiUrl string, wsId int, stage string, step int, w io.Writer) error {
-	endpoint := fmt.Sprintf("%s/workspaces/%d/logs/%s/%d", apiUrl, wsId, stage, step)
+func (c *Client) StreamLogs(ctx context.Context, wsId int, stage string, step int, w io.Writer) error {
+	endpoint := fmt.Sprintf("%s/workspaces/%d/logs/%s/%d", c.baseUrl.String(), wsId, stage, step)
 
 	req, err := http.NewRequestWithContext(ctx, "GET", endpoint, nil)
 	if err != nil {

cli/cmd/client.go

@@ -32,7 +32,7 @@ type Client interface {
 	GetPipelineState(wsId int, stage string) ([]api.PipelineStatus, error)
 	GitPull(wsId int, remote string, branch string) error
 	DeployLandscape(wsId int, profile string) error
-	StreamLogs(ctx context.Context, apiUrl string, wsId int, stage string, step int, w io.Writer) error
+	StreamLogs(ctx context.Context, wsId int, stage string, step int, w io.Writer) error
 }
 
 // CommandExecutor abstracts command execution for testing

cli/cmd/mocks.go

@@ -38,7 +38,7 @@ func (c *StartPipelineCmd) RunE(_ *cobra.Command, args []string) error {
 		return fmt.Errorf("failed to create Codesphere client: %w", err)
 	}
 
-	return c.StartPipelineStages(client, workspaceId, args, c.Opts.GetApiUrl())
+	return c.StartPipelineStages(client, workspaceId, args)
 }
 
 func AddStartPipelineCmd(start *cobra.Command, opts GlobalOptions) {
@@ -76,15 +76,10 @@ func AddStartPipelineCmd(start *cobra.Command, opts GlobalOptions) {
 	p.cmd.RunE = p.RunE
 }
 
-func (c *StartPipelineCmd) StartPipelineStages(client Client, wsId int, stages []string, apiUrl ...string) error {
-	url := ""
-	if len(apiUrl) > 0 {
-		url = apiUrl[0]
-	}
+func (c *StartPipelineCmd) StartPipelineStages(client Client, wsId int, stages []string) error {
 	runner := pipeline.NewRunner(client, c.Time)
 	return runner.RunStages(wsId, stages, pipeline.Config{
 		Profile: *c.Opts.Profile,
 		Timeout: *c.Opts.Timeout,
-		ApiUrl:  url,
 	})
 }

cli/cmd/start_pipeline.go

@@ -10,9 +10,7 @@ import (
 	"fmt"
 	"io"
 	"log"
-	"os"
 	"slices"
-	"sync"
 	"time"
 
 	"github.com/codesphere-cloud/cs-go/api"
@@ -25,14 +23,13 @@ type Client interface {
 	StartPipelineStage(wsId int, profile string, stage string) error
 	GetPipelineState(wsId int, stage string) ([]api.PipelineStatus, error)
 	DeployLandscape(wsId int, profile string) error
-	StreamLogs(ctx context.Context, apiUrl string, wsId int, stage string, step int, w io.Writer) error
+	StreamLogs(ctx context.Context, wsId int, stage string, step int, w io.Writer) error
 }
 
 // Config holds parameters for pipeline execution.
 type Config struct {
 	Profile string
 	Timeout time.Duration
-	ApiUrl  string
 }
 
 // Runner orchestrates pipeline stage execution.
@@ -61,11 +58,11 @@ func (r *Runner) RunStages(wsId int, stages []string, cfg Config) error {
 	for _, stage := range stages {
 		// Sync the landscape before the run stage
 		if stage == "run" {
-			fmt.Println("  🔄 Syncing landscape...")
+			log.Println("  🔄 Syncing landscape...")
 			if err := r.Client.DeployLandscape(wsId, cfg.Profile); err != nil {
 				return fmt.Errorf("syncing landscape: %w", err)
 			}
-			fmt.Println("  ✅ Landscape synced.")
+			log.Println("  ✅ Landscape synced.")
 		}
 
 		if err := r.runStage(wsId, stage, cfg); err != nil {
@@ -83,63 +80,13 @@ func (r *Runner) runStage(wsId int, stage string, cfg Config) error {
 		return fmt.Errorf("failed to start pipeline stage %s: %w", stage, err)
 	}
 
-	// Step-aware log streaming for non-run stages.
-	// Each step gets its own context; when a new step is discovered the
-	// previous step's stream is cancelled and drained before moving on.
-	streamEnabled := stage != "run" && cfg.ApiUrl != ""
-	streamingStep := -1
-	var stepCancel context.CancelFunc
-	var stepWg sync.WaitGroup
-
-	// drainStream waits for the current stream to deliver logs, then cancels.
-	drainStream := func() {
-		if stepCancel == nil {
-			return
-		}
-		done := make(chan struct{})
-		go func() {
-			stepWg.Wait()
-			close(done)
-		}()
-		select {
-		case <-done:
-		case <-time.After(3 * time.Second):
-			stepCancel()
-			stepWg.Wait()
-		}
-	}
-
-	startStreamForStep := func(step int, totalSteps int) {
-		if !streamEnabled || step <= streamingStep {
-			return
-		}
-
-		// Drain previous step before starting next
-		drainStream()
-
-		streamingStep = step
-		fmt.Printf("\n  📋 Step %d/%d\n", step+1, totalSteps)
-
-		ctx, cancel := context.WithCancel(context.Background())
-		stepCancel = cancel
-		stepWg.Add(1)
-		go func() {
-			defer stepWg.Done()
-			if err := r.Client.StreamLogs(ctx, cfg.ApiUrl, wsId, stage, step, os.Stdout); err != nil {
-				_, _ = fmt.Fprintf(os.Stderr, "⚠ log stream error (step %d): %v\n", step, err)
-			}
-		}()
-	}
-
-	err := r.waitForStageWithStepCallback(wsId, stage, cfg, startStreamForStep)
-
-	// Drain final step's logs
-	drainStream()
+	streamer := newStepStreamer(r.Client, wsId, stage)
+	defer streamer.drain()
 
-	return err
+	return r.waitForStage(wsId, stage, cfg, streamer)
 }
 
-func (r *Runner) waitForStageWithStepCallback(wsId int, stage string, cfg Config, onStep func(step int, total int)) error {
+func (r *Runner) waitForStage(wsId int, stage string, cfg Config, streamer *stepStreamer) error {
 	delay := 5 * time.Second
 	timeout := cfg.Timeout
 	if timeout == 0 {
@@ -156,17 +103,14 @@ func (r *Runner) waitForStageWithStepCallback(wsId int, stage string, cfg Config
 		}
 
 		// Discover active step from IDE server's Steps array
-		if onStep != nil {
-			for _, s := range status {
-				if s.Server == IdeServer {
-					total := len(s.Steps)
-					for i, step := range s.Steps {
-						if step.State == "running" || step.State == "success" {
-							onStep(i, total)
-						}
+		for _, s := range status {
+			if s.Server == IdeServer {
+				for i, step := range s.Steps {
+					if step.State == "running" {
+						streamer.startStep(i, len(s.Steps))
 					}
-					break
 				}
+				break
 			}
 		}