// backend/live.go package main import ( "bufio" "bytes" "context" "encoding/json" "fmt" "io" "net/http" "net/url" "os" "os/exec" "path" "path/filepath" "regexp" "strings" "sync" "syscall" "time" ) // ============================================================ // HLS Live Preview serving (+ m3u8 rewrite) // ============================================================ // // This file contains everything related to the HLS live preview stream: // - serving index*.m3u8 + segment files from a job's PreviewDir // - rewriting m3u8 segment URLs to a configurable base path // - starting/stopping the ffmpeg HLS preview process (per job) // - hover/play activation checks + preview "touch" + ensure-start logic // // It intentionally reuses existing globals/types from your backend (package main): // - jobs, jobsMu, RecordJob, JobRunning // - ffmpegPath, previewSem // - notifyJobsChanged() // - assetIDForJob(job *RecordJob) string // - startLiveThumbJPGLoop(ctx, job) // ============================================================ // Allowed files that may be served out of PreviewDir. var previewFileRe = regexp.MustCompile(`^(index(_hq)?\.m3u8|seg_(low|hq)_\d+\.ts|seg_\d+\.ts|init\.m4s|\w+\.m4s)$`) var cbHlsRefreshMu sync.Mutex var cbHlsRefreshAt = map[string]time.Time{} // key=userLower -> last refresh time func shouldRefreshHLS(userLower string, minInterval time.Duration) bool { userLower = strings.ToLower(strings.TrimSpace(userLower)) if userLower == "" { return false } cbHlsRefreshMu.Lock() defer cbHlsRefreshMu.Unlock() last := cbHlsRefreshAt[userLower] if !last.IsZero() && time.Since(last) < minInterval { return false } cbHlsRefreshAt[userLower] = time.Now() return true } // fetchCurrentBestHLS lädt die Room-Seite und parsed hls_source. func fetchCurrentBestHLS(ctx context.Context, username string, cookie string, userAgent string) (string, error) { u := strings.TrimSpace(username) if u == "" { return "", appErrorf("empty username") } if strings.TrimSpace(userAgent) == "" { userAgent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64)" } hc := NewHTTPClient(userAgent) pageURL := "https://chaturbate.com/" + strings.Trim(u, "/") + "/" body, err := hc.FetchPage(ctx, pageURL, cookie) if err != nil { return "", err } master, err := ParseStream(body) if err != nil { return "", err } master = strings.TrimSpace(master) if master == "" { return "", appErrorf("empty parsed hls url") } return master, nil } func serveLiveNotReady(w http.ResponseWriter, r *http.Request) { // ✅ Für HLS-Clients (hls.js) ist 204 beim Manifest "ein Fehler" -> aggressive Retries. // Deshalb: IMMER 200 + gültige (aber leere) m3u8 zurückgeben. w.Header().Set("Content-Type", "application/vnd.apple.mpegurl; charset=utf-8") w.Header().Set("Cache-Control", "no-store") w.Header().Set("X-Content-Type-Options", "nosniff") w.Header().Set("Retry-After", "1") if r.Method == http.MethodHead { w.WriteHeader(http.StatusOK) return } body := "#EXTM3U\n" + "#EXT-X-VERSION:3\n" + "#EXT-X-TARGETDURATION:2\n" + "#EXT-X-MEDIA-SEQUENCE:0\n" w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte(body)) } func maybeBlockHLSOnPreview(w http.ResponseWriter, r *http.Request, basePath, file string) bool { // Nur /api/preview (nicht /api/preview/live) if strings.TrimSpace(basePath) != "/api/preview" { return false } low := strings.ToLower(strings.TrimSpace(file)) if low == "" { return false } // Nur echte HLS-Dateien blocken (Manifest/Segmente) if strings.HasSuffix(low, ".m3u8") || strings.HasSuffix(low, ".ts") || strings.HasSuffix(low, ".m4s") { w.Header().Set("Cache-Control", "no-store") w.Header().Set("X-Preview-HLS-Disabled", "1") http.Error(w, "HLS disabled on /api/preview; use /api/preview/live", http.StatusGone) // 410 return true } return false } // stopPreview stops the running ffmpeg HLS preview process for a job and resets state. func stopPreview(job *RecordJob) { jobsMu.Lock() cmd := job.previewCmd cancel := job.previewCancel job.previewCmd = nil job.previewCancel = nil job.LiveThumbStarted = false job.PreviewDir = "" jobsMu.Unlock() if cancel != nil { cancel() } if cmd != nil && cmd.Process != nil { _ = cmd.Process.Kill() } } func refreshPreviewLiveSourceForJob(r *http.Request, job *RecordJob) error { if job == nil { return appErrorf("job nil") } jobsMu.RLock() sourceURL := strings.TrimSpace(job.SourceURL) cookie := strings.TrimSpace(job.PreviewCookie) ua := strings.TrimSpace(job.PreviewUA) jobsMu.RUnlock() if detectProvider(sourceURL) != "chaturbate" { return nil } // Optional: Cookie vom aktuellen Request bevorzugen if reqCookie := strings.TrimSpace(r.Header.Get("X-Chaturbate-Cookie")); reqCookie != "" { cookie = reqCookie } // UA vom Request bevorzugen if reqUA := strings.TrimSpace(r.Header.Get("User-Agent")); reqUA != "" { ua = reqUA } if ua == "" { ua = "Mozilla/5.0" } username := strings.TrimSpace(extractUsername(sourceURL)) if username == "" { return appErrorf("username not found") } // ✅ wichtig: doppelten Refresh verhindern if !shouldRefreshHLS(strings.ToLower(username), 10*time.Second) { return nil } ctxRefresh, cancelRefresh := context.WithTimeout(r.Context(), 8*time.Second) defer cancelRefresh() newHls, err := fetchCurrentBestHLS(ctxRefresh, username, cookie, ua) if err != nil { return err } newHls = strings.TrimSpace(newHls) if newHls == "" { return appErrorf("empty hls") } var oldHls string jobsMu.Lock() oldHls = strings.TrimSpace(job.PreviewM3U8) job.PreviewM3U8 = newHls job.PreviewCookie = cookie job.PreviewUA = ua job.PreviewState = "" job.PreviewStateAt = "" job.PreviewStateMsg = "" jobsMu.Unlock() if oldHls != "" && oldHls != newHls { stopPreview(job) } return nil } func recordPreviewLive(w http.ResponseWriter, r *http.Request) { prepare := strings.ToLower(strings.TrimSpace(r.URL.Query().Get("prepare"))) if prepare == "1" || prepare == "true" || prepare == "yes" { id := strings.TrimSpace(r.URL.Query().Get("id")) if id == "" { http.Error(w, "id fehlt", http.StatusBadRequest) return } jobsMu.RLock() job, ok := jobs[id] jobsMu.RUnlock() if !ok || job == nil { http.Error(w, "job nicht gefunden", http.StatusNotFound) return } err := refreshPreviewLiveSourceForJob(r, job) w.Header().Set("Content-Type", "application/json") w.Header().Set("Cache-Control", "no-store") type resp struct { OK bool `json:"ok"` Prepared bool `json:"prepared"` FetchedAt time.Time `json:"fetchedAt"` LastError string `json:"lastError,omitempty"` } if err != nil { _ = json.NewEncoder(w).Encode(resp{ OK: false, Prepared: false, FetchedAt: time.Now(), LastError: err.Error(), }) return } _ = json.NewEncoder(w).Encode(resp{ OK: true, Prepared: true, FetchedAt: time.Now(), }) return } // ✅ Route bleibt /api/preview/live file := strings.TrimSpace(r.URL.Query().Get("file")) if file == "" { recordPreviewLiveFMP4(w, r) return } id := strings.TrimSpace(r.URL.Query().Get("id")) if id == "" { http.Error(w, "id fehlt", http.StatusBadRequest) return } servePreviewHLSFileWithBase(w, r, id, file, "/api/preview/live") } // recordPreviewFile serves ONLY the HLS file requests for /api/preview?file=... // preview.jpg bleibt in preview.go (servePreviewJPGAlias). func recordPreviewFile(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet && r.Method != http.MethodHead { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } id := strings.TrimSpace(r.URL.Query().Get("id")) if id == "" { id = strings.TrimSpace(r.URL.Query().Get("name")) } if id == "" { http.Error(w, "id fehlt", http.StatusBadRequest) return } file := strings.TrimSpace(r.URL.Query().Get("file")) if file == "" { http.Error(w, "file fehlt", http.StatusBadRequest) return } // ✅ HLS auf /api/preview blocken (Manifest/Segmente), um Polling-Storm zu verhindern. // Wenn du es NICHT blocken willst, diese if-Zeile entfernen. if maybeBlockHLSOnPreview(w, r, "/api/preview", file) { return } // Alles andere (falls doch erlaubt) über die gemeinsame Serving-Funktion servePreviewHLSFileWithBase(w, r, id, file, "/api/preview") } // servePreviewHLSFileWithBase serves a single HLS file (index/segment) for a job. // If it's an m3u8, it is rewritten so that segment URIs point at basePath. func servePreviewHLSFileWithBase(w http.ResponseWriter, r *http.Request, id, file, basePath string) { file = strings.TrimSpace(file) if file == "" || filepath.Base(file) != file || !previewFileRe.MatchString(file) { http.Error(w, "ungültige file", http.StatusBadRequest) return } isIndex := file == "index.m3u8" || file == "index_hq.m3u8" jobsMu.RLock() job, ok := jobs[id] state := "" if ok && job != nil { state = strings.TrimSpace(job.PreviewState) } jobsMu.RUnlock() // HEAD: quick existence check if r.Method == http.MethodHead { if !ok || job == nil { w.WriteHeader(http.StatusNotFound) return } if state == "private" { w.WriteHeader(http.StatusForbidden) return } if state == "offline" { w.WriteHeader(http.StatusNotFound) return } jobsMu.RLock() previewDir := strings.TrimSpace(job.PreviewDir) jobsMu.RUnlock() if previewDir == "" { w.WriteHeader(http.StatusNotFound) return } p := filepath.Join(previewDir, file) if st, err := os.Stat(p); err == nil && !st.IsDir() { w.Header().Set("Cache-Control", "no-store") w.WriteHeader(http.StatusOK) return } w.WriteHeader(http.StatusNotFound) return } if !ok || job == nil { if isIndex { serveLiveNotReady(w, r) return } http.Error(w, "job nicht gefunden", http.StatusNotFound) return } ensurePreviewStarted(r, job) touchPreview(job) jobsMu.RLock() state = strings.TrimSpace(job.PreviewState) jobsMu.RUnlock() if state == "private" { http.Error(w, "model private", http.StatusForbidden) return } if state == "offline" { http.Error(w, "model offline", http.StatusNotFound) return } if state == "error" { http.Error(w, "preview error", http.StatusServiceUnavailable) return } previewDir := strings.TrimSpace(job.PreviewDir) if previewDir == "" { if isIndex { serveLiveNotReady(w, r) return } http.Error(w, "preview nicht verfügbar", http.StatusNotFound) return } p := filepath.Join(previewDir, file) st, err := os.Stat(p) if err != nil || st.IsDir() { if isIndex { serveLiveNotReady(w, r) return } http.Error(w, "datei nicht gefunden", http.StatusNotFound) return } ext := strings.ToLower(filepath.Ext(p)) w.Header().Set("Cache-Control", "no-store") w.Header().Set("X-Accel-Buffering", "no") // m3u8 -> rewrite if ext == ".m3u8" { raw, err := os.ReadFile(p) if err != nil { http.Error(w, "m3u8 read failed", http.StatusInternalServerError) return } rewritten := rewriteM3U8WithBase(raw, id, basePath) w.Header().Set("Content-Type", "application/vnd.apple.mpegurl; charset=utf-8") w.WriteHeader(http.StatusOK) _, _ = w.Write(rewritten) return } switch ext { case ".ts": w.Header().Set("Content-Type", "video/mp2t") case ".m4s": w.Header().Set("Content-Type", "video/iso.segment") default: w.Header().Set("Content-Type", "application/octet-stream") } // segments may still be written -> wait until size stabilizes if ext == ".ts" || ext == ".m4s" { if !waitForStableFile(p, 2, 120*time.Millisecond) { http.Error(w, "segment not ready", http.StatusNotFound) return } } f, err := os.Open(p) if err != nil { http.Error(w, "open failed", http.StatusNotFound) return } defer f.Close() http.ServeContent(w, r, file, st.ModTime(), f) } func waitForStableFile(path string, checks int, interval time.Duration) bool { var last int64 = -1 for i := 0; i < checks; i++ { st, err := os.Stat(path) if err != nil || st.IsDir() { return false } sz := st.Size() if last >= 0 && sz == last { return true } last = sz time.Sleep(interval) } return false } func classifyPreviewFFmpegStderr(stderr string) (state string, httpStatus int) { s := strings.ToLower(stderr) if strings.Contains(s, "403 forbidden") || strings.Contains(s, "http error 403") || strings.Contains(s, "server returned 403") { return "private", http.StatusForbidden } if strings.Contains(s, "404 not found") || strings.Contains(s, "http error 404") || strings.Contains(s, "server returned 404") { return "offline", http.StatusNotFound } return "", 0 } // startPreviewHLS starts ffmpeg to generate HLS segments in previewDir. // It also starts your existing live-thumb loop: startLiveThumbJPGLoop(ctx, job). func startPreviewHLS(ctx context.Context, job *RecordJob, m3u8URL, previewDir, httpCookie, userAgent string) error { if strings.TrimSpace(ffmpegPath) == "" { return appErrorf("kein ffmpeg gefunden – setze FFMPEG_PATH oder lege ffmpeg(.exe) neben das Backend") } if err := os.MkdirAll(previewDir, 0o755); err != nil { return err } jobsMu.Lock() job.PreviewState = "" job.PreviewStateAt = "" job.PreviewStateMsg = "" jobsMu.Unlock() commonIn := []string{"-y"} if strings.TrimSpace(userAgent) != "" { commonIn = append(commonIn, "-user_agent", userAgent) } if strings.TrimSpace(httpCookie) != "" { commonIn = append(commonIn, "-headers", fmt.Sprintf("Cookie: %s\r\n", httpCookie)) } commonIn = append(commonIn, "-i", m3u8URL) hqArgs := append(commonIn, "-map", "0:v:0", "-map", "0:a:0?", "-c:v", "copy", "-c:a", "copy", "-f", "hls", "-hls_time", "2", "-hls_list_size", "6", "-hls_allow_cache", "0", "-hls_flags", "delete_segments+append_list+independent_segments+temp_file", "-hls_segment_filename", filepath.Join(previewDir, "seg_hq_%05d.ts"), filepath.Join(previewDir, "index_hq.m3u8"), ) cmd := exec.CommandContext(ctx, ffmpegPath, hqArgs...) cmd.SysProcAttr = &syscall.SysProcAttr{ HideWindow: true, CreationFlags: 0x08000000, // CREATE_NO_WINDOW } var stderr bytes.Buffer cmd.Stderr = &stderr jobsMu.Lock() job.previewCmd = cmd jobsMu.Unlock() go func() { if err := previewSem.Acquire(ctx); err != nil { jobsMu.Lock() if job.previewCmd == cmd { job.previewCmd = nil } jobsMu.Unlock() return } defer previewSem.Release() if err := cmd.Run(); err != nil && ctx.Err() == nil { st := strings.TrimSpace(stderr.String()) state, code := classifyPreviewFFmpegStderr(st) jobsMu.Lock() if state != "" { job.PreviewState = state job.PreviewStateAt = time.Now().UTC().Format(time.RFC3339Nano) job.PreviewStateMsg = fmt.Sprintf("ffmpeg input returned HTTP %d", code) } else { job.PreviewState = "error" job.PreviewStateAt = time.Now().UTC().Format(time.RFC3339Nano) if len(st) > 280 { job.PreviewStateMsg = st[:280] + "…" } else { job.PreviewStateMsg = st } } jobsMu.Unlock() //fmt.Printf("⚠️ preview hq ffmpeg failed: %v (%s)\n", err, st) } jobsMu.Lock() if job.previewCmd == cmd { job.previewCmd = nil } jobsMu.Unlock() }() startLiveThumbJPGLoop(ctx, job) return nil } // rewriteM3U8WithBase rewrites all segment URIs inside an m3u8 to point at basePath. // // Example output line: // // /api/preview/live?id=&file=seg_hq_00001.ts&play=1 func rewriteM3U8WithBase(raw []byte, id string, basePath string) []byte { basePath = strings.TrimSpace(basePath) if basePath == "" { basePath = "/api/preview" } if !strings.HasPrefix(basePath, "/") { basePath = "/" + basePath } base := basePath + "?id=" + url.QueryEscape(id) + "&file=" var out bytes.Buffer sc := bufio.NewScanner(bytes.NewReader(raw)) for sc.Scan() { line := sc.Text() trim := strings.TrimSpace(line) if trim == "" { out.WriteByte('\n') continue } // tags: may contain URI="..." if strings.HasPrefix(trim, "#") { line = rewriteAttrURIWithBase(line, base, basePath) out.WriteString(line) out.WriteByte('\n') continue } u := trim // absolute URLs: keep if strings.HasPrefix(u, "http://") || strings.HasPrefix(u, "https://") { out.WriteString(line) out.WriteByte('\n') continue } // already points to our endpoint: keep if strings.Contains(u, basePath) || strings.Contains(u, "/api/preview") { out.WriteString(line) out.WriteByte('\n') continue } name := path.Base(u) out.WriteString(base + url.QueryEscape(name)) out.WriteByte('\n') } if err := sc.Err(); err != nil { return raw } return out.Bytes() } func rewriteAttrURIWithBase(line, base string, basePath string) string { const key = `URI="` i := strings.Index(line, key) if i < 0 { return line } j := strings.Index(line[i+len(key):], `"`) if j < 0 { return line } start := i + len(key) end := start + j val := line[start:end] valTrim := strings.TrimSpace(val) // keep absolute or already-rewritten URIs if strings.HasPrefix(valTrim, "http://") || strings.HasPrefix(valTrim, "https://") { return line } if strings.Contains(valTrim, basePath) || strings.Contains(valTrim, "/api/preview") { return line } name := path.Base(valTrim) repl := base + url.QueryEscape(name) return line[:start] + repl + line[end:] } // isHover decides whether this request should count as "active". func isHover(r *http.Request) bool { v := strings.ToLower(strings.TrimSpace(r.URL.Query().Get("hover"))) return v == "1" || v == "true" || v == "yes" } // touchPreview updates the last-hit timestamp so your cleanup/stop logic can use it. func touchPreview(job *RecordJob) { if job == nil { return } jobsMu.Lock() job.previewLastHit = time.Now() jobsMu.Unlock() } // ensurePreviewStarted starts the ffmpeg HLS preview if not running yet. func ensurePreviewStarted(r *http.Request, job *RecordJob) { if job == nil { return } job.previewStartMu.Lock() defer job.previewStartMu.Unlock() jobsMu.Lock() if job.previewCmd != nil && job.PreviewDir != "" { job.previewLastHit = time.Now() jobsMu.Unlock() return } m3u8 := strings.TrimSpace(job.PreviewM3U8) cookie := strings.TrimSpace(job.PreviewCookie) ua := strings.TrimSpace(job.PreviewUA) jobsMu.Unlock() if m3u8 == "" { return } pctx, cancel := context.WithCancel(context.Background()) assetID := assetIDForJob(job) pdir := filepath.Join(os.TempDir(), "rec_preview", assetID) jobsMu.Lock() job.PreviewDir = pdir job.previewCancel = cancel job.previewLastHit = time.Now() jobsMu.Unlock() _ = startPreviewHLS(pctx, job, m3u8, pdir, cookie, ua) } // ============================================================ // Live fMP4 (single request, chunked) via ffmpeg -> stdout // Route: /api/preview/live-fmp4?id=&hover=1 // ============================================================ func recordPreviewLiveFMP4(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "Nur GET erlaubt", http.StatusMethodNotAllowed) return } id := strings.TrimSpace(r.URL.Query().Get("id")) if id == "" { http.Error(w, "id fehlt", http.StatusBadRequest) return } jobsMu.RLock() job, ok := jobs[id] state := "" if ok && job != nil { state = strings.TrimSpace(job.PreviewState) } jobsMu.RUnlock() if !ok || job == nil { http.Error(w, "job nicht gefunden", http.StatusNotFound) return } _ = refreshPreviewLiveSourceForJob(r, job) // 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 } if state == "private" { http.Error(w, "model private", http.StatusForbidden) return } if state == "offline" { http.Error(w, "model offline", http.StatusNotFound) return } if state == "error" { http.Error(w, "preview error", http.StatusServiceUnavailable) return } w.Header().Set("Content-Type", "video/mp4") w.Header().Set("Cache-Control", "no-store") w.Header().Set("X-Accel-Buffering", "no") flusher, ok := w.(http.Flusher) if !ok { http.Error(w, "Streaming nicht unterstützt", http.StatusInternalServerError) return } ctx := r.Context() if ua == "" { ua = "Mozilla/5.0" } args := []string{"-hide_banner", "-loglevel", "error"} if ua != "" { args = append(args, "-user_agent", ua) } if cookie != "" { args = append(args, "-headers", fmt.Sprintf("Cookie: %s\r\n", cookie)) } args = append(args, "-i", m3u8) args = append(args, "-map", "0:v:0", "-map", "0:a:0?", "-c:v", "copy", "-c:a", "copy", ) args = append(args, "-f", "mp4", "-movflags", "frag_keyframe+empty_moov+default_base_moof", "-frag_duration", "2000000", "-min_frag_duration", "2000000", "pipe:1", ) cmd := exec.CommandContext(ctx, ffmpegPath, args...) cmd.SysProcAttr = &syscall.SysProcAttr{ HideWindow: true, CreationFlags: 0x08000000, // CREATE_NO_WINDOW } stdout, err := cmd.StdoutPipe() if err != nil { http.Error(w, "ffmpeg stdout pipe failed", http.StatusInternalServerError) return } var stderr bytes.Buffer cmd.Stderr = &stderr if err := cmd.Start(); err != nil { http.Error(w, "ffmpeg start failed: "+err.Error(), http.StatusInternalServerError) return } buf := make([]byte, 32*1024) for { select { case <-ctx.Done(): if cmd.Process != nil { _ = cmd.Process.Kill() } return default: } n, rerr := stdout.Read(buf) if n > 0 { _, _ = w.Write(buf[:n]) flusher.Flush() } if rerr != nil { if rerr == io.EOF { break } break } } _ = cmd.Wait() }