chris/nsfwapp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fixed autostart

Browse Source
This commit is contained in:
Linrador 2026-04-07 20:43:00 +02:00
parent f0541b9fe6
commit ff341755c6
11 changed files with 311 additions and 116 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
backend/chaturbate_autostart.go
View File

@ -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 {

123
backend/chaturbate_online.go
View File

@ -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

49
backend/postwork.go
View File

@ -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
}

5
backend/preview.go
View File

@ -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",

3
backend/record_stream_cb.go
View File

@ -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)
}
}

44
backend/server.go
View File

@ -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()
}

34
backend/sse.go
View File

@ -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()
}
}
}()
}

2
backend/tasks_regenerate_assets.go
View File

@ -14,7 +14,7 @@ import (
"time"
)
const regenerateAssetsMaxParallel = 3
const regenerateAssetsMaxParallel = 1
type regenerateAssetsRequest struct {
File string `json:"file"`

123
frontend/src/App.tsx
View File

@ -537,7 +537,10 @@ export default function App() {
const lastKnownRoomStatusByKeyRef = useRef<Record<string, string>>({})
const roomStatusBaselineReadyRef = useRef(false)
// ✅ Dedupe für "Cookies fehlen" Meldung (damit silent/autostarts nicht spammen)
const pendingJobEventsRef = useRef<JobEvent[]>([])
const pendingJobEventsTimerRef = useRef<number | null>(null)
// ✅ Dedupe für "Cookies fehlen" Meldung (damit silent/autostarts nicht spammen)
const cookieProblemLastAtRef = useRef(0)
const donePrefetchRef = useRef<DonePrefetch | null>(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<string, JobEvent>()
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<number | null>(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<string>('')
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((

14
frontend/src/components/ui/Downloads.tsx
View File

@ -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({
>
<ModelPreview
jobId={j.id}
live={true}
live={false}
thumbTick={thumbTick}
blur={blurPreviews}
roomStatus={effectiveRoomStatusOfJob(

5
frontend/src/components/ui/ModelPreview.tsx
View File

@ -105,11 +105,6 @@ export default function ModelPreview({
[jobId]
)
useEffect(() => {
setImgSrc(previewSrc)
setImgError(false)
}, [previewSrc])
useEffect(() => {
const ac = new AbortController()