nsfwapp/backend/tasks_regenerate_assets.go

// backend\tasks_regenerate_assets.go

package main

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

const regenerateAssetsMaxParallel = 3

type regenerateAssetsRequest struct {
	File string `json:"file"`
}

type regenerateAssetsPhase struct {
	Queue string `json:"queue"`
	Phase string `json:"phase"`
	State string `json:"state"`
	Label string `json:"label,omitempty"`
	TS    int64  `json:"ts"`
}

type regenerateAssetsJob struct {
	File      string
	VideoPath string
	AssetID   string

	Ctx    context.Context
	Cancel context.CancelFunc

	State string
	Error string

	EnqueuedAt time.Time
	StartedAt  time.Time
	EndedAt    time.Time

	Phases map[string]regenerateAssetsPhase
}

var (
	regenerateAssetsMu             sync.Mutex
	regenerateAssetsJobs           = map[string]*regenerateAssetsJob{} // key = file
	regenerateAssetsQueue          []string                            // file names
	regenerateAssetsWorkersStarted bool
)

func publishFinishedPostworkPhase(
	file string,
	assetID string,
	queue string,
	phase string,
	state string,
	label string,
) {
	setRegenerateAssetsJobPhase(file, queue, phase, state, label)

	ev := finishedPostworkEvent{
		Type:    "finished_postwork",
		File:    file,
		AssetID: assetID,
		Queue:   queue,
		State:   state,
		Phase:   phase,
		Label:   label,
		TS:      time.Now().UnixMilli(),
	}
	if b, err := json.Marshal(ev); err == nil {
		publishSSE("finishedPostwork", b)
	}
}

type regeneratePhaseTruth struct {
	MetaReady    bool
	ThumbReady   bool
	TeaserReady  bool
	SpritesReady bool
	AnalyzeReady bool
}

func regeneratePhaseTruthForVideo(videoPath string, assetID string) regeneratePhaseTruth {
	videoPath = strings.TrimSpace(videoPath)
	assetID = strings.TrimSpace(assetID)

	if assetID == "" && videoPath != "" {
		assetID = assetIDFromVideoPath(videoPath)
	}

	var out regeneratePhaseTruth
	if assetID == "" {
		return out
	}

	_, thumbPath, previewPath, spritePath, metaPath, err := assetPathsForID(assetID)
	if err != nil {
		return out
	}

	fileExists := func(p string) bool {
		fi, err := os.Stat(strings.TrimSpace(p))
		return err == nil && fi != nil && !fi.IsDir() && fi.Size() > 0
	}

	out.ThumbReady = fileExists(thumbPath)
	out.TeaserReady = fileExists(previewPath)
	out.SpritesReady = fileExists(spritePath)

	// Meta ist nur "ready", wenn die meta.json gültig zur aktuellen Videodatei passt
	if videoPath != "" {
		if vfi, err := os.Stat(videoPath); err == nil && vfi != nil && !vfi.IsDir() && vfi.Size() > 0 {
			if meta, ok := readVideoMetaIfValid(metaPath, vfi); ok && meta != nil {
				out.MetaReady = true
			}
		}

		// KI nur dann fertig, wenn Sprite existiert UND AI in meta.json vorhanden ist
		out.AnalyzeReady = out.SpritesReady && hasAIResultsForOutput(videoPath)
	}

	return out
}

func firstMissingPrimaryRegeneratePhaseError(videoPath string, assetID string) error {
	truth := regeneratePhaseTruthForVideo(videoPath, assetID)

	if !truth.MetaReady {
		return errors.New("meta.json fehlt oder ist noch unvollständig")
	}
	if !truth.ThumbReady {
		return errors.New("preview.jpg fehlt oder ist leer")
	}
	if !truth.TeaserReady {
		return errors.New("preview.mp4 fehlt oder ist leer")
	}
	return nil
}

func firstMissingDeferredRegeneratePhaseError(videoPath string, assetID string) error {
	truth := regeneratePhaseTruthForVideo(videoPath, assetID)

	if !truth.SpritesReady {
		return errors.New("preview-sprite.jpg fehlt oder ist leer")
	}
	if !truth.AnalyzeReady {
		return errors.New("KI-Analyse fehlt oder wurde nicht in meta.json gespeichert")
	}
	return nil
}

func ensureRegenerateAssetsWorkersStarted() {
	regenerateAssetsMu.Lock()
	defer regenerateAssetsMu.Unlock()

	if regenerateAssetsWorkersStarted {
		return
	}

	regenerateAssetsWorkersStarted = true

	for i := 0; i < regenerateAssetsMaxParallel; i++ {
		go regenerateAssetsWorker()
	}
}

