felhom-controller/controller/internal/web/media_events.go

package web

import (
	"context"
	"encoding/json"
	"errors"
	"net/http"
	"net/url"
	"path/filepath"
	"regexp"
	"sort"
	"strings"
	"time"

	"gitea.dooplex.hu/admin/felhom-controller/internal/agentapi"
	"gitea.dooplex.hu/admin/felhom-controller/internal/dockerexec"
	"gitea.dooplex.hu/admin/felhom-controller/internal/mediawire"
	"gitea.dooplex.hu/admin/felhom-controller/internal/settings"
	"gitea.dooplex.hu/admin/felhom-controller/internal/stacks"
)

// ── Media drive events (v0.313.0; `09` §3 decisions 238/239/245; felhom.eu `architecture/15-media-library.md` §10) ──
//
// On a drive ADDED, REMOVED or RETURNED the media apps must restart to see it (Docker resolves binds at container
// start). The box records the event on the drive (settings: MediaChangedAt), and the storage page and the start page say
// which apps must restart, with „restart now" and „later" (tonight 03:30). Each drive card says whether the apps were
// restarted since the drive changed. „Restart now" first counts what Plex and Jellyfin are playing (a restart cuts a
// stream — K5-4) and asks once more when something plays. After the restart the box wires every present drive into the
// apps (mediawire, R-1024). A RETURNED drive also recreates the file browser at once (R-1038 — it holds no stream).
//
// A SAFE REMOVAL (R-1044) pauses only that drive's torrents, recreates the media apps and the file browser WITHOUT the
// drive, then unmounts — and keeps the drive registered, so its return needs no household step; the paused torrents
// resume when it is back. The dead-mount rule holds throughout: no path here ever makes an empty folder under an app.

const mediaLaterHour, mediaLaterMinute = 3, 30

func (s *Server) wirer() *mediawire.Wirer {
	s.mediaMu.Lock()
	defer s.mediaMu.Unlock()
	if s.mediaWirer == nil {
		s.mediaWirer = mediawire.New(mediawire.DockerBox{}, s.logger)
	}
	return s.mediaWirer
}

func (s *Server) mediaApps() []string {
	if s.stackMgr == nil {
		return nil
	}
	return s.stackMgr.MediaApps()
}

func (s *Server) presentMediaDrives() []mediawire.Drive {
	if s.stackMgr == nil {
		return nil
	}
	var out []mediawire.Drive
	for _, p := range s.stackMgr.PresentMediaDrives() {
		out = append(out, mediawire.Drive{Path: p})
	}
	return out
}

// RunMediaWiring wires every present drive into the media apps and keeps the result for the page.
func (s *Server) RunMediaWiring(reason string) []mediawire.Result {
	if len(s.mediaApps()) == 0 {
		return nil
	}
	s.logger.Printf("[INFO] [media] wiring the drives into the media apps (%s)", reason)
	res := s.wirer().Run(s.presentMediaDrives())
	s.mediaMu.Lock()
	s.mediaWiring, s.mediaWiredAt = res, time.Now()
	s.mediaMu.Unlock()
	return res
}

// RunMediaWiringSoon debounces the wiring after media apps come up: one run 60 s after the LAST call (an install and
// a restart of six apps make one run, not six), once more 2 min later if an app answered with an error.
func (s *Server) RunMediaWiringSoon(app string) {
	s.mediaMu.Lock()
	if s.mediaSoonTmr != nil {
		s.mediaSoonTmr.Stop()
	}
	s.mediaSoonTmr = time.AfterFunc(60*time.Second, func() {
		for _, r := range s.RunMediaWiring("a media app came up: " + app) {
			if r.State == "error" {
				time.Sleep(2 * time.Minute)
				s.RunMediaWiring("retry after an app error")
				return
			}
		}
	})
	s.mediaMu.Unlock()
}

// mediaDriveEvent records a drive added / removed / returned — only when a media app exists to care.
func (s *Server) mediaDriveEvent(path, change string) {
	if s.settings == nil || len(s.mediaApps()) == 0 {
		return
	}
	if err := s.settings.MarkMediaChanged(path, change); err != nil {
		s.logger.Printf("[WARN] [media] recording the drive event %s %s: %v", change, path, err)
		return
	}
	if s.stackMgr != nil {
		// Write the new per-drive binds now (no restart): the page can say exactly which apps changed.
		if changed := s.stackMgr.MediaAppsNeedingRestart(); len(changed) > 0 {
			s.logger.Printf("[INFO] [media] drive %s %s: the binds of %v changed — they see it after a restart", path, change, changed)
		}
	}
}

