nsfwapp/backend/disk_guard.go

// backend\disk_guard.go

package main

import (
	"context"
	"fmt"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync/atomic"
	"time"

	godisk "github.com/shirou/gopsutil/v3/disk"
)

// -------------------------
// Low disk space guard
//  - pausiert Autostart
//  - stoppt laufende Downloads
// -------------------------

const (
	diskGuardInterval = 1 * time.Second
)

var diskEmergency int32 // 0=false, 1=true

type diskStatusResp struct {
	Emergency      bool   `json:"emergency"`
	PauseGB        int    `json:"pauseGB"`
	ResumeGB       int    `json:"resumeGB"`
	FreeBytes      uint64 `json:"freeBytes"`
	FreeBytesHuman string `json:"freeBytesHuman"`
	RecordPath     string `json:"recordPath"`
}

func diskStatusHandler(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet && r.Method != http.MethodHead {
		http.Error(w, "Nur GET/HEAD erlaubt", http.StatusMethodNotAllowed)
		return
	}

	s := getSettings()

	pauseGB, resumeGB, _, _, _ := computeDiskThresholds()

	free, dir, _ := minFreeBytesForDiskGuard(s)

	resp := diskStatusResp{
		Emergency:      atomic.LoadInt32(&diskEmergency) == 1,
		PauseGB:        pauseGB,
		ResumeGB:       resumeGB,
		FreeBytes:      free,
		FreeBytesHuman: formatBytesSI(int64(free)),
		RecordPath:     dir,
	}

	w.Header().Set("Cache-Control", "no-store")
	writeJSON(w, http.StatusOK, resp)
}

// stopJobsInternal markiert Jobs als "stopping" und cancelt sie (inkl. Preview-FFmpeg Kill).
// Nutzt 2 notify-Pushes, damit die UI Phase/Progress sofort sieht.
func stopJobsInternal(list []*RecordJob) {
	if len(list) == 0 {
		return
	}

	type payload struct {
		cmd           *exec.Cmd
		cancel        context.CancelFunc
		previewCancel context.CancelFunc
	}

	pl := make([]payload, 0, len(list))
	ids := make([]string, 0, len(list))
	nowMs := time.Now().UnixMilli()

	jobsMu.Lock()
	for _, job := range list {
		if job == nil {
			continue
		}
		if job.EndedAt != nil || isTerminalJobStatus(job.Status) || isPostworkJob(job) {
			continue
		}

		job.Phase = "stopping"
		job.Progress = 10
		job.StopRequestedAtMs = nowMs
		job.StopAttempts++

		pl = append(pl, payload{
			cmd:           job.previewCmd,
			cancel:        job.cancel,
			previewCancel: job.previewCancel,
		})
		ids = append(ids, job.ID)

		job.previewCmd = nil
		job.previewCancel = nil
		job.LiveThumbStarted = false
		job.PreviewDir = ""
	}
	jobsMu.Unlock()

	for _, id := range ids {
		_ = publishJobSnapshot(id)
		_ = publishModelJobUpsert(id)
	}

	for _, p := range pl {
		if p.previewCancel != nil {
			p.previewCancel()
		}
		if p.cmd != nil && p.cmd.Process != nil {
			_ = p.cmd.Process.Kill()
		}
		if p.cancel != nil {
			p.cancel()
		}
	}

	for _, id := range ids {
		_ = publishJobSnapshot(id)
		_ = publishModelJobUpsert(id)
	}
}

func stopAllStoppableJobs() int {
	stoppable := make([]*RecordJob, 0, 16)

	jobsMu.RLock()
	for _, j := range jobs {
		if j == nil {
			continue
		}

		if j.Status != JobRunning {
			continue
		}

		phase := strings.ToLower(strings.TrimSpace(j.Phase))

		// ✅ Im Disk-Notfall ALLES stoppen, was noch schreibt.
		// Wir skippen nur Jobs, die sowieso schon im "stopping" sind.
		if phase == "stopping" {
			continue
		}

		stoppable = append(stoppable, j)
	}
	jobsMu.RUnlock()

	stopJobsInternal(stoppable)
	return len(stoppable)
}

func sizeOfPathBestEffort(p string) uint64 {
	p = strings.TrimSpace(p)
	if p == "" {
		return 0
	}

	// relativ -> absolut versuchen
	if !filepath.IsAbs(p) {
		if abs, err := resolvePathRelativeToApp(p); err == nil && strings.TrimSpace(abs) != "" {
			p = abs
		}
	}

	fi, err := os.Stat(p)
	if err != nil || fi.IsDir() || fi.Size() <= 0 {
		return 0
	}
	return uint64(fi.Size())
}

