Turbine/pkg/launchpad/dev/dev.go

package dev

import (
	"bufio"
	"context"
	"fmt"
	"io"
	"net"
	"os"
	"os/exec"
	"os/signal"
	"strings"
	"sync"
	"syscall"
	"time"

	"git.solsynth.dev/goatworks/turbine/pkg/launchpad/config"
	"github.com/rs/zerolog/log"
)

var colors = []string{
	"\033[32m", "\033[33m", "\033[34m", "\033[35m", "\033[36m", "\033[31m",
}

const colorReset = "\033[0m"

// RunDev starts all services defined in the config in development mode.
func RunDev(cfg config.LaunchpadConfig) {
	log.Info().Msg("Starting services in development mode with dependency checks...")

	ctx, cancel := context.WithCancel(context.Background())
	var wg sync.WaitGroup

	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
	go func() {
		sig := <-sigChan
		log.Info().Msgf("Received signal: %v. Shutting down all services...", sig)
		cancel()
	}()

	devNetwork := "turbine-dev-net"
	if len(cfg.Networks) > 0 {
		for name := range cfg.Networks {
			devNetwork = name
			break
		}
	}
	createDockerNetwork(devNetwork)

	// --- Dependency-aware startup ---
	serviceMap := make(map[string]config.Service)
	for _, s := range cfg.Services {
		serviceMap[s.Name] = s
	}

	started := make(map[string]chan bool)
	for _, s := range cfg.Services {
		if _, exists := started[s.Name]; !exists {
			startServiceWithDeps(ctx, &wg, s, serviceMap, started, devNetwork)
		}
	}

	wg.Wait()
	log.Info().Msg("All services have been shut down.")
}

func startServiceWithDeps(ctx context.Context, wg *sync.WaitGroup, s config.Service, serviceMap map[string]config.Service, started map[string]chan bool, network string) {
	if _, exists := started[s.Name]; exists {
		return
	}

	// Create a channel that will be closed when this service is healthy
	healthyChan := make(chan bool)
	started[s.Name] = healthyChan

	// First, recursively start dependencies
	var depNames []string
	for depName := range s.Prod.DependsOn {
		depNames = append(depNames, depName)
	}

	for _, depName := range depNames {
		if dep, ok := serviceMap[depName]; ok {
			startServiceWithDeps(ctx, wg, dep, serviceMap, started, network)
		}
	}

	// Wait for dependencies to be healthy
	log.Info().Str("service", s.Name).Msgf("Waiting for dependencies to be healthy: %v", depNames)
	for _, depName := range depNames {
		if depChan, ok := started[depName]; ok {
			log.Info().Str("service", s.Name).Msgf("Waiting for %s...", depName)
			select {
			case <-depChan:
				log.Info().Str("service", s.Name).Msgf("Dependency %s is healthy.", depName)
			case <-ctx.Done():
				log.Warn().Str("service", s.Name).Msg("Shutdown signal received, aborting startup.")
				return
			}
		}
	}

	// Now, start the actual service
	wg.Add(1)
	go func(s config.Service, color string) {
		defer wg.Done()

		var healthCheckPorts []int
		if s.Type == "docker" {
			// For docker, we use the dev healthcheck ports to also map them
			healthCheckPorts = s.Dev.Healthcheck.TcpPorts
			startDockerService(ctx, s, color, network, healthCheckPorts)
		} else if s.Dev.Command != "" {
			healthCheckPorts = s.Dev.Healthcheck.TcpPorts
			startSourceService(ctx, s, color)
		} else {
			log.Warn().Str("service", s.Name).Msg("No dev.command or docker type, skipping.")
			close(healthyChan) // Mark as "healthy" so other things can proceed
			return
		}

		// Perform health check on the service we just started
		waitForHealth(ctx, s.Name, healthCheckPorts)
		close(healthyChan) // Signal that this service is now healthy

		// Block until context is cancelled to keep the service running
		// and ensure wg.Done is called at the right time.
		<-ctx.Done()
	}(s, colors[len(started)%len(colors)])
}

