package runtime

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"github.com/hexahost/gamecloud/node-agent/internal/archive"
	"github.com/hexahost/gamecloud/node-agent/internal/docker"
	"github.com/hexahost/gamecloud/node-agent/internal/files"
	"github.com/hexahost/gamecloud/node-agent/internal/protocol"
	"github.com/hexahost/gamecloud/node-agent/internal/transfer"
	worldpkg "github.com/hexahost/gamecloud/node-agent/internal/world"
)

const (
	ServerStateProvisioning = "provisioning"
	ServerStateStopped      = "stopped"
	ServerStateStarting     = "starting"
	ServerStateRunning      = "running"
	ServerStateStopping     = "stopping"
	ServerStateDeleting     = "deleting"
	ServerStateDeleted      = "deleted"
	ServerStateError        = "error"

	ResultCodeOK    = "ok"
	ResultCodeError = "error"
)

// 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
}

type serverRecord struct {
	ServerID    string
	ContainerID string
	HostPort    int
	DataPath    string
	RamMb       int
	State       string
}

// Manager tracks provisioned servers and executes lifecycle operations.
type Manager struct {
	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),
		logStreams: make(map[string]context.CancelFunc),
		resp:       resp,
		log:        log,
	}
}

// RunningCount returns the number of servers currently in the running state.
func (m *Manager) RunningCount() int {
	m.mu.RLock()
	defer m.mu.RUnlock()
	count := 0
	for _, rec := range m.servers {
		if rec.State == ServerStateRunning {
			count++
		}
	}
	return count
}

// HandleProvision creates a Minecraft container for the server.
func (m *Manager) HandleProvision(ctx context.Context, correlationID string, payload protocol.ServerProvisionPayload) {
	meta := payload.OperationMeta
	m.log.Info("provision server",
		"server_id", meta.ServerID,
		"host_port", payload.HostPort,
		"version", payload.MinecraftVersion,
	)

	if !payload.EulaAccepted {
		m.fail(ctx, correlationID, meta, "EULA_NOT_ACCEPTED", "EULA must be accepted before provisioning")
		return
	}

	m.mu.Lock()
	if existing, ok := m.servers[meta.ServerID]; ok {
		m.mu.Unlock()
		m.log.Info("provision idempotent hit", "server_id", meta.ServerID, "container_id", existing.ContainerID)
		m.complete(ctx, correlationID, meta, existing.State)
		return
	}
	m.mu.Unlock()

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

	containerID, err := m.docker.CreateMinecraftServer(ctx, docker.MinecraftServerSpec{
		ServerID:         meta.ServerID,
		MinecraftVersion: payload.MinecraftVersion,
		SoftwareFamily:   payload.SoftwareFamily,
		HostPort:         payload.HostPort,
		RamMb:            payload.RamMb,
		DataPath:         payload.DataPath,
		EulaAccepted:     payload.EulaAccepted,
	})
	if err != nil {
		m.fail(ctx, correlationID, meta, "PROVISION_FAILED", err.Error())
		return
	}

	rec := &serverRecord{
		ServerID:    meta.ServerID,
		ContainerID: containerID,
		HostPort:    payload.HostPort,
		DataPath:    payload.DataPath,
		RamMb:       payload.RamMb,
		State:       ServerStateStopped,
	}

	m.mu.Lock()
	m.servers[meta.ServerID] = rec
	m.mu.Unlock()

	m.emitState(ctx, correlationID, meta, ServerStateStopped, "", "")
	m.complete(ctx, correlationID, meta, ServerStateStopped)
}

// HandleStart starts a provisioned server container.
func (m *Manager) HandleStart(ctx context.Context, correlationID string, payload protocol.ServerStartPayload) {
	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
	}

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

	if err := m.docker.StartContainer(ctx, rec.ContainerID); err != nil {
		m.setState(meta.ServerID, ServerStateError)
		m.emitState(ctx, correlationID, meta, ServerStateError, "START_FAILED", err.Error())
		m.fail(ctx, correlationID, meta, "START_FAILED", err.Error())
		return
	}

	m.setState(meta.ServerID, ServerStateRunning)
	m.emitState(ctx, correlationID, meta, ServerStateRunning, "", "")
	m.complete(ctx, correlationID, meta, ServerStateRunning)
}