// staleMediaApps lists running media apps (and the file browser) whose own mount table still holds a DEAD drive
// (`shutdown` — K4: each container's mountinfo shows the old device after a return). No disk I/O.
func (s *Server) staleMediaApps() []string {
	var out []string
	for _, c := range append(s.mediaApps(), "filebrowser") {
		b, err := dockerexec.Command("docker", "exec", c, "cat", "/proc/self/mountinfo").Output()
		if err != nil {
			continue
		}
		for _, line := range strings.Split(string(b), "\n") {
			f := strings.Fields(line)
			if len(f) < 5 || !strings.HasPrefix(f[4], StableParentDir+"/") {
				continue
			}
			if i := strings.Index(line, " - "); i >= 0 {
				if t := strings.Fields(line[i+3:]); len(t) >= 3 && strings.Contains(","+t[2]+",", ",shutdown,") {
					out = append(out, c)
					break
				}
			}
		}
	}
	return out
}

// MediaDriveNotice is what the storage page and the start page show.
type MediaDriveNotice struct {
	Pending bool               `json:"pending"`
	Drives  []MediaNoticeDrive `json:"drives"`
	Apps    []string           `json:"apps"`
	Stale   []string           `json:"stale,omitempty"`
	LaterAt string             `json:"later_at,omitempty"`
	Wiring  []mediawire.Result `json:"wiring,omitempty"`
	WiredAt string             `json:"wired_at,omitempty"`
}

// MediaNoticeDrive is one drive's line.
type MediaNoticeDrive struct {
	Path      string `json:"path"`
	Label     string `json:"label"`
	Change    string `json:"change"`
	ChangedAt string `json:"changed_at"`
	Restarted bool   `json:"restarted"` // „apps restarted since this drive changed"
	Removed   bool   `json:"removed,omitempty"`
}

// MediaNotice computes the notice. Cheap enough for a page render: settings + one `docker exec cat` per media app.
func (s *Server) MediaNotice(checkStale bool) MediaDriveNotice {
	n := MediaDriveNotice{Apps: s.mediaApps()}
	if s.settings == nil || len(n.Apps) == 0 {
		return n
	}
	for _, sp := range s.settings.GetStoragePaths() {
		if sp.MediaChangedAt == "" {
			continue
		}
		d := MediaNoticeDrive{Path: sp.Path, Label: sp.Label, Change: sp.MediaChange, ChangedAt: sp.MediaChangedAt,
			Restarted: !sp.MediaRestartPending(), Removed: sp.EjectedAt != ""}
		if d.Label == "" {
			d.Label = filepath.Base(sp.Path)
		}
		n.Pending = n.Pending || !d.Restarted
		n.Drives = append(n.Drives, d)
	}
	sort.Slice(n.Drives, func(i, j int) bool { return n.Drives[i].Path < n.Drives[j].Path })
	if checkStale {
		n.Stale = s.staleMediaApps()
		n.Pending = n.Pending || len(n.Stale) > 0
	}
	s.mediaMu.Lock()
	if !s.mediaLaterAt.IsZero() {
		n.LaterAt = s.mediaLaterAt.Format("2006-01-02 15:04")
	}
	n.Wiring = s.mediaWiring
	if !s.mediaWiredAt.IsZero() {
		n.WiredAt = s.mediaWiredAt.Format(time.RFC3339)
	}
	s.mediaMu.Unlock()
	return n
}

// restartMediaApps restarts every media app for real (RestartStack: force-recreate with the stack's env — R-1026),
// recreates the file browser, records the restart on every drive, then wires the present drives in.
func (s *Server) restartMediaApps(reason string) []string {
	var failed []string
	for _, name := range s.mediaApps() {
		if s.appHeld(name) {
			continue
		}
		if err := s.stackMgr.RestartStack(name); err != nil {
			s.logger.Printf("[WARN] [media] restart %s (%s): %v", name, reason, err)
			failed = append(failed, name)
		}
	}
	s.SyncFileBrowserMountsForce()
	if len(failed) == 0 {
		if err := s.settings.MarkMediaRestarted(); err != nil {
			s.logger.Printf("[WARN] [media] recording the restart: %v", err)
		}
	}
	// R-1044: the torrents a safe removal paused resume now that the download apps see their drive again (K3-6: no
	// re-check after a restart).
	if ids := s.settings.TakeResumable(); len(ids) > 0 {
		go func() {
			time.Sleep(30 * time.Second)
			if err := s.wirer().Resume(ids); err != nil {
				s.logger.Printf("[WARN] [media] resuming %d torrent(s) paused by a safe removal: %v", len(ids), err)
			}
		}()
	}
	s.logger.Printf("[INFO] [media] media apps restarted (%s); failed: %v", reason, failed)
	return failed
}

