DockMV/internal/api/handlers.go

package api

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"time"

	"github.com/arescom/dockmv/internal/dkr"
	"github.com/arescom/dockmv/internal/job"
	"github.com/arescom/dockmv/internal/migrate"
	"github.com/arescom/dockmv/internal/spec"
	"github.com/arescom/dockmv/internal/sshx"
	"github.com/arescom/dockmv/internal/store"
)

func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
	ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
	defer cancel()

	body := map[string]any{
		"ok":           true,
		"dockerHost":   s.docker.Endpoint,
		"packageDir":   s.cfg.PackageDir,
		"dataDir":      s.cfg.DataDir,
		"knownHosts":   s.hosts.Path(),
		"authRequired": s.cfg.Token != "",
	}
	if v, err := s.docker.Ping(ctx); err != nil {
		body["ok"] = false
		body["dockerError"] = err.Error()
	} else {
		body["dockerVersion"] = v
	}
	writeJSON(w, http.StatusOK, body)
}

// handleSource returns the full inventory of the source daemon, plus the
// default selection for every container so the UI can render immediately.
func (s *Server) handleSource(w http.ResponseWriter, r *http.Request) {
	ctx, cancel := context.WithTimeout(r.Context(), 2*time.Minute)
	defer cancel()

	inv, err := s.docker.Inventory(ctx)
	if err != nil {
		writeError(w, http.StatusBadGateway, "%v", err)
		return
	}
	defaults := make(map[string]spec.ItemSelection, len(inv.Containers))
	for i := range inv.Containers {
		defaults[inv.Containers[i].ID] = spec.DefaultSelection(&inv.Containers[i])
	}
	writeJSON(w, http.StatusOK, map[string]any{
		"inventory": inv,
		"defaults":  defaults,
		"options":   spec.DefaultOptions(),
	})
}

// handleSourceSizes measures volume sizes, which is slow enough that the UI
// asks for it separately.
func (s *Server) handleSourceSizes(w http.ResponseWriter, r *http.Request) {
	ctx, cancel := context.WithTimeout(r.Context(), 5*time.Minute)
	defer cancel()

	sizes, err := s.docker.VolumeSizes(ctx)
	if err != nil {
		writeError(w, http.StatusBadGateway, "%v", err)
		return
	}
	writeJSON(w, http.StatusOK, map[string]any{"volumes": sizes})
}

func (s *Server) handleListConnections(w http.ResponseWriter, r *http.Request) {
	writeJSON(w, http.StatusOK, s.conns.List())
}

func (s *Server) handleSaveConnection(w http.ResponseWriter, r *http.Request) {
	var cfg sshx.Config
	if err := decode(r, &cfg); err != nil {
		writeError(w, http.StatusBadRequest, "%v", err)
		return
	}
	saved, err := s.conns.Save(cfg)
	if err != nil {
		writeError(w, http.StatusBadRequest, "%v", err)
		return
	}
	writeJSON(w, http.StatusOK, saved)
}

func (s *Server) handleDeleteConnection(w http.ResponseWriter, r *http.Request) {
	if err := s.conns.Delete(r.PathValue("id")); err != nil {
		writeError(w, http.StatusNotFound, "%v", err)
		return
	}
	w.WriteHeader(http.StatusNoContent)
}

// handleProbe reads the target's host key so the operator can compare the
// fingerprint before trusting it.
func (s *Server) handleProbe(w http.ResponseWriter, r *http.Request) {
	cfg, err := s.conns.Get(r.PathValue("id"))
	if err != nil {
		writeError(w, http.StatusNotFound, "%v", err)
		return
	}
	ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
	defer cancel()

	info, err := sshx.Probe(ctx, cfg, s.hosts)
	if err != nil {
		writeError(w, http.StatusBadGateway, "%v", err)
		return
	}
	writeJSON(w, http.StatusOK, info)
}

// handleTrust records the host key. The fingerprint the operator approved is
// echoed back and re-checked, so approving one key cannot trust another.
func (s *Server) handleTrust(w http.ResponseWriter, r *http.Request) {
	cfg, err := s.conns.Get(r.PathValue("id"))
	if err != nil {
		writeError(w, http.StatusNotFound, "%v", err)
		return
	}
	var body struct {
		Fingerprint string `json:"fingerprint"`
	}
	if err := decode(r, &body); err != nil {
		writeError(w, http.StatusBadRequest, "%v", err)
		return
	}
	ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
	defer cancel()

	if err := sshx.TrustFromProbe(ctx, cfg, s.hosts, body.Fingerprint); err != nil {
		writeError(w, http.StatusBadRequest, "%v", err)
		return
	}
	writeJSON(w, http.StatusOK, map[string]any{"trusted": true})
}

