rdev/internal/adapter/kubernetes/executor.go

// Package kubernetes provides Kubernetes-based implementations of port interfaces.
package kubernetes

import (
	"bufio"
	"context"
	"fmt"
	"io"
	"os/exec"
	"sync"
	"time"

	"github.com/orchard9/rdev/internal/domain"
	"github.com/orchard9/rdev/internal/port"
)

// Executor implements port.CommandExecutor using kubectl exec.
type Executor struct {
	namespace string
	mu        sync.RWMutex

	// Track active commands for cancellation
	activeCommands map[domain.CommandID]context.CancelFunc
	activeMu       sync.Mutex
}

// NewExecutor creates a new Kubernetes command executor.
func NewExecutor(namespace string) *Executor {
	return &Executor{
		namespace:      namespace,
		activeCommands: make(map[domain.CommandID]context.CancelFunc),
	}
}

// Ensure Executor implements port.CommandExecutor at compile time.
var _ port.CommandExecutor = (*Executor)(nil)

// Execute runs a command in the target pod and streams output to the handler.
func (e *Executor) Execute(ctx context.Context, cmd *domain.Command, podName string, handler domain.OutputHandler) (*domain.CommandResult, error) {
	e.mu.RLock()
	namespace := e.namespace
	e.mu.RUnlock()

	// Create cancellable context for this command
	cmdCtx, cancel := context.WithCancel(ctx)
	defer cancel()

	// Track for potential cancellation
	e.activeMu.Lock()
	e.activeCommands[cmd.ID] = cancel
	e.activeMu.Unlock()

	defer func() {
		e.activeMu.Lock()
		delete(e.activeCommands, cmd.ID)
		e.activeMu.Unlock()
	}()

	startTime := time.Now()
	var args []string

	switch cmd.Type {
	case domain.CommandTypeClaude:
		// claude "prompt"
		args = []string{
			"exec", "-n", namespace, podName, "--",
			"claude", cmd.Args[0], // prompt is first arg
		}
	case domain.CommandTypeShell:
		// bash -c "command"
		args = []string{
			"exec", "-n", namespace, podName, "--",
			"bash", "-c", cmd.Args[0], // command is first arg
		}
	case domain.CommandTypeGit:
		// git <args...>
		args = append([]string{
			"exec", "-n", namespace, podName, "--",
			"git", "-C", "/workspace",
		}, cmd.Args...)
	default:
		return &domain.CommandResult{
			CommandID: cmd.ID,
			ExitCode:  1,
			Error:     fmt.Errorf("unknown command type: %s", cmd.Type),
		}, nil
	}

	// Create the kubectl command
	kubectl := exec.CommandContext(cmdCtx, "kubectl", args...)

	// Get stdout and stderr pipes
	stdout, err := kubectl.StdoutPipe()
	if err != nil {
		return &domain.CommandResult{
			CommandID: cmd.ID,
			ExitCode:  1,
			Error:     fmt.Errorf("stdout pipe: %w", err),
		}, nil
	}
	stderr, err := kubectl.StderrPipe()
	if err != nil {
		return &domain.CommandResult{
			CommandID: cmd.ID,
			ExitCode:  1,
			Error:     fmt.Errorf("stderr pipe: %w", err),
		}, nil
	}

	// Start the command
	if err := kubectl.Start(); err != nil {
		return &domain.CommandResult{
			CommandID: cmd.ID,
			ExitCode:  1,
			Error:     fmt.Errorf("start: %w", err),
		}, nil
	}

	// Stream output concurrently
	var wg sync.WaitGroup
	wg.Add(2)

	go func() {
		defer wg.Done()
		streamOutput(stdout, "stdout", handler)
	}()

	go func() {
		defer wg.Done()
		streamOutput(stderr, "stderr", handler)
	}()

	// Wait for output to be consumed
	wg.Wait()

	// Wait for command to complete
	err = kubectl.Wait()
	duration := time.Since(startTime)

	result := &domain.CommandResult{
		CommandID:  cmd.ID,
		DurationMs: duration.Milliseconds(),
	}

	if err != nil {
		if exitError, ok := err.(*exec.ExitError); ok {
			result.ExitCode = exitError.ExitCode()
		} else {
			result.ExitCode = 1
			result.Error = err
		}
	}

	return result, nil
}

// streamOutput reads from a reader and sends each line to the handler.
func streamOutput(r io.Reader, stream string, handler domain.OutputHandler) {
	scanner := bufio.NewScanner(r)
	// Increase buffer size for long lines
	buf := make([]byte, 0, 64*1024)
	scanner.Buffer(buf, 1024*1024)

	for scanner.Scan() {
		handler(domain.OutputLine{
			Stream:    stream,
			Line:      scanner.Text(),
			Timestamp: time.Now(),
		})
	}
}

// Cancel attempts to cancel a running command.
func (e *Executor) Cancel(ctx context.Context, cmdID domain.CommandID) error {
	e.activeMu.Lock()
	defer e.activeMu.Unlock()

	cancel, exists := e.activeCommands[cmdID]
	if !exists {
		return domain.ErrCommandNotFound
	}

	cancel()
	return nil
}

// PodExists checks if a pod exists and is running.
func (e *Executor) PodExists(ctx context.Context, podName string) (bool, error) {
	e.mu.RLock()
	namespace := e.namespace
	e.mu.RUnlock()

	cmd := exec.CommandContext(ctx, "kubectl",
		"get", "pod", podName,
		"-n", namespace,
		"-o", "jsonpath={.status.phase}",
	)

	output, err := cmd.Output()
	if err != nil {
		// Pod doesn't exist or error
		return false, nil
	}

	return string(output) == "Running", nil
}

// CheckConnection verifies connectivity to the Kubernetes cluster.
func (e *Executor) CheckConnection(ctx context.Context) error {
	cmd := exec.CommandContext(ctx, "kubectl", "cluster-info", "--request-timeout=5s")
	return cmd.Run()
}