// HandleStop stops a running server container.
func (m *Manager) HandleStop(ctx context.Context, correlationID string, payload protocol.ServerStopPayload) {
	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
	}

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

	if err := m.docker.StopContainer(ctx, rec.ContainerID); err != nil {
		m.setState(meta.ServerID, ServerStateError)
		m.emitState(ctx, correlationID, meta, ServerStateError, "STOP_FAILED", err.Error())
		m.fail(ctx, correlationID, meta, "STOP_FAILED", err.Error())
		return
	}

	m.setState(meta.ServerID, ServerStateStopped)
	m.emitState(ctx, correlationID, meta, ServerStateStopped, "", "")
	m.complete(ctx, correlationID, meta, ServerStateStopped)
}

// HandleDelete removes a server container and drops local state.
func (m *Manager) HandleDelete(ctx context.Context, correlationID string, meta protocol.OperationMeta) {
	rec, ok := m.getServer(meta.ServerID)
	if !ok {
		m.complete(ctx, correlationID, meta, ServerStateDeleted)
		return
	}

	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())
			return
		}
	}

	if err := m.docker.RemoveContainer(ctx, rec.ContainerID); err != nil {
		m.fail(ctx, correlationID, meta, "DELETE_FAILED", err.Error())
		return
	}

	m.mu.Lock()
	delete(m.servers, meta.ServerID)
	m.mu.Unlock()

	m.emitState(ctx, correlationID, meta, ServerStateDeleted, "", "")
	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,
	})
}

// HandleFilesDelete removes a file or directory from the server data directory.
func (m *Manager) HandleFilesDelete(ctx context.Context, correlationID string, payload protocol.ServerFilesDeletePayload) {
	meta := payload.OperationMeta
	rec, ok := m.getServer(meta.ServerID)
	if !ok {
		m.failOperation(ctx, correlationID, meta, "files.delete", "SERVER_NOT_FOUND", "server is not provisioned on this node")
		return
	}

	if err := files.RemovePath(rec.DataPath, payload.Path); err != nil {
		m.failOperation(ctx, correlationID, meta, "files.delete", filesErrorCode(err), err.Error())
		return
	}

	m.completeOperation(ctx, correlationID, meta, "files.delete", map[string]string{
		"path": payload.Path,
	})
}

// HandleFilesArchive creates a tar.gz archive from selected paths.
func (m *Manager) HandleFilesArchive(ctx context.Context, correlationID string, payload protocol.ServerFilesArchivePayload) {
	meta := payload.OperationMeta
	rec, ok := m.getServer(meta.ServerID)
	if !ok {
		m.failOperation(ctx, correlationID, meta, "files.archive", "SERVER_NOT_FOUND", "server is not provisioned on this node")
		return
	}

	archiveName := payload.ArchiveName
	if archiveName == "" {
		archiveName = fmt.Sprintf("archive-%d.tar.gz", time.Now().Unix())
	}
	if !strings.HasSuffix(archiveName, ".tar.gz") {
		archiveName += ".tar.gz"
	}

	outputPath := filepath.Join(rec.DataPath, ".hexahost-archives", archiveName)
	checksum, size, err := archive.CreateTarGzFromPaths(rec.DataPath, payload.Paths, outputPath)
	if err != nil {
		m.failOperation(ctx, correlationID, meta, "files.archive", "ARCHIVE_FAILED", err.Error())
		return
	}

	rel, _ := filepath.Rel(rec.DataPath, outputPath)
	m.completeOperation(ctx, correlationID, meta, "files.archive", map[string]any{
		"archivePath": filepath.ToSlash(rel),
		"sha256":      checksum,
		"size":        size,
	})
}

// HandleFilesUnarchive extracts a tar.gz archive into a destination path.
func (m *Manager) HandleFilesUnarchive(ctx context.Context, correlationID string, payload protocol.ServerFilesUnarchivePayload) {
	meta := payload.OperationMeta
	rec, ok := m.getServer(meta.ServerID)
	if !ok {
		m.failOperation(ctx, correlationID, meta, "files.unarchive", "SERVER_NOT_FOUND", "server is not provisioned on this node")
		return
	}

	archiveAbs, err := files.ResolvePath(rec.DataPath, payload.ArchivePath)
	if err != nil {
		m.failOperation(ctx, correlationID, meta, "files.unarchive", filesErrorCode(err), err.Error())
		return
	}

	destAbs, err := files.ResolvePath(rec.DataPath, payload.DestinationPath)
	if err != nil {
		m.failOperation(ctx, correlationID, meta, "files.unarchive", filesErrorCode(err), err.Error())
		return
	}

	if err := archive.ExtractTarGz(archiveAbs, destAbs); err != nil {
		m.failOperation(ctx, correlationID, meta, "files.unarchive", "UNARCHIVE_FAILED", err.Error())
		return
	}

	m.completeOperation(ctx, correlationID, meta, "files.unarchive", map[string]string{
		"destinationPath": payload.DestinationPath,
	})
}

