nsfwapp/backend/tasks_assets.go

// backend\tasks_assets.go
package main

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"
)

// ---------------------------
// Tasks: Missing Assets erzeugen
// ---------------------------

type AssetsTaskState struct {
	Running           bool       `json:"running"`
	Total             int        `json:"total"`
	Done              int        `json:"done"`
	GeneratedThumbs   int        `json:"generatedThumbs"`
	GeneratedPreviews int        `json:"generatedPreviews"`
	Skipped           int        `json:"skipped"`
	StartedAt         time.Time  `json:"startedAt"`
	FinishedAt        *time.Time `json:"finishedAt,omitempty"`
	Error             string     `json:"error,omitempty"`
	CurrentFile       string     `json:"currentFile,omitempty"`
	CurrentQueue      string     `json:"currentQueue,omitempty"`
	CurrentPhase      string     `json:"currentPhase,omitempty"`
	CurrentLabel      string     `json:"currentLabel,omitempty"`
}

var assetsTaskMu sync.Mutex
var assetsTaskState AssetsTaskState
var assetsTaskCancel context.CancelFunc

// updateAssetsState mutiert den State atomar und triggert danach SSE notify.
// notifyAssetsChanged() muss außerhalb des Locks passieren.
func updateAssetsState(fn func(st *AssetsTaskState)) AssetsTaskState {
	assetsTaskMu.Lock()
	fn(&assetsTaskState)
	st := assetsTaskState
	assetsTaskMu.Unlock()

	notifyAssetsChanged()
	publishTaskState()
	return st
}

func snapshotAssetsState() AssetsTaskState {
	assetsTaskMu.Lock()
	st := assetsTaskState
	assetsTaskMu.Unlock()
	return st
}

func getGenerateAssetsTaskStatus() AssetsTaskState {
	return snapshotAssetsState()
}

func tasksGenerateAssets(w http.ResponseWriter, r *http.Request) {
	switch r.Method {
	case http.MethodGet:
		// GET bleibt als Fallback/Debug möglich (UI nutzt SSE)
		st := snapshotAssetsState()
		writeJSON(w, http.StatusOK, st)
		return

	case http.MethodPost:
		assetsTaskMu.Lock()
		if assetsTaskState.Running {
			st := assetsTaskState
			assetsTaskMu.Unlock()
			writeJSON(w, http.StatusOK, st)
			return
		}

		// cancelbarer Context (pro Run)
		ctx, cancel := context.WithCancel(context.Background())
		assetsTaskCancel = cancel

		now := time.Now()
		assetsTaskState = AssetsTaskState{
			Running:           true,
			Total:             0,
			Done:              0,
			GeneratedThumbs:   0,
			GeneratedPreviews: 0,
			Skipped:           0,
			StartedAt:         now,
			FinishedAt:        nil,
			Error:             "",
			CurrentFile:       "",
			CurrentQueue:      "",
			CurrentPhase:      "",
			CurrentLabel:      "",
		}
		st := assetsTaskState
		assetsTaskMu.Unlock()

		notifyAssetsChanged()
		publishTaskState()

		go runGenerateMissingAssets(ctx)

		writeJSON(w, http.StatusOK, st)
		return

	case http.MethodDelete:
		assetsTaskMu.Lock()
		cancel := assetsTaskCancel
		running := assetsTaskState.Running
		assetsTaskMu.Unlock()

		if !running || cancel == nil {
			w.WriteHeader(http.StatusNoContent)
			return
		}

		// canceln: Worker merkt das beim nächsten ctx.Err() und beendet sauber
		cancel()

		// UI sofort informieren (ohne Running künstlich auf false zu setzen —
		// das macht der Worker zuverlässig im finishWithErr(context.Canceled))
		st := updateAssetsState(func(st *AssetsTaskState) {
			if st.Running {
				st.Error = "Abgebrochen"
			}
		})

		writeJSON(w, http.StatusOK, st)
		return

	default:
		http.Error(w, "Nur GET/POST/DELETE", http.StatusMethodNotAllowed)
		return
	}
}

