nsfwapp/backend/performance.go

// backend\performance.go

package main

import (
	"bytes"
	"context"
	"encoding/json"
	"io"
	"math"
	"net/http"
	"os"
	"runtime"
	"strconv"
	"strings"
	"sync/atomic"
	"time"

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

// Server-Startzeit für Perf/Uptime
var serverStartedAt = time.Now()

var lastCPUUsageBits uint64 // atomic float64 bits

func setLastCPUUsage(v float64) {
	atomic.StoreUint64(&lastCPUUsageBits, math.Float64bits(v))
}

func getLastCPUUsage() float64 {
	return math.Float64frombits(atomic.LoadUint64(&lastCPUUsageBits))
}

func startCPUUsageSampler(ctx context.Context) {
	_, _ = gocpu.Percent(200*time.Millisecond, false)

	t := time.NewTicker(2 * time.Second)

	go func() {
		defer t.Stop()

		for {
			select {
			case <-ctx.Done():
				return
			case <-t.C:
				p, err := gocpu.Percent(0, false)
				if err != nil || len(p) == 0 {
					continue
				}

				usage := p[0]
				if math.IsNaN(usage) || math.IsInf(usage, 0) || usage < 0 {
					continue
				}
				if usage > 100 {
					usage = 100
				}

				setLastCPUUsage(usage)
			}
		}
	}()
}

// ---------- Feste ffmpeg/ffprobe Limiter ----------

type FixedLimiter chan struct{}

func NewFixedLimiter(n int) FixedLimiter {
	if n < 1 {
		n = 1
	}
	return make(FixedLimiter, n)
}

func (l FixedLimiter) Acquire(ctx context.Context) error {
	if l == nil {
		return nil
	}
	if ctx == nil {
		ctx = context.Background()
	}

	select {
	case l <- struct{}{}:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

func (l FixedLimiter) Release() {
	if l == nil {
		return
	}
	select {
	case <-l:
	default:
	}
}

func (l FixedLimiter) Limit() int {
	return cap(l)
}

func (l FixedLimiter) InUse() int {
	return len(l)
}

var (
	previewSem FixedLimiter
	thumbSem   FixedLimiter
	genSem     FixedLimiter
	durSem     FixedLimiter
)

func envInt(name string) (int, bool) {
	v := strings.TrimSpace(os.Getenv(name))
	if v == "" {
		return 0, false
	}
	n, err := strconv.Atoi(v)
	if err != nil {
		return 0, false
	}
	return n, true
}

func initFFmpegLimiters() {
	previewN := 2
	thumbN := 4
	genN := 2
	durN := 8

	if n, ok := envInt("PREVIEW_WORKERS"); ok && n > 0 {
		previewN = n
	}
	if n, ok := envInt("THUMB_WORKERS"); ok && n > 0 {
		thumbN = n
	}
	if n, ok := envInt("GEN_WORKERS"); ok && n > 0 {
		genN = n
	}
	if n, ok := envInt("DUR_WORKERS"); ok && n > 0 {
		durN = n
	}

	previewSem = NewFixedLimiter(previewN)
	thumbSem = NewFixedLimiter(thumbN)
	genSem = NewFixedLimiter(genN)
	durSem = NewFixedLimiter(durN)

	setLastCPUUsage(0)
	startCPUUsageSampler(context.Background())
}

func buildPerfSnapshot() map[string]any {
	var ms runtime.MemStats
	runtime.ReadMemStats(&ms)

	s := getSettings()
	recordDir, _ := resolvePathRelativeToApp(s.RecordDir)

	var diskFreeBytes uint64
	var diskTotalBytes uint64
	var diskUsedPercent float64
	diskPath := recordDir

	if recordDir != "" {
		if u, err := godisk.Usage(recordDir); err == nil && u != nil {
			diskFreeBytes = u.Free
			diskTotalBytes = u.Total
			diskUsedPercent = u.UsedPercent
		}
	}

	// ✅ Dynamische Disk-Schwellen (2× inFlight, Resume = +3GB)
	pauseGB, resumeGB, inFlight, pauseNeed, resumeNeed := computeDiskThresholds()

	resp := map[string]any{
		"ts":        time.Now().UTC().Format(time.RFC3339Nano),
		"serverMs":  time.Now().UTC().UnixMilli(),
		"uptimeSec": time.Since(serverStartedAt).Seconds(),
		"cpuPercent": func() float64 {
			v := getLastCPUUsage()
			if math.IsNaN(v) || math.IsInf(v, 0) || v < 0 {
				return 0
			}
			if v > 100 {
				return 100
			}
			return v
		}(),
		"diskPath":        diskPath,
		"diskFreeBytes":   diskFreeBytes,
		"diskTotalBytes":  diskTotalBytes,
		"diskUsedPercent": diskUsedPercent,
		"diskEmergency":   atomic.LoadInt32(&diskEmergency) == 1,

		"diskPauseBelowGB":  pauseGB,
		"diskResumeAboveGB": resumeGB,

		"diskInFlightBytes":   inFlight,
		"diskInFlightHuman":   formatBytesSI(u64ToI64(inFlight)),
		"diskPauseNeedBytes":  pauseNeed,
		"diskPauseNeedHuman":  formatBytesSI(u64ToI64(pauseNeed)),
		"diskResumeNeedBytes": resumeNeed,
		"diskResumeNeedHuman": formatBytesSI(u64ToI64(resumeNeed)),

		"goroutines": runtime.NumGoroutine(),
		"mem": map[string]any{
			"alloc":     ms.Alloc,
			"heapAlloc": ms.HeapAlloc,
			"heapInuse": ms.HeapInuse,
			"sys":       ms.Sys,
			"numGC":     ms.NumGC,
		},
	}

	sem := map[string]any{}
	if previewSem != nil {
		sem["preview"] = map[string]any{
			"inUse": previewSem.InUse(),
			"limit": previewSem.Limit(),
		}
	}
	if thumbSem != nil {
		sem["thumb"] = map[string]any{
			"inUse": thumbSem.InUse(),
			"limit": thumbSem.Limit(),
		}
	}
	if genSem != nil {
		sem["gen"] = map[string]any{
			"inUse": genSem.InUse(),
			"limit": genSem.Limit(),
		}
	}
	if durSem != nil {
		sem["dur"] = map[string]any{
			"inUse": durSem.InUse(),
			"limit": durSem.Limit(),
		}
	}
	if len(sem) > 0 {
		resp["sem"] = sem
	}

	return resp
}

func pingHandler(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
	}
	w.Header().Set("Cache-Control", "no-store")
	w.WriteHeader(http.StatusNoContent)
}

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

	resp := buildPerfSnapshot()

	w.Header().Set("Content-Type", "application/json")
	w.Header().Set("Cache-Control", "no-store")
	_ = json.NewEncoder(w).Encode(resp)
}

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

	fl, ok := w.(http.Flusher)
	if !ok {
		http.Error(w, "Streaming nicht unterstützt", http.StatusInternalServerError)
		return
	}

	ms := 5000
	if q := r.URL.Query().Get("ms"); q != "" {
		if v, err := strconv.Atoi(q); err == nil {
			if v < 1000 {
				v = 1000
			}
			if v > 30000 {
				v = 30000
			}
			ms = v
		}
	}

	w.Header().Set("Content-Type", "text/event-stream; charset=utf-8")
	w.Header().Set("Cache-Control", "no-store")
	w.Header().Set("Connection", "keep-alive")
	w.Header().Set("X-Accel-Buffering", "no")

	ctx := r.Context()

	send := func() error {
		payload := buildPerfSnapshot()

		var buf bytes.Buffer
		if err := json.NewEncoder(&buf).Encode(payload); err != nil {
			return err
		}

		_, _ = io.WriteString(w, "event: perf\n")
		_, _ = io.WriteString(w, "data: ")
		_, _ = w.Write(buf.Bytes())
		_, _ = io.WriteString(w, "\n")
		fl.Flush()
		return nil
	}

	_ = send()

	t := time.NewTicker(time.Duration(ms) * time.Millisecond)
	hb := time.NewTicker(15 * time.Second)
	defer t.Stop()
	defer hb.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			_ = send()
		case <-hb.C:
			_, _ = io.WriteString(w, ": keep-alive\n\n")
			fl.Flush()
		}
	}
}