func nextMediaLater(now time.Time) time.Time {
	loc, err := time.LoadLocation("Europe/Budapest")
	if err != nil {
		loc = time.Local
	}
	n := now.In(loc)
	t := time.Date(n.Year(), n.Month(), n.Day(), mediaLaterHour, mediaLaterMinute, 0, 0, loc)
	if !t.After(n) {
		t = t.Add(24 * time.Hour)
	}
	return t
}

// handleMediaRestart: POST /api/storage/media/restart {"when":"now"|"later","force":bool}.
func (s *Server) handleMediaRestart(w http.ResponseWriter, r *http.Request) {
	var req struct {
		When  string `json:"when"`
		Force bool   `json:"force"`
	}
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeDiskJSON(w, http.StatusBadRequest, false, s.msg(r, "disk.err.bad_request"), nil)
		return
	}
	if req.When == "later" {
		at := nextMediaLater(time.Now())
		s.mediaMu.Lock()
		if s.mediaLaterTmr != nil {
			s.mediaLaterTmr.Stop()
		}
		s.mediaLaterAt = at
		s.mediaLaterTmr = time.AfterFunc(time.Until(at), func() {
			s.restartMediaApps("the household's „later\"")
			s.mediaMu.Lock()
			s.mediaLaterAt = time.Time{}
			s.mediaMu.Unlock()
		})
		s.mediaMu.Unlock()
		s.logger.Printf("[INFO] [media] media restart scheduled for %s (the household's „later\"; a controller restart forgets it, the notice stays)", at.Format(time.RFC3339))
		writeDiskJSON(w, http.StatusOK, true, "", map[string]string{"later_at": at.Format("2006-01-02 15:04")})
		return
	}
	if !req.Force {
		if n, notes := s.wirer().PlayingStreams(); n > 0 {
			writeDiskJSON(w, http.StatusConflict, false, s.msg(r, "media.streams_playing"), map[string]interface{}{"streams": n, "where": notes})
			return
		}
	}
	failed := s.restartMediaApps("the household's „restart now\"")
	writeDiskJSON(w, http.StatusOK, len(failed) == 0, "", map[string]interface{}{"failed": failed})
}

// ServeMediaAPI answers /api/storage/media/* (dispatched by ServeStorageAPI).
func (s *Server) ServeMediaAPI(w http.ResponseWriter, r *http.Request) bool {
	switch {
	case r.URL.Path == "/api/storage/media/status" && r.Method == http.MethodGet:
		writeDiskJSON(w, http.StatusOK, true, "", s.MediaNotice(true))
	case r.URL.Path == "/api/storage/media/restart" && r.Method == http.MethodPost:
		s.handleMediaRestart(w, r)
	case r.URL.Path == "/api/storage/media/wire" && r.Method == http.MethodPost:
		writeDiskJSON(w, http.StatusOK, true, "", s.RunMediaWiring("the household's „connect now\""))
	default:
		return false
	}
	return true
}

// ── Safe removal and return (R-1044) ─────────────────────────────────────────────────────────────────────────────

// mediaSafeRemoval runs the media half of a safe removal BEFORE the agent unmounts: pause the drive's torrents, mark the
// drive removed (it stays registered), recreate the media apps and the file browser without it. Returns the paused
// torrents for the record.
func (s *Server) mediaSafeRemoval(path, durableID string) []string {
	d := mediawire.Drive{Path: path}
	paused, err := s.wirer().PauseOnDrive(d)
	if err != nil {
		s.logger.Printf("[WARN] [media] safe removal %s: pausing its torrents: %v (continuing — a paused torrent is a courtesy, the release below is the safety)", path, err)
	}
	if err := s.settings.SetEjected(path, durableID, paused); err != nil {
		s.logger.Printf("[WARN] [media] safe removal %s: recording it: %v", path, err)
	}
	for _, name := range s.mediaApps() {
		if s.appHeld(name) {
			continue
		}
		// The override no longer lists the removed drive, so the recreate lets go of it.
		if err := s.stackMgr.RestartStack(name); err != nil {
			s.logger.Printf("[WARN] [media] safe removal %s: recreating %s without it: %v", path, name, err)
		}
	}
	s.SyncFileBrowserMountsForce()
	s.logger.Printf("[INFO] [media] safe removal %s: %d torrent(s) paused, the media apps and the file browser let go of it", path, len(paused))
	return paused
}

