diff --git a/backend/chaturbate_autostart.go b/backend/chaturbate_autostart.go index 7d792de..984311e 100644 --- a/backend/chaturbate_autostart.go +++ b/backend/chaturbate_autostart.go @@ -143,7 +143,7 @@ func startChaturbateAutoStartWorker(store *ModelStore) { // running users (damit wir nicht doppelt starten) running := map[string]bool{} - jobsMu.Lock() + jobsMu.RLock() for _, j := range jobs { if j == nil || j.Status != JobRunning { continue @@ -153,7 +153,7 @@ func startChaturbateAutoStartWorker(store *ModelStore) { running[u] = true } } - jobsMu.Unlock() + jobsMu.RUnlock() // watched list aus DB watched := store.ListWatchedLite("chaturbate.com") diff --git a/backend/cors.go b/backend/cors.go index bb3485f..4b89986 100644 --- a/backend/cors.go +++ b/backend/cors.go @@ -14,8 +14,12 @@ func withCORS(next http.Handler) http.Handler { w.Header().Set("Access-Control-Allow-Origin", origin) w.Header().Set("Vary", "Origin") w.Header().Set("Access-Control-Allow-Methods", "GET,POST,DELETE,HEAD,OPTIONS") - w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Range, Last-Event-ID") + w.Header().Set( + "Access-Control-Allow-Headers", + "Content-Type, Range, Last-Event-ID, X-Chaturbate-Cookie, Authorization", + ) w.Header().Set("Access-Control-Expose-Headers", "Content-Length, Content-Range, Accept-Ranges") + w.Header().Set("Access-Control-Allow-Credentials", "true") // Nur wenn du wirklich Cookies/Authorization cross-origin brauchst: // w.Header().Set("Access-Control-Allow-Credentials", "true") } diff --git a/backend/disk_guard.go b/backend/disk_guard.go index 78460b7..9e4677f 100644 --- a/backend/disk_guard.go +++ b/backend/disk_guard.go @@ -81,8 +81,9 @@ func stopJobsInternal(list []*RecordJob) { } type payload struct { - cmd *exec.Cmd - cancel context.CancelFunc + cmd *exec.Cmd + cancel context.CancelFunc + previewCancel context.CancelFunc } pl := make([]payload, 0, len(list)) @@ -94,12 +95,24 @@ func stopJobsInternal(list []*RecordJob) { } job.Phase = "stopping" job.Progress = 10 - pl = append(pl, payload{cmd: job.previewCmd, cancel: job.cancel}) + + pl = append(pl, payload{ + cmd: job.previewCmd, + cancel: job.cancel, + previewCancel: job.previewCancel, + }) + job.previewCmd = nil + job.previewCancel = nil + job.LiveThumbStarted = false + job.PreviewDir = "" } jobsMu.Unlock() for _, p := range pl { + if p.previewCancel != nil { + p.previewCancel() + } if p.cmd != nil && p.cmd.Process != nil { _ = p.cmd.Process.Kill() } @@ -112,7 +125,7 @@ func stopJobsInternal(list []*RecordJob) { func stopAllStoppableJobs() int { stoppable := make([]*RecordJob, 0, 16) - jobsMu.Lock() + jobsMu.RLock() for _, j := range jobs { if j == nil { continue @@ -132,7 +145,7 @@ func stopAllStoppableJobs() int { stoppable = append(stoppable, j) } - jobsMu.Unlock() + jobsMu.RUnlock() stopJobsInternal(stoppable) return len(stoppable) @@ -184,7 +197,11 @@ const giB = uint64(1024 * 1024 * 1024) func computeDiskThresholds() (pauseGB int, resumeGB int, relevantInFlight uint64, pauseNeed uint64, resumeNeed uint64) { s := getSettings() - relevantInFlight = sumInFlightBytesAboveAutoDeleteThreshold() + runningBytes := sumInFlightBytesAboveAutoDeleteThreshold() + queuedPostworkBytes := reservedQueuedPostworkBytesAboveAutoDeleteThreshold() + + // Gesamtrelevanz = laufende Schreiblast + reservierter Bedarf der anstehenden Nacharbeit + relevantInFlight = runningBytes + queuedPostworkBytes configPauseGB := s.LowDiskPauseBelowGB if configPauseGB <= 0 { @@ -196,11 +213,8 @@ func computeDiskThresholds() (pauseGB int, resumeGB int, relevantInFlight uint64 dynamicPauseGB = int((relevantInFlight + giB - 1) / giB) // ceil } - // größere Schwelle nehmen: - // - manuelle Mindestreserve - // - dynamische Reserve für relevante laufende Downloads pauseGB = dynamicPauseGB - if pauseGB <= 0 { + if pauseGB < configPauseGB { pauseGB = configPauseGB } @@ -223,8 +237,8 @@ func computeDiskThresholds() (pauseGB int, resumeGB int, relevantInFlight uint64 func sumInFlightBytes() uint64 { var sum uint64 - jobsMu.Lock() - defer jobsMu.Unlock() + jobsMu.RLock() + defer jobsMu.RUnlock() for _, j := range jobs { if j == nil { @@ -252,8 +266,8 @@ func sumInFlightBytesAboveAutoDeleteThreshold() uint64 { var sum uint64 - jobsMu.Lock() - defer jobsMu.Unlock() + jobsMu.RLock() + defer jobsMu.RUnlock() for _, j := range jobs { if j == nil { @@ -280,6 +294,82 @@ func sumInFlightBytesAboveAutoDeleteThreshold() uint64 { return sum } +func shouldReserveQueuedPostworkJob(j *RecordJob) bool { + if j == nil { + return false + } + + // Bereits laufende Jobs werden schon über sumInFlight... erfasst. + if j.Status == JobRunning { + return false + } + + // Nur Jobs mit PostWork-Bezug + if !isPostworkJob(j) { + return false + } + + // Terminale Jobs nicht berücksichtigen + if isTerminalJobStatus(j.Status) { + return false + } + + // Nur wenn wirklich noch ein PostWorkKey existiert + if strings.TrimSpace(j.PostWorkKey) == "" { + return false + } + + // queued / running aus der Queue-Info berücksichtigen + if j.PostWork != nil { + st := strings.ToLower(strings.TrimSpace(j.PostWork.State)) + if st == "queued" || st == "running" { + return true + } + if j.PostWork.Position > 0 { + return true + } + } + + // Fallback-Heuristik + state := getEffectivePostworkState(j) + return state == "queued" || state == "running" +} + +func reservedQueuedPostworkBytesAboveAutoDeleteThreshold() uint64 { + s := getSettings() + + thresholdMB := s.AutoDeleteSmallDownloadsBelowMB + if thresholdMB < 0 { + thresholdMB = 0 + } + thresholdBytes := uint64(thresholdMB) * 1024 * 1024 + + var sum uint64 + + jobsMu.RLock() + defer jobsMu.RUnlock() + + for _, j := range jobs { + if !shouldReserveQueuedPostworkJob(j) { + continue + } + + size := inFlightBytesForJob(j) + if size == 0 { + continue + } + + // Kleine Dateien ignorieren, wenn sie ohnehin auto-gelöscht würden + if size <= thresholdBytes { + continue + } + + sum += size + } + + return sum +} + // startDiskSpaceGuard läuft im Backend und reagiert auch ohne offenen Browser. // Bei wenig freiem Platz: // - diskEmergency aktivieren (Autostart blockieren) @@ -339,7 +429,7 @@ func startDiskSpaceGuard() { broadcastAutostartPaused() fmt.Printf( - "🛑 [disk] Low space: free=%s (%dB) (< %s, %dB, pause=%dGB resume=%dGB, relevantInFlight=%s, %dB) -> stop jobs + block autostart via diskEmergency (path=%s)\n", + "🛑 [disk] Low space: free=%s (%dB) (< %s, %dB, pause=%dGB resume=%dGB, relevantBytes=%s, %dB) -> stop jobs + block autostart via diskEmergency (path=%s)\n", formatBytesSI(u64ToI64(free)), free, formatBytesSI(u64ToI64(pauseNeed)), pauseNeed, pauseGB, resumeGB, @@ -360,7 +450,7 @@ func startDiskSpaceGuard() { broadcastAutostartPaused() fmt.Printf( - "✅ [disk] Space recovered: free=%s (%dB) (>= %s, %dB, resume=%dGB, relevantInFlight=%s, %dB) -> unblock autostart (path=%s)\n", + "✅ [disk] Space recovered: free=%s (%dB) (>= %s, %dB, resume=%dGB, relevantBytes=%s, %dB) -> unblock autostart (path=%s)\n", formatBytesSI(u64ToI64(free)), free, formatBytesSI(u64ToI64(resumeNeed)), resumeNeed, resumeGB, diff --git a/backend/live.go b/backend/live.go index cef75ec..ff3b47a 100644 --- a/backend/live.go +++ b/backend/live.go @@ -165,13 +165,13 @@ func servePreviewHLSFileWithBase(w http.ResponseWriter, r *http.Request, id, fil isIndex := file == "index.m3u8" || file == "index_hq.m3u8" - jobsMu.Lock() + jobsMu.RLock() job, ok := jobs[id] state := "" if ok && job != nil { state = strings.TrimSpace(job.PreviewState) } - jobsMu.Unlock() + jobsMu.RUnlock() // HEAD: quick existence check if r.Method == http.MethodHead { @@ -187,7 +187,9 @@ func servePreviewHLSFileWithBase(w http.ResponseWriter, r *http.Request, id, fil w.WriteHeader(http.StatusNotFound) return } + jobsMu.RLock() previewDir := strings.TrimSpace(job.PreviewDir) + jobsMu.RUnlock() if previewDir == "" { w.WriteHeader(http.StatusNotFound) return @@ -214,9 +216,9 @@ func servePreviewHLSFileWithBase(w http.ResponseWriter, r *http.Request, id, fil ensurePreviewStarted(r, job) touchPreview(job) - jobsMu.Lock() + jobsMu.RLock() state = strings.TrimSpace(job.PreviewState) - jobsMu.Unlock() + jobsMu.RUnlock() if state == "private" { http.Error(w, "model private", http.StatusForbidden) @@ -574,23 +576,28 @@ func recordPreviewLiveFMP4(w http.ResponseWriter, r *http.Request) { return } - jobsMu.Lock() + jobsMu.RLock() job, ok := jobs[id] state := "" + sourceURL := "" if ok && job != nil { state = strings.TrimSpace(job.PreviewState) + sourceURL = strings.TrimSpace(job.SourceURL) } - jobsMu.Unlock() + jobsMu.RUnlock() if !ok || job == nil { http.Error(w, "job nicht gefunden", http.StatusNotFound) return } - username := extractUsername(job.SourceURL) + username := extractUsername(sourceURL) if strings.TrimSpace(username) != "" { + jobsMu.RLock() cookie := strings.TrimSpace(job.PreviewCookie) ua := strings.TrimSpace(job.PreviewUA) + jobsMu.RUnlock() + if ua == "" { ua = "Mozilla/5.0" } @@ -600,8 +607,10 @@ func recordPreviewLiveFMP4(w http.ResponseWriter, r *http.Request) { cancelRefresh() if err == nil && strings.TrimSpace(newHls) != "" { + var oldHls string + jobsMu.Lock() - oldHls := strings.TrimSpace(job.PreviewM3U8) + oldHls = strings.TrimSpace(job.PreviewM3U8) job.PreviewM3U8 = strings.TrimSpace(newHls) job.PreviewCookie = cookie job.PreviewUA = ua @@ -616,15 +625,19 @@ func recordPreviewLiveFMP4(w http.ResponseWriter, r *http.Request) { } } - // ensure ffmpeg preview input data exists - // (PreviewM3U8 + Cookie/UA werden beim Job gesetzt) + // Nach möglichem Refresh alles nochmal sauber lesen + jobsMu.RLock() + state = strings.TrimSpace(job.PreviewState) m3u8 := strings.TrimSpace(job.PreviewM3U8) + cookie := strings.TrimSpace(job.PreviewCookie) + ua := strings.TrimSpace(job.PreviewUA) + jobsMu.RUnlock() + if m3u8 == "" { http.Error(w, "preview m3u8 fehlt", http.StatusNotFound) return } - // states if state == "private" { http.Error(w, "model private", http.StatusForbidden) return @@ -638,30 +651,22 @@ func recordPreviewLiveFMP4(w http.ResponseWriter, r *http.Request) { return } - // Headers: fMP4 stream - w.Header().Set("Content-Type", `video/mp4`) + w.Header().Set("Content-Type", "video/mp4") w.Header().Set("Cache-Control", "no-store") w.Header().Set("X-Accel-Buffering", "no") - // Sehr wichtig: Flushbar? - flusher, okf := w.(http.Flusher) - if !okf { + flusher, ok := w.(http.Flusher) + if !ok { http.Error(w, "Streaming nicht unterstützt", http.StatusInternalServerError) return } - // Client disconnect => ffmpeg stoppen ctx := r.Context() - // Cookie/UA aus Job - cookie := strings.TrimSpace(job.PreviewCookie) - ua := strings.TrimSpace(job.PreviewUA) if ua == "" { ua = "Mozilla/5.0" } - // ffmpeg args: input = m3u8, output = fragmented mp4 to stdout - // ✅ Video + Audio für Browser-Playback args := []string{"-hide_banner", "-loglevel", "error"} if ua != "" { args = append(args, "-user_agent", ua) @@ -670,10 +675,8 @@ func recordPreviewLiveFMP4(w http.ResponseWriter, r *http.Request) { args = append(args, "-headers", fmt.Sprintf("Cookie: %s\r\n", cookie)) } - // Input args = append(args, "-i", m3u8) - // Video + Audio encode (low-latency-ish) args = append(args, "-map", "0:v:0", "-map", "0:a:0?", @@ -681,42 +684,38 @@ func recordPreviewLiveFMP4(w http.ResponseWriter, r *http.Request) { "-c:a", "copy", ) - // Output: fMP4 fragmented to stdout (single HTTP response) args = append(args, "-f", "mp4", "-movflags", "frag_keyframe+empty_moov+default_base_moof", - "-frag_duration", "2000000", // 2s (µs) + "-frag_duration", "2000000", "-min_frag_duration", "2000000", "pipe:1", ) cmd := exec.CommandContext(ctx, ffmpegPath, args...) - // stdout -> response stdout, err := cmd.StdoutPipe() if err != nil { http.Error(w, "ffmpeg stdout pipe failed", http.StatusInternalServerError) return } - // stderr nur für Debug (optional) var stderr bytes.Buffer cmd.Stderr = &stderr - // Start if err := cmd.Start(); err != nil { http.Error(w, "ffmpeg start failed: "+err.Error(), http.StatusInternalServerError) return } - // Wenn Client weg => Prozess killt CommandContext sowieso (ctx cancels), - // aber wir kopieren streaming-mäßig. buf := make([]byte, 32*1024) for { select { case <-ctx.Done(): - _ = cmd.Process.Kill() + if cmd.Process != nil { + _ = cmd.Process.Kill() + } return default: } @@ -734,6 +733,5 @@ func recordPreviewLiveFMP4(w http.ResponseWriter, r *http.Request) { } } - // Wait (verhindert Zombies) _ = cmd.Wait() } diff --git a/backend/main.go b/backend/main.go index edb555d..6a1b006 100644 --- a/backend/main.go +++ b/backend/main.go @@ -90,6 +90,25 @@ type RecordJob struct { cancel context.CancelFunc `json:"-"` } +type jobEventSnapshot struct { + ID string + SourceURL string + Output string + Status JobStatus + StartedAt time.Time + StartedAtMs int64 + EndedAt *time.Time + EndedAtMs int64 + Error string + Phase string + Progress int + SizeBytes int64 + DurationSeconds float64 + PreviewState string + PostWorkKey string + PostWork *PostWorkKeyStatus +} + type dummyResponseWriter struct { h http.Header } @@ -105,6 +124,42 @@ type ffprobeInfo struct { Streams []ffprobeStreamInfo `json:"streams"` } +func publishJobUpsertSnapshot(s jobEventSnapshot) { + ev := map[string]any{ + "type": "job_upsert", + "jobId": s.ID, + "status": string(s.Status), + "sourceUrl": s.SourceURL, + "output": s.Output, + "startedAt": s.StartedAt, + "startedAtMs": s.StartedAtMs, + "endedAt": s.EndedAt, + "endedAtMs": s.EndedAtMs, + "error": s.Error, + "phase": s.Phase, + "progress": s.Progress, + "sizeBytes": s.SizeBytes, + "durationSeconds": s.DurationSeconds, + "previewState": s.PreviewState, + "postWorkKey": s.PostWorkKey, + "ts": time.Now().UnixMilli(), + } + + if s.PostWork != nil { + ev["postWork"] = map[string]any{ + "state": s.PostWork.State, + "position": s.PostWork.Position, + "waiting": s.PostWork.Waiting, + "running": s.PostWork.Running, + "maxParallel": s.PostWork.MaxParallel, + } + } + + if b, err := json.Marshal(ev); err == nil { + publishSSE("job", b) + } +} + func jobMatchesModelKey(j *RecordJob, modelKey string) bool { if j == nil { return false @@ -244,7 +299,6 @@ func startPreviewIdleKiller() { func init() { initFFmpegLimiters() - startCPUUsageSampler(context.Background()) startPreviewIdleKiller() initSSE() @@ -511,8 +565,8 @@ func isAssetIDInUse(id string) bool { return false } - jobsMu.Lock() - defer jobsMu.Unlock() + jobsMu.RLock() + defer jobsMu.RUnlock() for _, j := range jobs { if j == nil { @@ -582,28 +636,61 @@ func purgeDurationCacheForPath(p string) { } func renameGenerated(oldID, newID string) { - thumbsRoot, _ := generatedThumbsRoot() - teaserRoot, _ := generatedTeaserRoot() + oldID = stripHotPrefix(strings.TrimSpace(oldID)) + newID = stripHotPrefix(strings.TrimSpace(newID)) - oldThumb := filepath.Join(thumbsRoot, oldID) - newThumb := filepath.Join(thumbsRoot, newID) - if _, err := os.Stat(oldThumb); err == nil { - if _, err2 := os.Stat(newThumb); os.IsNotExist(err2) { - _ = os.Rename(oldThumb, newThumb) + if oldID == "" || newID == "" || strings.EqualFold(oldID, newID) { + return + } + + var err error + oldID, err = sanitizeID(oldID) + if err != nil || oldID == "" { + return + } + newID, err = sanitizeID(newID) + if err != nil || newID == "" { + return + } + + oldDir, errOld := generatedDirForID(oldID) + newDir, errNew := generatedDirForID(newID) + if errOld != nil || errNew != nil || oldDir == "" || newDir == "" { + return + } + + if _, err := os.Stat(oldDir); err != nil { + return + } + + if _, err := os.Stat(newDir); os.IsNotExist(err) { + _ = renameWithRetry(oldDir, newDir) + return + } + + files := []string{ + "preview.jpg", + "preview.mp4", + "preview-sprite.jpg", + "meta.json", + } + + for _, name := range files { + src := filepath.Join(oldDir, name) + dst := filepath.Join(newDir, name) + + if _, err := os.Stat(src); err != nil { + continue + } + if _, err := os.Stat(dst); os.IsNotExist(err) { + _ = os.MkdirAll(filepath.Dir(dst), 0o755) + _ = renameWithRetry(src, dst) } else { - _ = os.RemoveAll(oldThumb) + _ = removeWithRetry(src) } } - oldTeaser := filepath.Join(teaserRoot, oldID+".mp4") - newTeaser := filepath.Join(teaserRoot, newID+".mp4") - if _, err := os.Stat(oldTeaser); err == nil { - if _, err2 := os.Stat(newTeaser); os.IsNotExist(err2) { - _ = os.Rename(oldTeaser, newTeaser) - } else { - _ = os.Remove(oldTeaser) - } - } + _ = os.Remove(oldDir) } // --- Gemeinsame Status-Werte für MFC --- diff --git a/backend/meta.go b/backend/meta.go index f270c08..ef6ef30 100644 --- a/backend/meta.go +++ b/backend/meta.go @@ -228,11 +228,6 @@ func attachMetaToJobBestEffort(ctx context.Context, job *RecordJob, fullPath str job.Meta = m - if strings.Contains(strings.TrimSpace(fullPath), "alice_kosmos_03_24_2026__03-48-34") { - b, _ := json.MarshalIndent(m, "", " ") - fmt.Println("ATTACH META SNAPSHOT:", string(b)) - } - if job.DurationSeconds <= 0 && m.Media.DurationSeconds > 0 { job.DurationSeconds = m.Media.DurationSeconds } diff --git a/backend/models_api.go b/backend/models_api.go index 20a2d43..52593b4 100644 --- a/backend/models_api.go +++ b/backend/models_api.go @@ -31,6 +31,7 @@ func setModelStore(store *ModelStore) { defer modelStoreMu.Unlock() modelStoreRef = store setCoverModelStore(store) + setChaturbateOnlineModelStore(store) } // ✅ umbenannt, damit es nicht mit models.go kollidiert diff --git a/backend/myfreecams_autostart.go b/backend/myfreecams_autostart.go index 785219f..d5e293c 100644 --- a/backend/myfreecams_autostart.go +++ b/backend/myfreecams_autostart.go @@ -15,18 +15,26 @@ func startMyFreeCamsAutoStartWorker(store *ModelStore) { return } - // pro Model: Retry-Cooldown, damit du nicht permanent die gleichen Models spamst + // pro Model: Retry-Cooldown, damit nicht ständig dieselben Models angestoßen werden const cooldown = 2 * time.Minute - // wie lange wir nach Start warten, ob eine Datei entsteht + // wie lange wir nach Start warten, ob eine Output-Datei entsteht const outputProbeMax = 20 * time.Second + // kleine Pause zwischen Starts, damit nicht zu viele Jobs auf einmal anlaufen + const startGap = 1200 * time.Millisecond + lastAttempt := map[string]time.Time{} tick := time.NewTicker(6 * time.Second) defer tick.Stop() for range tick.C { + // global früh abbrechen + if isAutostartPaused() { + continue + } + s := getSettings() if !s.UseMyFreeCamsWatcher { continue @@ -40,14 +48,11 @@ func startMyFreeCamsAutoStartWorker(store *ModelStore) { // langsam nacheinander starten (keine API -> einzelnes "Anprobieren") for _, m := range watched { - - // ✅ Wenn User den Switch während eines Ticks deaktiviert, sofort stoppen - if !getSettings().UseMyFreeCamsWatcher { + // während des Loops erneut prüfen, falls User live stoppt + if isAutostartPaused() { break } - - // ✅ Wenn im UI "Alle Stoppen" -> Autostart pausiert, sofort aufhören - if isAutostartPaused() { + if !getSettings().UseMyFreeCamsWatcher { break } @@ -61,29 +66,34 @@ func startMyFreeCamsAutoStartWorker(store *ModelStore) { // Fallback modelID = strings.TrimSpace(m.Host) + ":" + strings.TrimSpace(m.ModelKey) } + if modelID == "" { + continue + } // Cooldown if t, ok := lastAttempt[modelID]; ok && time.Since(t) < cooldown { continue } - // bereits als Job aktiv? + // Bereits aktiv? if isJobRunningForURL(u) { continue } lastAttempt[modelID] = time.Now() - job, err := startRecordingInternal(RecordRequest{URL: u, Hidden: true}) + job, err := startRecordingInternal(RecordRequest{ + URL: u, + Hidden: true, + }) if err != nil || job == nil { continue } - // Output prüfen: wenn nichts entsteht -> abbrechen + aus jobs entfernen + // Wenn keine echte Output-Datei entsteht -> Job wieder abbrechen/aufräumen go mfcAbortIfNoOutput(job.ID, outputProbeMax) - // kleine Pause, damit du nicht 20 Models in einem Tick startest - time.Sleep(1200 * time.Millisecond) + time.Sleep(startGap) } } } @@ -94,8 +104,8 @@ func isJobRunningForURL(u string) bool { return false } - jobsMu.Lock() - defer jobsMu.Unlock() + jobsMu.RLock() + defer jobsMu.RUnlock() for _, j := range jobs { if j == nil { @@ -114,7 +124,7 @@ func mfcAbortIfNoOutput(jobID string, maxWait time.Duration) { deadline := time.Now().Add(maxWait) for time.Now().Before(deadline) { - jobsMu.Lock() + jobsMu.RLock() job := jobs[jobID] status := JobStatus("") out := "" @@ -122,9 +132,9 @@ func mfcAbortIfNoOutput(jobID string, maxWait time.Duration) { status = job.Status out = strings.TrimSpace(job.Output) } - jobsMu.Unlock() + jobsMu.RUnlock() - // Job schon weg oder nicht mehr running -> nix tun + // Job schon weg oder nicht mehr running -> nichts tun if job == nil || status != JobRunning { return } @@ -132,7 +142,7 @@ func mfcAbortIfNoOutput(jobID string, maxWait time.Duration) { // Output schon da? if out != "" { if fi, err := os.Stat(out); err == nil && !fi.IsDir() && fi.Size() > 0 { - // ✅ jetzt ist es ein "echter" Download -> im UI sichtbar machen + // echter Download -> im UI sichtbar machen publishJob(jobID) return } @@ -149,30 +159,34 @@ func mfcAbortIfNoOutput(jobID string, maxWait time.Duration) { return } - // Snapshot: was wir ohne Lock beenden können pc := job.previewCmd job.previewCmd = nil - cancel := job.cancel + + previewCancel := job.previewCancel + job.previewCancel = nil + + recordCancel := job.cancel out := strings.TrimSpace(job.Output) + jobsMu.Unlock() - // preview kill + if previewCancel != nil { + previewCancel() + } if pc != nil && pc.Process != nil { _ = pc.Process.Kill() } - // recording cancel - if cancel != nil { - cancel() + if recordCancel != nil { + recordCancel() } - // 0-Byte Datei ggf. wegwerfen (damit "leere Starts" nicht im recordDir liegen bleiben) + // 0-Byte Datei ggf. wegwerfen if out != "" { if fi, err := os.Stat(out); err == nil && !fi.IsDir() && fi.Size() == 0 { _ = os.Remove(out) } } - // Job aus der Liste entfernen (UI bleibt sauber) jobsMu.Lock() j := jobs[jobID] wasVisible := (j != nil && !j.Hidden) @@ -180,7 +194,6 @@ func mfcAbortIfNoOutput(jobID string, maxWait time.Duration) { jobsMu.Unlock() if wasVisible { - publishJobRemove(job) + publishJobRemove(j) } - } diff --git a/backend/nsfw_assets.go b/backend/nsfw_assets.go index e4978af..0f937d2 100644 --- a/backend/nsfw_assets.go +++ b/backend/nsfw_assets.go @@ -102,7 +102,7 @@ func extractEmbeddedDir(srcRoot, dstRoot string) error { } defer in.Close() - out, err := os.OpenFile(dstPath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o755) + out, err := os.OpenFile(dstPath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644) if err != nil { return err } diff --git a/backend/nsfw_detector.go b/backend/nsfw_detector.go index 7c29a52..25b8b56 100644 --- a/backend/nsfw_detector.go +++ b/backend/nsfw_detector.go @@ -12,6 +12,7 @@ import ( "math" "os" "path/filepath" + "runtime" "sort" "strings" "sync" @@ -84,7 +85,19 @@ func initNSFWDetector() error { return err } - dllPath := filepath.Join(root, "onnxruntime.dll") + var libName string + switch runtime.GOOS { + case "windows": + libName = "onnxruntime.dll" + case "linux": + libName = "libonnxruntime.so" + case "darwin": + libName = "libonnxruntime.dylib" + default: + return fmt.Errorf("onnxruntime für OS %s nicht unterstützt", runtime.GOOS) + } + + dllPath := filepath.Join(root, libName) modelPath := filepath.Join(root, "320n.onnx") if _, err := os.Stat(dllPath); err != nil { diff --git a/backend/postwork.go b/backend/postwork.go index 20c68b2..9af7002 100644 --- a/backend/postwork.go +++ b/backend/postwork.go @@ -163,49 +163,49 @@ func (pq *PostWorkQueue) workerLoop(id int) { continue } - // Task startet jetzt wirklich → waiting -> running - pq.mu.Lock() - pq.removeWaitingKeyLocked(task.Key) - pq.runningKeys[task.Key] = struct{}{} - pq.mu.Unlock() - func() { defer func() { - // Panic-Schutz: Worker darf nicht sterben if r := recover(); r != nil { - // optional: hier loggen _ = r } - // States immer konsistent aufräumen (auch wenn er nie sauber run() beendet) pq.mu.Lock() - pq.removeWaitingKeyLocked(task.Key) // falls er noch als waiting drin hing + pq.removeWaitingKeyLocked(task.Key) delete(pq.runningKeys, task.Key) delete(pq.inflight, task.Key) + delete(pq.cancelByKey, task.Key) if pq.queued > 0 { pq.queued-- } pq.mu.Unlock() }() - // 3) Optional: Task timeout (gegen hängende ffmpeg) ctx, cancel := context.WithTimeout(context.Background(), 60*time.Minute) pq.mu.Lock() pq.cancelByKey[task.Key] = cancel pq.mu.Unlock() - defer func() { - cancel() + defer cancel() - pq.mu.Lock() - delete(pq.cancelByKey, task.Key) - pq.mu.Unlock() + // ffmpeg/heavy-slot holen + select { + case pq.ffmpegSem <- struct{}{}: + case <-ctx.Done(): + return + } + defer func() { + <-pq.ffmpegSem }() - // 4) Run + // jetzt wirklich von waiting -> running + pq.mu.Lock() + pq.removeWaitingKeyLocked(task.Key) + pq.runningKeys[task.Key] = struct{}{} + pq.mu.Unlock() + if task.Run != nil { - _ = task.Run(ctx) // optional: loggen + _ = task.Run(ctx) } }() } @@ -224,7 +224,7 @@ func (pq *PostWorkQueue) StartWorkers(n int) { func (pq *PostWorkQueue) Stats() (queued int, inflight int, maxParallel int) { pq.mu.Lock() defer pq.mu.Unlock() - return pq.queued, len(pq.inflight), cap(pq.ffmpegSem) + return len(pq.waitingKeys), len(pq.runningKeys), cap(pq.ffmpegSem) } // Optional für UI/Debug (praktisch für "Warte… Position X/Y") @@ -288,7 +288,7 @@ func startEnrichStatusRefresher() { defer t.Stop() for range t.C { - jobsMu.Lock() + jobsMu.RLock() snapshot := make([]*RecordJob, 0, len(jobs)) for _, job := range jobs { if job == nil { @@ -296,7 +296,7 @@ func startEnrichStatusRefresher() { } snapshot = append(snapshot, job) } - jobsMu.Unlock() + jobsMu.RUnlock() for _, job := range snapshot { if job == nil { @@ -348,20 +348,22 @@ func startPostWorkStatusRefresher() { defer t.Stop() for range t.C { - changedJobs := make([]*RecordJob, 0, 16) + snapshots := make([]jobEventSnapshot, 0, 16) jobsMu.Lock() for _, job := range jobs { + if job == nil { + continue + } + key := strings.TrimSpace(job.PostWorkKey) if key == "" { continue } st := postWorkQ.StatusForKey(key) - changed := false - // PostWork-Daten aktualisieren if job.PostWork == nil || job.PostWork.State != st.State || job.PostWork.Position != st.Position || @@ -373,7 +375,6 @@ func startPostWorkStatusRefresher() { changed = true } - // Status / Phase für UI vereinheitlichen switch st.State { case "queued": if job.Status != JobPostwork { @@ -399,11 +400,9 @@ func startPostWorkStatusRefresher() { } phaseLower := strings.TrimSpace(strings.ToLower(job.Phase)) - - // Konkrete Unterphasen NICHT überschreiben switch phaseLower { - case "probe", "remuxing", "moving", "assets": - // ok, so lassen + case "probe", "remuxing", "moving", "assets", "analyze": + // konkrete Unterphasen beibehalten default: if phaseLower == "" || phaseLower == "recording" || phaseLower == "postwork" { if phaseLower != "postwork" { @@ -412,12 +411,51 @@ func startPostWorkStatusRefresher() { } } } + + case "missing": + continue } - if changed { - changedJobs = append(changedJobs, job) + + if !changed { + continue } + + var pw *PostWorkKeyStatus + if job.PostWork != nil { + tmp := *job.PostWork + pw = &tmp + } + + var endedAtCopy *time.Time + if job.EndedAt != nil { + tm := *job.EndedAt + endedAtCopy = &tm + } + + snapshots = append(snapshots, jobEventSnapshot{ + ID: job.ID, + SourceURL: job.SourceURL, + Output: job.Output, + Status: job.Status, + StartedAt: job.StartedAt, + StartedAtMs: job.StartedAtMs, + EndedAt: endedAtCopy, + EndedAtMs: job.EndedAtMs, + Error: job.Error, + Phase: job.Phase, + Progress: job.Progress, + SizeBytes: job.SizeBytes, + DurationSeconds: job.DurationSeconds, + PreviewState: job.PreviewState, + PostWorkKey: job.PostWorkKey, + PostWork: pw, + }) } jobsMu.Unlock() + + for _, snap := range snapshots { + publishJobUpsertSnapshot(snap) + } } }() } diff --git a/backend/preview.go b/backend/preview.go index 91ca494..2a3a015 100644 --- a/backend/preview.go +++ b/backend/preview.go @@ -44,7 +44,7 @@ import ( // - resolvePathRelativeToApp, getSettings, ensureAssetsForVideo, generatedThumbFile // - atomicWriteFile, ensureGeneratedDirs, ensureGeneratedDir // - durationSecondsCached, parseFFmpegOutTime, ffmpegInputTol -// - jobs, jobsMu, RecordJob, previewSem, thumbSem, JobRunning, notifyJobsChanged +// - jobs, jobsMu, RecordJob, previewSem, thumbSem, JobRunning // - sanitizeID, findFinishedFileByID, stripHotPrefix, assetIDForJob, generatedThumbJPGFile // Bitte diese Abhängigkeiten NICHT löschen – preview.go nutzt sie. @@ -1599,9 +1599,9 @@ func serveProviderFallbackOrStatus(w http.ResponseWriter, r *http.Request, job * } func servePreviewJPGAlias(w http.ResponseWriter, r *http.Request, id string) { - jobsMu.Lock() + jobsMu.RLock() job := jobs[id] - jobsMu.Unlock() + jobsMu.RUnlock() if job != nil { assetID := assetIDForJob(job) diff --git a/backend/record.go b/backend/record.go index 81b3931..1957f98 100644 --- a/backend/record.go +++ b/backend/record.go @@ -1069,26 +1069,22 @@ func applyFinishedPhaseTruthToRecordJobMeta(j *RecordJob) { // ---------------- Handlers ---------------- type recordListItem struct { - ID string `json:"id"` - SourceURL string `json:"sourceUrl"` - Status string `json:"status"` - StartedAt time.Time `json:"startedAt"` - StartedAtMs int64 `json:"startedAtMs"` - EndedAt *time.Time `json:"endedAt,omitempty"` - EndedAtMs int64 `json:"endedAtMs,omitempty"` - Output string `json:"output"` - Error string `json:"error,omitempty"` - Phase string `json:"phase,omitempty"` - Progress int `json:"progress,omitempty"` - SizeBytes int64 `json:"sizeBytes,omitempty"` - DurationSeconds float64 `json:"durationSeconds,omitempty"` - PreviewState string `json:"previewState,omitempty"` - RoomStatus string `json:"roomStatus,omitempty"` - IsOnline bool `json:"isOnline,omitempty"` - ModelImageURL string `json:"modelImageUrl,omitempty"` - ModelChatRoomURL string `json:"modelChatRoomUrl,omitempty"` - PostWorkKey string `json:"postWorkKey,omitempty"` - PostWork *PostWorkKeyStatus `json:"postWork,omitempty"` + ID string `json:"id"` + SourceURL string `json:"sourceUrl"` + Status string `json:"status"` + StartedAt time.Time `json:"startedAt"` + StartedAtMs int64 `json:"startedAtMs"` + EndedAt *time.Time `json:"endedAt,omitempty"` + EndedAtMs int64 `json:"endedAtMs,omitempty"` + Output string `json:"output"` + Error string `json:"error,omitempty"` + Phase string `json:"phase,omitempty"` + Progress int `json:"progress,omitempty"` + SizeBytes int64 `json:"sizeBytes,omitempty"` + DurationSeconds float64 `json:"durationSeconds,omitempty"` + PreviewState string `json:"previewState,omitempty"` + PostWorkKey string `json:"postWorkKey,omitempty"` + PostWork *PostWorkKeyStatus `json:"postWork,omitempty"` } func snapshotRecordJobForList(j *RecordJob) recordListItem { @@ -1463,15 +1459,21 @@ func recordStatus(w http.ResponseWriter, r *http.Request) { jobsMu.RLock() job, ok := jobs[id] - jobsMu.RUnlock() - - if !ok { + if !ok || job == nil { + jobsMu.RUnlock() http.Error(w, "job nicht gefunden", http.StatusNotFound) return } - applyPreviewSpriteTruthToRecordJobMeta(job) - respondJSON(w, job) + c := *job + if job.PostWork != nil { + pw := *job.PostWork + c.PostWork = &pw + } + jobsMu.RUnlock() + + applyPreviewSpriteTruthToRecordJobMeta(&c) + respondJSON(w, &c) } func recordStop(w http.ResponseWriter, r *http.Request) { @@ -2847,7 +2849,6 @@ func recordToggleHot(w http.ResponseWriter, r *http.Request) { return } - // optional, aber sinnvoll cancelDeferredEnrichForFile(file) s := getSettings() @@ -2861,7 +2862,6 @@ func recordToggleHot(w http.ResponseWriter, r *http.Request) { return } - // Quelle kann in done/ oder keep/ liegen src, from, fi, err := resolveDoneFileByName(doneAbs, file) if err != nil { http.Error(w, "datei nicht gefunden", http.StatusNotFound) @@ -2899,6 +2899,8 @@ func recordToggleHot(w http.ResponseWriter, r *http.Request) { return } + // Wichtig: + // Generated-Assets / Meta bleiben auf der kanonischen ID OHNE "HOT ". canonicalID := stripHotPrefix(strings.TrimSuffix(file, filepath.Ext(file))) renameJobsOutputBasename(file, newFile) diff --git a/backend/record_paths.go b/backend/record_paths.go index 1940458..c18521c 100644 --- a/backend/record_paths.go +++ b/backend/record_paths.go @@ -330,9 +330,9 @@ func resolvePlayablePathFromQuery(r *http.Request) (string, bool, int, string) { return "", false, http.StatusBadRequest, "id fehlt" } - jobsMu.Lock() + jobsMu.RLock() job, ok := jobs[id] - jobsMu.Unlock() + jobsMu.RUnlock() if !ok { return "", false, http.StatusNotFound, "job nicht gefunden" } diff --git a/backend/record_stream_cb.go b/backend/record_stream_cb.go index 475454a..d7448ab 100644 --- a/backend/record_stream_cb.go +++ b/backend/record_stream_cb.go @@ -566,21 +566,35 @@ func addCookiesFromString(req *http.Request, cookieStr string) { // --- helper --- func extractUsername(input string) string { - input = strings.TrimSpace(input) - input = strings.TrimPrefix(input, "https://") - input = strings.TrimPrefix(input, "http://") - input = strings.TrimPrefix(input, "www.") - if strings.HasPrefix(input, "chaturbate.com/") { - input = strings.TrimPrefix(input, "chaturbate.com/") + s := strings.TrimSpace(input) + if s == "" { + return "" } - // alles nach dem ersten Slash abschneiden (Pfadteile, /, etc.) - if idx := strings.IndexAny(input, "/?#"); idx != -1 { - input = input[:idx] + // Falls volle URL + if u, err := url.Parse(s); err == nil && u != nil { + host := strings.ToLower(strings.TrimSpace(u.Hostname())) + if strings.Contains(host, "chaturbate.com") { + p := strings.Trim(u.Path, "/") + if p == "" { + return "" + } + parts := strings.Split(p, "/") + return strings.TrimSpace(parts[0]) + } } - // zur Sicherheit evtl. übrig gebliebene Slash/Backslash trimmen - return strings.Trim(input, "/\\") + // Fallback: rohe Eingabe + s = strings.TrimPrefix(s, "https://") + s = strings.TrimPrefix(s, "http://") + s = strings.TrimPrefix(s, "www.") + if strings.HasPrefix(strings.ToLower(s), "chaturbate.com/") { + s = s[len("chaturbate.com/"):] + } + if idx := strings.IndexAny(s, "/?#"); idx != -1 { + s = s[:idx] + } + return strings.Trim(s, "/\\") } func hasChaturbateCookies(cookieStr string) bool { diff --git a/backend/recorder.go b/backend/recorder.go index d4006a7..f3f861a 100644 --- a/backend/recorder.go +++ b/backend/recorder.go @@ -278,8 +278,8 @@ func isActiveRecordingJob(job *RecordJob) bool { } func activeRecordingJobsCount() int { - jobsMu.Lock() - defer jobsMu.Unlock() + jobsMu.RLock() + defer jobsMu.RUnlock() n := 0 for _, j := range jobs { diff --git a/backend/sse.go b/backend/sse.go index e89b90a..77b82cb 100644 --- a/backend/sse.go +++ b/backend/sse.go @@ -232,8 +232,8 @@ func visibleJobEventsJSON() []ssePublishItem { nowTs := time.Now().UnixMilli() out := make([]ssePublishItem, 0, 64) - jobsMu.Lock() - defer jobsMu.Unlock() + jobsMu.RLock() + defer jobsMu.RUnlock() for _, j := range jobs { if j == nil || j.Hidden { @@ -476,7 +476,7 @@ func notifyAssetsChanged() { // -------------------- snapshots -------------------- func jobsSnapshotJSON() []byte { - jobsMu.Lock() + jobsMu.RLock() list := make([]*RecordJob, 0, len(jobs)) for _, j := range jobs { if j == nil || j.Hidden { @@ -486,7 +486,7 @@ func jobsSnapshotJSON() []byte { c.cancel = nil list = append(list, &c) } - jobsMu.Unlock() + jobsMu.RUnlock() sort.Slice(list, func(i, j int) bool { return list[i].StartedAt.After(list[j].StartedAt) diff --git a/backend/tasks_cleanup.go b/backend/tasks_cleanup.go index fa09aca..0cf19d6 100644 --- a/backend/tasks_cleanup.go +++ b/backend/tasks_cleanup.go @@ -312,7 +312,7 @@ func runGeneratedGarbageCollector() generatedGCStats { // 1b) Auch aktuell laufende / nachbearbeitete Jobs als "live" behandeln. // Sonst räumt der GC generated/meta/ weg, obwohl RecordStream dort noch schreibt. - jobsMu.Lock() + jobsMu.RLock() for _, j := range jobs { if j == nil { continue @@ -325,7 +325,7 @@ func runGeneratedGarbageCollector() generatedGCStats { live[id] = struct{}{} } - jobsMu.Unlock() + jobsMu.RUnlock() // 2) /generated/meta/ prüfen metaRoot, err := generatedMetaRoot() diff --git a/frontend/src/components/ui/Downloads.tsx b/frontend/src/components/ui/Downloads.tsx index 4dada61..d1a985c 100644 --- a/frontend/src/components/ui/Downloads.tsx +++ b/frontend/src/components/ui/Downloads.tsx @@ -3,7 +3,6 @@ 'use client' import { useMemo, useState, useCallback, useEffect, useRef } from 'react' -import type { ReactNode } from 'react' import Table, { type Column } from './Table' import Card from './Card' import Button from './Button' diff --git a/frontend/src/components/ui/DownloadsCardRow.tsx b/frontend/src/components/ui/DownloadsCardRow.tsx index f4d49a5..6de42b2 100644 --- a/frontend/src/components/ui/DownloadsCardRow.tsx +++ b/frontend/src/components/ui/DownloadsCardRow.tsx @@ -2,7 +2,6 @@ 'use client' -import { useEffect, useRef, useState, type ReactNode } from 'react' import Button from './Button' import ModelPreview from './ModelPreview' import ProgressBar from './ProgressBar' diff --git a/frontend/src/components/ui/FinishedDownloads.tsx b/frontend/src/components/ui/FinishedDownloads.tsx index 2a959f6..424b0c7 100644 --- a/frontend/src/components/ui/FinishedDownloads.tsx +++ b/frontend/src/components/ui/FinishedDownloads.tsx @@ -1664,7 +1664,6 @@ export default function FinishedDownloads({ const emptyFolder = !effectiveAllMode && totalItemsForPagination === 0 const emptyByFilter = globalFilterActive && visibleRows.length === 0 - const isEmptyState = emptyFolder || emptyByFilter const selectedCount = selectedFiles.length const allPageSelected =