func enqueueRegenerateAssetsJob(job *regenerateAssetsJob) {
	regenerateAssetsMu.Lock()
	defer regenerateAssetsMu.Unlock()

	regenerateAssetsJobs[job.File] = job
	regenerateAssetsQueue = append(regenerateAssetsQueue, job.File)
}

func nextQueuedRegenerateAssetsJob() *regenerateAssetsJob {
	regenerateAssetsMu.Lock()
	defer regenerateAssetsMu.Unlock()

	for len(regenerateAssetsQueue) > 0 {
		file := regenerateAssetsQueue[0]
		regenerateAssetsQueue = regenerateAssetsQueue[1:]

		job := regenerateAssetsJobs[file]
		if job == nil {
			continue
		}
		if job.State != "queued" {
			continue
		}

		job.State = "running"
		job.StartedAt = time.Now()
		return job
	}

	return nil
}

func finishRegenerateAssetsJob(file string, state string, errMsg string) {
	regenerateAssetsMu.Lock()
	defer regenerateAssetsMu.Unlock()

	job := regenerateAssetsJobs[file]
	if job == nil {
		return
	}

	job.State = state
	job.Error = strings.TrimSpace(errMsg)
	job.EndedAt = time.Now()
}

func setRegenerateAssetsJobPhase(
	file string,
	queue string,
	phase string,
	state string,
	label string,
) {
	regenerateAssetsMu.Lock()
	defer regenerateAssetsMu.Unlock()

	job := regenerateAssetsJobs[file]
	if job == nil {
		return
	}
	if job.Phases == nil {
		job.Phases = make(map[string]regenerateAssetsPhase)
	}

	key := strings.ToLower(strings.TrimSpace(queue)) + ":" + strings.ToLower(strings.TrimSpace(phase))
	job.Phases[key] = regenerateAssetsPhase{
		Queue: strings.TrimSpace(queue),
		Phase: strings.TrimSpace(phase),
		State: strings.TrimSpace(state),
		Label: strings.TrimSpace(label),
		TS:    time.Now().UnixMilli(),
	}
}

func removeRegenerateAssetsJobLater(file string, after time.Duration) {
	time.AfterFunc(after, func() {
		regenerateAssetsMu.Lock()
		defer regenerateAssetsMu.Unlock()

		delete(regenerateAssetsJobs, file)
	})
}

func getRegenerateAssetsJobsSnapshot() []map[string]any {
	regenerateAssetsMu.Lock()
	defer regenerateAssetsMu.Unlock()

	out := make([]map[string]any, 0, len(regenerateAssetsJobs))
	for _, job := range regenerateAssetsJobs {
		phases := make([]map[string]any, 0, len(job.Phases))
		for _, p := range job.Phases {
			phases = append(phases, map[string]any{
				"queue": p.Queue,
				"phase": p.Phase,
				"state": p.State,
				"label": p.Label,
				"ts":    p.TS,
			})
		}

		out = append(out, map[string]any{
			"file":       job.File,
			"assetId":    job.AssetID,
			"state":      job.State,
			"error":      job.Error,
			"phases":     phases,
			"enqueuedAt": job.EnqueuedAt.UnixMilli(),
			"startedAt": func() int64 {
				if job.StartedAt.IsZero() {
					return 0
				}
				return job.StartedAt.UnixMilli()
			}(),
			"endedAt": func() int64 {
				if job.EndedAt.IsZero() {
					return 0
				}
				return job.EndedAt.UnixMilli()
			}(),
		})
	}

	return out
}

func regenerateAssetsWorker() {
	for {
		job := nextQueuedRegenerateAssetsJob()
		if job == nil {
			time.Sleep(250 * time.Millisecond)
			continue
		}

		runRegenerateAssetsJob(job)
	}
}

