CosmicClash/server/supervisor/supervisor.go

// Package supervisor contains the small PID-1 lifecycle boundary around an
// allocated Godot process. The Agones client is HTTP-only so local/Compose
// execution remains independent of the cloud SDK.
package supervisor

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"os"
	"os/exec"
	"strconv"
	"strings"
	"time"
)

type GameServer struct {
	Status struct {
		Address string `json:"address"`
		Ports   []struct {
			Name string `json:"name"`
			Port int    `json:"port"`
		} `json:"ports"`
	} `json:"status"`
}

type Config struct {
	Command      []string
	Environment  []string
	SDKBaseURL   string
	ReadyURL     string
	Transport    string
	ReadyTimeout time.Duration
	PollInterval time.Duration
	HTTPClient   *http.Client
}

type Supervisor struct {
	config Config
	client *http.Client
	cmd    *exec.Cmd
}

func New(config Config) (*Supervisor, error) {
	if len(config.Command) == 0 || config.Command[0] == "" {
		return nil, fmt.Errorf("supervisor command is required")
	}
	if config.ReadyTimeout <= 0 {
		config.ReadyTimeout = 30 * time.Second
	}
	if config.PollInterval <= 0 {
		config.PollInterval = 100 * time.Millisecond
	}
	if config.Transport == "" {
		config.Transport = "enet"
	}
	if config.Transport != "enet" && config.Transport != "steam_sdr" {
		return nil, fmt.Errorf("unsupported transport %q", config.Transport)
	}
	if config.HTTPClient == nil {
		config.HTTPClient = http.DefaultClient
	}
	return &Supervisor{config: config, client: config.HTTPClient}, nil
}

// Start launches the process and marks Agones Ready only after the explicit
// readiness probe succeeds. No stdout/log scraping is used. With no SDK URL,
// this is direct/Compose mode and the command is simply started.
func (s *Supervisor) Start(ctx context.Context) error {
	env := append([]string(nil), os.Environ()...)
	env = append(env, s.config.Environment...)
	if s.config.SDKBaseURL != "" {
		port, address, err := s.assignedEndpoint(ctx)
		if err != nil {
			return err
		}
		if s.config.Transport == "steam_sdr" {
			env = append(env, "SDR_LISTEN_PORT="+strconv.Itoa(port), "SDR_IP="+address+":"+strconv.Itoa(port))
		}
		command := withPort(s.config.Command, port)
		s.cmd = exec.CommandContext(ctx, command[0], command[1:]...)
	} else {
		s.cmd = exec.CommandContext(ctx, s.config.Command[0], s.config.Command[1:]...)
	}
	s.cmd.Env = env
	if err := s.cmd.Start(); err != nil {
		return err
	}
	if s.config.SDKBaseURL == "" {
		return nil
	}
	if err := s.waitReady(ctx); err != nil {
		_ = s.cmd.Process.Kill()
		return err
	}
	return s.sdkPost(ctx, "/ready")
}

func withPort(command []string, port int) []string {
	result := append([]string(nil), command...)
	for i, arg := range result {
		if strings.HasPrefix(arg, "--port=") {
			result[i] = "--port=" + strconv.Itoa(port)
			return result
		}
	}
	return append(result, "--port="+strconv.Itoa(port))
}

func (s *Supervisor) Wait() error {
	if s.cmd == nil {
		return fmt.Errorf("supervisor has not started")
	}
	return s.cmd.Wait()
}

func (s *Supervisor) assignedEndpoint(ctx context.Context) (int, string, error) {
	var server GameServer
	if err := s.sdkGet(ctx, "/gameserver", &server); err != nil {
		return 0, "", err
	}
	if len(server.Status.Ports) == 0 || server.Status.Address == "" {
		return 0, "", fmt.Errorf("Agones returned no assigned endpoint")
	}
	for _, port := range server.Status.Ports {
		if port.Port > 0 && (port.Name == "game" || len(server.Status.Ports) == 1) {
			return port.Port, server.Status.Address, nil
		}
	}
	return 0, "", fmt.Errorf("Agones returned no usable game port")
}

func (s *Supervisor) waitReady(ctx context.Context) error {
	if s.config.ReadyURL == "" {
		return fmt.Errorf("allocated mode requires an explicit readiness URL")
	}
	deadline := time.NewTimer(s.config.ReadyTimeout)
	defer deadline.Stop()
	for {
		request, err := http.NewRequestWithContext(ctx, http.MethodGet, s.config.ReadyURL, nil)
		if err == nil {
			response, requestErr := s.client.Do(request)
			if requestErr == nil {
				_ = response.Body.Close()
				if response.StatusCode >= 200 && response.StatusCode < 300 {
					return nil
				}
			}
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-deadline.C:
			return fmt.Errorf("process-ready probe timed out")
		case <-time.After(s.config.PollInterval):
		}
	}
}

func (s *Supervisor) sdkGet(ctx context.Context, path string, target any) error {
	request, err := http.NewRequestWithContext(ctx, http.MethodGet, strings.TrimRight(s.config.SDKBaseURL, "/")+path, nil)
	if err != nil {
		return err
	}
	response, err := s.client.Do(request)
	if err != nil {
		return err
	}
	defer response.Body.Close()
	if response.StatusCode/100 != 2 {
		return fmt.Errorf("Agones GET %s returned %s", path, response.Status)
	}
	return json.NewDecoder(response.Body).Decode(target)
}

func (s *Supervisor) sdkPost(ctx context.Context, path string) error {
	request, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(s.config.SDKBaseURL, "/")+path, nil)
	if err != nil {
		return err
	}
	response, err := s.client.Do(request)
	if err != nil {
		return err
	}
	defer response.Body.Close()
	if response.StatusCode/100 != 2 {
		return fmt.Errorf("Agones POST %s returned %s", path, response.Status)
	}
	return nil
}