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 regenerateAssetsJob struct {
	File      string
	VideoPath string
	AssetID   string

	Ctx    context.Context
	Cancel context.CancelFunc

	State string // queued | running | done | error | cancelled
	Error string

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

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,
) {
	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)
	}
}

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 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 {
		out = append(out, map[string]any{
			"file":       job.File,
			"assetId":    job.AssetID,
			"state":      job.State,
			"error":      job.Error,
			"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)

	select {
	case <-ctx.Done():
		finishRegenerateAssetsJob(file, "cancelled", "Abgebrochen")
		setRegenerateAssetsTaskError(file, "Abgebrochen")

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

		removeRegenerateAssetsJobLater(file, 2*time.Second)
		return
	default:
	}

	publishFinishedPostworkPhase(file, id, "enrich", "analyze", "missing", "")

	publishFinishedPostworkPhase(file, id, "postwork", "meta", "running", "Meta")
	publishFinishedPostworkPhase(file, id, "postwork", "teaser", "running", "Teaser")
	publishFinishedPostworkPhase(file, id, "postwork", "thumb", "running", "Vorschaubild")
	publishFinishedPostworkPhase(file, id, "postwork", "sprites", "running", "Sprites")

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

	assetWatchDone := make(chan struct{})

	go func(ctx context.Context, file string, id string) {
		ticker := time.NewTicker(300 * time.Millisecond)
		defer ticker.Stop()

		sentMetaDone := false
		sentTeaserDone := false
		sentThumbDone := false
		sentSpritesDone := false

		checkAndEmit := func() {
			truth := finishedPhaseTruthForID(id)

			if truth.MetaReady && !sentMetaDone {
				sentMetaDone = true
				publishFinishedPostworkPhase(file, id, "postwork", "meta", "done", "Meta")
			}
			if truth.TeaserReady && !sentTeaserDone {
				sentTeaserDone = true
				publishFinishedPostworkPhase(file, id, "postwork", "teaser", "done", "Teaser")
			}
			if truth.ThumbReady && !sentThumbDone {
				sentThumbDone = true
				publishFinishedPostworkPhase(file, id, "postwork", "thumb", "done", "Vorschaubild")
			}
			if truth.SpritesReady && !sentSpritesDone {
				sentSpritesDone = true
				publishFinishedPostworkPhase(file, id, "postwork", "sprites", "done", "Sprites")
			}
		}

		for {
			select {
			case <-ctx.Done():
				return
			case <-assetWatchDone:
				checkAndEmit()
				return
			case <-ticker.C:
				checkAndEmit()
			}
		}
	}(ctx, file, id)

	_, errAssets := ensureAssetsForVideoWithProgressCtx(ctx, videoPath, "", nil)
	close(assetWatchDone)

	if errors.Is(errAssets, context.Canceled) || ctx.Err() == context.Canceled {
		finishRegenerateAssetsJob(file, "cancelled", "Abgebrochen")
		setRegenerateAssetsTaskError(file, "Abgebrochen")

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

		removeRegenerateAssetsJobLater(file, 2*time.Second)
		return
	}

	if errAssets == nil {
		publishFinishedPostworkPhase(file, id, "enrich", "analyze", "running", "KI-Analyse")
	}

	errAI := ensureDeferredAssetsAndAI(ctx, videoPath, "")

	if errors.Is(errAI, context.Canceled) || ctx.Err() == context.Canceled {
		finishRegenerateAssetsJob(file, "cancelled", "Abgebrochen")
		setRegenerateAssetsTaskError(file, "Abgebrochen")
		publishFinishedPostworkPhase(file, id, "enrich", "analyze", "missing", "")

		removeRegenerateAssetsJobLater(file, 2*time.Second)
		return
	}

	if errAssets == nil && errAI == nil {
		publishFinishedPostworkPhase(file, id, "enrich", "analyze", "done", "KI-Analyse")
	}

	if errAssets != nil || errAI != nil {
		errMsg := ""
		if errAssets != nil {
			errMsg = errAssets.Error()
		} else {
			errMsg = errAI.Error()
		}

		finishRegenerateAssetsJob(file, "error", errMsg)
		setRegenerateAssetsTaskError(file, errMsg)

		errEv := finishedPostworkEvent{
			Type:    "finished_postwork",
			File:    file,
			AssetID: id,
			Queue:   "enrich",
			State:   "error",
			Phase:   "assets",
			Label:   "Fehler",
			TS:      time.Now().UnixMilli(),
		}
		if b, err := json.Marshal(errEv); err == nil {
			publishSSE("finishedPostwork", b)
		}

		removeRegenerateAssetsJobLater(file, 5*time.Second)
		return
	}

	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 == "" {
			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")
		_ = json.NewEncoder(w).Encode(map[string]any{
			"ok":    true,
			"file":  file,
			"state": existing.State,
		})
		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(),
	}

	enqueueRegenerateAssetsJob(job)

	publishFinishedPostworkPhase(file, id, "postwork", "meta", "queued", "Meta")
	publishFinishedPostworkPhase(file, id, "postwork", "teaser", "queued", "Teaser")
	publishFinishedPostworkPhase(file, id, "postwork", "thumb", "queued", "Vorschaubild")
	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",
	})
}