initial commit

apps/node-agent/internal/agent/agent.go

@@ -0,0 +1,92 @@
+package agent
+
+import (
+	"context"
+	"log/slog"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/hexahost/gamecloud/node-agent/internal/config"
+	"github.com/hexahost/gamecloud/node-agent/internal/health"
+)
+
+const Version = "0.1.0-phase0"
+
+// Agent coordinates the node agent lifecycle until shutdown.
+type Agent struct {
+	cfg    *config.Config
+	log    *slog.Logger
+	health *health.Server
+}
+
+// New creates an agent instance from configuration.
+func New(cfg *config.Config, log *slog.Logger) *Agent {
+	return &Agent{
+		cfg:    cfg,
+		log:    log,
+		health: health.NewServer(cfg.HealthListenAddr, Version),
+	}
+}
+
+// Run starts the agent loop and blocks until context cancellation or signal.
+func (a *Agent) Run(ctx context.Context) error {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	sigCh := make(chan os.Signal, 1)
+	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
+	defer signal.Stop(sigCh)
+
+	errCh := make(chan error, 1)
+	go func() {
+		a.log.Info("health server listening", "addr", a.cfg.HealthListenAddr)
+		if err := a.health.ListenAndServe(); err != nil {
+			errCh <- err
+		}
+	}()
+
+	a.health.SetReady(true)
+	a.log.Info("agent started",
+		"node_id", a.cfg.NodeID,
+		"control_plane", a.cfg.ControlPlaneURL,
+		"version", Version,
+	)
+
+	ticker := time.NewTicker(a.cfg.HeartbeatInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return a.shutdown(ctx.Err())
+		case sig := <-sigCh:
+			a.log.Info("shutdown signal received", "signal", sig.String())
+			return a.shutdown(nil)
+		case err := <-errCh:
+			return err
+		case <-ticker.C:
+			a.log.Debug("heartbeat tick",
+				"node_id", a.cfg.NodeID,
+				"interval", a.cfg.HeartbeatInterval.String(),
+			)
+		}
+	}
+}
+
+func (a *Agent) shutdown(reason error) error {
+	a.health.SetReady(false)
+	a.log.Info("agent shutting down", "grace", a.cfg.ShutdownGracePeriod.String())
+
+	timer := time.NewTimer(a.cfg.ShutdownGracePeriod)
+	defer timer.Stop()
+	<-timer.C
+
+	if reason != nil && reason != context.Canceled {
+		a.log.Warn("shutdown with error", "error", reason)
+		return reason
+	}
+	a.log.Info("agent stopped")
+	return nil
+}

apps/node-agent/internal/config/config.go