// HandleBackupPrepare runs save-off and save-all flush via RCON.
func (m *Manager) HandleBackupPrepare(ctx context.Context, correlationID string, meta protocol.OperationMeta) {
	rec, ok := m.getServer(meta.ServerID)
	if !ok {
		m.failOperation(ctx, correlationID, meta, "backup.prepare", "SERVER_NOT_FOUND", "server is not provisioned on this node")
		return
	}

	if rec.State != ServerStateRunning {
		m.completeOperation(ctx, correlationID, meta, "backup.prepare", map[string]string{"skipped": "not_running"})
		return
	}

	for _, command := range []string{"save-off", "save-all flush"} {
		if _, _, err := m.docker.ExecRcon(ctx, rec.ContainerID, command); err != nil {
			m.failOperation(ctx, correlationID, meta, "backup.prepare", "BACKUP_PREPARE_FAILED", err.Error())
			return
		}
	}

	m.completeOperation(ctx, correlationID, meta, "backup.prepare", map[string]string{"status": "ok"})
}

// HandleBackupRelease runs save-on via RCON.
func (m *Manager) HandleBackupRelease(ctx context.Context, correlationID string, meta protocol.OperationMeta) {
	rec, ok := m.getServer(meta.ServerID)
	if !ok {
		m.failOperation(ctx, correlationID, meta, "backup.release", "SERVER_NOT_FOUND", "server is not provisioned on this node")
		return
	}

	if rec.State != ServerStateRunning {
		m.completeOperation(ctx, correlationID, meta, "backup.release", map[string]string{"skipped": "not_running"})
		return
	}

	if _, _, err := m.docker.ExecRcon(ctx, rec.ContainerID, "save-on"); err != nil {
		m.failOperation(ctx, correlationID, meta, "backup.release", "BACKUP_RELEASE_FAILED", err.Error())
		return
	}

	m.completeOperation(ctx, correlationID, meta, "backup.release", map[string]string{"status": "ok"})
}

// HandleWorldValidate validates a world directory and lists worlds.
func (m *Manager) HandleWorldValidate(ctx context.Context, correlationID string, payload protocol.ServerWorldValidatePayload) {
	meta := payload.OperationMeta
	rec, ok := m.getServer(meta.ServerID)
	if !ok {
		m.failOperation(ctx, correlationID, meta, "world.validate", "SERVER_NOT_FOUND", "server is not provisioned on this node")
		return
	}

	worldName := payload.WorldName
	if worldName == "" {
		worldName = "world"
	}

	if err := worldpkg.ValidateWorld(rec.DataPath, worldName); err != nil {
		m.failOperation(ctx, correlationID, meta, "world.validate", "WORLD_INVALID", err.Error())
		return
	}

	worlds, err := worldpkg.ListWorlds(rec.DataPath, worldName)
	if err != nil {
		m.failOperation(ctx, correlationID, meta, "world.validate", "WORLD_LIST_FAILED", err.Error())
		return
	}

	m.completeOperation(ctx, correlationID, meta, "world.validate", map[string]any{
		"valid":     true,
		"worldName": worldName,
		"worlds":    worlds,
	})
}