func (s *Server) handleTestConnection(w http.ResponseWriter, r *http.Request) {
	ctx, cancel := context.WithTimeout(r.Context(), 60*time.Second)
	defer cancel()

	rd, closeFn, err := s.dialTarget(ctx, r.PathValue("id"))
	if err != nil {
		s.writeDialError(w, err)
		return
	}
	defer closeFn()

	pre, err := rd.Preflight(ctx)
	if err != nil {
		writeError(w, http.StatusBadGateway, "%v", err)
		return
	}
	writeJSON(w, http.StatusOK, pre)
}

func (s *Server) handleTargetInventory(w http.ResponseWriter, r *http.Request) {
	ctx, cancel := context.WithTimeout(r.Context(), 2*time.Minute)
	defer cancel()

	rd, closeFn, err := s.dialTarget(ctx, r.PathValue("id"))
	if err != nil {
		s.writeDialError(w, err)
		return
	}
	defer closeFn()

	inv, err := rd.Inventory(ctx)
	if err != nil {
		writeError(w, http.StatusBadGateway, "%v", err)
		return
	}
	writeJSON(w, http.StatusOK, inv)
}

// PreviewItem is what the UI shows when the operator asks "what will this do?".
type PreviewItem struct {
	ContainerID string   `json:"containerId"`
	Name        string   `json:"name"`
	TargetName  string   `json:"targetName"`
	Image       string   `json:"image"`
	Commands    []string `json:"commands"`
	Transfers   []string `json:"transfers"`
	Notes       []string `json:"notes"`
	Warnings    []string `json:"warnings"`
	TotalBytes  int64    `json:"totalBytes"`
}

// handlePreview renders the exact docker commands a plan would run, without
// touching either host.
func (s *Server) handlePreview(w http.ResponseWriter, r *http.Request) {
	var plan spec.Plan
	if err := decode(r, &plan); err != nil {
		writeError(w, http.StatusBadRequest, "%v", err)
		return
	}
	ctx, cancel := context.WithTimeout(r.Context(), 2*time.Minute)
	defer cancel()

	inv, err := s.docker.Inventory(ctx)
	if err != nil {
		writeError(w, http.StatusBadGateway, "%v", err)
		return
	}
	byID := map[string]*spec.Container{}
	for i := range inv.Containers {
		byID[inv.Containers[i].ID] = &inv.Containers[i]
	}

	out := []PreviewItem{}
	netSeen := map[string]bool{}
	var netCmds []string

	for _, sel := range plan.Items {
		if !sel.Include {
			continue
		}
		p, err := migrate.Prepare(byID[sel.ContainerID], sel, inv.Volumes, inv.Networks)
		if err != nil {
			writeError(w, http.StatusBadRequest, "container %s: %v", sel.ContainerID, err)
			return
		}
		item := PreviewItem{
			ContainerID: p.Source.ID,
			Name:        p.Source.Name,
			TargetName:  p.ContainerName(),
			Image:       p.Target.Image,
			Notes:       p.Notes,
			Warnings:    p.Source.Warnings,
		}
		for _, n := range p.Networks {
			if !netSeen[n.Name] {
				netSeen[n.Name] = true
				netCmds = append(netCmds, "docker "+spec.ShellQuoteAll(n.CreateArgs()))
			}
		}
		for _, v := range p.Volumes {
			item.Commands = append(item.Commands, "docker "+spec.ShellQuoteAll(v.CreateArgs()))
		}
		item.Commands = append(item.Commands, "docker "+spec.ShellQuoteAll(p.Target.CreateArgs(p.Render)))
		for _, args := range p.Target.NetworkConnectArgs(p.Render) {
			item.Commands = append(item.Commands, "docker "+spec.ShellQuoteAll(args))
		}
		for _, t := range p.Transfers {
			item.Transfers = append(item.Transfers, t.Label)
			item.Commands = append(item.Commands,
				fmt.Sprintf("docker cp -a - %s:%s   # contents of %s", p.ContainerName(), t.RestoreInto, t.SourcePath))
			if t.SizeBytes > 0 {
				item.TotalBytes += t.SizeBytes
			}
		}
		if p.Selection.StartAfter {
			item.Commands = append(item.Commands, "docker start "+spec.ShellQuote(p.ContainerName()))
		}
		out = append(out, item)
	}

	writeJSON(w, http.StatusOK, map[string]any{
		"networkCommands": netCmds,
		"items":           out,
	})
}

