ci/internal/process/manager.go

package process

import (
	"bufio"
	"fmt"
	"io"
	"os"
	"os/exec"
	"sync"
	"syscall"
	"time"
)

// LineSubscriber receives individual log lines as they arrive.
type LineSubscriber func(line string)

// ManagedProcess wraps a running child process with log streaming.
type ManagedProcess struct {
	mu          sync.RWMutex
	cmd         *exec.Cmd
	PID         int
	ProjectID   uint
	Status      string // "running", "stopped"
	subscribers []LineSubscriber
	logFile     *os.File
	cancel      chan struct{}
}

// Manager manages multiple running processes keyed by project ID.
type Manager struct {
	mu       sync.RWMutex
	procs    map[uint]*ManagedProcess // projectID -> process
	stopChan chan struct{}
}

// NewManager creates a new process Manager.
func NewManager() *Manager {
	return &Manager{
		procs:    make(map[uint]*ManagedProcess),
		stopChan: make(chan struct{}),
	}
}

// Start launches a command for the given project and streams stdout/stderr.
func (m *Manager) Start(projectID uint, workDir string, command string, envVars map[string]string, logWriter io.Writer) (*ManagedProcess, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	// Stop existing process for this project if any
	if existing, ok := m.procs[projectID]; ok {
		existing.Stop()
	}

	// Parse command: support simple "cmd arg1 arg2" or shell execution
	var cmd *exec.Cmd
	if isShellCommand(command) {
		cmd = exec.Command("sh", "-c", command)
	} else {
		parts := splitCommand(command)
		cmd = exec.Command(parts[0], parts[1:]...)
	}

	cmd.Dir = workDir
	cmd.Env = os.Environ()
	for k, v := range envVars {
		cmd.Env = append(cmd.Env, fmt.Sprintf("%s=%s", k, v))
	}

	// Pipe stdout and stderr
	stdout, err := cmd.StdoutPipe()
	if err != nil {
		return nil, err
	}
	stderr, err := cmd.StderrPipe()
	if err != nil {
		return nil, err
	}

	if err := cmd.Start(); err != nil {
		return nil, fmt.Errorf("failed to start process: %w", err)
	}

	mp := &ManagedProcess{
		cmd:       cmd,
		PID:       cmd.Process.Pid,
		ProjectID: projectID,
		Status:    "running",
		cancel:    make(chan struct{}),
	}

	// Stream stdout/stderr to subscribers and log writer
	go mp.streamOutput(stdout, logWriter)
	go mp.streamOutput(stderr, logWriter)

	// Monitor process exit
	go func() {
		err := cmd.Wait()
		m.mu.Lock()
		defer m.mu.Unlock()
		mp.mu.Lock()
		mp.Status = "stopped"
		if err != nil {
			mp.broadcast(fmt.Sprintf("[Process exited with error: %v]", err))
		} else {
			mp.broadcast("[Process exited successfully]")
		}
		mp.mu.Unlock()
		delete(m.procs, projectID)
	}()

	m.procs[projectID] = mp
	return mp, nil
}

// Stop terminates a running process by project ID.
func (m *Manager) Stop(projectID uint) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	mp, ok := m.procs[projectID]
	if !ok {
		return fmt.Errorf("no running process for project %d", projectID)
	}
	return mp.Stop()
}

// Stop terminates the managed process.
func (mp *ManagedProcess) Stop() error {
	mp.mu.Lock()
	defer mp.mu.Unlock()

	if mp.Status == "stopped" {
		return nil
	}

	close(mp.cancel)

	// Try graceful shutdown first
	if err := mp.cmd.Process.Signal(syscall.SIGTERM); err != nil {
		// On Windows, SIGTERM is not supported; use Kill
		mp.cmd.Process.Kill()
		return nil
	}

	// Wait with timeout
	done := make(chan error, 1)
	go func() {
		_, err := mp.cmd.Process.Wait()
		done <- err
	}()

	select {
	case <-done:
		mp.Status = "stopped"
		return nil
	case <-time.After(10 * time.Second):
		mp.cmd.Process.Kill()
		mp.Status = "stopped"
		return fmt.Errorf("process %d killed after timeout", mp.PID)
	}
}

// GetStatus returns the status of a project's process.
func (m *Manager) GetStatus(projectID uint) string {
	m.mu.RLock()
	defer m.mu.RUnlock()

	mp, ok := m.procs[projectID]
	if !ok {
		return "stopped"
	}

	mp.mu.RLock()
	defer mp.mu.RUnlock()
	return mp.Status
}

// GetProcess returns the managed process for a project, or nil.
func (m *Manager) GetProcess(projectID uint) *ManagedProcess {
	m.mu.RLock()
	defer m.mu.RUnlock()
	return m.procs[projectID]
}

// Subscribe adds a log line subscriber to a running process.
func (mp *ManagedProcess) Subscribe(fn LineSubscriber) {
	mp.mu.Lock()
	defer mp.mu.Unlock()
	mp.subscribers = append(mp.subscribers, fn)
}

// UnsubscribeAll removes all subscribers.
func (mp *ManagedProcess) UnsubscribeAll() {
	mp.mu.Lock()
	defer mp.mu.Unlock()
	mp.subscribers = nil
}

// IsRunning returns whether the process is currently running.
func (mp *ManagedProcess) IsRunning() bool {
	mp.mu.RLock()
	defer mp.mu.RUnlock()
	return mp.Status == "running"
}

func (mp *ManagedProcess) broadcast(line string) {
	for _, sub := range mp.subscribers {
		sub(line)
	}
}

func (mp *ManagedProcess) streamOutput(r io.Reader, logWriter io.Writer) {
	scanner := bufio.NewScanner(r)
	scanner.Buffer(make([]byte, 1024*1024), 1024*1024) // 1MB buffer
	for scanner.Scan() {
		line := scanner.Text()
		mp.mu.RLock()
		mp.broadcast(line)
		mp.mu.RUnlock()
		if logWriter != nil {
			fmt.Fprintln(logWriter, line)
		}
	}
}

// Shutdown stops all managed processes gracefully.
func (m *Manager) Shutdown() {
	m.mu.Lock()
	defer m.mu.Unlock()

	for id, mp := range m.procs {
		mp.Stop()
		delete(m.procs, id)
	}
}

// isShellCommand returns true if the command contains shell metacharacters.
func isShellCommand(cmd string) bool {
	for _, c := range cmd {
		switch c {
		case '|', '&', ';', '$', '>', '<', '`', '*', '?', '[', ']', '(', ')', '{', '}':
			return true
		}
	}
	return false
}

// splitCommand splits a simple command string into parts, respecting quotes.
func splitCommand(cmd string) []string {
	var parts []string
	var current []rune
	inQuote := false
	quoteChar := rune(0)

	for _, r := range cmd {
		switch {
		case r == '"' || r == '\'':
			if inQuote && r == quoteChar {
				inQuote = false
			} else if !inQuote {
				inQuote = true
				quoteChar = r
			} else {
				current = append(current, r)
			}
		case r == ' ' && !inQuote:
			if len(current) > 0 {
				parts = append(parts, string(current))
				current = nil
			}
		default:
			current = append(current, r)
		}
	}
	if len(current) > 0 {
		parts = append(parts, string(current))
	}
	return parts
}