Phase3

2026-06-26 12:17:26 +02:00
parent c4077d4673
commit 9b061c3ee7
92 changed files with 4128 additions and 146 deletions
--- a/apps/node-agent/internal/runtime/manager.go
+++ b/apps/node-agent/internal/runtime/manager.go
@@ -2,11 +2,13 @@ package runtime

 import (
 	"context"
-	"fmt"
+	"encoding/json"
+	"errors"
 	"log/slog"
 	"sync"

 	"github.com/hexahost/gamecloud/node-agent/internal/docker"
+	"github.com/hexahost/gamecloud/node-agent/internal/files"
 	"github.com/hexahost/gamecloud/node-agent/internal/protocol"
 )

@@ -27,6 +29,7 @@ const (
 // Responder sends protocol responses back to the control plane.
 type Responder interface {
 	SendState(ctx context.Context, correlationID string, payload protocol.ServerStatePayload) error
+	SendLog(ctx context.Context, correlationID string, payload protocol.ServerLogPayload) error
 	SendOperationComplete(ctx context.Context, correlationID string, payload protocol.ServerOperationResultPayload) error
 	SendOperationFailed(ctx context.Context, correlationID string, payload protocol.ServerOperationResultPayload) error
 }
@@ -42,20 +45,22 @@ type serverRecord struct {

 // Manager tracks provisioned servers and executes lifecycle operations.
 type Manager struct {
-	mu      sync.RWMutex
-	docker  *docker.Client
-	servers map[string]*serverRecord
-	resp    Responder
-	log     *slog.Logger
+	mu         sync.RWMutex
+	docker     *docker.Client
+	servers    map[string]*serverRecord
+	logStreams map[string]context.CancelFunc
+	resp       Responder
+	log        *slog.Logger
 }

 // NewManager creates a runtime manager backed by the given Docker client.
 func NewManager(dockerClient *docker.Client, resp Responder, log *slog.Logger) *Manager {
 	return &Manager{
-		docker:  dockerClient,
-		servers: make(map[string]*serverRecord),
-		resp:    resp,
-		log:     log,
+		docker:     dockerClient,
+		servers:    make(map[string]*serverRecord),
+		logStreams: make(map[string]context.CancelFunc),
+		resp:       resp,
+		log:        log,
 	}
 }

@@ -184,6 +189,8 @@ func (m *Manager) HandleDelete(ctx context.Context, correlationID string, meta p

 	m.emitState(ctx, correlationID, meta, ServerStateDeleting, "", "")

+	m.stopLogStream(meta.ServerID)
+
 	if state, err := m.docker.GetContainerState(ctx, rec.ContainerID); err == nil && state.Running {
 		if err := m.docker.StopContainer(ctx, rec.ContainerID); err != nil {
 			m.fail(ctx, correlationID, meta, "DELETE_STOP_FAILED", err.Error())
@@ -204,6 +211,147 @@ func (m *Manager) HandleDelete(ctx context.Context, correlationID string, meta p
 	m.complete(ctx, correlationID, meta, ServerStateDeleted)
 }

+// HandleLogsSubscribe starts streaming container logs for a server.
+func (m *Manager) HandleLogsSubscribe(ctx context.Context, correlationID string, payload protocol.ServerLogsSubscribePayload) {
+	meta := payload.OperationMeta
+	rec, ok := m.getServer(meta.ServerID)
+	if !ok {
+		m.fail(ctx, correlationID, meta, "SERVER_NOT_FOUND", "server is not provisioned on this node")
+		return
+	}
+
+	streamCtx, cancel := context.WithCancel(ctx)
+	m.mu.Lock()
+	if existing, ok := m.logStreams[meta.ServerID]; ok {
+		existing()
+	}
+	m.logStreams[meta.ServerID] = cancel
+	m.mu.Unlock()
+
+	go func() {
+		defer func() {
+			m.mu.Lock()
+			if current, ok := m.logStreams[meta.ServerID]; ok && current == cancel {
+				delete(m.logStreams, meta.ServerID)
+			}
+			m.mu.Unlock()
+		}()
+
+		err := m.docker.StreamLogs(streamCtx, rec.ContainerID, payload.TailLines, payload.Follow, func(stream, line string) {
+			logPayload := protocol.ServerLogPayload{
+				OperationMeta: meta,
+				Line:          line,
+				Stream:        stream,
+			}
+			if err := m.resp.SendLog(ctx, correlationID, logPayload); err != nil {
+				m.log.Warn("send server.log failed",
+					"server_id", meta.ServerID,
+					"error", err,
+				)
+			}
+		})
+		if err != nil && streamCtx.Err() == nil {
+			m.log.Warn("log stream ended with error",
+				"server_id", meta.ServerID,
+				"error", err,
+			)
+		}
+	}()
+}
+
+// HandleCommand executes a console command via RCON.
+func (m *Manager) HandleCommand(ctx context.Context, correlationID string, payload protocol.ServerCommandPayload) {
+	meta := payload.OperationMeta
+	rec, ok := m.getServer(meta.ServerID)
+	if !ok {
+		m.failOperation(ctx, correlationID, meta, "command", "SERVER_NOT_FOUND", "server is not provisioned on this node")
+		return
+	}
+
+	stdout, stderr, err := m.docker.ExecRcon(ctx, rec.ContainerID, payload.Command)
+	if err != nil {
+		m.failOperation(ctx, correlationID, meta, "command", "COMMAND_FAILED", err.Error())
+		return
+	}
+
+	m.completeOperation(ctx, correlationID, meta, "command", map[string]string{
+		"stdout": stdout,
+		"stderr": stderr,
+	})
+}
+
+// HandleFilesList lists files in the server data directory.
+func (m *Manager) HandleFilesList(ctx context.Context, correlationID string, payload protocol.ServerFilesListPayload) {
+	meta := payload.OperationMeta
+	rec, ok := m.getServer(meta.ServerID)
+	if !ok {
+		m.failOperation(ctx, correlationID, meta, "files.list", "SERVER_NOT_FOUND", "server is not provisioned on this node")
+		return
+	}
+
+	entries, err := files.ListDir(rec.DataPath, payload.Path)
+	if err != nil {
+		m.failOperation(ctx, correlationID, meta, "files.list", filesErrorCode(err), err.Error())
+		return
+	}
+
+	result := protocol.ServerFilesListResultPayload{
+		Entries: make([]protocol.FileEntry, 0, len(entries)),
+	}
+	for _, entry := range entries {
+		result.Entries = append(result.Entries, protocol.FileEntry{
+			Name:       entry.Name,
+			Path:       entry.Path,
+			IsDir:      entry.IsDir,
+			Size:       entry.Size,
+			ModifiedAt: entry.ModifiedAt.Format("2006-01-02T15:04:05Z"),
+		})
+	}
+
+	m.completeOperation(ctx, correlationID, meta, "files.list", result)
+}
+
+// HandleFilesRead reads a file from the server data directory.
+func (m *Manager) HandleFilesRead(ctx context.Context, correlationID string, payload protocol.ServerFilesReadPayload) {
+	meta := payload.OperationMeta
+	rec, ok := m.getServer(meta.ServerID)
+	if !ok {
+		m.failOperation(ctx, correlationID, meta, "files.read", "SERVER_NOT_FOUND", "server is not provisioned on this node")
+		return
+	}
+
+	content, truncated, err := files.ReadFile(rec.DataPath, payload.Path, payload.MaxBytes)
+	if err != nil {
+		m.failOperation(ctx, correlationID, meta, "files.read", filesErrorCode(err), err.Error())
+		return
+	}
+
+	m.completeOperation(ctx, correlationID, meta, "files.read", protocol.ServerFilesReadResultPayload{
+		Path:      payload.Path,
+		Content:   content,
+		Truncated: truncated,
+	})
+}
+
+// HandleFilesWrite writes a file in the server data directory.
+func (m *Manager) HandleFilesWrite(ctx context.Context, correlationID string, payload protocol.ServerFilesWritePayload) {
+	meta := payload.OperationMeta
+	rec, ok := m.getServer(meta.ServerID)
+	if !ok {
+		m.failOperation(ctx, correlationID, meta, "files.write", "SERVER_NOT_FOUND", "server is not provisioned on this node")
+		return
+	}
+
+	if err := files.WriteFile(rec.DataPath, payload.Path, payload.Content, payload.Create); err != nil {
+		m.failOperation(ctx, correlationID, meta, "files.write", filesErrorCode(err), err.Error())
+		return
+	}
+
+	m.completeOperation(ctx, correlationID, meta, "files.write", map[string]string{
+		"path": payload.Path,
+	})
+}
+
 func (m *Manager) getServer(serverID string) (*serverRecord, bool) {
 	m.mu.RLock()
 	defer m.mu.RUnlock()
@@ -223,6 +371,15 @@ func (m *Manager) setState(serverID, state string) {
 	}
 }

+func (m *Manager) stopLogStream(serverID string) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if cancel, ok := m.logStreams[serverID]; ok {
+		cancel()
+		delete(m.logStreams, serverID)
+	}
+}
+
 func (m *Manager) emitState(ctx context.Context, correlationID string, meta protocol.OperationMeta, state, errorCode, errorMessage string) {
 	payload := protocol.ServerStatePayload{
 		OperationMeta: meta,
@@ -267,3 +424,59 @@ func (m *Manager) fail(ctx context.Context, correlationID string, meta protocol.
 		)
 	}
 }
+
+func (m *Manager) completeOperation(ctx context.Context, correlationID string, meta protocol.OperationMeta, operation string, result any) {
+	raw, err := json.Marshal(result)
+	if err != nil {
+		m.failOperation(ctx, correlationID, meta, operation, "RESULT_MARSHAL_FAILED", err.Error())
+		return
+	}
+
+	payload := protocol.ServerOperationResultPayload{
+		OperationMeta: meta,
+		ResultCode:    ResultCodeOK,
+		Operation:     operation,
+		Result:        raw,
+	}
+	if err := m.resp.SendOperationComplete(ctx, correlationID, payload); err != nil {
+		m.log.Warn("send operation complete failed",
+			"server_id", meta.ServerID,
+			"operation", operation,
+			"error", err,
+		)
+	}
+}
+
+func (m *Manager) failOperation(ctx context.Context, correlationID string, meta protocol.OperationMeta, operation, errorCode, errorMessage string) {
+	payload := protocol.ServerOperationResultPayload{
+		OperationMeta: meta,
+		ResultCode:    ResultCodeError,
+		Operation:     operation,
+		ErrorCode:     errorCode,
+		ErrorMessage:  errorMessage,
+	}
+	if err := m.resp.SendOperationFailed(ctx, correlationID, payload); err != nil {
+		m.log.Warn("send operation failed failed",
+			"server_id", meta.ServerID,
+			"operation", operation,
+			"error", err,
+		)
+	}
+}
+
+func filesErrorCode(err error) string {
+	switch {
+	case errors.Is(err, files.ErrPathTraversal):
+		return "PATH_TRAVERSAL"
+	case errors.Is(err, files.ErrSymlink):
+		return "SYMLINK_NOT_ALLOWED"
+	case errors.Is(err, files.ErrNotFound):
+		return "FILE_NOT_FOUND"
+	case errors.Is(err, files.ErrNotDirectory):
+		return "NOT_A_DIRECTORY"
+	case errors.Is(err, files.ErrIsDirectory):
+		return "IS_DIRECTORY"
+	default:
+		return "FILES_ERROR"
+	}
+}