// reconcileEjectedDrives handles a safely removed drive: once its disk has LEFT (not attached any more) and is BACK
// (attached again), re-bind it under the parent, clear the removal, resume its torrents, and raise the notice.
func (s *Server) reconcileEjectedDrives(ctx context.Context, agent *agentapi.Client, disks []agentapi.DiskInfo) {
	for _, sp := range s.settings.GetStoragePaths() {
		if sp.EjectedAt == "" {
			continue
		}
		step, disk := ejectStep(sp, disks)
		switch step {
		case ejectWaitGone, ejectWaitBack:
			continue
		case ejectGone:
			_ = s.settings.SetEjectSeenGone(sp.Path)
			s.logger.Printf("[INFO] [media] safely removed drive %s: its disk has left the box", sp.Path)
			continue
		}
		if !disk.BoundUnderParent && disk.MountPath != "" {
			if err := agent.GuestAttach(ctx, disk.MountPath); err != nil {
				s.logger.Printf("[WARN] [media] safely removed drive %s is back: re-attach (raw %s): %v", sp.Path, disk.MountPath, err)
				continue
			}
		}
		if err := s.settings.ClearEjected(sp.Path); err != nil {
			s.logger.Printf("[WARN] [media] safely removed drive %s is back: clearing the removal: %v", sp.Path, err)
			continue
		}
		s.logger.Printf("[INFO] [media] safely removed drive %s is BACK — registered as before, no household step", sp.Path)
		s.mediaDriveEvent(sp.Path, "returned")
		// The file browser holds no stream: it gets the drive back at once. The torrents resume after the media apps'
		// restart (restartMediaApps → TakeResumable).
		go s.SyncFileBrowserMountsForce()
	}
}

// ejectDurableID finds the durable id of the disk behind a registered path (for finding it again after the return).
func ejectDurableID(disks []agentapi.DiskInfo, path string) string {
	for _, d := range disks {
		if d.GuestPath == path {
			return d.DurableID
		}
	}
	return ""
}

var _ = settings.StoragePath{}

// plexClaimCode is the shape of a plex.tv claim code (claim-<20-ish>); anything else is refused before a call.
var plexClaimCode = regexp.MustCompile(`^claim-[A-Za-z0-9_-]{8,64}$`)

// appPlexClaimHandler (R-1028, `15` §10 K6): the household's code claims the RUNNING Plex — POST
// POST <plex by address>:32400/myplex/claim?token=<code> (no Plex token needed: the server is unclaimed; R-1060). The code is never logged
// and never stored. The setup gate opens only when Plex itself then says claimed (the probe), never on the press alone.
func (s *Server) appPlexClaimHandler(w http.ResponseWriter, r *http.Request) {
	var req struct {
		Code string `json:"code"`
	}
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		escrowJSON(w, http.StatusBadRequest, nil, s.msg(r, "err.plex_claim.bad_code"))
		return
	}
	if err := s.applyPlexClaim(strings.TrimSpace(req.Code)); err != nil {
		var ae *afterStartError
		if errors.As(err, &ae) {
			escrowJSON(w, ae.Status, nil, s.msg(r, ae.MsgKey))
			return
		}
		escrowJSON(w, http.StatusConflict, nil, s.msg(r, "err.plex_claim.refused"))
		return
	}
	escrowJSON(w, http.StatusOK, map[string]any{"claimed": true}, "")
}