@@ -0,0 +1,88 @@
+package config
+
+import (
+	"fmt"
+	"os"
+	"strconv"
+	"time"
+)
+
+// Config holds runtime configuration loaded from environment variables.
+type Config struct {
+	AppName              string
+	NodeID               string
+	ControlPlaneURL      string
+	NodeToken            string
+	HealthListenAddr     string
+	HeartbeatInterval    time.Duration
+	ReconnectBackoff     time.Duration
+	DataDir              string
+	DockerSocket         string
+	LogLevel             string
+	TLSCertFile          string
+	TLSKeyFile           string
+	TLSCAFile            string
+	TLSSkipVerify        bool
+	ShutdownGracePeriod  time.Duration
+}
+
+// Load reads configuration from environment variables with sensible defaults for development.
+func Load() (*Config, error) {
+	cfg := &Config{
+		AppName:             envOrDefault("APP_NAME", "HexaHost GameCloud"),
+		NodeID:              os.Getenv("NODE_ID"),
+		ControlPlaneURL:     os.Getenv("NODE_AGENT_PUBLIC_URL"),
+		NodeToken:           os.Getenv("NODE_TOKEN"),
+		HealthListenAddr:    envOrDefault("NODE_AGENT_HEALTH_ADDR", ":9100"),
+		HeartbeatInterval:   durationFromEnv("NODE_HEARTBEAT_INTERVAL", 10*time.Second),
+		ReconnectBackoff:    durationFromEnv("NODE_RECONNECT_BACKOFF", 5*time.Second),
+		DataDir:             envOrDefault("NODE_DATA_DIR", "/var/lib/hexahost-gamecloud"),
+		DockerSocket:        envOrDefault("DOCKER_HOST", "unix:///var/run/docker.sock"),
+		LogLevel:            envOrDefault("LOG_LEVEL", "info"),
+		TLSCertFile:         os.Getenv("NODE_TLS_CERT"),
+		TLSKeyFile:          os.Getenv("NODE_TLS_KEY"),
+		TLSCAFile:           os.Getenv("NODE_CA_CERT"),
+		TLSSkipVerify:       boolFromEnv("NODE_TLS_SKIP_VERIFY", false),
+		ShutdownGracePeriod: durationFromEnv("NODE_SHUTDOWN_GRACE", 30*time.Second),
+	}
+
+	if cfg.NodeID == "" {
+		return nil, fmt.Errorf("NODE_ID is required")
+	}
+	if cfg.ControlPlaneURL == "" {
+		return nil, fmt.Errorf("NODE_AGENT_PUBLIC_URL is required")
+	}
+
+	return cfg, nil
+}
+
+func envOrDefault(key, fallback string) string {
+	if v := os.Getenv(key); v != "" {
+		return v
+	}
+	return fallback
+}
+
+func durationFromEnv(key string, fallback time.Duration) time.Duration {
+	raw := os.Getenv(key)
+	if raw == "" {
+		return fallback
+	}
+	d, err := time.ParseDuration(raw)
+	if err != nil {
+		return fallback
+	}
+	return d
+}
+
+func boolFromEnv(key string, fallback bool) bool {
+	raw := os.Getenv(key)
+	if raw == "" {
+		return fallback
+	}
+	v, err := strconv.ParseBool(raw)
+	if err != nil {
+		return fallback
+	}
+	return v
+}

apps/node-agent/internal/health/health.go

@@ -0,0 +1,78 @@
+package health
+
+import (
+	"encoding/json"
+	"net/http"
+	"sync/atomic"
+	"time"
+)
+
+// Status represents agent health for liveness and readiness probes.
+type Status struct {
+	Status    string    `json:"status"`
+	Agent     string    `json:"agent"`
+	Version   string    `json:"version"`
+	Uptime    string    `json:"uptime"`
+	Timestamp time.Time `json:"timestamp"`
+}
+
+// Server exposes HTTP health endpoints for the node agent.
+type Server struct {
+	addr      string
+	version   string
+	startedAt time.Time
+	ready     atomic.Bool
+	mux       *http.ServeMux
+}
+
+// NewServer creates a health HTTP server bound to addr.
+func NewServer(addr, version string) *Server {
+	s := &Server{
+		addr:      addr,
+		version:   version,
+		startedAt: time.Now().UTC(),
+		mux:       http.NewServeMux(),
+	}
+	s.mux.HandleFunc("GET /healthz", s.handleLiveness)
+	s.mux.HandleFunc("GET /readyz", s.handleReadiness)
+	return s
+}
+
+// SetReady toggles readiness for orchestration probes.
+func (s *Server) SetReady(ready bool) {
+	s.ready.Store(ready)
+}
+
+// Handler returns the HTTP handler for embedding in tests.
+func (s *Server) Handler() http.Handler {
+	return s.mux
+}
+
+// ListenAndServe starts the health server.
+func (s *Server) ListenAndServe() error {
+	return http.ListenAndServe(s.addr, s.mux)
+}
+
+func (s *Server) handleLiveness(w http.ResponseWriter, _ *http.Request) {
+	s.writeStatus(w, http.StatusOK, "alive")
+}
+
+func (s *Server) handleReadiness(w http.ResponseWriter, _ *http.Request) {
+	if !s.ready.Load() {
+		s.writeStatus(w, http.StatusServiceUnavailable, "not_ready")
+		return
+	}
+	s.writeStatus(w, http.StatusOK, "ready")
+}
+
+func (s *Server) writeStatus(w http.ResponseWriter, code int, status string) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(code)
+	_ = json.NewEncoder(w).Encode(Status{
+		Status:    status,
+		Agent:     "node-agent",
+		Version:   s.version,
+		Uptime:    time.Since(s.startedAt).Round(time.Second).String(),
+		Timestamp: time.Now().UTC(),
+	})
+}

apps/node-agent/internal/protocol/messages.go