func inFlightBytesForJob(j *RecordJob) uint64 {
	if j == nil {
		return 0
	}

	// Dateisystem ist die harte Wahrheit. SizeBytes kann bei laufenden Downloads
	// verzögert oder gar nicht aktualisiert sein.
	statSize := sizeOfPathBestEffort(j.Output)

	liveSize := uint64(0)
	if j.SizeBytes > 0 {
		liveSize = uint64(j.SizeBytes)
	}

	if statSize > liveSize {
		return statSize
	}
	return liveSize
}

const giB = uint64(1024 * 1024 * 1024)

// computeDiskThresholds:
//
// Wir brauchen:
//  1. die konfigurierte Mindestreserve
//  2. zusätzlich ungefähr die Größe aller aktiven/queued TS-Dateien,
//     weil TS -> MP4 temporär nochmal ungefähr dieselbe Größe benötigt.
//
// PauseNeed = configReserve + conversionReserve
// ResumeNeed = PauseNeed + 3 GiB Hysterese
func computeDiskThresholds() (pauseGB int, resumeGB int, relevantInFlight uint64, pauseNeed uint64, resumeNeed uint64) {
	s := getSettings()

	configPauseGB := s.LowDiskPauseBelowGB
	if configPauseGB <= 0 {
		configPauseGB = 5
	}

	runningBytes := sumActiveRecordingBytes()
	postworkBytes := reservedPostworkBytes()

	relevantInFlight = runningBytes + postworkBytes

	configReserve := uint64(configPauseGB) * giB

	pauseNeed = configReserve + relevantInFlight
	resumeNeed = pauseNeed + 3*giB

	pauseGB = int((pauseNeed + giB - 1) / giB)
	resumeGB = int((resumeNeed + giB - 1) / giB)

	if pauseGB > 10_000 {
		pauseGB = 10_000
	}
	if resumeGB > 10_000 {
		resumeGB = 10_000
	}

	return
}

// ✅ Summe der "wachsenden" Daten (running + remuxing etc.)
// Idee: Für TS->MP4 Peak brauchst du grob nochmal die Größe der aktuellen Datei als Reserve.
func sumInFlightBytes() uint64 {
	var sum uint64

	jobsMu.RLock()
	defer jobsMu.RUnlock()

	for _, j := range jobs {
		if j == nil {
			continue
		}
		if j.Status != JobRunning {
			continue
		}

		sum += inFlightBytesForJob(j)
	}

	return sum
}

func sumActiveRecordingBytes() uint64 {
	var sum uint64

	jobsMu.RLock()
	defer jobsMu.RUnlock()

	for _, j := range jobs {
		if j == nil {
			continue
		}

		// Nur echte aktive Recorder. Postwork wird separat gezählt.
		if !isActiveRecordingJob(j) {
			continue
		}

		size := inFlightBytesForJob(j)
		if size == 0 {
			continue
		}

		sum += size
	}

	return sum
}

func shouldReservePostworkJob(j *RecordJob) bool {
	if j == nil {
		return false
	}

	if isTerminalJobStatus(j.Status) {
		return false
	}

	if !isPostworkJob(j) {
		return false
	}

	if strings.TrimSpace(j.PostWorkKey) == "" {
		return false
	}

	if j.PostWork != nil {
		st := strings.ToLower(strings.TrimSpace(j.PostWork.State))
		if st == "queued" || st == "running" {
			return true
		}
		if j.PostWork.Position > 0 {
			return true
		}
	}

	state := getEffectivePostworkState(j)
	return state == "queued" || state == "running"
}

func reservedPostworkBytes() uint64 {
	var sum uint64

	jobsMu.RLock()
	defer jobsMu.RUnlock()

	for _, j := range jobs {
		if !shouldReservePostworkJob(j) {
			continue
		}

		size := inFlightBytesForJob(j)
		if size == 0 {
			continue
		}

		sum += size
	}

	return sum
}

func resolveDiskGuardDirs(s RecorderSettings) []string {
	rawDirs := []string{s.RecordDir, s.DoneDir}

	seen := map[string]bool{}
	out := make([]string, 0, len(rawDirs))

	for _, raw := range rawDirs {
		raw = strings.TrimSpace(raw)
		if raw == "" {
			continue
		}

		dir, _ := resolvePathRelativeToApp(raw)
		dir = strings.TrimSpace(dir)
		if dir == "" {
			dir = raw
		}
		if dir == "" {
			continue
		}

		dir = filepath.Clean(dir)
		key := strings.ToLower(dir)

		if seen[key] {
			continue
		}
		seen[key] = true
		out = append(out, dir)
	}

	return out
}