// applyPlexClaim is the claim itself, unchanged in behaviour since v0.314.0 and now reachable two
// ways: the after-start registry's „plex/PLEX_CLAIM" entry (the app page's card) and the original
// /apps/plex/plex-claim route, which the page kept so an open tab from before the update still works.
// ONE body, so the two cannot drift — the gate still opens only when Plex ITSELF says claimed.
//
// The code is a credential and is never logged, here or in the caller.
func (s *Server) applyPlexClaim(code string) error {
	if !plexClaimCode.MatchString(code) {
		return &afterStartError{Status: http.StatusBadRequest, MsgKey: "err.plex_claim.bad_code"}
	}
	status, err := plexClaim(code)
	s.logger.Printf("[INFO] [web] plex: the household's claim code given to the running server -> HTTP %d (err %v; the code is not logged)", status, err)
	if err != nil || status != http.StatusOK {
		return &afterStartError{Status: http.StatusConflict, MsgKey: "err.plex_claim.refused"}
	}
	for i := 0; i < 10; i++ {
		if plexClaimed() {
			if oerr := s.stackMgr.OpenSetupGate("plex", stacks.SetupGateByProbe); oerr != nil && !errors.Is(oerr, stacks.ErrSetupGateNotClosed) {
				s.logger.Printf("[WARN] [web] plex: claimed, but opening the setup gate failed: %v", oerr)
			}
			s.logger.Printf("[INFO] [web] plex: claimed — the setup gate is open (R-1028)")
			return nil
		}
		time.Sleep(time.Second)
	}
	return &afterStartError{Status: http.StatusConflict, MsgKey: "err.plex_claim.not_claimed"}
}

// plexHTTP is the client for the box's own calls to Plex (a seam: tests dial a fake that refuses the name „plex").
var plexHTTP = &http.Client{Timeout: 30 * time.Second}

// plexClaim is the call (a seam for tests). R-1060 (v0.314.0): by ADDRESS — Plex answered 401 to the name „plex".
var plexClaim = func(code string) (int, error) {
	base, err := mediawire.PlexURL(32400)
	if err != nil {
		return 0, err
	}
	req, err := http.NewRequest(http.MethodPost, base+"/myplex/claim?token="+url.QueryEscape(code), nil)
	if err != nil {
		return 0, err
	}
	req.Header.Set("Accept", "application/json")
	resp, err := plexHTTP.Do(req)
	if err != nil {
		return 0, err
	}
	resp.Body.Close()
	return resp.StatusCode, nil
}

// The safe-removal states (R-1044).
const (
	ejectWaitGone = iota // removed, the disk is still in the box
	ejectGone            // the disk has just been seen gone
	ejectWaitBack        // gone, not back yet
	ejectBack            // back: re-attach and clear
)

// ejectStep is the PURE decision of reconcileEjectedDrives. A disk is PHYSICALLY present when the agent reports a backing
// device: its `state` stays „attached" for a registered disk that is gone (measured on Tester 1, Part G4: Media D pulled,
// state attached, backing_device "", total_bytes 0). Pinned by TestEjectStep_G4.
func ejectStep(sp settings.StoragePath, disks []agentapi.DiskInfo) (int, *agentapi.DiskInfo) {
	var disk *agentapi.DiskInfo
	for i := range disks {
		if (sp.EjectDurableID != "" && disks[i].DurableID == sp.EjectDurableID) || (sp.EjectDurableID == "" && disks[i].GuestPath == sp.Path) {
			disk = &disks[i]
		}
	}
	present := disk != nil && disk.BackingDevice != ""
	switch {
	case !sp.EjectSeenGone && present:
		return ejectWaitGone, disk
	case !sp.EjectSeenGone:
		return ejectGone, disk
	case !present:
		return ejectWaitBack, disk
	}
	return ejectBack, disk
}

// plexClaimed asks the running Plex itself whether it is claimed (`/identity`, JSON only with this Accept header —
// measured 2026-10-15). A seam for tests. The live template carries NO setup_done_probe on purpose: a controller older
// than 0.313.0 would send no Accept header, read XML as „not JSON" and refuse the household's „Kész" forever.
var plexClaimed = func() bool {
	base, err := mediawire.PlexURL(32400)
	if err != nil {
		return false
	}
	req, err := http.NewRequest(http.MethodGet, base+"/identity", nil)
	if err != nil {
		return false
	}
	req.Header.Set("Accept", "application/json")
	ctx, cancel := context.WithTimeout(req.Context(), 5*time.Second)
	defer cancel()
	resp, err := plexHTTP.Do(req.WithContext(ctx))
	if err != nil {
		return false
	}
	defer resp.Body.Close()
	var id struct {
		MediaContainer struct {
			Claimed bool `json:"claimed"`
		} `json:"MediaContainer"`
	}
	return resp.StatusCode == http.StatusOK && json.NewDecoder(resp.Body).Decode(&id) == nil && id.MediaContainer.Claimed
}