@@ -0,0 +1,209 @@
+package protocol
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+	"time"
+)
+
+const CurrentProtocolVersion = 1
+
+// MessageType identifies a versioned agent/control-plane message.
+type MessageType string
+
+const (
+	// Agent → Control Plane
+	TypeAgentHello              MessageType = "agent.hello"
+	TypeAgentHeartbeat          MessageType = "agent.heartbeat"
+	TypeAgentInventory          MessageType = "agent.inventory"
+	TypeServerState             MessageType = "server.state"
+	TypeServerMetrics           MessageType = "server.metrics"
+	TypeServerLog               MessageType = "server.log"
+	TypeServerOperationProgress MessageType = "server.operation.progress"
+	TypeServerOperationComplete MessageType = "server.operation.completed"
+	TypeServerOperationFailed   MessageType = "server.operation.failed"
+
+	// Control Plane → Agent
+	TypeServerProvision       MessageType = "server.provision"
+	TypeServerInstall         MessageType = "server.install"
+	TypeServerStart           MessageType = "server.start"
+	TypeServerStop            MessageType = "server.stop"
+	TypeServerKill            MessageType = "server.kill"
+	TypeServerDelete          MessageType = "server.delete"
+	TypeServerInspect         MessageType = "server.inspect"
+	TypeServerCommand         MessageType = "server.command"
+	TypeServerBackupPrepare   MessageType = "server.backup.prepare"
+	TypeServerBackupRelease   MessageType = "server.backup.release"
+	TypeServerFilesList       MessageType = "server.files.list"
+	TypeServerFilesRead       MessageType = "server.files.read"
+	TypeServerFilesWrite      MessageType = "server.files.write"
+	TypeServerFilesUploadDone MessageType = "server.files.upload.complete"
+	TypeServerWorldValidate   MessageType = "server.world.validate"
+	TypeServerMetricsSub      MessageType = "server.metrics.subscribe"
+	TypeServerLogsSubscribe   MessageType = "server.logs.subscribe"
+)
+
+var agentToControlTypes = map[MessageType]struct{}{
+	TypeAgentHello:              {},
+	TypeAgentHeartbeat:          {},
+	TypeAgentInventory:          {},
+	TypeServerState:             {},
+	TypeServerMetrics:           {},
+	TypeServerLog:               {},
+	TypeServerOperationProgress: {},
+	TypeServerOperationComplete: {},
+	TypeServerOperationFailed:   {},
+}
+
+var controlToAgentTypes = map[MessageType]struct{}{
+	TypeServerProvision:       {},
+	TypeServerInstall:         {},
+	TypeServerStart:           {},
+	TypeServerStop:            {},
+	TypeServerKill:            {},
+	TypeServerDelete:          {},
+	TypeServerInspect:         {},
+	TypeServerCommand:         {},
+	TypeServerBackupPrepare:   {},
+	TypeServerBackupRelease:   {},
+	TypeServerFilesList:       {},
+	TypeServerFilesRead:       {},
+	TypeServerFilesWrite:      {},
+	TypeServerFilesUploadDone: {},
+	TypeServerWorldValidate:   {},
+	TypeServerMetricsSub:      {},
+	TypeServerLogsSubscribe:   {},
+}
+
+// Envelope is the common wrapper for all protocol messages.
+type Envelope struct {
+	ProtocolVersion int             `json:"protocolVersion"`
+	MessageID       string          `json:"messageId"`
+	CorrelationID   string          `json:"correlationId,omitempty"`
+	Type            MessageType     `json:"type"`
+	Timestamp       time.Time       `json:"timestamp"`
+	Payload         json.RawMessage `json:"payload"`
+}
+
+// OperationMeta is shared metadata for lifecycle operations.
+type OperationMeta struct {
+	ServerID       string     `json:"serverId"`
+	Generation     int64      `json:"generation"`
+	Deadline       *time.Time `json:"deadline,omitempty"`
+	IdempotencyKey string     `json:"idempotencyKey,omitempty"`
+}
+
+// AgentHelloPayload is sent when the agent connects to the control plane.
+type AgentHelloPayload struct {
+	NodeID         string `json:"nodeId"`
+	AgentVersion   string `json:"agentVersion"`
+	ProtocolVersion int   `json:"protocolVersion"`
+	Hostname       string `json:"hostname,omitempty"`
+	Capabilities   []string `json:"capabilities,omitempty"`
+}
+
+// AgentHeartbeatPayload reports node health and resource usage.
+type AgentHeartbeatPayload struct {
+	NodeID           string  `json:"nodeId"`
+	CPUUsagePercent  float64 `json:"cpuUsagePercent"`
+	MemoryUsedBytes  int64   `json:"memoryUsedBytes"`
+	MemoryTotalBytes int64   `json:"memoryTotalBytes"`
+	DiskUsedBytes    int64   `json:"diskUsedBytes"`
+	DiskTotalBytes   int64   `json:"diskTotalBytes"`
+	RunningServers   int     `json:"runningServers"`
+}
+
+// ServerStartPayload instructs the agent to start a game server.
+type ServerStartPayload struct {
+	OperationMeta
+	DesiredGeneration int64 `json:"desiredGeneration"`
+}
+
+// ServerStatePayload reports the observed server state.
+type ServerStatePayload struct {
+	OperationMeta
+	State        string `json:"state"`
+	ErrorCode    string `json:"errorCode,omitempty"`
+	ErrorMessage string `json:"errorMessage,omitempty"`
+}
+
+// Validate checks envelope fields and known message types.
+func (e *Envelope) Validate() error {
+	if e == nil {
+		return fmt.Errorf("envelope is nil")
+	}
+	if e.ProtocolVersion < 1 {
+		return fmt.Errorf("protocolVersion must be >= 1")
+	}
+	if e.ProtocolVersion > CurrentProtocolVersion {
+		return fmt.Errorf("unsupported protocolVersion %d", e.ProtocolVersion)
+	}
+	if strings.TrimSpace(e.MessageID) == "" {
+		return fmt.Errorf("messageId is required")
+	}
+	if strings.TrimSpace(string(e.Type)) == "" {
+		return fmt.Errorf("type is required")
+	}
+	if e.Timestamp.IsZero() {
+		return fmt.Errorf("timestamp is required")
+	}
+	if !e.Timestamp.UTC().Equal(e.Timestamp) {
+		return fmt.Errorf("timestamp must be UTC")
+	}
+	if _, ok := agentToControlTypes[e.Type]; !ok {
+		if _, ok := controlToAgentTypes[e.Type]; !ok {
+			return fmt.Errorf("unknown message type: %s", e.Type)
+		}
+	}
+	if len(e.Payload) == 0 {
+		return fmt.Errorf("payload is required")
+	}
+	if !json.Valid(e.Payload) {
+		return fmt.Errorf("payload must be valid JSON")
+	}
+	return nil
+}
+
+// IsAgentToControl returns true when the message flows from agent to control plane.
+func (t MessageType) IsAgentToControl() bool {
+	_, ok := agentToControlTypes[t]
+	return ok
+}
+
+// IsControlToAgent returns true when the message flows from control plane to agent.
+func (t MessageType) IsControlToAgent() bool {
+	_, ok := controlToAgentTypes[t]
+	return ok
+}
+
+// DecodeEnvelope parses and validates a JSON envelope.
+func DecodeEnvelope(data []byte) (*Envelope, error) {
+	var env Envelope
+	if err := json.Unmarshal(data, &env); err != nil {
+		return nil, fmt.Errorf("decode envelope: %w", err)
+	}
+	if err := env.Validate(); err != nil {
+		return nil, err
+	}
+	return &env, nil
+}
+
+// NewEnvelope creates a validated outbound envelope.
+func NewEnvelope(msgType MessageType, messageID string, payload any) (*Envelope, error) {
+	raw, err := json.Marshal(payload)
+	if err != nil {
+		return nil, fmt.Errorf("marshal payload: %w", err)
+	}
+	env := &Envelope{
+		ProtocolVersion: CurrentProtocolVersion,
+		MessageID:       messageID,
+		Type:            msgType,
+		Timestamp:       time.Now().UTC(),
+		Payload:         raw,
+	}
+	if err := env.Validate(); err != nil {
+		return nil, err
+	}
+	return env, nil
+}