func minFreeBytesForDiskGuard(s RecorderSettings) (uint64, string, bool) {
	dirs := resolveDiskGuardDirs(s)
	if len(dirs) == 0 {
		return 0, "", false
	}

	var minFree uint64
	var minDir string
	okAny := false

	for _, dir := range dirs {
		u, err := godisk.Usage(dir)
		if err != nil || u == nil {
			continue
		}

		if !okAny || u.Free < minFree {
			minFree = u.Free
			minDir = dir
			okAny = true
		}
	}

	return minFree, minDir, okAny
}

// startDiskSpaceGuard läuft im Backend und reagiert auch ohne offenen Browser.
// Bei wenig freiem Platz:
//   - diskEmergency aktivieren (Autostart blockieren)
//   - laufende Jobs stoppen
//
// Bei Erholung (Resume-Schwelle):
//   - diskEmergency automatisch wieder freigeben
func startDiskSpaceGuard() {
	checkOnce := func() {
		s := getSettings()

		free, dir, ok := minFreeBytesForDiskGuard(s)
		if !ok {
			return
		}

		pauseGB, resumeGB, relevantInFlight, pauseNeed, resumeNeed := computeDiskThresholds()

		wasEmergency := atomic.LoadInt32(&diskEmergency) == 1

		isLowForPause := free < pauseNeed
		isHighEnoughForResume := free >= resumeNeed

		if !wasEmergency {
			if !isLowForPause {
				return
			}

			atomic.StoreInt32(&diskEmergency, 1)
			broadcastAutostartPaused()

			fmt.Printf(
				"🛑 [disk] Low space: free=%s (%dB) (< need=%s, %dB, pause=%dGB resume=%dGB, reservedForTS2MP4=%s, %dB) -> stop downloads + block autostart (path=%s)\n",
				formatBytesSI(u64ToI64(free)), free,
				formatBytesSI(u64ToI64(pauseNeed)), pauseNeed,
				pauseGB, resumeGB,
				formatBytesSI(u64ToI64(relevantInFlight)), relevantInFlight,
				dir,
			)

			stopped := stopAllStoppableJobs()
			if stopped > 0 {
				fmt.Printf("🛑 [disk] Stop requested for %d download job(s)\n", stopped)
			}

			return
		}

		if isHighEnoughForResume {
			atomic.StoreInt32(&diskEmergency, 0)
			broadcastAutostartPaused()

			fmt.Printf(
				"✅ [disk] Space recovered: free=%s (%dB) (>= resume=%s, %dB, resume=%dGB, reservedForTS2MP4=%s, %dB) -> unblock autostart (path=%s)\n",
				formatBytesSI(u64ToI64(free)), free,
				formatBytesSI(u64ToI64(resumeNeed)), resumeNeed,
				resumeGB,
				formatBytesSI(u64ToI64(relevantInFlight)), relevantInFlight,
				dir,
			)
		}
	}

	// Sofort beim Start prüfen, nicht erst nach dem ersten Tick.
	checkOnce()

	t := time.NewTicker(diskGuardInterval)
	defer t.Stop()

	for range t.C {
		checkOnce()
	}
}

func ensureDiskGuardAllowsStart() error {
	if atomic.LoadInt32(&diskEmergency) == 1 {
		return appErrorf("Speicherplatz-Notbremse aktiv: neuer Download wird blockiert")
	}

	s := getSettings()

	free, dir, ok := minFreeBytesForDiskGuard(s)
	if !ok {
		return nil
	}

	pauseGB, resumeGB, relevantInFlight, pauseNeed, _ := computeDiskThresholds()
	if free >= pauseNeed {
		return nil
	}

	atomic.StoreInt32(&diskEmergency, 1)
	broadcastAutostartPaused()

	stopped := stopAllStoppableJobs()

	return appErrorf(
		"Speicherplatz-Notbremse aktiv: frei=%s, benötigt=%s, pause=%dGB, resume=%dGB, reserve=%s, gestoppt=%d, pfad=%s",
		formatBytesSI(u64ToI64(free)),
		formatBytesSI(u64ToI64(pauseNeed)),
		pauseGB,
		resumeGB,
		formatBytesSI(u64ToI64(relevantInFlight)),
		stopped,
		dir,
	)
}