nsfwapp/backend/tasks_status.go

// backend\tasks_status.go

package main

import (
	"encoding/json"
	"fmt"
	"net/http"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

type CleanupTaskState struct {
	ID          string     `json:"id"`
	Queued      bool       `json:"queued"`
	Running     bool       `json:"running"`
	QueuedAt    time.Time  `json:"queuedAt"`
	StartedAt   time.Time  `json:"startedAt"`
	FinishedAt  *time.Time `json:"finishedAt,omitempty"`
	Text        string     `json:"text,omitempty"`
	Error       string     `json:"error,omitempty"`
	CurrentFile string     `json:"currentFile,omitempty"`
	Done        int        `json:"done"`
	Total       int        `json:"total"`
}

type CheckVideosTaskState struct {
	ID          string     `json:"id"`
	Queued      bool       `json:"queued"`
	Running     bool       `json:"running"`
	QueuedAt    time.Time  `json:"queuedAt"`
	StartedAt   time.Time  `json:"startedAt"`
	FinishedAt  *time.Time `json:"finishedAt,omitempty"`
	Text        string     `json:"text,omitempty"`
	Error       string     `json:"error,omitempty"`
	CurrentFile string     `json:"currentFile,omitempty"`
	Done        int        `json:"done"`
	Total       int        `json:"total"`
}

type RegenerateAssetsTaskState struct {
	Running bool   `json:"running"`
	File    string `json:"file,omitempty"`
	Text    string `json:"text,omitempty"`
	Error   string `json:"error,omitempty"`
}

type GenerateAssetsLiveStatus struct {
	Running      bool    `json:"running"`
	Queued       bool    `json:"queued"`
	CurrentFile  string  `json:"currentFile,omitempty"`
	CurrentQueue string  `json:"currentQueue,omitempty"`
	CurrentPhase string  `json:"currentPhase,omitempty"`
	CurrentLabel string  `json:"currentLabel,omitempty"`
	Progress     float64 `json:"progress,omitempty"`
	Current      int     `json:"current,omitempty"`
	Total        int     `json:"total,omitempty"`
	TS           int64   `json:"ts,omitempty"`
}

type TasksStatusResponse struct {
	GenerateAssets       GenerateAssetsLiveStatus `json:"generateAssets"`
	GenerateAssetsJobs   []AssetsTaskState        `json:"generateAssetsJobs"`
	RegenerateAssetsJobs []map[string]any         `json:"regenerateAssetsJobs"`
	CheckVideosJobs      []CheckVideosTaskState   `json:"checkVideosJobs"`
	CleanupJobs          []CleanupTaskState       `json:"cleanupJobs"`
	AnyRunning           bool                     `json:"anyRunning"`
}

var (
	cleanupTaskMu    sync.Mutex
	cleanupTaskState CleanupTaskState

	regenerateAssetsTaskMu    sync.Mutex
	regenerateAssetsTaskState RegenerateAssetsTaskState
)

var nextTaskID atomic.Uint64

func newTaskID(prefix string) string {
	n := nextTaskID.Add(1)
	return fmt.Sprintf("%s-%d-%d", prefix, time.Now().UnixNano(), n)
}

func cleanupNowTime() *time.Time {
	t := time.Now()
	return &t
}

func anyToString(v any) string {
	switch x := v.(type) {
	case string:
		return x
	case []byte:
		return string(x)
	default:
		return ""
	}
}

func anyToBool(v any) bool {
	switch x := v.(type) {
	case bool:
		return x
	case string:
		s := strings.TrimSpace(strings.ToLower(x))
		return s == "1" || s == "true" || s == "yes" || s == "running" || s == "queued"
	default:
		return false
	}
}

func anyToInt(v any) int {
	switch x := v.(type) {
	case int:
		return x
	case int64:
		return int(x)
	case float64:
		return int(x)
	case json.Number:
		n, _ := x.Int64()
		return int(n)
	default:
		return 0
	}
}

func anyToFloat64(v any) float64 {
	switch x := v.(type) {
	case float64:
		return x
	case float32:
		return float64(x)
	case int:
		return float64(x)
	case int64:
		return float64(x)
	case json.Number:
		n, _ := x.Float64()
		return n
	default:
		return 0
	}
}

func anyToMap(v any) map[string]any {
	if m, ok := v.(map[string]any); ok {
		return m
	}

	b, err := json.Marshal(v)
	if err != nil {
		return nil
	}

	var out map[string]any
	if err := json.Unmarshal(b, &out); err != nil {
		return nil
	}

	return out
}

func taskFirstString(m map[string]any, keys ...string) string {
	for _, key := range keys {
		if s := strings.TrimSpace(anyToString(m[key])); s != "" {
			return s
		}
	}
	return ""
}

func firstFloat(m map[string]any, keys ...string) float64 {
	for _, key := range keys {
		if v, ok := m[key]; ok {
			f := anyToFloat64(v)
			if f != 0 {
				return f
			}
		}
	}
	return 0
}

func normalizeTaskProgress(v float64) float64 {
	if v > 1 && v <= 100 {
		v = v / 100
	}
	if v < 0 {
		return 0
	}
	if v > 1 {
		return 1
	}
	return v
}

func taskStepLabel(queue, phase string) string {
	queue = strings.TrimSpace(strings.ToLower(queue))
	phase = strings.TrimSpace(strings.ToLower(phase))

	switch queue + ":" + phase {
	case "postwork:meta":
		return "Meta"
	case "postwork:thumb":
		return "Vorschaubild"
	case "postwork:teaser", "postwork:assets":
		return "Teaser"
	case "postwork:sprites":
		return "Sprites"
	case "enrich:analyze", "enrich:assets", "enrich:":
		return "Analyse"
	default:
		if queue == "enrich" {
			return "Analyse"
		}
		if phase != "" {
			return phase
		}
		return "Meta"
	}
}

func taskLabelWithProgress(queue, phase, label string, progress float64) string {
	queue = strings.TrimSpace(strings.ToLower(queue))
	phase = strings.TrimSpace(strings.ToLower(phase))
	label = strings.TrimSpace(label)
	progress = normalizeTaskProgress(progress)

	isAnalyze := queue == "enrich" && (phase == "" || phase == "analyze" || phase == "assets")

	if isAnalyze && progress > 0 {
		return fmt.Sprintf("Analyse %d%%", int(progress*100+0.5))
	}

	if label != "" {
		return label
	}

	return taskStepLabel(queue, phase)
}

func liveStatusFromJobMap(m map[string]any) GenerateAssetsLiveStatus {
	if m == nil {
		return GenerateAssetsLiveStatus{}
	}

	state := strings.TrimSpace(strings.ToLower(taskFirstString(m, "state", "status")))
	running := anyToBool(m["running"]) || state == "running"
	queued := anyToBool(m["queued"]) || state == "queued"

	if !running && !queued {
		return GenerateAssetsLiveStatus{}
	}

	file := taskFirstString(m, "currentFile", "file")
	queue := strings.TrimSpace(strings.ToLower(taskFirstString(m, "currentQueue", "queue")))
	phase := strings.TrimSpace(strings.ToLower(taskFirstString(m, "currentPhase", "phase")))
	label := taskFirstString(m, "currentLabel", "label", "text", "message")

	current := anyToInt(m["current"])
	total := anyToInt(m["total"])

	progress := firstFloat(m, "progress", "overallProgress", "overall")
	progress = normalizeTaskProgress(progress)

	if progress == 0 && total > 0 && current > 0 {
		progress = normalizeTaskProgress(float64(current) / float64(total))
	}

	if queue == "" {
		queue = "postwork"
	}
	if queue == "enrich" && (phase == "" || phase == "assets") {
		phase = "analyze"
	}

	return GenerateAssetsLiveStatus{
		Running:      running,
		Queued:       queued,
		CurrentFile:  file,
		CurrentQueue: queue,
		CurrentPhase: phase,
		CurrentLabel: taskLabelWithProgress(queue, phase, label, progress),
		Progress:     progress,
		Current:      current,
		Total:        total,
		TS:           time.Now().UnixMilli(),
	}
}

func liveStatusFromRegenerateJob(job map[string]any) GenerateAssetsLiveStatus {
	if job == nil {
		return GenerateAssetsLiveStatus{}
	}

	file := taskFirstString(job, "currentFile", "file")
	assetID := taskFirstString(job, "assetId")

	rawPhases, _ := job["phases"].([]any)

	for _, raw := range rawPhases {
		p := anyToMap(raw)
		if p == nil {
			continue
		}

		state := strings.TrimSpace(strings.ToLower(taskFirstString(p, "state", "status")))
		if state != "running" && state != "queued" {
			continue
		}

		if file != "" {
			p["file"] = file
		}
		if assetID != "" {
			p["assetId"] = assetID
		}

		return liveStatusFromJobMap(p)
	}

	return liveStatusFromJobMap(job)
}

func buildGenerateAssetsLiveStatus(
	assetsJobs []AssetsTaskState,
	regenerateJobs []map[string]any,
) GenerateAssetsLiveStatus {
	// Running gewinnt vor queued.
	for _, st := range assetsJobs {
		status := liveStatusFromJobMap(anyToMap(st))
		if status.Running {
			return status
		}
	}

	for _, job := range regenerateJobs {
		status := liveStatusFromRegenerateJob(job)
		if status.Running {
			return status
		}
	}

	for _, st := range assetsJobs {
		status := liveStatusFromJobMap(anyToMap(st))
		if status.Queued {
			return status
		}
	}

	for _, job := range regenerateJobs {
		status := liveStatusFromRegenerateJob(job)
		if status.Queued {
			return status
		}
	}

	return GenerateAssetsLiveStatus{}
}

func snapshotRegenerateAssetsTaskState() RegenerateAssetsTaskState {
	regenerateAssetsTaskMu.Lock()
	st := regenerateAssetsTaskState
	regenerateAssetsTaskMu.Unlock()
	return st
}

func setRegenerateAssetsTaskRunning(file, text string) {
	regenerateAssetsTaskMu.Lock()
	regenerateAssetsTaskState = RegenerateAssetsTaskState{
		Running: true,
		File:    strings.TrimSpace(file),
		Text:    strings.TrimSpace(text),
		Error:   "",
	}
	regenerateAssetsTaskMu.Unlock()

	publishTaskState()
}

func setRegenerateAssetsTaskDone(file, text string) {
	regenerateAssetsTaskMu.Lock()
	regenerateAssetsTaskState = RegenerateAssetsTaskState{
		Running: false,
		File:    strings.TrimSpace(file),
		Text:    strings.TrimSpace(text),
		Error:   "",
	}
	regenerateAssetsTaskMu.Unlock()

	publishTaskState()
}

func setRegenerateAssetsTaskError(file, errText string) {
	regenerateAssetsTaskMu.Lock()
	regenerateAssetsTaskState = RegenerateAssetsTaskState{
		Running: false,
		File:    strings.TrimSpace(file),
		Text:    "",
		Error:   strings.TrimSpace(errText),
	}
	regenerateAssetsTaskMu.Unlock()

	publishTaskState()
}

func clearRegenerateAssetsTaskState() {
	regenerateAssetsTaskMu.Lock()
	regenerateAssetsTaskState = RegenerateAssetsTaskState{}
	regenerateAssetsTaskMu.Unlock()

	publishTaskState()
}

func snapshotCleanupTaskState() CleanupTaskState {
	cleanupTaskMu.Lock()
	st := cleanupTaskState
	cleanupTaskMu.Unlock()
	return st
}

func setCleanupTaskQueued(text string) {
	cleanupTaskMu.Lock()
	cleanupTaskState = CleanupTaskState{
		Queued:      true,
		Running:     false,
		QueuedAt:    time.Now(),
		StartedAt:   time.Time{},
		FinishedAt:  nil,
		Text:        strings.TrimSpace(text),
		Error:       "",
		CurrentFile: "",
		Done:        0,
		Total:       0,
	}
	cleanupTaskMu.Unlock()

	publishTaskState()
}

func setCleanupTaskRunning(text string) {
	cleanupTaskMu.Lock()

	now := time.Now()
	queuedAt := cleanupTaskState.QueuedAt
	if queuedAt.IsZero() {
		queuedAt = now
	}

	cleanupTaskState = CleanupTaskState{
		Queued:      false,
		Running:     true,
		QueuedAt:    queuedAt,
		StartedAt:   now,
		FinishedAt:  nil,
		Text:        strings.TrimSpace(text),
		Error:       "",
		CurrentFile: "",
		Done:        0,
		Total:       0,
	}
	cleanupTaskMu.Unlock()

	publishTaskState()
}

func setCleanupTaskProgress(done, total int, currentFile, text string) {
	cleanupTaskMu.Lock()

	now := time.Now()
	if cleanupTaskState.QueuedAt.IsZero() {
		cleanupTaskState.QueuedAt = now
	}
	if cleanupTaskState.StartedAt.IsZero() {
		cleanupTaskState.StartedAt = now
	}

	cleanupTaskState.Queued = false
	cleanupTaskState.Running = true
	cleanupTaskState.Done = done
	cleanupTaskState.Total = total
	cleanupTaskState.CurrentFile = strings.TrimSpace(currentFile)
	cleanupTaskState.Text = strings.TrimSpace(text)
	cleanupTaskState.Error = ""
	cleanupTaskState.FinishedAt = nil

	cleanupTaskMu.Unlock()

	publishTaskState()
}

func setCleanupTaskDone(text string) {
	cleanupTaskMu.Lock()

	cleanupTaskState.Queued = false
	cleanupTaskState.Running = false
	cleanupTaskState.Text = strings.TrimSpace(text)
	cleanupTaskState.CurrentFile = ""
	cleanupTaskState.Error = ""
	cleanupTaskState.FinishedAt = cleanupNowTime()

	if cleanupTaskState.Total > 0 {
		cleanupTaskState.Done = cleanupTaskState.Total
	}

	cleanupTaskMu.Unlock()

	publishTaskState()
}

func setCleanupTaskError(errText string) {
	cleanupTaskMu.Lock()

	cleanupTaskState.Queued = false
	cleanupTaskState.Running = false
	cleanupTaskState.Text = ""
	cleanupTaskState.Error = strings.TrimSpace(errText)
	cleanupTaskState.CurrentFile = ""
	cleanupTaskState.FinishedAt = cleanupNowTime()

	cleanupTaskMu.Unlock()

	publishTaskState()
}

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

	assetsJobs := snapshotAssetsJobs()
	regenerateJobs := getRegenerateAssetsJobsSnapshot()
	checkVideosJobs := snapshotCheckVideosJobs()
	cleanupJobs := snapshotCleanupJobs()
	generateAssets := buildGenerateAssetsLiveStatus(assetsJobs, regenerateJobs)

	anyRunning := false

	if !anyRunning {
		for _, st := range assetsJobs {
			if st.Queued || st.Running {
				anyRunning = true
				break
			}
		}
	}

	if !anyRunning {
		for _, st := range checkVideosJobs {
			if st.Queued || st.Running {
				anyRunning = true
				break
			}
		}
	}

	if !anyRunning {
		for _, st := range cleanupJobs {
			if st.Queued || st.Running {
				anyRunning = true
				break
			}
		}
	}

	if !anyRunning {
		for _, job := range regenerateJobs {
			state := strings.TrimSpace(strings.ToLower(anyToString(job["state"])))
			if state == "queued" || state == "running" {
				anyRunning = true
				break
			}
		}
	}

	resp := TasksStatusResponse{
		GenerateAssets:       generateAssets,
		GenerateAssetsJobs:   assetsJobs,
		RegenerateAssetsJobs: regenerateJobs,
		CheckVideosJobs:      checkVideosJobs,
		CleanupJobs:          cleanupJobs,
		AnyRunning:           anyRunning,
	}

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