func waitForHealth(ctx context.Context, serviceName string, ports []int) {
	if len(ports) == 0 {
		log.Info().Str("service", serviceName).Msg("No healthcheck ports defined, assuming healthy immediately.")
		return
	}

	for _, port := range ports {
		address := fmt.Sprintf("127.0.0.1:%d", port)
		log.Info().Str("service", serviceName).Msgf("Waiting for %s to be available...", address)
		for {
			select {
			case <-ctx.Done():
				log.Warn().Str("service", serviceName).Msg("Shutdown signal received, aborting health check.")
				return
			default:
				conn, err := net.DialTimeout("tcp", address, 1*time.Second)
				if err == nil {
					conn.Close()
					log.Info().Str("service", serviceName).Msgf("%s is healthy!", address)
					goto nextPort
				}
				time.Sleep(2 * time.Second)
			}
		}
	nextPort:
	}
}

// startSourceService runs a service from its source code.
func startSourceService(ctx context.Context, s config.Service, color string) {
	prefix := fmt.Sprintf("%s[%-10s]%s ", color, s.Name, colorReset)
	log.Info().Str("service", s.Name).Str("command", s.Dev.Command).Msg("Starting from source")

	parts := strings.Fields(s.Dev.Command)
	cmd := exec.CommandContext(ctx, parts[0], parts[1:]...)
	cmd.Dir = s.Path

	runAndMonitorCommand(ctx, cmd, s.Name, prefix)
}

// startDockerService runs a pre-built Docker image.
func startDockerService(ctx context.Context, s config.Service, color string, network string, portsToMap []int) {
	prefix := fmt.Sprintf("%s[%-10s]%s ", color, s.Name, colorReset)
	log.Info().Str("service", s.Name).Str("image", s.Prod.Image).Msg("Starting from Docker image")

	containerName := fmt.Sprintf("%s-dev", s.Name)
	exec.Command("docker", "rm", "-f", containerName).Run()

	args := []string{"run", "--rm", "-i", "--name", containerName}
	if network != "" {
		args = append(args, "--network", network)
	}
	for _, p := range portsToMap {
		args = append(args, "-p", fmt.Sprintf("%d:%d", p, p))
	}
	for _, e := range s.Prod.Environment {
		args = append(args, "-e", os.ExpandEnv(e))
	}

	args = append(args, os.ExpandEnv(s.Prod.Image))

	if s.Prod.Command != nil {
		switch v := s.Prod.Command.(type) {
		case string:
			args = append(args, strings.Fields(v)...)
		case []interface{}:
			for _, item := range v {
				args = append(args, fmt.Sprintf("%v", item))
			}
		}
	}

	go func() {
		<-ctx.Done()
		log.Info().Str("service", s.Name).Msg("Stopping docker container...")
		stopCmd := exec.Command("docker", "stop", containerName)
		if err := stopCmd.Run(); err != nil {
			log.Warn().Err(err).Str("service", s.Name).Msg("Failed to stop container.")
		}
	}()

	cmd := exec.Command("docker", args...)
	runAndMonitorCommand(ctx, cmd, s.Name, prefix)
}

func runAndMonitorCommand(ctx context.Context, cmd *exec.Cmd, serviceName, prefix string) {
	stdout, _ := cmd.StdoutPipe()
	stderr, _ := cmd.StderrPipe()

	if err := cmd.Start(); err != nil {
		log.Error().Err(err).Str("service", serviceName).Msg("Failed to start")
		return
	}

	go streamOutput(stdout, prefix)
	go streamOutput(stderr, prefix)

	go func() {
		err := cmd.Wait()

		if ctx.Err() != nil {
			log.Info().Str("service", serviceName).Msg("Process stopped.")
		} else if err != nil {
			log.Error().Err(err).Str("service", serviceName).Msg("Exited with error")
		} else {
			log.Info().Str("service", serviceName).Msg("Exited")
		}
	}()
}

func streamOutput(pipe io.ReadCloser, prefix string) {
	scanner := bufio.NewScanner(pipe)
	for scanner.Scan() {
		fmt.Printf("%s%s\n", prefix, scanner.Text())
	}
}

func createDockerNetwork(networkName string) {
	log.Info().Str("network", networkName).Msg("Ensuring docker network exists")
	cmd := exec.Command("docker", "network", "inspect", networkName)
	if err := cmd.Run(); err == nil {
		log.Info().Str("network", networkName).Msg("Network already exists.")
		return
	}

	cmd = exec.Command("docker", "network", "create", networkName)
	if err := cmd.Run(); err != nil {
		log.Warn().Err(err).Msg("Could not create docker network.")
	}
}