type sshMigrateRequest struct {
	ConnectionID string    `json:"connectionId"`
	Plan         spec.Plan `json:"plan"`
}

func (s *Server) handleMigrateSSH(w http.ResponseWriter, r *http.Request) {
	var req sshMigrateRequest
	if err := decode(r, &req); err != nil {
		writeError(w, http.StatusBadRequest, "%v", err)
		return
	}
	if req.ConnectionID == "" {
		writeError(w, http.StatusBadRequest, "connectionId is required")
		return
	}
	if countIncluded(req.Plan) == 0 {
		writeError(w, http.StatusBadRequest, "no containers selected")
		return
	}

	cfg, err := s.conns.Get(req.ConnectionID)
	if err != nil {
		writeError(w, http.StatusNotFound, "%v", err)
		return
	}

	// The inventory is re-read now so the plan is applied to current state
	// rather than to whatever the browser last loaded.
	invCtx, cancel := context.WithTimeout(r.Context(), 2*time.Minute)
	inv, err := s.docker.Inventory(invCtx)
	cancel()
	if err != nil {
		writeError(w, http.StatusBadGateway, "%v", err)
		return
	}

	// Dial before the job starts so a bad credential is an immediate error in
	// the UI instead of a failed job.
	dialCtx, dialCancel := context.WithTimeout(r.Context(), 45*time.Second)
	client, err := sshx.Dial(dialCtx, cfg, s.hosts)
	dialCancel()
	if err != nil {
		s.writeDialError(w, err)
		return
	}

	title := fmt.Sprintf("%d container(s) to %s", countIncluded(req.Plan), cfg.Name)
	runner := &migrate.SSHRunner{
		Src:        s.docker,
		Dst:        sshx.NewRemoteDocker(client),
		Containers: inv.Containers,
		Volumes:    inv.Volumes,
		Networks:   inv.Networks,
		Plan:       req.Plan,
	}

	j := s.jobs.Run(context.Background(), job.KindSSH, title, req.Plan.Options.DryRun,
		func(ctx context.Context, j *job.Job) error {
			defer client.Close()
			j.Logf(job.LevelInfo, "", "migrating to %s@%s over ssh", cfg.User, cfg.Host)
			return runner.Run(ctx, j)
		})

	writeJSON(w, http.StatusAccepted, j.Snapshot())
}

type packageRequest struct {
	Plan   spec.Plan             `json:"plan"`
	Format migrate.PackageFormat `json:"format"`
}

func (s *Server) handleBuildPackage(w http.ResponseWriter, r *http.Request) {
	var req packageRequest
	if err := decode(r, &req); err != nil {
		writeError(w, http.StatusBadRequest, "%v", err)
		return
	}
	if countIncluded(req.Plan) == 0 {
		writeError(w, http.StatusBadRequest, "no containers selected")
		return
	}
	if req.Format == "" {
		req.Format = migrate.FormatTar
	}
	if req.Format != migrate.FormatTar && req.Format != migrate.FormatDir {
		writeError(w, http.StatusBadRequest, "format must be %q or %q", migrate.FormatTar, migrate.FormatDir)
		return
	}

	invCtx, cancel := context.WithTimeout(r.Context(), 2*time.Minute)
	inv, err := s.docker.Inventory(invCtx)
	cancel()
	if err != nil {
		writeError(w, http.StatusBadGateway, "%v", err)
		return
	}

	packager := &migrate.Packager{
		Src:        s.docker,
		Containers: inv.Containers,
		Volumes:    inv.Volumes,
		Networks:   inv.Networks,
		Plan:       req.Plan,
		OutputDir:  s.cfg.PackageDir,
		Format:     req.Format,
		SourceHost: inv.Host,
	}

	title := fmt.Sprintf("package of %d container(s)", countIncluded(req.Plan))
	j := s.jobs.Run(context.Background(), job.KindPackage, title, req.Plan.Options.DryRun,
		func(ctx context.Context, j *job.Job) error {
			res, err := packager.Run(ctx, j)
			if err != nil {
				return err
			}
			j.Logf(job.LevelInfo, "", "package ready: %s (%s)", res.Name, humanBytes(res.Bytes))
			return nil
		})

	writeJSON(w, http.StatusAccepted, j.Snapshot())
}

