// backend/recorder.go package main import ( "context" "errors" "fmt" "math" "net/http" "net/url" "os" "path/filepath" "strconv" "strings" "time" "github.com/google/uuid" ) // ---------------- Progress mapping ---------------- func setJobProgress(job *RecordJob, phase string, pct int) { phase = strings.TrimSpace(phase) phaseLower := strings.ToLower(phase) if pct < 0 { pct = 0 } if pct > 100 { pct = 100 } type rng struct{ start, end int } rangeFor := func(ph string) rng { switch ph { case "postwork": return rng{0, 8} case "remuxing": return rng{8, 38} case "moving": return rng{38, 54} case "probe": return rng{54, 70} case "assets": return rng{70, 88} case "analyze": return rng{88, 99} default: return rng{0, 100} } } jobsMu.Lock() defer jobsMu.Unlock() oldPhase := job.Phase oldProgress := job.Progress inPostwork := job.EndedAt != nil || (strings.TrimSpace(job.Phase) != "" && strings.ToLower(strings.TrimSpace(job.Phase)) != "recording") if inPostwork { if phaseLower == "" || phaseLower == "recording" { return } } newPhase := job.Phase if phase != "" { newPhase = phase } newProgress := job.Progress if !inPostwork { if pct < newProgress { pct = newProgress } newProgress = pct } else { r := rangeFor(phaseLower) width := float64(r.end - r.start) mapped := r.start if width > 0 { mapped = r.start + int(math.Round((float64(pct)/100.0)*width)) } if mapped < r.start { mapped = r.start } if mapped > r.end { mapped = r.end } if mapped < newProgress { mapped = newProgress } newProgress = mapped } if oldPhase == newPhase && oldProgress == newProgress { return } job.Phase = newPhase job.Progress = newProgress } // ---------------- Preview scrubber ---------------- const defaultScrubberCount = 18 // /api/preview-scrubber/{index}?id=... (oder ?file=...) func recordPreviewScrubberFrame(w http.ResponseWriter, r *http.Request) { const prefix = "/api/preview-scrubber/" if !strings.HasPrefix(r.URL.Path, prefix) { http.NotFound(w, r) return } idxPart := strings.Trim(strings.TrimPrefix(r.URL.Path, prefix), "/") if idxPart == "" { http.Error(w, "missing scrubber frame index", http.StatusBadRequest) return } idx, err := strconv.Atoi(idxPart) if err != nil || idx < 0 { http.Error(w, "invalid scrubber frame index", http.StatusBadRequest) return } q := r.URL.Query() id := strings.TrimSpace(q.Get("id")) file := strings.TrimSpace(q.Get("file")) if id == "" && file == "" { http.Error(w, "missing id or file", http.StatusBadRequest) return } durSec, err := lookupDurationForScrubber(r) if err != nil || durSec <= 0 { durSec = 60 } count := defaultScrubberCount if idx >= count { idx = count - 1 } if count < 1 { count = 1 } t := scrubberIndexToTime(idx, count, durSec) targetQ := url.Values{} if id != "" { targetQ.Set("id", id) } if file != "" { targetQ.Set("file", file) } targetQ.Set("t", fmt.Sprintf("%.3f", t)) w.Header().Set("Cache-Control", "private, max-age=300") http.Redirect(w, r, "/api/preview?"+targetQ.Encode(), http.StatusFound) } // Gleichmäßig über die Videolänge sampeln (Mitte des Segments) func scrubberIndexToTime(index, count int, durationSec float64) float64 { if count <= 1 { return 0.1 } if durationSec <= 0 { return 0.1 } maxT := math.Max(0.1, durationSec-0.1) ratio := (float64(index) + 0.5) / float64(count) t := ratio * maxT if t < 0.1 { t = 0.1 } if t > maxT { t = maxT } return t } func lookupDurationForScrubber(r *http.Request) (float64, error) { path, ok, _, _ := resolvePlayablePathFromQuery(r) if !ok || strings.TrimSpace(path) == "" { return 0, fmt.Errorf("unable to resolve file") } // best-effort meta ensureMetaJSONForPlayback(r.Context(), path) ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second) defer cancel() sec, err := durationSecondsCached(ctx, path) if err != nil { return 0, err } return sec, nil } // ---------------- Preview sprite file handler ---------------- func recordPreviewSprite(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet && r.Method != http.MethodHead { http.Error(w, "Nur GET/HEAD", http.StatusMethodNotAllowed) return } id := strings.TrimPrefix(r.URL.Path, "/api/record/preview-sprite/") if id == r.URL.Path { id = strings.TrimPrefix(r.URL.Path, "/api/preview-sprite/") } id = strings.TrimSpace(id) id = strings.Trim(id, "/") if id == "" { http.Error(w, "id fehlt", http.StatusBadRequest) return } var err error id, err = sanitizeID(id) if err != nil { http.Error(w, "ungültige id", http.StatusBadRequest) return } dir, err := generatedDirForID(id) if err != nil { http.Error(w, "ungültige id", http.StatusBadRequest) return } spritePath := filepath.Join(dir, "preview-sprite.jpg") fi, err := os.Stat(spritePath) if err != nil || fi.IsDir() || fi.Size() <= 0 { http.NotFound(w, r) return } f, err := os.Open(spritePath) if err != nil { http.NotFound(w, r) return } defer f.Close() w.Header().Set("Content-Type", "image/jpeg") w.Header().Set("Cache-Control", "private, max-age=31536000, immutable") w.Header().Set("X-Content-Type-Options", "nosniff") http.ServeContent(w, r, "preview-sprite.jpg", fi.ModTime(), f) } // ---------------- Start + run job ---------------- func markModelPublicWhileRecording(job *RecordJob) { if job == nil { return } rawURL := strings.TrimSpace(job.SourceURL) if rawURL == "" { return } if detectProvider(rawURL) != "chaturbate" { return } modelKey := strings.TrimSpace(extractUsername(rawURL)) if modelKey == "" { return } store := getModelStore() if store == nil { return } now := time.Now().UTC() _ = store.SetChaturbateRoomState( "chaturbate.com", modelKey, "public", true, rawURL, "", now, ) } func safeFileSize(path string) int64 { path = strings.TrimSpace(path) if path == "" { return 0 } fi, err := os.Stat(path) if err != nil || fi == nil || fi.IsDir() { return 0 } return fi.Size() } func statOutputForLog(path string) string { path = strings.TrimSpace(path) if path == "" { return "path-empty" } fi, err := os.Stat(path) if err != nil { return "stat-error=" + err.Error() } if fi == nil { return "stat-nil" } if fi.IsDir() { return "is-dir" } return fmt.Sprintf("exists size=%d mod=%s", fi.Size(), fi.ModTime().Format(time.RFC3339Nano)) } func waitForUsableOutput(path string, timeout time.Duration) (os.FileInfo, error) { path = strings.TrimSpace(path) if path == "" { return nil, fmt.Errorf("path empty") } deadline := time.Now().Add(timeout) var lastErr error for { fi, err := os.Stat(path) if err == nil && fi != nil && !fi.IsDir() && fi.Size() > 0 { return fi, nil } if err != nil { lastErr = err } else if fi == nil { lastErr = fmt.Errorf("stat returned nil") } else if fi.IsDir() { lastErr = fmt.Errorf("path is dir") } else { lastErr = fmt.Errorf("file size is 0") } if time.Now().After(deadline) { break } time.Sleep(150 * time.Millisecond) } if lastErr == nil { lastErr = fmt.Errorf("usable output not ready in time") } return nil, lastErr } func shouldKeepStoppedJobForInspection(target JobStatus, out string) bool { if target != JobStopped { return false } fi, err := os.Stat(strings.TrimSpace(out)) if err != nil || fi == nil || fi.IsDir() { return false } // auch kleine Fragmente sichtbar behalten, damit du sie debuggen kannst return fi.Size() > 0 } func isActiveRecordingJob(job *RecordJob) bool { if job == nil { return false } if job.EndedAt != nil { return false } if job.Status != JobRunning { return false } phase := strings.ToLower(strings.TrimSpace(job.Phase)) return phase == "" || phase == "recording" } func activeRecordingJobsCount() int { jobsMu.RLock() defer jobsMu.RUnlock() n := 0 for _, j := range jobs { if isActiveRecordingJob(j) { n++ } } return n } func concurrentDownloadsLimitState() (enabled bool, max int, active int, reached bool) { s := getSettings() enabled = s.EnableConcurrentDownloadsLimit max = s.MaxConcurrentDownloads if max < 1 { max = 1 } active = activeRecordingJobsCount() reached = enabled && active >= max return } func startRecordingInternal(req RecordRequest) (*RecordJob, error) { url := strings.TrimSpace(req.URL) if url == "" { return nil, errors.New("url fehlt") } provider := detectProvider(url) if provider != "chaturbate" && provider != "mfc" { return nil, fmt.Errorf("unsupported provider: %q", url) } jobsMu.Lock() defer jobsMu.Unlock() // Duplicate-Check for _, j := range jobs { if j != nil && j.Status == JobRunning && j.EndedAt == nil && strings.TrimSpace(j.SourceURL) == url { if j.Hidden && !req.Hidden { j.Hidden = false return j, nil } return j, nil } } // Limit-Check ATOMAR unter jobsMu s := getSettings() if s.EnableConcurrentDownloadsLimit { max := s.MaxConcurrentDownloads if max < 1 { max = 1 } active := 0 for _, j := range jobs { if isActiveRecordingJob(j) { active++ } } if active >= max { return nil, fmt.Errorf("Download-Limit erreicht (%d/%d aktiv)", active, max) } } startedAt := time.Now() username := "" switch provider { case "chaturbate": username = extractUsername(url) case "mfc": username = extractMFCUsername(url) } if strings.TrimSpace(username) == "" { username = "unknown" } filename := fmt.Sprintf("%s_%s.mp4", username, startedAt.Format("01_02_2006__15-04-05")) recordDirAbs, _ := resolvePathRelativeToApp(s.RecordDir) recordDir := strings.TrimSpace(recordDirAbs) if recordDir == "" { recordDir = strings.TrimSpace(s.RecordDir) } outPath := filepath.Join(recordDir, filename) jobID := uuid.NewString() ctx, cancel := context.WithCancel(context.Background()) job := &RecordJob{ ID: jobID, SourceURL: url, Status: JobRunning, StartedAt: startedAt, StartedAtMs: startedAt.UnixMilli(), Output: outPath, Hidden: req.Hidden, cancel: cancel, } jobs[jobID] = job go runJob(ctx, job, req) return job, nil } func runJob(ctx context.Context, job *RecordJob, req RecordRequest) { hc := NewHTTPClient(req.UserAgent) provider := detectProvider(req.URL) var err error now := job.StartedAt if now.IsZero() { now = time.Now() } if job.StartedAtMs == 0 { base := job.StartedAt if base.IsZero() { base = time.Now() jobsMu.Lock() job.StartedAt = base jobsMu.Unlock() } jobsMu.Lock() job.StartedAtMs = base.UnixMilli() jobsMu.Unlock() } setJobProgress(job, "recording", 0) switch provider { case "chaturbate": if !hasChaturbateCookies(req.Cookie) { err = errors.New("cf_clearance und session_id (oder sessionid) Cookies sind für Chaturbate erforderlich") break } s := getSettings() recordDirAbs, rerr := resolvePathRelativeToApp(s.RecordDir) if rerr != nil || strings.TrimSpace(recordDirAbs) == "" { err = fmt.Errorf("recordDir auflösung fehlgeschlagen: %v", rerr) break } _ = os.MkdirAll(recordDirAbs, 0o755) username := extractUsername(req.URL) filename := fmt.Sprintf("%s_%s.mp4", username, now.Format("01_02_2006__15-04-05")) jobsMu.Lock() existingOut := strings.TrimSpace(job.Output) jobsMu.Unlock() outPath := existingOut if outPath == "" || !filepath.IsAbs(outPath) { outPath = filepath.Join(recordDirAbs, filename) } if strings.TrimSpace(existingOut) != strings.TrimSpace(outPath) { jobsMu.Lock() job.Output = outPath jobsMu.Unlock() } markModelPublicWhileRecording(job) err = RecordStream(ctx, hc, "https://chaturbate.com/", username, outPath, req.Cookie, job) case "mfc": s := getSettings() recordDirAbs, rerr := resolvePathRelativeToApp(s.RecordDir) if rerr != nil || strings.TrimSpace(recordDirAbs) == "" { err = fmt.Errorf("recordDir auflösung fehlgeschlagen: %v", rerr) break } _ = os.MkdirAll(recordDirAbs, 0o755) username := extractMFCUsername(req.URL) filename := fmt.Sprintf("%s_%s.mp4", username, now.Format("01_02_2006__15-04-05")) outPath := filepath.Join(recordDirAbs, filename) jobsMu.Lock() job.Output = outPath jobsMu.Unlock() err = RecordStreamMFC(ctx, hc, username, outPath, job) default: err = errors.New("unsupported provider") } end := time.Now() target := JobFinished var errText string if err != nil { if errors.Is(err, context.Canceled) { target = JobStopped } else { target = JobFailed errText = err.Error() } } stopPreview(job) jobsMu.Lock() job.EndedAt = &end job.EndedAtMs = end.UnixMilli() job.Status = target if errText != "" { job.Error = errText } job.Phase = "postwork" out := strings.TrimSpace(job.Output) jobsMu.Unlock() // ------------------------------------------------------------ // HARTE SCHRANKE 1: gar kein Output-Pfad // ------------------------------------------------------------ if out == "" { jobsMu.Lock() job.Phase = "" job.Progress = 100 job.PostWorkKey = "" job.PostWork = nil jobsMu.Unlock() jobsMu.Lock() delete(jobs, job.ID) jobsMu.Unlock() publishJobRemove(job) return } // ------------------------------------------------------------ // HARTE SCHRANKE 2: Output ungültig / leer vor Postwork // WICHTIG: bei JobStopped kurz warten, damit Windows/File-Flush // nach context canceled noch fertig werden kann // ------------------------------------------------------------ { var ( fi os.FileInfo serr error ) if target == JobStopped { fi, serr = waitForUsableOutput(out, 2*time.Second) } else { fi, serr = os.Stat(out) } if serr != nil || fi == nil || fi.IsDir() || fi.Size() <= 0 { _ = removeWithRetry(out) purgeDurationCacheForPath(out) jobsMu.Lock() delete(jobs, job.ID) jobsMu.Unlock() publishJobRemove(job) return } } // ------------------------------------------------------------ // AUTO DELETE SMALL DOWNLOADS // ------------------------------------------------------------ { fi, serr := os.Stat(out) if serr == nil && fi != nil && !fi.IsDir() { jobsMu.Lock() job.SizeBytes = fi.Size() jobsMu.Unlock() s := getSettings() minMB := s.AutoDeleteSmallDownloadsBelowMB if s.AutoDeleteSmallDownloads && minMB > 0 { threshold := int64(minMB) * 1024 * 1024 if fi.Size() > 0 && fi.Size() < threshold { base := filepath.Base(out) id := stripHotPrefix(strings.TrimSuffix(base, filepath.Ext(base))) derr := removeWithRetry(out) if derr == nil || os.IsNotExist(derr) { removeGeneratedForID(id) purgeDurationCacheForPath(out) jobsMu.Lock() delete(jobs, job.ID) jobsMu.Unlock() publishJobRemove(job) return } fmt.Println("⚠️ auto-delete before enqueue failed:", derr) } } } } // ------------------------------------------------------------ // POSTWORK QUEUE // ------------------------------------------------------------ postOut := out postTarget := target postKey := "postwork:" + job.ID postFile := filepath.Base(out) postAssetID := assetIDFromVideoPath(out) jobsMu.Lock() job.Phase = "postwork" job.PostWorkKey = postKey { s := postWorkQ.StatusForKey(postKey) job.PostWork = &s } jobsMu.Unlock() okQueued := postWorkQ.Enqueue(PostWorkTask{ Key: postKey, Added: time.Now(), Run: func(ctx context.Context) error { { st := postWorkQ.StatusForKey(postKey) jobsMu.Lock() job.PostWork = &st jobsMu.Unlock() setJobProgress(job, "postwork", 0) publishFinishedPostworkStatusForJob(job, finishedPostworkEvent{ File: postFile, AssetID: postAssetID, Queue: "postwork", State: "running", Phase: "postwork", Label: "Nachbearbeitung läuft…", Position: st.Position, Waiting: st.Waiting, Running: st.Running, MaxParallel: st.MaxParallel, }) } out := strings.TrimSpace(postOut) if out == "" { jobsMu.Lock() job.Status = postTarget job.Phase = "" job.Progress = 100 job.PostWorkKey = "" job.PostWork = nil jobsMu.Unlock() jobsMu.Lock() delete(jobs, job.ID) jobsMu.Unlock() publishJobRemove(job) return nil } setPhase := func(phase string, pct int) { setJobProgress(job, phase, pct) st := postWorkQ.StatusForKey(postKey) jobsMu.Lock() job.PostWork = &st jobsMu.Unlock() label := "Nachbearbeitung läuft…" switch strings.ToLower(strings.TrimSpace(phase)) { case "remuxing": label = "Remux läuft…" case "moving": label = "Verschiebe Datei…" case "probe": label = "Analysiere Datei…" case "assets": label = "Erstelle Preview…" case "analyze": label = "Analysiere Inhalt…" } publishFinishedPostworkStatusForJob(job, finishedPostworkEvent{ File: filepath.Base(out), AssetID: assetIDFromVideoPath(out), Queue: "postwork", State: "running", Phase: phase, Label: label, Position: st.Position, Waiting: st.Waiting, Running: st.Running, MaxParallel: st.MaxParallel, }) } setPhase("moving", 10) moved, err2 := moveToDoneDir(out) if err2 != nil || strings.TrimSpace(moved) == "" { jobsMu.Lock() job.Status = JobFailed if err2 != nil { job.Error = "moveToDoneDir failed: " + err2.Error() } else { job.Error = "moveToDoneDir returned empty path" } job.Phase = "" job.Progress = 100 job.PostWorkKey = "" job.PostWork = nil jobsMu.Unlock() jobsMu.Lock() delete(jobs, job.ID) jobsMu.Unlock() publishJobRemove(job) publishFinishedPostworkStatusForJob(job, finishedPostworkEvent{ File: filepath.Base(out), AssetID: assetIDFromVideoPath(out), Queue: "postwork", State: "error", Phase: "moving", Label: "Verschieben nach done fehlgeschlagen", }) return nil } out = strings.TrimSpace(moved) jobsMu.Lock() job.Output = out jobsMu.Unlock() notifyDoneChanged() // 3) Datei nach move trotzdem kaputt { fi, serr := os.Stat(out) if serr != nil || fi == nil || fi.IsDir() || fi.Size() <= 0 { _ = removeWithRetry(out) purgeDurationCacheForPath(out) jobsMu.Lock() delete(jobs, job.ID) jobsMu.Unlock() publishJobRemove(job) publishFinishedPostworkStatusForJob(job, finishedPostworkEvent{ File: filepath.Base(out), AssetID: assetIDFromVideoPath(out), Queue: "postwork", State: "error", Phase: "moving", Label: "Nachbearbeitung fehlgeschlagen", }) return nil } } // 4) DURATION setPhase("probe", 35) { dctx, cancel := context.WithTimeout(ctx, 6*time.Second) if sec, derr := durationSecondsCached(dctx, out); derr == nil && sec > 0 { jobsMu.Lock() job.DurationSeconds = sec jobsMu.Unlock() } cancel() } // 5) VIDEO PROPS setPhase("probe", 75) { pctx, cancel := context.WithTimeout(ctx, 6*time.Second) w, h, fps, perr := probeVideoProps(pctx, out) cancel() if perr == nil { jobsMu.Lock() job.VideoWidth = w job.VideoHeight = h job.FPS = fps jobsMu.Unlock() } } // 6) PRIMARY ASSETS setPhase("assets", 0) lastPct := -1 lastTick := time.Time{} update := func(r float64) { if r < 0 { r = 0 } if r > 1 { r = 1 } pct := int(math.Round(r * 100)) if pct < 0 { pct = 0 } if pct > 100 { pct = 100 } if pct == lastPct { return } if !lastTick.IsZero() && time.Since(lastTick) < 150*time.Millisecond { return } lastPct = pct lastTick = time.Now() setPhase("assets", pct) } if _, err := ensurePrimaryAssetsForVideoWithProgressCtx(ctx, out, job.SourceURL, update); err != nil { fmt.Println("⚠️ ensurePrimaryAssetsForVideo:", err) } setPhase("assets", 100) // FINALIZE PRIMARY POSTWORK finalFile := filepath.Base(out) finalAssetID := assetIDFromVideoPath(out) jobsMu.Lock() job.Status = postTarget job.Phase = "" job.Progress = 100 job.PostWorkKey = "" job.PostWork = nil jobsMu.Unlock() jobsMu.Lock() delete(jobs, job.ID) jobsMu.Unlock() publishJobRemove(job) publishFinishedPostworkStatusForJob(job, finishedPostworkEvent{ File: finalFile, AssetID: finalAssetID, Queue: "postwork", State: "done", Label: "Primäre Nachbearbeitung fertig", }) if ok := enqueueDeferredAssetsAndAI(job, out, job.SourceURL); !ok { fmt.Println("⚠️ deferred enrichment enqueue failed:", out) publishFinishedPostworkStatusForJob(job, finishedPostworkEvent{ File: finalFile, AssetID: finalAssetID, Queue: "enrich", State: "error", Phase: "assets", Label: "Späte Nachbearbeitung konnte nicht gestartet werden", }) } return nil }, }) if okQueued { st := postWorkQ.StatusForKey(postKey) jobsMu.Lock() job.PostWork = &st jobsMu.Unlock() publishFinishedPostworkStatusForJob(job, finishedPostworkEvent{ File: postFile, AssetID: postAssetID, Queue: "postwork", State: "queued", Phase: "postwork", Label: "Warte auf Nachbearbeitung…", Position: st.Position, Waiting: st.Waiting, Running: st.Running, MaxParallel: st.MaxParallel, }) } else { jobsMu.Lock() job.Status = postTarget job.Phase = "" job.Progress = 100 job.PostWorkKey = "" job.PostWork = nil jobsMu.Unlock() jobsMu.Lock() delete(jobs, job.ID) jobsMu.Unlock() publishJobRemove(job) publishFinishedPostworkStatusForJob(job, finishedPostworkEvent{ File: postFile, AssetID: postAssetID, Queue: "postwork", State: "error", Phase: "postwork", Label: "Nachbearbeitung konnte nicht eingeplant werden", }) } }