apps/node-agent/internal/protocol/messages_test.go

@@ -0,0 +1,151 @@
+package protocol
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+)
+
+func TestEnvelopeValidate_AgentHello(t *testing.T) {
+	payload := AgentHelloPayload{
+		NodeID:          "node-01",
+		AgentVersion:    "0.1.0",
+		ProtocolVersion: CurrentProtocolVersion,
+	}
+	raw, err := json.Marshal(payload)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	env := &Envelope{
+		ProtocolVersion: CurrentProtocolVersion,
+		MessageID:       "msg-001",
+		Type:            TypeAgentHello,
+		Timestamp:       time.Now().UTC(),
+		Payload:         raw,
+	}
+
+	if err := env.Validate(); err != nil {
+		t.Fatalf("expected valid envelope, got %v", err)
+	}
+	if !env.Type.IsAgentToControl() {
+		t.Fatal("agent.hello should be agent-to-control")
+	}
+}
+
+func TestEnvelopeValidate_ServerStart(t *testing.T) {
+	payload := ServerStartPayload{
+		OperationMeta: OperationMeta{
+			ServerID:   "srv-01",
+			Generation: 4,
+		},
+		DesiredGeneration: 4,
+	}
+	raw, err := json.Marshal(payload)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	env := &Envelope{
+		ProtocolVersion: CurrentProtocolVersion,
+		MessageID:       "msg-002",
+		CorrelationID:   "corr-002",
+		Type:            TypeServerStart,
+		Timestamp:       time.Now().UTC(),
+		Payload:         raw,
+	}
+
+	if err := env.Validate(); err != nil {
+		t.Fatalf("expected valid envelope, got %v", err)
+	}
+	if !env.Type.IsControlToAgent() {
+		t.Fatal("server.start should be control-to-agent")
+	}
+}
+
+func TestEnvelopeValidate_RejectsInvalid(t *testing.T) {
+	tests := []struct {
+		name string
+		env  Envelope
+	}{
+		{
+			name: "missing message id",
+			env: Envelope{
+				ProtocolVersion: 1,
+				Type:            TypeAgentHeartbeat,
+				Timestamp:       time.Now().UTC(),
+				Payload:         json.RawMessage(`{"nodeId":"n1"}`),
+			},
+		},
+		{
+			name: "unknown type",
+			env: Envelope{
+				ProtocolVersion: 1,
+				MessageID:       "m1",
+				Type:            "agent.unknown",
+				Timestamp:       time.Now().UTC(),
+				Payload:         json.RawMessage(`{}`),
+			},
+		},
+		{
+			name: "non-utc timestamp",
+			env: Envelope{
+				ProtocolVersion: 1,
+				MessageID:       "m1",
+				Type:            TypeAgentHeartbeat,
+				Timestamp:       time.Now().Local(),
+				Payload:         json.RawMessage(`{"nodeId":"n1"}`),
+			},
+		},
+		{
+			name: "invalid payload json",
+			env: Envelope{
+				ProtocolVersion: 1,
+				MessageID:       "m1",
+				Type:            TypeAgentHeartbeat,
+				Timestamp:       time.Now().UTC(),
+				Payload:         json.RawMessage(`{not-json`),
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if err := tc.env.Validate(); err == nil {
+				t.Fatal("expected validation error")
+			}
+		})
+	}
+}
+
+func TestDecodeEnvelope(t *testing.T) {
+	data := []byte(`{
+		"protocolVersion": 1,
+		"messageId": "01HXYZ",
+		"type": "agent.heartbeat",
+		"timestamp": "2026-01-01T00:00:00Z",
+		"payload": {"nodeId":"node-01","runningServers":0}
+	}`)
+
+	env, err := DecodeEnvelope(data)
+	if err != nil {
+		t.Fatalf("decode failed: %v", err)
+	}
+	if env.Type != TypeAgentHeartbeat {
+		t.Fatalf("unexpected type %s", env.Type)
+	}
+}
+
+func TestNewEnvelope(t *testing.T) {
+	env, err := NewEnvelope(TypeAgentHello, "msg-new", AgentHelloPayload{
+		NodeID:          "n1",
+		AgentVersion:    "0.1.0",
+		ProtocolVersion: 1,
+	})
+	if err != nil {
+		t.Fatalf("new envelope: %v", err)
+	}
+	if env.ProtocolVersion != CurrentProtocolVersion {
+		t.Fatalf("protocol version %d", env.ProtocolVersion)
+	}
+}