func (s *Server) handleListJobs(w http.ResponseWriter, r *http.Request) {
	writeJSON(w, http.StatusOK, s.jobs.List())
}

func (s *Server) handleGetJob(w http.ResponseWriter, r *http.Request) {
	j, err := s.jobs.Get(r.PathValue("id"))
	if err != nil {
		writeError(w, http.StatusNotFound, "%v", err)
		return
	}
	writeJSON(w, http.StatusOK, j.Snapshot())
}

// handleJobEvents streams job snapshots over server-sent events, coalescing
// rapid updates so a fast transfer does not saturate the browser.
func (s *Server) handleJobEvents(w http.ResponseWriter, r *http.Request) {
	j, err := s.jobs.Get(r.PathValue("id"))
	if err != nil {
		writeError(w, http.StatusNotFound, "%v", err)
		return
	}
	flusher, ok := w.(http.Flusher)
	if !ok {
		writeError(w, http.StatusInternalServerError, "streaming unsupported")
		return
	}

	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")
	w.Header().Set("X-Accel-Buffering", "no")
	w.WriteHeader(http.StatusOK)

	changes, unsubscribe := j.Subscribe()
	defer unsubscribe()

	send := func() bool {
		b, err := json.Marshal(j.Snapshot())
		if err != nil {
			return false
		}
		if _, err := fmt.Fprintf(w, "data: %s\n\n", b); err != nil {
			return false
		}
		flusher.Flush()
		return true
	}
	if !send() {
		return
	}

	// Updates are batched: at most one frame every 250ms while work is busy.
	ticker := time.NewTicker(250 * time.Millisecond)
	defer ticker.Stop()
	keepalive := time.NewTicker(20 * time.Second)
	defer keepalive.Stop()

	dirty := false
	for {
		select {
		case <-r.Context().Done():
			return
		case _, open := <-changes:
			if !open {
				return
			}
			dirty = true
		case <-ticker.C:
			if !dirty {
				continue
			}
			dirty = false
			if !send() {
				return
			}
			if j.Snapshot().State.Terminal() {
				fmt.Fprint(w, "event: done\ndata: {}\n\n")
				flusher.Flush()
				return
			}
		case <-keepalive.C:
			fmt.Fprint(w, ": keepalive\n\n")
			flusher.Flush()
		case <-j.Done():
			// Drain one last snapshot so the client sees the final state.
			send()
			fmt.Fprint(w, "event: done\ndata: {}\n\n")
			flusher.Flush()
			return
		}
	}
}

func (s *Server) handleCancelJob(w http.ResponseWriter, r *http.Request) {
	j, err := s.jobs.Get(r.PathValue("id"))
	if err != nil {
		writeError(w, http.StatusNotFound, "%v", err)
		return
	}
	j.Cancel()
	writeJSON(w, http.StatusOK, map[string]any{"canceled": true})
}

func (s *Server) handleDeleteJob(w http.ResponseWriter, r *http.Request) {
	if err := s.jobs.Delete(r.PathValue("id")); err != nil {
		writeError(w, http.StatusNotFound, "%v", err)
		return
	}
	w.WriteHeader(http.StatusNoContent)
}

// PackageInfo describes one built package on disk.
type PackageInfo struct {
	Name      string    `json:"name"`
	Path      string    `json:"path"`
	Bytes     int64     `json:"bytes"`
	IsDir     bool      `json:"isDir"`
	CreatedAt time.Time `json:"createdAt"`
}

func (s *Server) handleListPackages(w http.ResponseWriter, r *http.Request) {
	entries, err := os.ReadDir(s.cfg.PackageDir)
	if err != nil {
		writeError(w, http.StatusInternalServerError, "read package directory: %v", err)
		return
	}
	out := []PackageInfo{}
	for _, e := range entries {
		info, err := e.Info()
		if err != nil {
			continue
		}
		p := PackageInfo{
			Name:      e.Name(),
			Path:      filepath.Join(s.cfg.PackageDir, e.Name()),
			IsDir:     e.IsDir(),
			CreatedAt: info.ModTime(),
		}
		if e.IsDir() {
			p.Bytes, _ = dirBytes(p.Path)
		} else {
			p.Bytes = info.Size()
		}
		out = append(out, p)
	}
	sort.Slice(out, func(i, j int) bool { return out[i].CreatedAt.After(out[j].CreatedAt) })
	writeJSON(w, http.StatusOK, out)
}

