// backend/recorder.go package main import ( "context" "encoding/json" "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, appErrorf("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, appErrorf("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 = appErrorf("stat returned nil") } else if fi.IsDir() { lastErr = appErrorf("path is dir") } else { lastErr = appErrorf("file size is 0") } if time.Now().After(deadline) { break } time.Sleep(150 * time.Millisecond) } if lastErr == nil { lastErr = appErrorf("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 } type downloadLimitReachedError struct { Active int Max int } func (e downloadLimitReachedError) Error() string { return fmt.Sprintf("Download-Limit erreicht (%d/%d aktiv)", e.Active, e.Max) } func isDownloadLimitReachedError(err error) bool { var e downloadLimitReachedError return errors.As(err, &e) } var lastDownloadLimitLogAt time.Time func logDownloadLimitReachedThrottled(err error) { if err == nil { return } if !isDownloadLimitReachedError(err) { appLogln("❌", err) return } now := time.Now() // Nur alle 60 Sekunden ins Log schreiben. if !lastDownloadLimitLogAt.IsZero() && now.Sub(lastDownloadLimitLogAt) < 60*time.Second { return } lastDownloadLimitLogAt = now appLogln("❌", err) } 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, appErrorf("unsupported provider: %q", url) } if err := ensureDiskGuardAllowsStart(); err != nil { return nil, err } 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 && !req.IgnoreConcurrentLimit { max := s.MaxConcurrentDownloads if max < 1 { max = 1 } active := 0 for _, j := range jobs { if isActiveRecordingJob(j) { active++ } } if active >= max { return nil, downloadLimitReachedError{ Active: active, Max: max, } } } startedAt := time.Now() username := "" switch provider { case "chaturbate": username = extractUsername(url) case "mfc": username = extractMFCUsername(url) } if strings.TrimSpace(username) == "" { username = "unknown" } // ✅ Erst in TS aufnehmen filename := fmt.Sprintf("%s_%s.ts", 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 requireNonEmptyRegularFile(path string, label string) error { path = strings.TrimSpace(path) if path == "" { return appErrorf("%s: path empty", label) } fi, err := os.Stat(path) if err != nil { return appErrorf("%s: %w", label, err) } if fi == nil || fi.IsDir() { return appErrorf("%s: not a file", label) } if fi.Size() <= 0 { return appErrorf("%s: empty file", label) } return nil } func buildPostworkTempMP4Path(tsPath string) (string, error) { s := getSettings() doneAbs, err := resolvePathRelativeToApp(s.DoneDir) if err != nil { return "", appErrorf("doneDir auflösung fehlgeschlagen: %w", err) } doneAbs = strings.TrimSpace(doneAbs) if doneAbs == "" { return "", appErrorf("doneDir ist leer") } tmpDir := filepath.Join(doneAbs, ".postwork_tmp") if err := os.MkdirAll(tmpDir, 0o755); err != nil { return "", appErrorf("postwork tmp dir erstellen fehlgeschlagen: %w", err) } base := strings.TrimSuffix(filepath.Base(strings.TrimSpace(tsPath)), filepath.Ext(tsPath)) + ".mp4" return uniqueDestPath(tmpDir, base) } func moveRepairSourceTSBestEffort(tsPath string, finalMP4Path string) string { tsPath = strings.TrimSpace(tsPath) if tsPath == "" { return "" } if err := requireNonEmptyRegularFile(tsPath, "repair source ts"); err != nil { appLogln("⚠️ repair source ts not usable:", err) return tsPath } s := getSettings() doneAbs, err := resolvePathRelativeToApp(s.DoneDir) if err != nil || strings.TrimSpace(doneAbs) == "" { doneAbs = filepath.Dir(strings.TrimSpace(finalMP4Path)) } doneAbs = strings.TrimSpace(doneAbs) if doneAbs == "" { return tsPath } repairDir := filepath.Join(doneAbs, ".repair_sources") if err := os.MkdirAll(repairDir, 0o755); err != nil { appLogln("⚠️ repair source dir create failed:", err) return tsPath } dst, err := uniqueDestPath(repairDir, filepath.Base(tsPath)) if err != nil { appLogln("⚠️ repair source unique path failed:", err) return tsPath } if err := renameWithRetryAggressive(tsPath, dst); err != nil { appLogln("⚠️ repair source move failed:", err) return tsPath } purgeDurationCacheForPath(tsPath) return dst } func markVideoNeedsRepairBestEffort(videoPath string, sourceTSPath string, reason string, detail string) { videoPath = strings.TrimSpace(videoPath) if videoPath == "" { return } assetID := strings.TrimSpace(assetIDFromVideoPath(videoPath)) if assetID == "" { assetID = canonicalAssetIDFromName(filepath.Base(videoPath)) } if assetID == "" { return } marker := map[string]any{ "needsRepair": true, "reason": strings.TrimSpace(reason), "detail": strings.TrimSpace(detail), "videoPath": videoPath, "sourceTSPath": strings.TrimSpace(sourceTSPath), "createdAt": time.Now().UTC().Format(time.RFC3339Nano), } b, err := json.MarshalIndent(marker, "", " ") if err != nil { appLogln("⚠️ repair marker marshal failed:", err) return } repairPath := "" if _, _, _, _, metaPath, err := assetPathsForID(assetID); err == nil { metaPath = strings.TrimSpace(metaPath) if metaPath != "" { dir := filepath.Dir(metaPath) if err := os.MkdirAll(dir, 0o755); err == nil { repairPath = filepath.Join(dir, "repair.json") } } } if repairPath == "" { stem := strings.TrimSuffix(filepath.Base(videoPath), filepath.Ext(videoPath)) repairPath = filepath.Join(filepath.Dir(videoPath), stem+".repair.json") } if err := os.WriteFile(repairPath, b, 0o644); err != nil { appLogln("⚠️ repair marker write failed:", err) return } appLogln("⚠️ video marked for deferred repair:", filepath.Base(videoPath), "marker=", repairPath) } func runPrimaryPostworkPipeline( ctx context.Context, job *RecordJob, out string, setPhase func(string, int), ) (string, error) { out = strings.TrimSpace(out) if err := requireNonEmptyRegularFile(out, "record output"); err != nil { return out, err } cleanupTSAfterMove := "" removeTempOutOnMoveFailure := false needsRepair := false repairReason := "" repairDetail := "" repairSourceTS := "" // 1) TS -> MP4 zuerst if strings.EqualFold(filepath.Ext(out), ".ts") { tsPath := out if setPhase != nil { setPhase("remuxing", 0) } fi, err := os.Stat(tsPath) if err != nil || fi == nil || fi.IsDir() || fi.Size() <= 0 { return out, appErrorf("ts invalid before remux") } inSize := fi.Size() mp4Path, err := buildPostworkTempMP4Path(tsPath) if err != nil { return tsPath, appErrorf("postwork temp mp4 path failed: %w", err) } _ = removeWithRetry(mp4Path) dur := 0.0 dctx, cancel := context.WithTimeout(ctx, 6*time.Second) if d, derr := durationSecondsCached(dctx, tsPath); derr == nil && d > 0 { dur = d } cancel() err = remuxTSToMP4WithProgress( ctx, tsPath, mp4Path, dur, inSize, func(r float64) { if setPhase == nil { return } if r < 0 { r = 0 } if r > 1 { r = 1 } setPhase("remuxing", int(math.Round(r*100))) }, ) if err != nil { _ = removeWithRetry(mp4Path) return tsPath, appErrorf("ts remux failed: %w", err) } if err := requireNonEmptyRegularFile(mp4Path, "remux result"); err != nil { _ = removeWithRetry(mp4Path) return tsPath, err } if err := validateVideoDecodesNearEnd(ctx, mp4Path); err != nil { // Wichtig: // Nicht mehr inline reencoden. Reencode ist CPU-teuer und blockiert die postWorkQ. // Das remuxte MP4 bleibt erhalten und wird als "needs repair" markiert. appLogln("⚠️ remux mp4 decode validation failed; keeping remux result and deferring repair:", err) needsRepair = true repairReason = "decode_near_end_failed" repairDetail = err.Error() repairSourceTS = tsPath } // TS erst NACH erfolgreichem Move löschen. // Bei Repair-Verdacht behalten wir die TS als Reparaturquelle. if !needsRepair { cleanupTSAfterMove = tsPath } else { cleanupTSAfterMove = "" repairSourceTS = tsPath } removeTempOutOnMoveFailure = true out = mp4Path } // 2) Danach nach done verschieben if setPhase != nil { setPhase("moving", 0) } fallbackOut := out if cleanupTSAfterMove != "" { fallbackOut = cleanupTSAfterMove } moved, err := moveToDoneDir(out) if err != nil || strings.TrimSpace(moved) == "" { if removeTempOutOnMoveFailure { _ = removeWithRetry(out) purgeDurationCacheForPath(out) } if err != nil { return fallbackOut, appErrorf("moveToDoneDir failed: %w", err) } return fallbackOut, appErrorf("moveToDoneDir returned empty path") } out = strings.TrimSpace(moved) if err := requireNonEmptyRegularFile(out, "moved output"); err != nil { if cleanupTSAfterMove != "" { _ = removeWithRetry(out) purgeDurationCacheForPath(out) return cleanupTSAfterMove, err } return out, err } if needsRepair { keptSource := strings.TrimSpace(repairSourceTS) if keptSource != "" { keptSource = moveRepairSourceTSBestEffort(keptSource, out) } markVideoNeedsRepairBestEffort(out, keptSource, repairReason, repairDetail) } if cleanupTSAfterMove != "" { if err := removeWithRetry(cleanupTSAfterMove); err != nil && !os.IsNotExist(err) { appLogln("⚠️ ts cleanup after successful move:", err) } purgeDurationCacheForPath(cleanupTSAfterMove) } jobsMu.Lock() job.Output = out if fi, err := os.Stat(out); err == nil && fi != nil && !fi.IsDir() { job.SizeBytes = fi.Size() } jobsMu.Unlock() notifyDoneChanged() // 3) Danach meta.json sicher erzeugen if setPhase != nil { setPhase("probe", 15) } if _, err := ensureMetaForVideoCtx(ctx, out, job.SourceURL); err != nil { return out, appErrorf("meta.json generation failed: %w", err) } attachMetaToJobBestEffort(ctx, job, out) assetID := assetIDFromVideoPath(out) if assetID == "" { return out, appErrorf("asset id missing after move") } _, thumbPath, previewPath, _, metaPath, err := assetPathsForID(assetID) if err != nil { return out, err } if err := requireNonEmptyRegularFile(metaPath, "meta.json"); err != nil { return out, err } // 4) Danach preview.jpg und 5) preview.mp4 if setPhase != nil { setPhase("assets", 0) } lastPct := -1 lastTick := time.Time{} update := func(r float64) { if setPhase == nil { return } if r < 0 { r = 0 } if r > 1 { r = 1 } pct := int(math.Round(r * 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 { return out, appErrorf("primary assets failed: %w", err) } if err := requireNonEmptyRegularFile(thumbPath, "preview.jpg"); err != nil { return out, err } if err := requireNonEmptyRegularFile(previewPath, "preview.mp4"); err != nil { return out, err } attachMetaToJobBestEffort(ctx, job, out) if setPhase != nil { setPhase("assets", 100) } return out, nil } func runJob(ctx context.Context, job *RecordJob, req RecordRequest) { now := ensureJobStartInitialized(job) hc := NewHTTPClient(req.UserAgent) err := runProviderRecording(ctx, job, req, hc, now) target, out := finalizeRecordingAttempt(job, err) if !prepareOutputForPostwork(job, target, out) { return } out = strings.TrimSpace(job.Output) if out == "" { removeJobAndPublish(job) return } enqueuePostworkOrFail(job, out, target) } func ensureJobStartInitialized(job *RecordJob) time.Time { 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) return now } func runProviderRecording( ctx context.Context, job *RecordJob, req RecordRequest, hc *HTTPClient, now time.Time, ) error { switch detectProvider(req.URL) { case "chaturbate": return runChaturbateRecording(ctx, job, req, hc, now) case "mfc": return runMFCRecording(ctx, job, req, hc, now) default: return errors.New("unsupported provider") } } func runChaturbateRecording( ctx context.Context, job *RecordJob, req RecordRequest, hc *HTTPClient, now time.Time, ) error { if !hasChaturbateCookies(req.Cookie) { return errors.New("cf_clearance und session_id (oder sessionid) Cookies sind für Chaturbate erforderlich") } s := getSettings() recordDirAbs, err := resolvePathRelativeToApp(s.RecordDir) if err != nil || strings.TrimSpace(recordDirAbs) == "" { return appErrorf("recordDir auflösung fehlgeschlagen: %v", err) } _ = os.MkdirAll(recordDirAbs, 0o755) username := extractUsername(req.URL) filename := fmt.Sprintf("%s_%s.ts", 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) return RecordStream(ctx, hc, "https://chaturbate.com/", username, outPath, req.Cookie, job) } func runMFCRecording( ctx context.Context, job *RecordJob, req RecordRequest, hc *HTTPClient, now time.Time, ) error { s := getSettings() recordDirAbs, err := resolvePathRelativeToApp(s.RecordDir) if err != nil || strings.TrimSpace(recordDirAbs) == "" { return appErrorf("recordDir auflösung fehlgeschlagen: %v", err) } _ = os.MkdirAll(recordDirAbs, 0o755) username := extractMFCUsername(req.URL) filename := fmt.Sprintf("%s_%s.ts", username, now.Format("01_02_2006__15-04-05")) outPath := filepath.Join(recordDirAbs, filename) jobsMu.Lock() job.Output = outPath jobsMu.Unlock() return RecordStreamMFC(ctx, hc, username, outPath, job) } func finalizeRecordingAttempt(job *RecordJob, runErr error) (JobStatus, string) { end := time.Now() target := JobFinished var errText string if runErr != nil { if errors.Is(runErr, context.Canceled) { target = JobStopped } else { target = JobFailed errText = runErr.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() return target, out } func prepareOutputForPostwork(job *RecordJob, target JobStatus, out string) bool { if strings.TrimSpace(out) == "" { clearJobPostworkState(job) removeJobAndPublish(job) return false } if !ensureUsableRecordedOutput(job, target, out) { return false } if maybeDeleteSmallRecordedOutput(job, out) { return false } return true } func ensureUsableRecordedOutput(job *RecordJob, target JobStatus, out string) bool { var ( fi os.FileInfo err error ) if target == JobStopped { fi, err = waitForUsableOutput(out, 3*time.Second) } else { fi, err = os.Stat(out) } if err == nil && fi != nil && !fi.IsDir() && fi.Size() > 0 { return true } if target == JobStopped && shouldKeepStoppedJobForInspection(target, out) { clearJobPostworkState(job) return false } _ = removeWithRetry(out) purgeDurationCacheForPath(out) removeJobAndPublish(job) return false } func favoriteProtectionModelKeyForDownload(out string, sourceURL string) (host string, modelKey string) { sourceURL = strings.TrimSpace(sourceURL) if sourceURL != "" { switch detectProvider(sourceURL) { case "chaturbate": host = "chaturbate.com" modelKey = strings.TrimSpace(extractUsername(sourceURL)) case "mfc": host = "myfreecams.com" modelKey = strings.TrimSpace(extractMFCUsername(sourceURL)) } } if strings.TrimSpace(modelKey) == "" { stem := strings.TrimSuffix(filepath.Base(strings.TrimSpace(out)), filepath.Ext(out)) stem = stripHotPrefix(stem) modelKey = sanitizeModelKey(strings.TrimSpace(modelNameFromFilename(stem))) } return strings.TrimSpace(host), strings.ToLower(strings.TrimSpace(modelKey)) } func isFavoriteProtectedDownload(out string, sourceURL string) bool { s := getSettings() if !s.AutoDeleteSmallDownloadsKeepFavorites { return false } host, modelKey := favoriteProtectionModelKeyForDownload(out, sourceURL) if modelKey == "" { return false } store := getModelStore() if store == nil { return false } return store.IsFavoriteModel(host, modelKey) } func maybeDeleteSmallRecordedOutput(job *RecordJob, out string) bool { fi, err := os.Stat(out) if err != nil || fi == nil || fi.IsDir() { return false } jobsMu.Lock() job.SizeBytes = fi.Size() jobsMu.Unlock() s := getSettings() minMB := s.AutoDeleteSmallDownloadsBelowMB if !s.AutoDeleteSmallDownloads || minMB <= 0 { return false } threshold := int64(minMB) * 1024 * 1024 if fi.Size() <= 0 || fi.Size() >= threshold { return false } if isFavoriteProtectedDownload(out, job.SourceURL) { appLogln("⭐ auto-delete skipped for favorite:", filepath.Base(out)) return false } base := filepath.Base(out) id := stripHotPrefix(strings.TrimSuffix(base, filepath.Ext(base))) if derr := removeWithRetry(out); derr == nil || os.IsNotExist(derr) { removeGeneratedForID(id) purgeDurationCacheForPath(out) removeJobAndPublish(job) return true } else { appLogln("⚠️ auto-delete before enqueue failed:", derr) } return false } func clearJobPostworkState(job *RecordJob) { jobsMu.Lock() job.Phase = "" job.Progress = 100 job.PostWorkKey = "" job.PostWork = nil jobsMu.Unlock() } func removeJobAndPublish(job *RecordJob) { jobsMu.Lock() delete(jobs, job.ID) jobsMu.Unlock() publishJobRemove(job) } func enqueuePostworkOrFail(job *RecordJob, out string, postTarget JobStatus) { postKey := "postwork:" + job.ID postFile := filepath.Base(out) postAssetID := assetIDFromVideoPath(out) _, _, maxParallel := postWorkQ.Stats() sortBucket, sortName := postWorkSortForVideoPath(out) addedAt := time.Now() jobsMu.Lock() job.Phase = "postwork" job.PostWorkKey = postKey job.PostWork = &PostWorkKeyStatus{ State: "queued", Position: 0, Waiting: 0, Running: 0, MaxParallel: maxParallel, } jobsMu.Unlock() okQueued := postWorkQ.Enqueue(PostWorkTask{ Key: postKey, Path: out, Added: addedAt, SortBucket: sortBucket, SortName: sortName, Run: func(ctx context.Context) error { return runQueuedPostwork(ctx, job, out, postTarget, postKey) }, }) if okQueued { publishQueuedPostworkState(job, postKey, postFile, postAssetID) return } handlePostworkEnqueueFailure(job, out, postTarget, postFile, postAssetID) } func publishQueuedPostworkState(job *RecordJob, postKey, postFile, postAssetID string) { 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, }) } func handlePostworkEnqueueFailure(job *RecordJob, out string, postTarget JobStatus, postFile, postAssetID string) { jobsMu.Lock() job.Status = postTarget job.Phase = "" job.Progress = 100 job.PostWorkKey = "" job.PostWork = nil job.Error = "Nachbearbeitung konnte nicht eingeplant werden" jobsMu.Unlock() if shouldKeepStoppedJobForInspection(postTarget, out) { _ = publishJob(job.ID) publishFinishedPostworkStatusForJob(job, finishedPostworkEvent{ File: postFile, AssetID: postAssetID, Queue: "postwork", State: "error", Phase: "postwork", Label: "Nachbearbeitung konnte nicht eingeplant werden", }) return } removeJobAndPublish(job) publishFinishedPostworkStatusForJob(job, finishedPostworkEvent{ File: postFile, AssetID: postAssetID, Queue: "postwork", State: "error", Phase: "postwork", Label: "Nachbearbeitung konnte nicht eingeplant werden", }) } func runQueuedPostwork( ctx context.Context, job *RecordJob, initialOut string, postTarget JobStatus, postKey string, ) error { out := strings.TrimSpace(initialOut) if out == "" { clearJobPostworkState(job) removeJobAndPublish(job) return nil } publishRunningPostworkStart(job, postKey, out) setPhase := buildPostworkPhaseUpdater(job, postKey, &out) var err error out, err = runPrimaryPostworkPipeline(ctx, job, out, setPhase) if err != nil { handlePrimaryPostworkFailure(job, out, postTarget, err) return nil } handlePrimaryPostworkSuccess(job, out, postTarget) return nil } func publishRunningPostworkStart(job *RecordJob, postKey, out string) { st := postWorkQ.StatusForKey(postKey) jobsMu.Lock() job.PostWork = &st jobsMu.Unlock() setJobProgress(job, "postwork", 0) publishFinishedPostworkStatusForJob(job, finishedPostworkEvent{ File: filepath.Base(out), AssetID: assetIDFromVideoPath(out), Queue: "postwork", State: "running", Phase: "postwork", Label: "Nachbearbeitung läuft…", Position: st.Position, Waiting: st.Waiting, Running: st.Running, MaxParallel: st.MaxParallel, }) } func buildPostworkPhaseUpdater(job *RecordJob, postKey string, out *string) func(string, int) { return 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…" } currentOut := "" if out != nil { currentOut = strings.TrimSpace(*out) } publishFinishedPostworkStatusForJob(job, finishedPostworkEvent{ File: filepath.Base(currentOut), AssetID: assetIDFromVideoPath(currentOut), Queue: "postwork", State: "running", Phase: phase, Label: label, Position: st.Position, Waiting: st.Waiting, Running: st.Running, MaxParallel: st.MaxParallel, }) } } func publishDeferredPhase(fileName, assetID, queue, phase, state, label string, err error) { errText := "" if err != nil { errText = strings.TrimSpace(err.Error()) } publishFinishedPostworkPhase( fileName, assetID, queue, phase, state, label, errText, ) } func runDeferredEnrichPipeline(ctx context.Context, outPath string, sourceURL string) error { outPath = strings.TrimSpace(outPath) if outPath == "" { return appErrorf("output path leer") } fileName := filepath.Base(outPath) assetID := assetIDFromVideoPath(outPath) if assetID == "" { return appErrorf("asset id fehlt") } requiredGoals := requiredAnalyzeGoals() // best effort sourceURL aus meta.json nachziehen, falls leer if strings.TrimSpace(sourceURL) == "" { if _, _, _, _, metaPath, err := assetPathsForID(assetID); err == nil { if fi, serr := os.Stat(outPath); serr == nil && fi != nil && !fi.IsDir() && fi.Size() > 0 { if u, ok := readVideoMetaSourceURL(metaPath, fi); ok { sourceURL = u } } } } abortIfCanceled := func(stepErr error, queue string, phase string) error { if errors.Is(stepErr, context.Canceled) || errors.Is(stepErr, context.DeadlineExceeded) || ctx.Err() != nil { publishDeferredPhase(fileName, assetID, queue, phase, "missing", "", nil) return context.Canceled } return nil } waitBeforeHeavyStep := func(queue string, phase string) error { if err := ctx.Err(); err != nil { publishDeferredPhase(fileName, assetID, queue, phase, "missing", "", nil) return context.Canceled } if err := waitUntilEnrichMayRun(ctx); err != nil { publishDeferredPhase(fileName, assetID, queue, phase, "missing", "", nil) return err } if err := ctx.Err(); err != nil { publishDeferredPhase(fileName, assetID, queue, phase, "missing", "", nil) return context.Canceled } return nil } truth := assetsTruthForVideo(outPath) // ------------------------------------------------- // Fehlende Vorstufen nur nachziehen, NICHT alles neu // ------------------------------------------------- if !truth.MetaReady { if err := waitBeforeHeavyStep("postwork", "meta"); err != nil { return err } publishDeferredPhase(fileName, assetID, "postwork", "meta", "running", "Meta", nil) okMeta, err := ensureMetaForVideoCtx(ctx, outPath, sourceURL) if cerr := abortIfCanceled(err, "postwork", "meta"); cerr != nil { return cerr } if err != nil || !okMeta || !assetsTruthForVideo(outPath).MetaReady { if err == nil { err = appErrorf("meta konnte nicht erzeugt werden") } publishDeferredPhase(fileName, assetID, "postwork", "meta", "error", "Meta fehlgeschlagen", err) return err } publishDeferredPhase(fileName, assetID, "postwork", "meta", "done", "Meta", nil) truth = assetsTruthForVideo(outPath) } if !truth.ThumbReady { if err := waitBeforeHeavyStep("postwork", "thumb"); err != nil { return err } publishDeferredPhase(fileName, assetID, "postwork", "thumb", "running", "Vorschaubild", nil) _, err := ensurePrimaryAssetsForVideoWithProgressCtx(ctx, outPath, sourceURL, nil) if cerr := abortIfCanceled(err, "postwork", "thumb"); cerr != nil { return cerr } if err != nil || !assetsTruthForVideo(outPath).ThumbReady { if err == nil { err = appErrorf("vorschaubild konnte nicht erzeugt werden") } publishDeferredPhase(fileName, assetID, "postwork", "thumb", "error", "Vorschaubild fehlgeschlagen", err) return err } publishDeferredPhase(fileName, assetID, "postwork", "thumb", "done", "Vorschaubild", nil) truth = assetsTruthForVideo(outPath) } if !truth.TeaserReady { if err := waitBeforeHeavyStep("postwork", "teaser"); err != nil { return err } publishDeferredPhase(fileName, assetID, "postwork", "teaser", "running", "Teaser", nil) _, err := ensureTeaserForVideoCtx(ctx, outPath, sourceURL) if cerr := abortIfCanceled(err, "postwork", "teaser"); cerr != nil { return cerr } if err != nil || !assetsTruthForVideo(outPath).TeaserReady { if err == nil { err = appErrorf("teaser konnte nicht erzeugt werden") } publishDeferredPhase(fileName, assetID, "postwork", "teaser", "error", "Teaser fehlgeschlagen", err) return err } publishDeferredPhase(fileName, assetID, "postwork", "teaser", "done", "Teaser", nil) truth = assetsTruthForVideo(outPath) } var firstErr error // ------------------------------------------------- // Langsame Hintergrundphase: Sprite // ------------------------------------------------- if !truth.SpritesReady { if err := waitBeforeHeavyStep("postwork", "sprites"); err != nil { return err } publishDeferredPhase(fileName, assetID, "postwork", "sprites", "running", "Sprites", nil) _, err := ensureSpritesForVideoCtx(ctx, outPath, sourceURL) if cerr := abortIfCanceled(err, "postwork", "sprites"); cerr != nil { return cerr } if err != nil || !assetsTruthForVideo(outPath).SpritesReady { if err == nil { err = appErrorf("sprites konnten nicht erzeugt werden") } publishDeferredPhase(fileName, assetID, "postwork", "sprites", "error", "Sprites fehlgeschlagen", err) if firstErr == nil { firstErr = err } } else { // Best effort. Wenn währenddessen postWork reinkommt, beendet ctx diesen Check. if _, checkErr := runTailBlackoutCheck(ctx, outPath); checkErr != nil { appLogln("⚠️ sprite tail blackout check failed:", outPath, checkErr) } publishDeferredPhase(fileName, assetID, "postwork", "sprites", "done", "Sprites", nil) } } truth = assetsTruthForVideo(outPath) // Ohne Sprite kann keine Sprite-basierte Analyse laufen. if !truth.SpritesReady { if firstErr == nil { firstErr = appErrorf("Analysen übersprungen: preview-sprite.jpg fehlt") } return firstErr } // ------------------------------------------------- // Langsame Hintergrundphase: NSFW + Highlights zusammen // ------------------------------------------------- analyzeReady := hasAIResultsForAllOutputGoals(outPath, requiredGoals) if !analyzeReady { if err := waitBeforeHeavyStep("enrich", "analyze"); err != nil { return err } publishDeferredPhase(fileName, assetID, "enrich", "analyze", "running", "Analysen", nil) _, err := ensureAnalyzeAllGoalsForVideoCtx(ctx, outPath, sourceURL) if cerr := abortIfCanceled(err, "enrich", "analyze"); cerr != nil { return cerr } analyzeReady = hasAIResultsForAllOutputGoals(outPath, requiredGoals) if err != nil || !analyzeReady { if err == nil { err = appErrorf("Analysen konnten nicht erzeugt werden") } publishDeferredPhase(fileName, assetID, "enrich", "analyze", "error", "Analysen fehlgeschlagen", err) if firstErr == nil { firstErr = err } } else { publishDeferredPhase(fileName, assetID, "enrich", "analyze", "done", "Analysen", nil) } } return firstErr } func handlePrimaryPostworkFailure(job *RecordJob, out string, postTarget JobStatus, err error) { finalStatus := JobFailed if postTarget == JobStopped { finalStatus = JobStopped } jobsMu.Lock() job.Status = finalStatus job.Error = err.Error() job.Output = out job.Phase = "" job.Progress = 100 job.PostWorkKey = "" job.PostWork = nil jobsMu.Unlock() if shouldKeepStoppedJobForInspection(postTarget, out) { _ = publishJob(job.ID) publishFinishedPostworkStatusForJob(job, finishedPostworkEvent{ File: filepath.Base(out), AssetID: assetIDFromVideoPath(out), Queue: "postwork", State: "error", Phase: "postwork", Label: "Primäre Nachbearbeitung fehlgeschlagen", }) return } publishFinishedPostworkStatusForJob(job, finishedPostworkEvent{ File: filepath.Base(out), AssetID: assetIDFromVideoPath(out), Queue: "postwork", State: "error", Phase: "postwork", Label: "Primäre Nachbearbeitung fehlgeschlagen", }) removeJobAndPublish(job) } func handlePrimaryPostworkSuccess(job *RecordJob, out string, postTarget JobStatus) { finalFile := filepath.Base(out) finalAssetID := assetIDFromVideoPath(out) jobsMu.Lock() job.Status = postTarget job.Phase = "" job.Progress = 100 job.PostWorkKey = "" job.PostWork = nil jobsMu.Unlock() publishFinishedPostworkStatusForJob(job, finishedPostworkEvent{ File: finalFile, AssetID: finalAssetID, Queue: "postwork", State: "done", Label: "Primäre Nachbearbeitung fertig", }) if ok := enqueueDeferredAssetsAndAI(job, out, job.SourceURL); !ok { appLogln("⚠️ deferred enrichment enqueue failed:", out) } if postTarget == JobStopped { _ = publishJob(job.ID) return } removeJobAndPublish(job) }