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 = 5 * 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()

	recordDirAbs, _ := resolvePathRelativeToApp(s.RecordDir)
	dir := strings.TrimSpace(recordDirAbs)
	if dir == "" {
		dir = strings.TrimSpace(s.RecordDir)
	}

	free := uint64(0)
	if dir != "" {
		if u, err := godisk.Usage(dir); err == nil && u != nil {
			free = u.Free
		}
	}

	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))

	jobsMu.Lock()
	for _, job := range list {
		if job == nil {
			continue
		}
		job.Phase = "stopping"
		job.Progress = 10

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

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

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

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
	}
	// Prefer live-tracked bytes if available (accurate & cheap).
	if j.SizeBytes > 0 {
		return uint64(j.SizeBytes)
	}
	return sizeOfPathBestEffort(j.Output)
}

const giB = uint64(1024 * 1024 * 1024)

// computeDiskThresholds:
// Pause = max(lowDiskPauseBelowGB, ceil(relevantInFlightBytes / GiB))
// Resume = Pause + 3 GB (Hysterese)
//
// relevantInFlightBytes = Summe aller laufenden Downloads,
// deren aktuelle Dateigröße über AutoDeleteSmallDownloadsBelowMB liegt.
//
// Damit greift immer mindestens die konfigurierte Mindestschwelle,
// zusätzlich aber auch eine dynamische Schwelle basierend auf den
// "relevanten" Downloads, die nicht automatisch gelöscht würden.
func computeDiskThresholds() (pauseGB int, resumeGB int, relevantInFlight uint64, pauseNeed uint64, resumeNeed uint64) {
	s := getSettings()

	runningBytes := sumInFlightBytesAboveAutoDeleteThreshold()
	queuedPostworkBytes := reservedQueuedPostworkBytesAboveAutoDeleteThreshold()

	// Gesamtrelevanz = laufende Schreiblast + reservierter Bedarf der anstehenden Nacharbeit
	relevantInFlight = runningBytes + queuedPostworkBytes

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

	dynamicPauseGB := 0
	if relevantInFlight > 0 {
		dynamicPauseGB = int((relevantInFlight + giB - 1) / giB) // ceil
	}

	pauseGB = dynamicPauseGB
	if pauseGB < configPauseGB {
		pauseGB = configPauseGB
	}

	if pauseGB > 10_000 {
		pauseGB = 10_000
	}

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

	pauseNeed = uint64(pauseGB) * giB
	resumeNeed = uint64(resumeGB) * giB
	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 sumInFlightBytesAboveAutoDeleteThreshold() uint64 {
	s := getSettings()

	thresholdMB := s.AutoDeleteSmallDownloadsBelowMB
	if thresholdMB < 0 {
		thresholdMB = 0
	}

	thresholdBytes := uint64(thresholdMB) * 1024 * 1024

	var sum uint64

	jobsMu.RLock()
	defer jobsMu.RUnlock()

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

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

		// Nur Downloads berücksichtigen, die über der Auto-Delete-Grenze liegen.
		// Kleine Dateien würden später ohnehin automatisch entfernt.
		if size <= thresholdBytes {
			continue
		}

		sum += size
	}

	return sum
}

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

	// Bereits laufende Jobs werden schon über sumInFlight... erfasst.
	if j.Status == JobRunning {
		return false
	}

	// Nur Jobs mit PostWork-Bezug
	if !isPostworkJob(j) {
		return false
	}

	// Terminale Jobs nicht berücksichtigen
	if isTerminalJobStatus(j.Status) {
		return false
	}

	// Nur wenn wirklich noch ein PostWorkKey existiert
	if strings.TrimSpace(j.PostWorkKey) == "" {
		return false
	}

	// queued / running aus der Queue-Info berücksichtigen
	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
		}
	}

	// Fallback-Heuristik
	state := getEffectivePostworkState(j)
	return state == "queued" || state == "running"
}

func reservedQueuedPostworkBytesAboveAutoDeleteThreshold() uint64 {
	s := getSettings()

	thresholdMB := s.AutoDeleteSmallDownloadsBelowMB
	if thresholdMB < 0 {
		thresholdMB = 0
	}
	thresholdBytes := uint64(thresholdMB) * 1024 * 1024

	var sum uint64

	jobsMu.RLock()
	defer jobsMu.RUnlock()

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

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

		// Kleine Dateien ignorieren, wenn sie ohnehin auto-gelöscht würden
		if size <= thresholdBytes {
			continue
		}

		sum += size
	}

	return sum
}

// 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() {
	t := time.NewTicker(diskGuardInterval)
	defer t.Stop()

	for range t.C {
		s := getSettings()

		// Pfad bestimmen, auf dem wir freien Speicher prüfen
		recordDirAbs, _ := resolvePathRelativeToApp(s.RecordDir)
		dir := strings.TrimSpace(recordDirAbs)
		if dir == "" {
			dir = strings.TrimSpace(s.RecordDir)
		}
		if dir == "" {
			continue
		}

		u, err := godisk.Usage(dir)
		if err != nil || u == nil {
			continue
		}
		free := u.Free

		// ✅ Schwellen:
		// Pause = max(config lowDiskPauseBelowGB, ceil((2 * inFlight) / GiB))
		// Resume = Pause + 3 GB
		// pauseNeed/resumeNeed sind die benötigten freien Bytes
		pauseGB, resumeGB, relevantInFlight, pauseNeed, resumeNeed := computeDiskThresholds()

		// ✅ diskEmergency NICHT sticky behalten.
		// Stattdessen dynamisch mit Hysterese setzen/löschen:
		//
		// - triggern bei free < pauseNeed
		// - freigeben erst bei free >= resumeNeed
		//
		// So kann die Notbremse später erneut greifen.

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

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

		if !wasEmergency {
			// Normalzustand: nur triggern, wenn unter Pause-Schwelle
			if !isLowForPause {
				continue
			}

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

			fmt.Printf(
				"🛑 [disk] Low space: free=%s (%dB) (< %s, %dB, pause=%dGB resume=%dGB, relevantBytes=%s, %dB) -> stop jobs + block autostart via diskEmergency (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 job(s)\n", stopped)
			}
			continue
		}

		// ✅ Emergency ist aktiv: nur freigeben, wenn Resume-Schwelle erreicht ist
		if isHighEnoughForResume {
			atomic.StoreInt32(&diskEmergency, 0)
			broadcastAutostartPaused()

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