// handleDownloadPackage streams a package file. Directory packages are not
// downloadable as-is; the operator picks the tar format for that.
func (s *Server) handleDownloadPackage(w http.ResponseWriter, r *http.Request) {
	path, err := s.packagePath(r.PathValue("name"))
	if err != nil {
		writeError(w, http.StatusBadRequest, "%v", err)
		return
	}
	info, err := os.Stat(path)
	if err != nil {
		writeError(w, http.StatusNotFound, "package not found")
		return
	}
	if info.IsDir() {
		writeError(w, http.StatusBadRequest,
			"this package is a directory; copy it from %s, or rebuild with the tar format", path)
		return
	}
	f, err := os.Open(path)
	if err != nil {
		writeError(w, http.StatusInternalServerError, "%v", err)
		return
	}
	defer f.Close()
	w.Header().Set("Content-Type", "application/x-tar")
	w.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=%q", filepath.Base(path)))
	http.ServeContent(w, r, filepath.Base(path), info.ModTime(), f)
}

func (s *Server) handleDeletePackage(w http.ResponseWriter, r *http.Request) {
	path, err := s.packagePath(r.PathValue("name"))
	if err != nil {
		writeError(w, http.StatusBadRequest, "%v", err)
		return
	}
	if err := os.RemoveAll(path); err != nil {
		writeError(w, http.StatusInternalServerError, "%v", err)
		return
	}
	w.WriteHeader(http.StatusNoContent)
}

// packagePath resolves a package name against the package directory, refusing
// anything that would escape it.
func (s *Server) packagePath(name string) (string, error) {
	if name == "" || strings.ContainsAny(name, `/\`) || name == "." || name == ".." {
		return "", errors.New("invalid package name")
	}
	base, err := filepath.Abs(s.cfg.PackageDir)
	if err != nil {
		return "", err
	}
	full := filepath.Join(base, name)
	if !strings.HasPrefix(full, base+string(os.PathSeparator)) {
		return "", errors.New("invalid package name")
	}
	return full, nil
}

// dialTarget opens a short-lived connection for an interactive request.
func (s *Server) dialTarget(ctx context.Context, id string) (*sshx.RemoteDocker, func(), error) {
	cfg, err := s.conns.Get(id)
	if err != nil {
		return nil, nil, err
	}
	client, err := sshx.Dial(ctx, cfg, s.hosts)
	if err != nil {
		return nil, nil, err
	}
	return sshx.NewRemoteDocker(client), func() { client.Close() }, nil
}

// writeDialError turns an untrusted host key into a structured response the UI
// can turn into a trust prompt.
func (s *Server) writeDialError(w http.ResponseWriter, err error) {
	var hk *sshx.HostKeyError
	if errors.As(err, &hk) {
		writeJSON(w, http.StatusPreconditionRequired, map[string]any{
			"error":      err.Error(),
			"hostKey":    hk.Fingerprint,
			"keyType":    hk.KeyType,
			"changed":    hk.Changed,
			"needsTrust": true,
			"host":       hk.Host,
		})
		return
	}
	if errors.Is(err, store.ErrNotFound) {
		writeError(w, http.StatusNotFound, "%v", err)
		return
	}
	writeError(w, http.StatusBadGateway, "%v", err)
}

func countIncluded(p spec.Plan) int {
	n := 0
	for _, i := range p.Items {
		if i.Include {
			n++
		}
	}
	return n
}

func dirBytes(dir string) (int64, error) {
	var total int64
	err := filepath.Walk(dir, func(_ string, info os.FileInfo, err error) error {
		if err != nil {
			return err
		}
		if !info.IsDir() {
			total += info.Size()
		}
		return nil
	})
	return total, err
}

func humanBytes(n int64) string {
	const unit = 1024
	if n < unit {
		return fmt.Sprintf("%d B", n)
	}
	div, exp := int64(unit), 0
	for v := n / unit; v >= unit; v /= unit {
		div *= unit
		exp++
	}
	return fmt.Sprintf("%.1f %ciB", float64(n)/float64(div), "KMGTPE"[exp])
}

var _ = dkr.RestorePath