diff --git a/backend/chaturbate_autostart.go b/backend/chaturbate_autostart.go index d9f317c..1b09b62 100644 --- a/backend/chaturbate_autostart.go +++ b/backend/chaturbate_autostart.go @@ -217,6 +217,8 @@ func startChaturbateAutoStartWorker(store *ModelStore) { const scanInterval = 1 * time.Second const startInterval = 5 * time.Second + const snapshotMaxAge = 90 * time.Second + // normaler Retry für API-public const retryCooldown = 25 * time.Second @@ -260,10 +262,17 @@ func startChaturbateAutoStartWorker(store *ModelStore) { } lastCookieHdr = cookieHdr - // online snapshot aus cache - cbMu.RLock() - rooms := append([]ChaturbateRoom(nil), cb.Rooms...) - cbMu.RUnlock() + rooms, fetchedAt, fresh := getChaturbateRoomsSnapshot(snapshotMaxAge) + if !fresh { + if verboseLogs() { + if fetchedAt.IsZero() { + fmt.Println("⚠️ [autostart] chaturbate snapshot missing") + } else { + fmt.Println("⚠️ [autostart] chaturbate snapshot stale:", time.Since(fetchedAt).Round(time.Second)) + } + } + continue + } showByUser := map[string]string{} seenInAPI := map[string]bool{} @@ -445,10 +454,10 @@ func startChaturbateAutoStartWorker(store *ModelStore) { continue } - // unterscheiden: API-public oder Blind-Try - cbMu.RLock() - rooms := append([]ChaturbateRoom(nil), cb.Rooms...) - cbMu.RUnlock() + rooms, _, fresh := getChaturbateRoomsSnapshot(snapshotMaxAge) + if !fresh { + continue + } showByUser := map[string]string{} for _, r := range rooms { diff --git a/backend/chaturbate_online.go b/backend/chaturbate_online.go index ccbc7e4..cdecd97 100644 --- a/backend/chaturbate_online.go +++ b/backend/chaturbate_online.go @@ -478,14 +478,32 @@ func cbApplySnapshot(rooms []ChaturbateRoom) time.Time { return fetchedAtNow } +func getChaturbateRoomsSnapshot(maxAge time.Duration) ([]ChaturbateRoom, time.Time, bool) { + cbMu.RLock() + defer cbMu.RUnlock() + + fetchedAt := cb.FetchedAt + if fetchedAt.IsZero() { + return nil, time.Time{}, false + } + if maxAge > 0 && time.Since(fetchedAt) > maxAge { + return nil, fetchedAt, false + } + + rooms := append([]ChaturbateRoom(nil), cb.Rooms...) + return rooms, fetchedAt, true +} + func refreshChaturbateSnapshotNow(ctx context.Context) (time.Time, error) { + cbMu.Lock() + cb.LastAttempt = time.Now() + cbMu.Unlock() + rooms, err := fetchChaturbateOnlineRooms(ctx) if err != nil { cbMu.Lock() cb.LastErr = err.Error() - cb.Rooms = nil - cb.RoomsByUser = nil - cb.LiteByUser = nil + // ✅ KEIN Cache-Reset hier cbMu.Unlock() return time.Time{}, err } @@ -513,68 +531,55 @@ func startChaturbateOnlinePoller(store *ModelStore) { const interval = 10 * time.Second const tagsFillEvery = 10 * time.Minute + if store != nil { + setChaturbateOnlineModelStore(store) + } + lastLoggedCount := -1 lastLoggedErr := "" var tagsMu sync.Mutex var tagsLast time.Time - first := time.NewTimer(0) - defer first.Stop() - - ticker := time.NewTicker(interval) - defer ticker.Stop() - - for { - select { - case <-first.C: - case <-ticker.C: - } - + refreshOnce := func() { if !getSettings().UseChaturbateAPI { - continue + return } - // immer merken: wir haben es versucht - cbMu.Lock() - cb.LastAttempt = time.Now() - cbMu.Unlock() - ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) - rooms, err := fetchChaturbateOnlineRooms(ctx) + fetchedAtNow, err := refreshChaturbateSnapshotNow(ctx) cancel() if err != nil { - cbMu.Lock() - cb.LastErr = err.Error() + cbMu.RLock() + msg := cb.LastErr + prevFetchedAt := cb.FetchedAt + prevCount := len(cb.Rooms) + cbMu.RUnlock() - // Fehler => Cache leeren (damit offline nicht hängen bleibt) - cb.Rooms = nil - cb.RoomsByUser = nil - cb.LiteByUser = nil - cbMu.Unlock() - - if cb.LastErr != lastLoggedErr { - fmt.Println("❌ [chaturbate] online rooms fetch failed:", cb.LastErr) - lastLoggedErr = cb.LastErr + if msg != lastLoggedErr { + if prevFetchedAt.IsZero() { + fmt.Println("❌ [chaturbate] online rooms fetch failed (no snapshot yet):", msg) + } else { + fmt.Printf( + "⚠️ [chaturbate] refresh failed, keeping snapshot from %s (%d rooms): %s\n", + prevFetchedAt.Format(time.RFC3339), + prevCount, + msg, + ) + } + lastLoggedErr = msg } - continue + return } - fetchedAtNow := cbApplySnapshot(rooms) + cbMu.RLock() + roomsCopy := append([]ChaturbateRoom(nil), cb.Rooms...) + roomsCount := len(cb.Rooms) + cbMu.RUnlock() - // ✅ Alle bekannten Chaturbate-Models in der DB mit aktuellem Online-Snapshot synchronisieren - if cbModelStore != nil { - copiedRooms := append([]ChaturbateRoom(nil), rooms...) - - go func(roomsCopy []ChaturbateRoom, ts time.Time) { - syncChaturbateRoomStateIntoModelStore(cbModelStore, roomsCopy, ts) - _ = cbModelStore.SyncChaturbateOnlineForKnownModels(roomsCopy, ts) - }(copiedRooms, fetchedAtNow) - } - - // Tags übernehmen ist teuer -> nur selten + im Hintergrund - if cbModelStore != nil && len(rooms) > 0 { + // Tags nur selten und asynchron + if cbModelStore != nil && len(roomsCopy) > 0 { shouldFill := false tagsMu.Lock() @@ -585,20 +590,34 @@ func startChaturbateOnlinePoller(store *ModelStore) { tagsMu.Unlock() if shouldFill { - go cbModelStore.FillMissingTagsFromChaturbateOnline(rooms) + go cbModelStore.FillMissingTagsFromChaturbateOnline(roomsCopy) } } - // success logging only on changes if lastLoggedErr != "" { fmt.Println("✅ [chaturbate] online rooms fetch recovered") lastLoggedErr = "" } - if verboseLogs() && len(rooms) != lastLoggedCount { - fmt.Println("✅ [chaturbate] online rooms:", len(rooms)) - lastLoggedCount = len(rooms) + + if verboseLogs() && roomsCount != lastLoggedCount { + fmt.Printf( + "✅ [chaturbate] online rooms: %d (snapshot %s)\n", + roomsCount, + fetchedAtNow.Format(time.RFC3339), + ) + lastLoggedCount = roomsCount } } + + // ✅ sofort einmal versuchen + refreshOnce() + + ticker := time.NewTicker(interval) + defer ticker.Stop() + + for range ticker.C { + refreshOnce() + } } var onlineCacheMu sync.Mutex diff --git a/backend/postwork.go b/backend/postwork.go index 9af7002..a869cd9 100644 --- a/backend/postwork.go +++ b/backend/postwork.go @@ -277,13 +277,51 @@ func (pq *PostWorkQueue) StatusForKey(key string) PostWorkKeyStatus { } // global (oder in deinem app struct halten) -var postWorkQ = NewPostWorkQueue(512, 6) // schneller kritischer Pfad -var enrichQ = NewPostWorkQueue(512, 2) // langsame Hintergrundaufgaben +var postWorkQ = NewPostWorkQueue(512, 2) // schneller kritischer Pfad +var enrichQ = NewPostWorkQueue(512, 1) // langsame Hintergrundaufgaben // --- Status Refresher (ehemals postwork_refresh.go) --- +type enrichStatusSnapshot struct { + State string + Position int + Waiting int + Running int + MaxParallel int +} + +var enrichStatusLastMu sync.Mutex +var enrichStatusLast = map[string]enrichStatusSnapshot{} + +func shouldPublishEnrichStatus(key string, st PostWorkKeyStatus) bool { + enrichStatusLastMu.Lock() + defer enrichStatusLastMu.Unlock() + + cur := enrichStatusSnapshot{ + State: st.State, + Position: st.Position, + Waiting: st.Waiting, + Running: st.Running, + MaxParallel: st.MaxParallel, + } + + prev, ok := enrichStatusLast[key] + if ok && prev == cur { + return false + } + + enrichStatusLast[key] = cur + return true +} + +func clearEnrichStatusSnapshot(key string) { + enrichStatusLastMu.Lock() + delete(enrichStatusLast, key) + enrichStatusLastMu.Unlock() +} + func startEnrichStatusRefresher() { - t := time.NewTicker(1 * time.Second) + t := time.NewTicker(2 * time.Second) go func() { defer t.Stop() @@ -318,6 +356,11 @@ func startEnrichStatusRefresher() { // Nur senden, wenn der Job tatsächlich gerade queued/running ist if st.State != "queued" && st.State != "running" { + clearEnrichStatusSnapshot(key) + continue + } + + if !shouldPublishEnrichStatus(key, st) { continue } diff --git a/backend/preview.go b/backend/preview.go index 123318c..b229df9 100644 --- a/backend/preview.go +++ b/backend/preview.go @@ -2003,7 +2003,7 @@ func generateTeaserChunkMP4(ctx context.Context, src, out string, start, dur flo "-level", "4.2", "-preset", opts.Preset, "-crf", strconv.Itoa(opts.CRF), - "-threads", "4", + "-threads", "2", ) if opts.UseVsync2 { @@ -2145,7 +2145,7 @@ func generateTeaserPreviewMP4WithProgress(ctx context.Context, srcPath, outPath "-level", "4.2", "-preset", opts.Preset, "-crf", strconv.Itoa(opts.CRF), - "-threads", "4", + "-threads", "2", ) if opts.UseVsync2 { args = append(args, "-vsync", "2") @@ -2301,6 +2301,7 @@ func generateTeaserMP4(ctx context.Context, srcPath, outPath string, startSec, d "-c:v", "libx264", "-preset", "veryfast", + "-threads", "2", "-crf", "28", "-pix_fmt", "yuv420p", diff --git a/backend/record_stream_cb.go b/backend/record_stream_cb.go index 193b6b9..993aecd 100644 --- a/backend/record_stream_cb.go +++ b/backend/record_stream_cb.go @@ -92,7 +92,8 @@ func RecordStream( job.PreviewUA = hc.userAgent jobsMu.Unlock() - if assetID != "" { + // Live-Thumbs sind teuer: bei vielen parallelen Downloads nicht für alle starten. + if assetID != "" && activeRecordingJobsCount() <= 2 { startLiveThumbJPGLoop(ctx, job) } } diff --git a/backend/server.go b/backend/server.go index 4ee6cbf..efc1540 100644 --- a/backend/server.go +++ b/backend/server.go @@ -1,5 +1,4 @@ // backend\server.go - package main import ( @@ -8,6 +7,7 @@ import ( "net/http" "os" "os/signal" + "sync" "syscall" "time" ) @@ -17,38 +17,59 @@ func main() { loadSettings() fixKeepRootFilesIntoModelSubdirs() - postWorkQ.StartWorkers(1) - enrichQ.StartWorkers(2) + + postWorkQ.StartWorkers(2) + enrichQ.StartWorkers(1) startPostWorkStatusRefresher() startEnrichStatusRefresher() go startGeneratedGarbageCollector() - // ✅ NSFW-ONNX Detector initialisieren if err := initNSFWDetector(); err != nil { fmt.Println("❌ NSFW-ONNX Fehler:", err) os.Exit(1) } - defer func() { - _ = closeNSFWDetector() - }() + + var closeNSFWOnce sync.Once + closeNSFW := func() { + closeNSFWOnce.Do(func() { + _ = closeNSFWDetector() + }) + } + defer closeNSFW() mux := http.NewServeMux() auth, err := NewAuthManager() if err != nil { fmt.Println("❌ auth init:", err) - _ = closeNSFWDetector() + closeNSFW() os.Exit(1) } store := registerRoutes(mux, auth) + setChaturbateOnlineModelStore(store) + // Hintergrund-Worker go startChaturbateOnlinePoller(store) go startChaturbateAutoStartWorker(store) go startMyFreeCamsAutoStartWorker(store) go startDiskSpaceGuard() + // optional: Bootstrap nur im Hintergrund + if getSettings().UseChaturbateAPI { + go func() { + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + + if _, err := refreshChaturbateSnapshotNow(ctx); err != nil { + fmt.Println("⚠️ [chaturbate] startup snapshot failed:", err) + } else { + fmt.Println("✅ [chaturbate] startup snapshot loaded") + } + }() + } + if _, err := ensureCoversDir(); err != nil { fmt.Println("⚠️ covers dir:", err) } @@ -61,7 +82,6 @@ func main() { Handler: handler, } - // Shutdown-Signale stopSig := make(chan os.Signal, 1) signal.Notify(stopSig, os.Interrupt, syscall.SIGTERM) @@ -73,14 +93,14 @@ func main() { defer cancel() _ = srv.Shutdown(ctx) - _ = closeNSFWDetector() + closeNSFW() }() if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { fmt.Println("❌ HTTP-Server Fehler:", err) - _ = closeNSFWDetector() + closeNSFW() os.Exit(1) } - _ = closeNSFWDetector() + closeNSFW() } diff --git a/backend/sse.go b/backend/sse.go index c9bd33b..5c3552f 100644 --- a/backend/sse.go +++ b/backend/sse.go @@ -51,6 +51,7 @@ type jobEvent struct { type ssePublishItem struct { EventName string + CacheKey string Data []byte } @@ -230,7 +231,6 @@ func shouldPublishModelEventForJob(j *RecordJob) bool { } func visibleJobEventsJSON() []ssePublishItem { - nowTs := time.Now().UnixMilli() out := make([]ssePublishItem, 0, 64) jobsMu.RLock() @@ -265,7 +265,7 @@ func visibleJobEventsJSON() []ssePublishItem { PreviewState: j.PreviewState, PostWorkKey: strings.TrimSpace(j.PostWorkKey), PostWork: j.PostWork, - TS: nowTs, + TS: 0, // wichtig: stabil halten, damit Dedupe funktioniert } if sm := sseStoredModelForJob(j); sm != nil { @@ -287,6 +287,7 @@ func visibleJobEventsJSON() []ssePublishItem { out = append(out, ssePublishItem{ EventName: eventName, + CacheKey: eventName + "|" + j.ID, Data: b, }) } @@ -422,17 +423,42 @@ func initSSE() { // Per running job: 1 SSE event per second go func() { - t := time.NewTicker(1 * time.Second) + t := time.NewTicker(2 * time.Second) defer t.Stop() + lastByKey := map[string][]byte{} + lastHeartbeat := time.Now() + for range t.C { events := visibleJobEventsJSON() + liveKeys := make(map[string]struct{}, len(events)) + forceHeartbeat := time.Since(lastHeartbeat) >= 15*time.Second + for _, ev := range events { - if len(ev.Data) == 0 || ev.EventName == "" { + if len(ev.Data) == 0 || ev.EventName == "" || ev.CacheKey == "" { continue } + + liveKeys[ev.CacheKey] = struct{}{} + + prev, ok := lastByKey[ev.CacheKey] + if !forceHeartbeat && ok && bytes.Equal(prev, ev.Data) { + continue + } + + lastByKey[ev.CacheKey] = append([]byte(nil), ev.Data...) publishSSE(ev.EventName, ev.Data) } + + for k := range lastByKey { + if _, ok := liveKeys[k]; !ok { + delete(lastByKey, k) + } + } + + if forceHeartbeat { + lastHeartbeat = time.Now() + } } }() } diff --git a/backend/tasks_regenerate_assets.go b/backend/tasks_regenerate_assets.go index de367fb..7a94929 100644 --- a/backend/tasks_regenerate_assets.go +++ b/backend/tasks_regenerate_assets.go @@ -14,7 +14,7 @@ import ( "time" ) -const regenerateAssetsMaxParallel = 3 +const regenerateAssetsMaxParallel = 1 type regenerateAssetsRequest struct { File string `json:"file"` diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 5ce0bb9..add413c 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -537,7 +537,10 @@ export default function App() { const lastKnownRoomStatusByKeyRef = useRef>({}) const roomStatusBaselineReadyRef = useRef(false) - // ✅ Dedupe für "Cookies fehlen" Meldung (damit silent/autostarts nicht spammen) + const pendingJobEventsRef = useRef([]) + const pendingJobEventsTimerRef = useRef(null) + + // ✅ Dedupe für "Cookies fehlen" Meldung (damit silent/autostarts nicht spammen) const cookieProblemLastAtRef = useRef(0) const donePrefetchRef = useRef(null) @@ -1118,25 +1121,23 @@ export default function App() { lastSeenAtByJobIdRef.current[jobId] = now setJobs((prev) => { - const sameId = prev.filter((j) => String((j as any).id ?? '').trim() === jobId) - const prevJob = sameId.length > 0 ? sameId[0] : null + const idx = prev.findIndex( + (j) => String((j as any).id ?? '').trim() === jobId + ) + + const prevJob = idx >= 0 ? prev[idx] : null const merged = mergeJob(prevJob) - const next = [ - merged, - ...prev.filter((j) => String((j as any).id ?? '').trim() !== jobId), - ].sort((a, b) => { - const aMs = - Number((a as any)?.startedAtMs ?? 0) || - (a?.startedAt ? Date.parse(a.startedAt) || 0 : 0) - - const bMs = - Number((b as any)?.startedAtMs ?? 0) || - (b?.startedAt ? Date.parse(b.startedAt) || 0 : 0) - - return bMs - aMs - }) + // bestehender Job: nur ersetzen, Reihenfolge beibehalten + if (idx >= 0) { + const next = [...prev] + next[idx] = merged + jobsRef.current = next + return next + } + // neuer Job: vorne einfügen + const next = [merged, ...prev] jobsRef.current = next return next }) @@ -1148,6 +1149,52 @@ export default function App() { }) }, []) + const flushQueuedJobEvents = useCallback(() => { + if (pendingJobEventsTimerRef.current != null) { + window.clearTimeout(pendingJobEventsTimerRef.current) + pendingJobEventsTimerRef.current = null + } + + const batch = pendingJobEventsRef.current + if (batch.length === 0) return + + pendingJobEventsRef.current = [] + + // pro Job nur das letzte Event behalten + const byJobId = new Map() + const passthrough: JobEvent[] = [] + + for (const msg of batch) { + const type = String(msg?.type ?? '').trim() + const jobId = String(msg?.jobId ?? '').trim() + + if ((type === 'job_upsert' || type === 'job_remove') && jobId) { + byJobId.set(jobId, msg) + continue + } + + passthrough.push(msg) + } + + for (const msg of byJobId.values()) { + applyJobDelta(msg) + } + + for (const msg of passthrough) { + applyJobDelta(msg) + } + }, [applyJobDelta]) + + const queueJobDelta = useCallback((msg: JobEvent) => { + pendingJobEventsRef.current.push(msg) + + if (pendingJobEventsTimerRef.current != null) return + + pendingJobEventsTimerRef.current = window.setTimeout(() => { + flushQueuedJobEvents() + }, 120) + }, [flushQueuedJobEvents]) + const ensureModelEventListener = useCallback((name: string) => { const es = eventSourceRef.current const handler = onModelJobEventRef.current @@ -1275,6 +1322,14 @@ export default function App() { onlineLikedCount: 0, }) + useEffect(() => { + return () => { + if (pendingJobEventsTimerRef.current != null) { + window.clearTimeout(pendingJobEventsTimerRef.current) + } + } + }, []) + useEffect(() => { setModelsCount(headerStats.totalCount) }, [headerStats.totalCount]) @@ -1282,6 +1337,7 @@ export default function App() { const overviewInFlightRef = useRef(false) const overviewLastSigRef = useRef('') const overviewLastAtRef = useRef(0) + const overviewTimerRef = useRef(null) const loadAppModelsSnapshot = useCallback(async (keys: string[]) => { const uniqKeys = Array.from( @@ -1475,12 +1531,12 @@ export default function App() { } } - applyJobDelta(msg) + queueJobDelta(msg) } catch { // ignore } } - }, [applyJobDelta, applyRoomStateToModel, enqueueStart, maybeNotifyWatchedModelStatusChange]) + }, [applyRoomStateToModel, enqueueStart, maybeNotifyWatchedModelStatusChange, queueJobDelta]) useEffect(() => { const onOpen = (ev: Event) => { @@ -1782,9 +1838,19 @@ export default function App() { const lastOverviewSigRef = useRef('') useEffect(() => { + const scheduleOverviewLoad = () => { + if (overviewTimerRef.current != null) { + window.clearTimeout(overviewTimerRef.current) + } + + overviewTimerRef.current = window.setTimeout(() => { + void loadAppModelsSnapshot(requiredModelKeys) + }, 250) + } + if (lastOverviewSigRef.current !== requiredModelKeysSig) { lastOverviewSigRef.current = requiredModelKeysSig - void loadAppModelsSnapshot(requiredModelKeys) + scheduleOverviewLoad() } const onChanged = (ev: Event) => { @@ -1794,6 +1860,7 @@ export default function App() { if (updated && typeof updated === 'object') { const k = String(updated.modelKey ?? '').toLowerCase().trim() + if (k) { setModelsByKey((prev) => ({ ...prev, [k]: updated })) } @@ -1803,6 +1870,9 @@ export default function App() { } catch {} setPlayerModel((prev) => (prev?.id === updated.id ? updated : prev)) + + // counts/headerStats trotzdem später sauber nachziehen + scheduleOverviewLoad() return } @@ -1821,14 +1891,23 @@ export default function App() { setPlayerModel((prev) => (prev?.id === removedId ? null : prev)) } + scheduleOverviewLoad() return } - void loadAppModelsSnapshot(requiredModelKeys) + scheduleOverviewLoad() } window.addEventListener('models-changed', onChanged as any) - return () => window.removeEventListener('models-changed', onChanged as any) + + return () => { + window.removeEventListener('models-changed', onChanged as any) + + if (overviewTimerRef.current != null) { + window.clearTimeout(overviewTimerRef.current) + overviewTimerRef.current = null + } + } }, [loadAppModelsSnapshot, requiredModelKeysSig, requiredModelKeys, upsertModelCache]) const queuePendingAutoStart = useCallback(( diff --git a/frontend/src/components/ui/Downloads.tsx b/frontend/src/components/ui/Downloads.tsx index 95f746e..b8992f5 100644 --- a/frontend/src/components/ui/Downloads.tsx +++ b/frontend/src/components/ui/Downloads.tsx @@ -578,16 +578,18 @@ export default function Downloads({ }, []) useEffect(() => { - const hasLiveJobs = jobs.some((j) => { + const activeLiveJobs = jobs.filter((j) => { if ((j as any).endedAt) return false return String(j.status ?? '').trim().toLowerCase() === 'running' - }) + }).length - if (!hasLiveJobs || !pageVisible) return + if (activeLiveJobs === 0 || !pageVisible) return + + const intervalMs = activeLiveJobs > 2 ? 30000 : 10000 const id = window.setInterval(() => { setThumbTick((t) => t + 1) - }, 10000) + }, intervalMs) return () => window.clearInterval(id) }, [jobs, pageVisible]) @@ -909,7 +911,7 @@ export default function Downloads({ useEffect(() => { if (!hasActive) return - const t = window.setInterval(() => setNowMs(Date.now()), 1000) + const t = window.setInterval(() => setNowMs(Date.now()), 5000) return () => window.clearInterval(t) }, [hasActive]) @@ -946,7 +948,7 @@ export default function Downloads({ > { - setImgSrc(previewSrc) - setImgError(false) - }, [previewSrc]) - useEffect(() => { const ac = new AbortController()