// HandleWorldArchive creates a tar.gz archive of a world directory.
func (m *Manager) HandleWorldArchive(ctx context.Context, correlationID string, payload protocol.ServerWorldArchivePayload) {
	meta := payload.OperationMeta
	rec, ok := m.getServer(meta.ServerID)
	if !ok {
		m.failOperation(ctx, correlationID, meta, "world.archive", "SERVER_NOT_FOUND", "server is not provisioned on this node")
		return
	}

	worldName := payload.WorldName
	if worldName == "" {
		worldName = "world"
	}

	if err := worldpkg.ValidateWorld(rec.DataPath, worldName); err != nil {
		m.failOperation(ctx, correlationID, meta, "world.archive", "WORLD_INVALID", err.Error())
		return
	}

	archiveDir := filepath.Join(rec.DataPath, ".hgc", "archives")
	archiveID := payload.ArchiveID
	if archiveID == "" {
		archiveID = meta.ServerID
	}
	outputPath := filepath.Join(archiveDir, archiveID+".tar.gz")
	sourceDir := filepath.Join(rec.DataPath, worldName)

	checksum, size, err := archive.CreateTarGz(sourceDir, outputPath)
	if err != nil {
		m.failOperation(ctx, correlationID, meta, "world.archive", "ARCHIVE_FAILED", err.Error())
		return
	}

	m.completeOperation(ctx, correlationID, meta, "world.archive", map[string]any{
		"localPath": outputPath,
		"sha256":    checksum,
		"sizeBytes": size,
		"worldName": worldName,
	})
}

// HandleStorageUpload uploads a local archive to object storage.
func (m *Manager) HandleStorageUpload(ctx context.Context, correlationID string, payload protocol.ServerStorageUploadPayload) {
	meta := payload.OperationMeta

	if err := transfer.UploadFile(payload.LocalPath, payload.UploadURL); err != nil {
		m.failOperation(ctx, correlationID, meta, "storage.upload", "UPLOAD_FAILED", err.Error())
		return
	}

	m.completeOperation(ctx, correlationID, meta, "storage.upload", map[string]string{
		"localPath": payload.LocalPath,
	})
}

// HandleStorageDownload downloads an archive from object storage.
func (m *Manager) HandleStorageDownload(ctx context.Context, correlationID string, payload protocol.ServerStorageDownloadPayload) {
	meta := payload.OperationMeta

	if err := transfer.DownloadFile(payload.DownloadURL, payload.LocalPath); err != nil {
		m.failOperation(ctx, correlationID, meta, "storage.download", "DOWNLOAD_FAILED", err.Error())
		return
	}

	m.completeOperation(ctx, correlationID, meta, "storage.download", map[string]string{
		"localPath": payload.LocalPath,
	})
}

// HandleWorldReplace replaces a world directory from a local archive.
func (m *Manager) HandleWorldReplace(ctx context.Context, correlationID string, payload protocol.ServerWorldReplacePayload) {
	meta := payload.OperationMeta
	rec, ok := m.getServer(meta.ServerID)
	if !ok {
		m.failOperation(ctx, correlationID, meta, "world.replace", "SERVER_NOT_FOUND", "server is not provisioned on this node")
		return
	}

	if rec.State == ServerStateRunning {
		m.failOperation(ctx, correlationID, meta, "world.replace", "SERVER_RUNNING", "server must be stopped before world replace")
		return
	}

	worldName := payload.WorldName
	if worldName == "" {
		worldName = "world"
	}

	worldDir := filepath.Join(rec.DataPath, worldName)
	tempDir := filepath.Join(rec.DataPath, ".hgc", "restore", meta.ServerID)
	if err := os.RemoveAll(tempDir); err != nil {
		m.failOperation(ctx, correlationID, meta, "world.replace", "RESTORE_PREP_FAILED", err.Error())
		return
	}
	if err := os.MkdirAll(tempDir, 0o755); err != nil {
		m.failOperation(ctx, correlationID, meta, "world.replace", "RESTORE_PREP_FAILED", err.Error())
		return
	}

	if err := archive.ExtractTarGz(payload.ArchivePath, tempDir); err != nil {
		m.failOperation(ctx, correlationID, meta, "world.replace", "EXTRACT_FAILED", err.Error())
		return
	}

	extractedWorld := filepath.Join(tempDir, worldName)
	if info, err := os.Stat(extractedWorld); err != nil || !info.IsDir() {
		extractedWorld = tempDir
	}

	if err := worldpkg.ValidateWorld(filepath.Dir(extractedWorld), filepath.Base(extractedWorld)); err != nil {
		m.failOperation(ctx, correlationID, meta, "world.replace", "WORLD_INVALID", err.Error())
		return
	}

	backupDir := worldDir + ".bak"
	_ = os.RemoveAll(backupDir)
	if _, err := os.Stat(worldDir); err == nil {
		if err := os.Rename(worldDir, backupDir); err != nil {
			m.failOperation(ctx, correlationID, meta, "world.replace", "REPLACE_FAILED", err.Error())
			return
		}
	}

	if err := os.Rename(extractedWorld, worldDir); err != nil {
		_ = os.Rename(backupDir, worldDir)
		m.failOperation(ctx, correlationID, meta, "world.replace", "REPLACE_FAILED", err.Error())
		return
	}

	_ = os.RemoveAll(tempDir)
	_ = os.RemoveAll(backupDir)

	m.completeOperation(ctx, correlationID, meta, "world.replace", map[string]string{
		"worldName": worldName,
	})
}

