feat(KLEF-216): Daemon mod file injection and bound mounts

cmd/kleffd/main.go

@@ -69,6 +69,19 @@ func main() {
 	// --- Repository ---
 	repo := memrepo.NewServerRepository()
 
+	// Reseed the in-memory repository from any workloads that were running
+	// before this daemon process started (e.g. after a daemon restart).
+	if existing, err := runtime.ListRunning(context.Background()); err != nil {
+		daemonLog.Warn("Failed to recover running workloads on startup", "error", err)
+	} else {
+		for _, srv := range existing {
+			_ = repo.Save(context.Background(), srv)
+		}
+		if len(existing) > 0 {
+			daemonLog.Info("Recovered running workloads", "count", len(existing))
+		}
+	}
+
 	// --- Platform registration + status reporting ---
 	platformClient := platformadapter.NewClient(cfg.PlatformURL, cfg.SharedSecret, cfg.NodeID, daemonLog)
 	if err := platformClient.RegisterNode(context.Background()); err != nil {
@@ -84,6 +97,10 @@ func main() {
 	dispatcher.Register(jobs.JobTypeServerDelete, workers.NewDeleteWorker(runtime, repo, daemonLog, platformClient).Handle)
 	dispatcher.Register(jobs.JobTypeServerRestart, workers.NewRestartWorker(runtime, repo, daemonLog, platformClient).Handle)
 
+	modWorker := workers.NewModWorker(runtime, daemonLog)
+	dispatcher.Register(jobs.JobTypeModInstall, modWorker.HandleInstall)
+	dispatcher.Register(jobs.JobTypeModUninstall, modWorker.HandleUninstall)
+
 	daemonLog.Info("Daemon started", "node_id", cfg.NodeID, "grpc_port", cfg.GRPCPort)
 
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
@@ -124,7 +141,7 @@ func detectRuntime(cfg *config.Config, logger ports.Logger) (ports.RuntimeAdapte
 	}
 
 	// No Kubernetes — check if Docker is actually reachable before using it.
-	dockerAdapter, err := dockeradapter.New(cfg.NodeID)
+	dockerAdapter, err := dockeradapter.New(cfg.NodeID, cfg.StoragePath)
 	if err != nil {
 		return nil, fmt.Errorf("no runtime available: kubernetes not detected, docker client failed: %w", err)
 	}

cmd/testprovision/main.go

@@ -79,7 +79,7 @@ func main() {
 }
 
 func runDocker(cleanup bool, serverID string, spec ports.WorkloadSpec) {
-	adapter, err := dockeradapter.New("test-node")
+	adapter, err := dockeradapter.New("test-node", "/var/lib/kleffd/servers")
 	if err != nil {
 		log.Fatalf("failed to create docker adapter: %v", err)
 	}

internal/adapters/out/queue/redis.go

@@ -141,6 +141,8 @@ func (q *RedisQueue) Retry(jobID string) error {
 	var job jobs.Job
 	json.Unmarshal([]byte(raw), &job)
 
+	job.Attempts++
+
 	if job.Attempts >= job.MaxAttempts {
 		job.Status = jobs.JobStatusFailed
 		newRaw, _ := json.Marshal(job)

internal/adapters/out/runtime/docker/docker.go

@@ -6,8 +6,12 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"net/http"
+	"os"
+	"path/filepath"
 	"sort"
 	"strings"
+	"time"
 
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/filters"
@@ -24,18 +28,43 @@ import (
 // All three strategies (agones, statefulset, deployment) map to the same
 // Docker container lifecycle — the strategy hint is ignored here.
 type Adapter struct {
-	client *client.Client
-	nodeID string
+	client           *client.Client
+	nodeID           string
+	storageLocalPath string // path inside this container where server data is mounted
+	storageHostPath  string // corresponding host path passed to Docker for bind mounts
 }
 
 var errContainerNotFound = errors.New("container not found")
 
-func New(nodeID string) (*Adapter, error) {
+// New creates a Docker adapter. storageLocalPath is the directory inside the
+// daemon container where game server data lives (e.g. /var/lib/kleffd/servers).
+// The adapter auto-detects the matching host path by inspecting its own container
+// mounts so that bind mount sources are always valid host filesystem paths.
+func New(nodeID, storageLocalPath string) (*Adapter, error) {
 	c, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
 	if err != nil {
 		return nil, fmt.Errorf("failed to create docker client: %w", err)
 	}
-	return &Adapter{client: c, nodeID: nodeID}, nil
+
+	// Default: host path == local path (works when daemon runs directly on host).
+	storageHostPath := storageLocalPath
+	if hostname, err := os.Hostname(); err == nil {
+		if info, err := c.ContainerInspect(context.Background(), hostname); err == nil {
+			for _, m := range info.Mounts {
+				if m.Destination == storageLocalPath {
+					storageHostPath = m.Source
+					break
+				}
+			}
+		}
+	}
+
+	return &Adapter{
+		client:           c,
+		nodeID:           nodeID,
+		storageLocalPath: storageLocalPath,
+		storageHostPath:  storageHostPath,
+	}, nil
 }
 
 // Ping checks if the Docker daemon is reachable.
@@ -268,7 +297,7 @@ func (a *Adapter) Stop(ctx context.Context, projectID, workloadID string) error
 	return nil
 }
 
-// Remove stops and removes the container.
+// Remove stops and removes the container. Data directory is preserved on disk.
 func (a *Adapter) Remove(ctx context.Context, projectID, workloadID string) error {
 	containerID, err := a.findContainer(ctx, projectID, workloadID)
 	if err != nil {
@@ -385,22 +414,66 @@ func (a *Adapter) Endpoint(ctx context.Context, projectID, workloadID string) (s
 }
 
 // Logs streams the container's stdout/stderr.
+// When follow is true, Since is set to the container's last start time so that
+// only the logs from the current run are streamed — avoiding replaying the full
+// history across restarts before reaching live output.
 func (a *Adapter) Logs(ctx context.Context, projectID, workloadID string, follow bool) (io.ReadCloser, error) {
 	containerID, err := a.findContainer(ctx, projectID, workloadID)
 	if err != nil {
 		return nil, err
 	}
-	rc, err := a.client.ContainerLogs(ctx, containerID, container.LogsOptions{
+
+	opts := container.LogsOptions{
 		ShowStdout: true,
 		ShowStderr: true,
 		Follow:     follow,
-	})
+	}
+	if follow {
+		if info, err := a.client.ContainerInspect(ctx, containerID); err == nil && info.State != nil {
+			if t, err := time.Parse(time.RFC3339Nano, info.State.StartedAt); err == nil && !t.IsZero() {
+				opts.Since = t.UTC().Format(time.RFC3339Nano)
+			}
+		}
+	}
+
+	rc, err := a.client.ContainerLogs(ctx, containerID, opts)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get logs: %w", err)
 	}
 	return rc, nil
 }
 
+// ListRunning returns a ServerRecord for each container currently managed by
+// this daemon that is in the "running" state. Used to reseed the in-memory
+// repository after a daemon restart.
+func (a *Adapter) ListRunning(ctx context.Context) ([]*ports.ServerRecord, error) {
+	containers, err := a.client.ContainerList(ctx, container.ListOptions{
+		Filters: filters.NewArgs(
+			filters.Arg("label", labels.ManagedBy+"="+labels.ManagedByValue),
+			filters.Arg("status", "running"),
+		),
+	})
+	if err != nil {
+		return nil, fmt.Errorf("list managed containers: %w", err)
+	}
+
+	records := make([]*ports.ServerRecord, 0, len(containers))
+	for _, c := range containers {
+		wl := labels.FromMap(c.Labels)
+		if wl.ServerID == "" {
+			continue
+		}
+		records = append(records, &ports.ServerRecord{
+			ID:         wl.ServerID,
+			Status:     "Running",
+			NodeID:     wl.NodeID,
+			RuntimeRef: c.ID,
+			ProjectID:  wl.ProjectID,
+		})
+	}
+	return records, nil
+}
+
 // --- Helpers ---
 
 func (a *Adapter) createContainer(ctx context.Context, spec ports.WorkloadSpec, scope *ports.ProjectScope) (string, error) {
@@ -443,9 +516,17 @@ func (a *Adapter) createContainer(ctx context.Context, spec ports.WorkloadSpec,
 
 	var mounts []mount.Mount
 	if spec.RuntimeHints.PersistentStorage && spec.RuntimeHints.StoragePath != "" {
+		localDir := filepath.Join(a.storageLocalPath, projectDataDir(spec.ProjectID, spec.ServerID))
+		if err := os.MkdirAll(localDir, 0777); err != nil {
+			return "", fmt.Errorf("create server storage directory: %w", err)
+		}
+		// MkdirAll respects the process umask, so chmod explicitly to ensure
+		// any user inside the game server container can write to /data.
+		_ = os.Chmod(localDir, 0777)
+		hostDir := filepath.Join(a.storageHostPath, projectDataDir(spec.ProjectID, spec.ServerID))
 		mounts = append(mounts, mount.Mount{
-			Type:   mount.TypeVolume,
-			Source: projectVolumeName(spec.ProjectID, spec.ServerID),
+			Type:   mount.TypeBind,
+			Source: hostDir,
 			Target: spec.RuntimeHints.StoragePath,
 		})
 	}
@@ -549,10 +630,72 @@ func projectNetworkName(projectID string) string {
 	return "kleff_proj_" + shortID(projectID)
 }
 
-// projectVolumeName namespaces persistent volumes by project so a stray
-// workload id collision across projects cannot cross-mount data.
-func projectVolumeName(projectID, workloadID string) string {
-	return fmt.Sprintf("kleff_proj_%s_%s_data", shortID(projectID), workloadID)
+// projectDataDir returns a unique directory name for a workload's persistent data.
+func projectDataDir(projectID, workloadID string) string {
+	return fmt.Sprintf("kleff_proj_%s_%s", shortID(projectID), workloadID)
+}
+
+// contentTypeDirs maps a mod content type to its subdirectory inside the storage volume.
+var contentTypeDirs = map[string]string{
+	"mod":          "mods",
+	"plugin":       "plugins",
+	"datapack":     "world/datapacks",
+	"resourcepack": "resourcepacks",
+}
+
+// InjectFile downloads a file from downloadURL directly into the workload's
+// bind-mounted storage directory on the host filesystem.
+func (a *Adapter) InjectFile(ctx context.Context, projectID, workloadID, storagePath, contentType, downloadURL, fileName string) error {
+	subDir, ok := contentTypeDirs[contentType]
+	if !ok {
+		return fmt.Errorf("%w: unsupported content type %q", ports.ErrPermanent, contentType)
+	}
+
+	destDir := filepath.Join(a.storageLocalPath, projectDataDir(projectID, workloadID), subDir)
+	if err := os.MkdirAll(destDir, 0755); err != nil {
+		return fmt.Errorf("create mod directory: %w", err)
+	}
+
+	return downloadFile(ctx, downloadURL, filepath.Join(destDir, fileName))
+}
+
+// RemoveFile deletes a previously injected file from the workload's storage directory.
+func (a *Adapter) RemoveFile(ctx context.Context, projectID, workloadID, storagePath, contentType, fileName string) error {
+	subDir, ok := contentTypeDirs[contentType]
+	if !ok {
+		return fmt.Errorf("%w: unsupported content type %q", ports.ErrPermanent, contentType)
+	}
+
+	filePath := filepath.Join(a.storageLocalPath, projectDataDir(projectID, workloadID), subDir, fileName)
+	if err := os.Remove(filePath); err != nil && !os.IsNotExist(err) {
+		return fmt.Errorf("remove file: %w", err)
+	}
+	return nil
+}
+
+func downloadFile(ctx context.Context, url, destPath string) error {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+	if err != nil {
+		return fmt.Errorf("build request: %w", err)
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return fmt.Errorf("download: %w", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("download: HTTP %d", resp.StatusCode)
+	}
+	f, err := os.Create(destPath)
+	if err != nil {
+		return fmt.Errorf("create file: %w", err)
+	}
+	defer f.Close()
+	if _, err := io.Copy(f, resp.Body); err != nil {
+		os.Remove(destPath)
+		return fmt.Errorf("write file: %w", err)
+	}
+	return nil
 }
 
 func shortID(s string) string {

internal/adapters/out/runtime/kubernetes/kubernetes.go

@@ -569,3 +569,16 @@ func buildService(spec ports.WorkloadSpec, selectorLabels map[string]string) *co
 		},
 	}
 }
+
+
+func (a *Adapter) InjectFile(_ context.Context, _, _, _, _, _, _ string) error {
+	return fmt.Errorf("mod file injection is not yet supported on Kubernetes")
+}
+
+func (a *Adapter) RemoveFile(_ context.Context, _, _, _, _, _ string) error {
+	return fmt.Errorf("mod file removal is not yet supported on Kubernetes")
+}
+
+func (a *Adapter) ListRunning(_ context.Context) ([]*ports.ServerRecord, error) {
+	return nil, nil
+}

internal/app/config/config.go

@@ -34,6 +34,7 @@ type Config struct {
 	RedisTLS      bool         `mapstructure:"redis.tls"`
 	PlatformURL   string       `mapstructure:"platform.url"`
 	SharedSecret  string       `mapstructure:"shared_secret"`
+	StoragePath   string       `mapstructure:"storage.path"`
 }
 
 func (c *Config) Validate() error {
@@ -81,6 +82,7 @@ func Load() (*Config, error) {
 	v.SetDefault("redis.tls", false)
 	v.SetDefault("platform.url", "")
 	v.SetDefault("shared_secret", "")
+	v.SetDefault("storage.path", "/var/lib/kleffd/servers")
 
 	v.SetEnvPrefix("kleff")
 	v.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
@@ -126,6 +128,7 @@ func Load() (*Config, error) {
 	cfg.RedisTLS = v.GetBool("redis.tls")
 	cfg.PlatformURL = v.GetString("platform.url")
 	cfg.SharedSecret = v.GetString("shared_secret")
+	cfg.StoragePath = v.GetString("storage.path")
 
 	if err := cfg.Validate(); err != nil {
 		return nil, fmt.Errorf("configuration validation failed: %w", err)

internal/application/ports/runtime_adapter.go

@@ -43,4 +43,17 @@ type RuntimeAdapter interface {
 	Endpoint(ctx context.Context, projectID, workloadID string) (string, error)
 	// Logs streams the workload's stdout/stderr.
 	Logs(ctx context.Context, projectID, workloadID string, follow bool) (io.ReadCloser, error)
+
+	// InjectFile downloads a file from downloadURL and writes it into the
+	// workload's persistent volume under the appropriate content-type subdirectory.
+	// storagePath is the volume's mount point inside the container (e.g. "/data").
+	InjectFile(ctx context.Context, projectID, workloadID, storagePath, contentType, downloadURL, fileName string) error
+
+	// RemoveFile deletes a previously injected file from the workload's volume.
+	RemoveFile(ctx context.Context, projectID, workloadID, storagePath, contentType, fileName string) error
+
+	// ListRunning returns server records for all workloads currently managed by
+	// this daemon that are in a running state. Used to reseed the in-memory
+	// repository after a daemon restart.
+	ListRunning(ctx context.Context) ([]*ServerRecord, error)
 }

internal/application/ports/workload.go

@@ -62,6 +62,25 @@ type RuntimeHints struct {
 	StorageGB          int    `json:"storage_gb,omitempty"`
 }
 
+// ModInstallSpec is the payload for server.install_mod jobs.
+type ModInstallSpec struct {
+	ServerID    string `json:"server_id"`
+	ProjectID   string `json:"project_id"`
+	DownloadURL string `json:"download_url"`
+	FileName    string `json:"file_name"`
+	ContentType string `json:"content_type"` // "mod", "plugin", "datapack", "resourcepack"
+	StoragePath string `json:"storage_path"` // base mount path, e.g. "/data"
+}
+
+// ModUninstallSpec is the payload for server.uninstall_mod jobs.
+type ModUninstallSpec struct {
+	ServerID    string `json:"server_id"`
+	ProjectID   string `json:"project_id"`
+	FileName    string `json:"file_name"`
+	ContentType string `json:"content_type"`
+	StoragePath string `json:"storage_path"`
+}
+
 // WorkloadHealth is the per-workload status reported in heartbeats.
 type WorkloadHealth struct {
 	WorkloadID    string `json:"workload_id"`