func runRegenerateAssetsJob(job *regenerateAssetsJob) {
	file := job.File
	videoPath := job.VideoPath
	id := job.AssetID
	ctx := job.Ctx

	setRegenerateAssetsTaskRunning(file, file)
	cancelDeferredEnrichForFile(file)

	cancelJob := func() {
		finishRegenerateAssetsJob(file, "cancelled", "Abgebrochen")
		setRegenerateAssetsTaskError(file, "Abgebrochen")

		publishFinishedPostworkPhase(file, id, "postwork", "meta", "missing", "")
		publishFinishedPostworkPhase(file, id, "postwork", "thumb", "missing", "")
		publishFinishedPostworkPhase(file, id, "postwork", "teaser", "missing", "")
		publishFinishedPostworkPhase(file, id, "postwork", "sprites", "missing", "")
		publishFinishedPostworkPhase(file, id, "enrich", "analyze", "missing", "")

		removeRegenerateAssetsJobLater(file, 2*time.Second)
	}

	failJob := func(queue, phase, msg string) {
		if queue != "" && phase != "" {
			publishFinishedPostworkPhase(file, id, queue, phase, "error", "Fehler")
		}

		finishRegenerateAssetsJob(file, "error", msg)
		setRegenerateAssetsTaskError(file, msg)
		removeRegenerateAssetsJobLater(file, 5*time.Second)
	}

	if ctx.Err() == context.Canceled {
		cancelJob()
		return
	}

	// Initialstatus
	publishFinishedPostworkPhase(file, id, "postwork", "meta", "running", "Meta")
	publishFinishedPostworkPhase(file, id, "postwork", "thumb", "queued", "Vorschaubild")
	publishFinishedPostworkPhase(file, id, "postwork", "teaser", "queued", "Teaser")
	publishFinishedPostworkPhase(file, id, "postwork", "sprites", "queued", "Sprites")
	publishFinishedPostworkPhase(file, id, "enrich", "analyze", "queued", "KI-Analyse")

	// Alte Assets löschen
	if id != "" {
		if assetDir, _, _, _, metaPath, err := assetPathsForID(id); err == nil {
			_ = os.Remove(filepath.Join(assetDir, "preview.jpg"))
			_ = os.Remove(filepath.Join(assetDir, "preview.mp4"))
			_ = os.Remove(filepath.Join(assetDir, "preview-sprite.jpg"))
			_ = os.Remove(metaPath)
		}
	}

	// 1) Meta
	if _, err := ensureMetaForVideoCtx(ctx, videoPath, ""); err != nil {
		if errors.Is(err, context.Canceled) || ctx.Err() == context.Canceled {
			cancelJob()
			return
		}
		failJob("postwork", "meta", err.Error())
		return
	}

	if !regeneratePhaseTruthForVideo(videoPath, id).MetaReady {
		failJob("postwork", "meta", "meta.json fehlt oder ist noch unvollständig")
		return
	}

	publishFinishedPostworkPhase(file, id, "postwork", "meta", "done", "Meta")

	// 2) Thumb
	publishFinishedPostworkPhase(file, id, "postwork", "thumb", "running", "Vorschaubild")

	if _, err := ensurePrimaryAssetsForVideoWithProgressCtx(ctx, videoPath, "", nil); err != nil {
		if errors.Is(err, context.Canceled) || ctx.Err() == context.Canceled {
			cancelJob()
			return
		}
		failJob("postwork", "thumb", err.Error())
		return
	}

	if !regeneratePhaseTruthForVideo(videoPath, id).ThumbReady {
		failJob("postwork", "thumb", "preview.jpg fehlt oder ist leer")
		return
	}

	publishFinishedPostworkPhase(file, id, "postwork", "thumb", "done", "Vorschaubild")

	// 3) Teaser
	publishFinishedPostworkPhase(file, id, "postwork", "teaser", "running", "Teaser")

	if _, err := ensureTeaserForVideoCtx(ctx, videoPath, ""); err != nil {
		if errors.Is(err, context.Canceled) || ctx.Err() == context.Canceled {
			cancelJob()
			return
		}
		failJob("postwork", "teaser", err.Error())
		return
	}

	if !regeneratePhaseTruthForVideo(videoPath, id).TeaserReady {
		failJob("postwork", "teaser", "preview.mp4 fehlt oder ist leer")
		return
	}

	publishFinishedPostworkPhase(file, id, "postwork", "teaser", "done", "Teaser")

	// 4) Sprites
	publishFinishedPostworkPhase(file, id, "postwork", "sprites", "running", "Sprites")

	if _, err := ensureSpritesForVideoCtx(ctx, videoPath, ""); err != nil {
		if errors.Is(err, context.Canceled) || ctx.Err() == context.Canceled {
			cancelJob()
			return
		}
		failJob("postwork", "sprites", err.Error())
		return
	}

	if !regeneratePhaseTruthForVideo(videoPath, id).SpritesReady {
		failJob("postwork", "sprites", "preview-sprite.jpg fehlt oder ist leer")
		return
	}

	publishFinishedPostworkPhase(file, id, "postwork", "sprites", "done", "Sprites")

	// 5) KI-Analyse
	publishFinishedPostworkPhase(file, id, "enrich", "analyze", "running", "KI-Analyse")

	if _, err := ensureAnalyzeForVideoCtx(ctx, videoPath, "", "nsfw"); err != nil {
		if errors.Is(err, context.Canceled) || ctx.Err() == context.Canceled {
			cancelJob()
			return
		}
		failJob("enrich", "analyze", err.Error())
		return
	}

	if !regeneratePhaseTruthForVideo(videoPath, id).AnalyzeReady {
		failJob("enrich", "analyze", "KI-Analyse fehlt oder wurde nicht in meta.json gespeichert")
		return
	}

	publishFinishedPostworkPhase(file, id, "enrich", "analyze", "done", "KI-Analyse")

	finishRegenerateAssetsJob(file, "done", "")
	setRegenerateAssetsTaskDone(file, "Fertig")
	removeRegenerateAssetsJobLater(file, 3*time.Second)
}