// HandleAddonInstall downloads and installs an addon JAR.
func (m *Manager) HandleAddonInstall(ctx context.Context, correlationID string, payload protocol.ServerAddonInstallPayload) {
	meta := payload.OperationMeta
	rec, ok := m.getServer(meta.ServerID)
	if !ok {
		m.failOperation(ctx, correlationID, meta, "addon.install", "SERVER_NOT_FOUND", "server is not provisioned on this node")
		return
	}

	if rec.State == ServerStateRunning {
		m.failOperation(ctx, correlationID, meta, "addon.install", "SERVER_RUNNING", "server must be stopped before installing addons")
		return
	}

	localPath, err := files.ResolvePath(rec.DataPath, payload.RelativePath)
	if err != nil {
		m.failOperation(ctx, correlationID, meta, "addon.install", filesErrorCode(err), err.Error())
		return
	}

	if err := transfer.DownloadFile(payload.DownloadURL, localPath); err != nil {
		m.failOperation(ctx, correlationID, meta, "addon.install", "DOWNLOAD_FAILED", err.Error())
		return
	}

	if err := transfer.VerifySha512File(localPath, payload.Sha512); err != nil {
		_ = os.Remove(localPath)
		m.failOperation(ctx, correlationID, meta, "addon.install", "HASH_MISMATCH", err.Error())
		return
	}

	m.completeOperation(ctx, correlationID, meta, "addon.install", map[string]string{
		"path": payload.RelativePath,
	})
}

// HandleAddonRemove deletes an installed addon file.
func (m *Manager) HandleAddonRemove(ctx context.Context, correlationID string, payload protocol.ServerAddonRemovePayload) {
	meta := payload.OperationMeta
	rec, ok := m.getServer(meta.ServerID)
	if !ok {
		m.failOperation(ctx, correlationID, meta, "addon.remove", "SERVER_NOT_FOUND", "server is not provisioned on this node")
		return
	}

	if rec.State == ServerStateRunning {
		m.failOperation(ctx, correlationID, meta, "addon.remove", "SERVER_RUNNING", "server must be stopped before removing addons")
		return
	}

	if err := files.DeleteFile(rec.DataPath, payload.RelativePath); err != nil {
		m.failOperation(ctx, correlationID, meta, "addon.remove", filesErrorCode(err), err.Error())
		return
	}

	m.completeOperation(ctx, correlationID, meta, "addon.remove", map[string]string{
		"path": payload.RelativePath,
	})
}

func (m *Manager) getServer(serverID string) (*serverRecord, bool) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	rec, ok := m.servers[serverID]
	if !ok {
		return nil, false
	}
	copy := *rec
	return &copy, true
}

func (m *Manager) setState(serverID, state string) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if rec, ok := m.servers[serverID]; ok {
		rec.State = state
	}
}

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,
		State:         state,
		ErrorCode:     errorCode,
		ErrorMessage:  errorMessage,
	}
	if err := m.resp.SendState(ctx, correlationID, payload); err != nil {
		m.log.Warn("send server.state failed",
			"server_id", meta.ServerID,
			"state", state,
			"error", err,
		)
	}
}

func (m *Manager) complete(ctx context.Context, correlationID string, meta protocol.OperationMeta, _ string) {
	payload := protocol.ServerOperationResultPayload{
		OperationMeta: meta,
		ResultCode:    ResultCodeOK,
	}
	if err := m.resp.SendOperationComplete(ctx, correlationID, payload); err != nil {
		m.log.Warn("send operation complete failed",
			"server_id", meta.ServerID,
			"error", err,
		)
	}
}

func (m *Manager) fail(ctx context.Context, correlationID string, meta protocol.OperationMeta, errorCode, errorMessage string) {
	m.emitState(ctx, correlationID, meta, ServerStateError, errorCode, errorMessage)
	payload := protocol.ServerOperationResultPayload{
		OperationMeta: meta,
		ResultCode:    ResultCodeError,
		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,
			"error", err,
		)
	}
}

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"
	}
}