func publishAssetsTaskPhase(fileName string, queue string, phase string, state string, label string) {
	fileName = strings.TrimSpace(fileName)
	if fileName == "" {
		return
	}

	base := strings.TrimSuffix(fileName, filepath.Ext(fileName))
	assetID := stripHotPrefix(base)

	publishFinishedPostworkPhase(
		fileName,
		assetID,
		queue,
		phase,
		state,
		label,
		"",
	)
}

func clearAssetsTaskFinishedPostworkStates(fileName string) {
	defs := []struct {
		queue string
		phase string
	}{
		{queue: "postwork", phase: "meta"},
		{queue: "postwork", phase: "thumb"},
		{queue: "postwork", phase: "teaser"},
		{queue: "postwork", phase: "sprites"},
		{queue: "enrich", phase: "analyze"},
	}

	for _, def := range defs {
		publishAssetsTaskPhase(fileName, def.queue, def.phase, "missing", "")
	}
}

func runGenerateMissingAssets(ctx context.Context) {
	// Worker-Ende: CancelFunc zurücksetzen (pro Run)
	defer func() {
		assetsTaskMu.Lock()
		assetsTaskCancel = nil
		assetsTaskMu.Unlock()
	}()

	finishWithErr := func(err error) {
		now := time.Now()

		updateAssetsState(func(st *AssetsTaskState) {
			st.Running = false
			st.FinishedAt = &now
			st.CurrentFile = ""
			st.CurrentQueue = ""
			st.CurrentPhase = ""
			st.CurrentLabel = ""

			if err == nil {
				st.Error = ""
				return
			}

			if errors.Is(err, context.Canceled) {
				st.Error = "Abgebrochen"
			} else {
				st.Error = err.Error()
			}
		})
	}

	s := getSettings()
	doneAbs, err := resolvePathRelativeToApp(s.DoneDir)
	if err != nil || strings.TrimSpace(doneAbs) == "" {
		finishWithErr(fmt.Errorf("doneDir auflösung fehlgeschlagen: %v", err))
		return
	}

	type item struct {
		name string
		path string
	}

	indexedItems, sortedIdx := buildDoneIndex(doneAbs)

	// gleiche Grundreihenfolge wie FinishedDownloads-Default:
	// includeKeep = true, sort = completed_desc
	order := sortedIdx["1|completed_desc"]

	seen := map[string]struct{}{}
	items := make([]item, 0, len(order))

	for _, idx := range order {
		if idx < 0 || idx >= len(indexedItems) {
			continue
		}

		j := indexedItems[idx].job
		if j == nil {
			continue
		}

		full := strings.TrimSpace(j.Output)
		if full == "" {
			continue
		}

		if _, ok := seen[full]; ok {
			continue
		}
		seen[full] = struct{}{}

		items = append(items, item{
			name: filepath.Base(full),
			path: full,
		})
	}

	// Initialisierung: Total etc. + SSE Push
	type assetCandidate struct {
		name        string
		path        string
		id          string
		beforeTruth finishedPhaseTruth
	}

	candidates := make([]assetCandidate, 0, len(items))
	skippedCount := 0

	for _, it := range items {
		if err := ctx.Err(); err != nil {
			finishWithErr(err)
			return
		}

		base := strings.TrimSuffix(it.name, filepath.Ext(it.name))
		id := stripHotPrefix(base)
		if strings.TrimSpace(id) == "" {
			clearAssetsTaskFinishedPostworkStates(it.name)
			skippedCount++
			continue
		}

		vfi, verr := os.Stat(it.path)
		if verr != nil || vfi == nil || vfi.IsDir() || vfi.Size() <= 0 {
			clearAssetsTaskFinishedPostworkStates(it.name)
			skippedCount++
			continue
		}

		beforeTruth := finishedPhaseTruthForID(id)

		needMeta := !beforeTruth.MetaReady
		needThumb := !beforeTruth.ThumbReady
		needTeaser := !beforeTruth.TeaserReady
		needSprites := !beforeTruth.SpritesReady
		needAnalyze := !beforeTruth.AnalyzeReady

		// Bereits vollständig -> gar nicht erst in die Arbeitsliste aufnehmen
		if !needMeta && !needThumb && !needTeaser && !needSprites && !needAnalyze {
			clearAssetsTaskFinishedPostworkStates(it.name)
			skippedCount++
			continue
		}

		candidates = append(candidates, assetCandidate{
			name:        it.name,
			path:        it.path,
			id:          id,
			beforeTruth: beforeTruth,
		})
	}

	updateAssetsState(func(st *AssetsTaskState) {
		st.Total = len(candidates)
		st.Done = 0
		st.GeneratedThumbs = 0
		st.GeneratedPreviews = 0
		st.Skipped = skippedCount
		st.Error = ""
		st.CurrentFile = ""
		st.CurrentQueue = ""
		st.CurrentPhase = ""
		st.CurrentLabel = ""
	})

	if len(candidates) == 0 {
		finishWithErr(nil)
		return
	}

	for i, it := range candidates {
		if err := ctx.Err(); err != nil {
			finishWithErr(err)
			return
		}

		updateAssetsState(func(st *AssetsTaskState) {
			st.CurrentFile = it.name
		})

		id := it.id

		vfi, verr := os.Stat(it.path)
		if verr != nil || vfi == nil || vfi.IsDir() || vfi.Size() <= 0 {
			clearAssetsTaskFinishedPostworkStates(it.name)

			updateAssetsState(func(st *AssetsTaskState) {
				st.Done = i + 1
			})
			continue
		}

		beforeTruth := it.beforeTruth

		needMeta := !beforeTruth.MetaReady
		needThumb := !beforeTruth.ThumbReady
		needTeaser := !beforeTruth.TeaserReady
		needSprites := !beforeTruth.SpritesReady
		needAnalyze := !beforeTruth.AnalyzeReady

		// Pfade einmalig über zentralen Helper
		_, _, _, _, metaPath, perr := assetPathsForID(id)
		if perr != nil {
			if needMeta {
				publishAssetsTaskPhase(it.name, "postwork", "meta", "error", "Meta fehlgeschlagen")
			}
			if needThumb {
				publishAssetsTaskPhase(it.name, "postwork", "thumb", "error", "Vorschaubild fehlgeschlagen")
			}
			if needTeaser {
				publishAssetsTaskPhase(it.name, "postwork", "teaser", "error", "Teaser fehlgeschlagen")
			}
			if needSprites {
				publishAssetsTaskPhase(it.name, "postwork", "sprites", "error", "Sprites fehlgeschlagen")
			}
			if needAnalyze {
				publishAssetsTaskPhase(it.name, "enrich", "analyze", "error", "KI-Analyse fehlgeschlagen")
			}

			updateAssetsState(func(st *AssetsTaskState) {
				st.Error = "mindestens ein Eintrag konnte nicht verarbeitet werden (siehe Logs)"
				st.Done = i + 1
			})
			fmt.Println("⚠️ assetPathsForID:", perr)
			continue
		}

		// SourceURL best-effort: aus bestehender meta.json
		sourceURL := ""
		if u, ok := readVideoMetaSourceURL(metaPath, vfi); ok {
			sourceURL = u
		}

		thumbGenerated := false
		previewGenerated := false

		// ----------------
		// META
		// ----------------
		if needMeta {
			updateAssetsState(func(st *AssetsTaskState) {
				st.CurrentFile = it.name
				st.CurrentQueue = "postwork"
				st.CurrentPhase = "meta"
				st.CurrentLabel = "Meta"
			})

			publishAssetsTaskPhase(it.name, "postwork", "meta", "running", "Meta")

			okMeta, merr := ensureMetaForVideoCtx(ctx, it.path, sourceURL)
			if merr != nil || !okMeta {
				publishAssetsTaskPhase(it.name, "postwork", "meta", "error", "Meta fehlgeschlagen")
			} else {
				publishAssetsTaskPhase(it.name, "postwork", "meta", "done", "Meta")
			}
		}

		// ----------------
		// THUMB
		// ----------------
		if needThumb {
			updateAssetsState(func(st *AssetsTaskState) {
				st.CurrentFile = it.name
				st.CurrentQueue = "postwork"
				st.CurrentPhase = "thumb"
				st.CurrentLabel = "Vorschaubild"
			})

			publishAssetsTaskPhase(it.name, "postwork", "thumb", "running", "Vorschaubild")

			resThumb, terr := ensurePrimaryAssetsForVideoWithProgressCtx(ctx, it.path, sourceURL, nil)
			if terr != nil {
				publishAssetsTaskPhase(it.name, "postwork", "thumb", "error", "Vorschaubild fehlgeschlagen")
			} else if finishedPhaseTruthForID(id).ThumbReady {
				if resThumb.ThumbGenerated {
					thumbGenerated = true
				}
				publishAssetsTaskPhase(it.name, "postwork", "thumb", "done", "Vorschaubild")
			} else {
				publishAssetsTaskPhase(it.name, "postwork", "thumb", "error", "Vorschaubild fehlgeschlagen")
			}
		}

		// ----------------
		// TEASER
		// ----------------
		if needTeaser {
			updateAssetsState(func(st *AssetsTaskState) {
				st.CurrentFile = it.name
				st.CurrentQueue = "postwork"
				st.CurrentPhase = "teaser"
				st.CurrentLabel = "Teaser"
			})

			publishAssetsTaskPhase(it.name, "postwork", "teaser", "running", "Teaser")

			genTeaser, terr := ensureTeaserForVideoCtx(ctx, it.path, sourceURL)
			if terr != nil {
				publishAssetsTaskPhase(it.name, "postwork", "teaser", "error", "Teaser fehlgeschlagen")
			} else if finishedPhaseTruthForID(id).TeaserReady {
				if genTeaser {
					previewGenerated = true
				}
				publishAssetsTaskPhase(it.name, "postwork", "teaser", "done", "Teaser")
			} else {
				publishAssetsTaskPhase(it.name, "postwork", "teaser", "error", "Teaser fehlgeschlagen")
			}
		}

		// ----------------
		// SPRITES
		// ----------------
		if needSprites {
			updateAssetsState(func(st *AssetsTaskState) {
				st.CurrentFile = it.name
				st.CurrentQueue = "postwork"
				st.CurrentPhase = "sprites"
				st.CurrentLabel = "Sprites"
			})

			publishAssetsTaskPhase(it.name, "postwork", "sprites", "running", "Sprites")

			_, serr := ensureSpritesForVideoCtx(ctx, it.path, sourceURL)
			if serr != nil {
				publishAssetsTaskPhase(it.name, "postwork", "sprites", "error", "Sprites fehlgeschlagen")
			} else if finishedPhaseTruthForID(id).SpritesReady {
				publishAssetsTaskPhase(it.name, "postwork", "sprites", "done", "Sprites")
			} else {
				publishAssetsTaskPhase(it.name, "postwork", "sprites", "error", "Sprites fehlgeschlagen")
			}
		}

		// ----------------
		// ANALYZE
		// ----------------
		if needAnalyze {
			updateAssetsState(func(st *AssetsTaskState) {
				st.CurrentFile = it.name
				st.CurrentQueue = "enrich"
				st.CurrentPhase = "analyze"
				st.CurrentLabel = "KI-Analyse"
			})

			publishAssetsTaskPhase(it.name, "enrich", "analyze", "running", "KI-Analyse")

			_, aerr := ensureAnalyzeForVideoCtx(ctx, it.path, sourceURL, "nsfw")
			if aerr != nil {
				publishAssetsTaskPhase(it.name, "enrich", "analyze", "error", "KI-Analyse fehlgeschlagen")
			} else if finishedPhaseTruthForID(id).AnalyzeReady {
				publishAssetsTaskPhase(it.name, "enrich", "analyze", "done", "KI-Analyse")
			} else {
				publishAssetsTaskPhase(it.name, "enrich", "analyze", "error", "KI-Analyse fehlgeschlagen")
			}
		}

		updateAssetsState(func(st *AssetsTaskState) {
			if thumbGenerated {
				st.GeneratedThumbs++
			}
			if previewGenerated {
				st.GeneratedPreviews++
			}
			st.Done = i + 1
		})
	}

	finishWithErr(nil)
}