func handleRegenerateAssets(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost && r.Method != http.MethodDelete {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}

	if r.Method == http.MethodDelete {
		var req regenerateAssetsRequest
		_ = json.NewDecoder(r.Body).Decode(&req)

		file := strings.TrimSpace(req.File)

		if file == "" {
			regenerateAssetsMu.Lock()
			for _, j := range regenerateAssetsJobs {
				if j != nil && (j.State == "running" || j.State == "queued") {
					file = j.File
					break
				}
			}
			regenerateAssetsMu.Unlock()
		}

		if file == "" {
			http.Error(w, "missing file", http.StatusBadRequest)
			return
		}

		regenerateAssetsMu.Lock()
		job := regenerateAssetsJobs[file]
		regenerateAssetsMu.Unlock()

		if job != nil && job.Cancel != nil {
			job.Cancel()
		}

		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(map[string]any{
			"ok":        true,
			"cancelled": true,
			"file":      file,
		})
		return
	}

	var req regenerateAssetsRequest
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		http.Error(w, "invalid json body", http.StatusBadRequest)
		return
	}

	file := strings.TrimSpace(req.File)
	if file == "" {
		http.Error(w, "missing file", http.StatusBadRequest)
		return
	}
	if !isAllowedVideoExt(file) {
		http.Error(w, "nicht erlaubt", http.StatusForbidden)
		return
	}

	s := getSettings()
	doneAbs, err := resolvePathRelativeToApp(s.DoneDir)
	if err != nil {
		http.Error(w, "doneDir auflösung fehlgeschlagen: "+err.Error(), http.StatusInternalServerError)
		return
	}
	if strings.TrimSpace(doneAbs) == "" {
		http.Error(w, "doneDir ist leer", http.StatusBadRequest)
		return
	}

	videoPath, _, fi, err := resolveDoneFileByName(doneAbs, file)
	if err != nil {
		http.Error(w, "datei nicht gefunden", http.StatusNotFound)
		return
	}
	if fi == nil || fi.IsDir() || fi.Size() <= 0 {
		http.Error(w, "invalid output path", http.StatusBadRequest)
		return
	}

	ensureRegenerateAssetsWorkersStarted()

	regenerateAssetsMu.Lock()
	if existing := regenerateAssetsJobs[file]; existing != nil {
		regenerateAssetsMu.Unlock()

		w.Header().Set("Content-Type", "application/json")
		phases := make([]map[string]any, 0, len(existing.Phases))
		for _, p := range existing.Phases {
			phases = append(phases, map[string]any{
				"queue": p.Queue,
				"phase": p.Phase,
				"state": p.State,
				"label": p.Label,
				"ts":    p.TS,
			})
		}

		_ = json.NewEncoder(w).Encode(map[string]any{
			"ok":     true,
			"file":   file,
			"state":  existing.State,
			"error":  existing.Error,
			"phases": phases,
		})
		return
	}
	regenerateAssetsMu.Unlock()

	runCtx, runCancel := context.WithCancel(context.Background())

	id := assetIDFromVideoPath(videoPath)
	if strings.TrimSpace(id) == "" {
		id = stripHotPrefix(strings.TrimSuffix(filepath.Base(file), filepath.Ext(file)))
	}

	job := &regenerateAssetsJob{
		File:       file,
		VideoPath:  videoPath,
		AssetID:    id,
		Ctx:        runCtx,
		Cancel:     runCancel,
		State:      "queued",
		EnqueuedAt: time.Now(),
		Phases:     make(map[string]regenerateAssetsPhase),
	}

	enqueueRegenerateAssetsJob(job)

	publishFinishedPostworkPhase(file, id, "postwork", "meta", "queued", "Meta")
	publishFinishedPostworkPhase(file, id, "postwork", "thumb", "queued", "Vorschaubild")
	publishFinishedPostworkPhase(file, id, "postwork", "teaser", "queued", "Teaser")
	publishFinishedPostworkPhase(file, id, "postwork", "sprites", "queued", "Sprites")
	publishFinishedPostworkPhase(file, id, "enrich", "analyze", "queued", "KI-Analyse")

	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(map[string]any{
		"ok":    true,
		"file":  file,
		"state": "queued",
	})
}