/** * Session Processor * * Core processing logic for the poller: * - processServerSessions: Process sessions from a single server * - pollServers: Orchestrate polling across all servers * - Lifecycle management: start, stop, trigger */ import { eq, and, desc, isNull, gte, lte, inArray } from 'drizzle-orm'; import { POLLING_INTERVALS, TIME_MS, REDIS_KEYS, SESSION_LIMITS, type ActiveSession, type SessionState, type Rule } from '@tracearr/shared'; import { db } from '../../db/client.js'; import { servers, serverUsers, sessions, users } from '../../db/schema.js'; import { createMediaServerClient } from '../../services/mediaServer/index.js'; import { geoipService, type GeoLocation } from '../../services/geoip.js'; import { ruleEngine } from '../../services/rules.js'; import type { CacheService, PubSubService } from '../../services/cache.js'; import type { Redis } from 'ioredis'; import { sseManager } from '../../services/sseManager.js'; import type { PollerConfig, ServerWithToken, ServerProcessingResult } from './types.js'; import { mapMediaSession } from './sessionMapper.js'; import { batchGetRecentUserSessions, getActiveRules } from './database.js'; import { calculatePauseAccumulation, calculateStopDuration, checkWatchCompletion, shouldForceStopStaleSession, shouldRecordSession, } from './stateTracker.js'; import { createViolationInTransaction, broadcastViolations, doesRuleApplyToUser, isDuplicateViolation, type ViolationInsertResult } from './violations.js'; import { enqueueNotification } from '../notificationQueue.js'; // ============================================================================ // Module State // ============================================================================ let pollingInterval: NodeJS.Timeout | null = null; let staleSweepInterval: NodeJS.Timeout | null = null; let cacheService: CacheService | null = null; let pubSubService: PubSubService | null = null; let redisClient: Redis | null = null; const defaultConfig: PollerConfig = { enabled: true, intervalMs: POLLING_INTERVALS.SESSIONS, }; // ============================================================================ // Server Session Processing // ============================================================================ /** * Process a single server's sessions * * This function: * 1. Fetches current sessions from the media server * 2. Creates/updates users as needed * 3. Creates new session records for new playbacks * 4. Updates existing sessions with state changes * 5. Marks stopped sessions as stopped * 6. Evaluates rules and creates violations * * @param server - Server to poll * @param activeRules - Active rules for evaluation * @param cachedSessionKeys - Set of currently cached session keys * @returns Processing results (new, updated, stopped sessions) */ async function processServerSessions( server: ServerWithToken, activeRules: Rule[], cachedSessionKeys: Set ): Promise { const newSessions: ActiveSession[] = []; const updatedSessions: ActiveSession[] = []; const currentSessionKeys = new Set(); try { // Fetch sessions from server using unified adapter const client = createMediaServerClient({ type: server.type, url: server.url, token: server.token, }); const mediaSessions = await client.getSessions(); const processedSessions = mediaSessions.map((s) => mapMediaSession(s, server.type)); // OPTIMIZATION: Early return if no active sessions from media server if (processedSessions.length === 0) { // Still need to handle stopped sessions detection const stoppedSessionKeys: string[] = []; for (const cachedKey of cachedSessionKeys) { if (cachedKey.startsWith(`${server.id}:`)) { stoppedSessionKeys.push(cachedKey); // Mark session as stopped in database const sessionKey = cachedKey.replace(`${server.id}:`, ''); const stoppedRows = await db .select() .from(sessions) .where( and( eq(sessions.serverId, server.id), eq(sessions.sessionKey, sessionKey), isNull(sessions.stoppedAt) ) ) .limit(1); const stoppedSession = stoppedRows[0]; if (stoppedSession) { const stoppedAt = new Date(); const { durationMs, finalPausedDurationMs } = calculateStopDuration( { startedAt: stoppedSession.startedAt, lastPausedAt: stoppedSession.lastPausedAt, pausedDurationMs: stoppedSession.pausedDurationMs || 0, }, stoppedAt ); const watched = stoppedSession.watched || checkWatchCompletion( stoppedSession.progressMs, stoppedSession.totalDurationMs ); await db .update(sessions) .set({ state: 'stopped', stoppedAt, durationMs, pausedDurationMs: finalPausedDurationMs, lastPausedAt: null, watched, }) .where(eq(sessions.id, stoppedSession.id)); } } } return { success: true, newSessions: [], stoppedSessionKeys, updatedSessions: [] }; } // OPTIMIZATION: Only load server users that match active sessions (not all users for server) // Collect unique externalIds from current sessions const sessionExternalIds = [...new Set(processedSessions.map((s) => s.externalUserId))]; const serverUsersList = await db .select({ id: serverUsers.id, userId: serverUsers.userId, serverId: serverUsers.serverId, externalId: serverUsers.externalId, username: serverUsers.username, email: serverUsers.email, thumbUrl: serverUsers.thumbUrl, isServerAdmin: serverUsers.isServerAdmin, trustScore: serverUsers.trustScore, sessionCount: serverUsers.sessionCount, createdAt: serverUsers.createdAt, updatedAt: serverUsers.updatedAt, identityName: users.name, }) .from(serverUsers) .innerJoin(users, eq(serverUsers.userId, users.id)) .where( and( eq(serverUsers.serverId, server.id), inArray(serverUsers.externalId, sessionExternalIds) ) ); // Build server user caches: externalId -> serverUser and id -> serverUser const serverUserByExternalId = new Map(); const serverUserById = new Map(); for (const serverUser of serverUsersList) { if (serverUser.externalId) { serverUserByExternalId.set(serverUser.externalId, serverUser); } serverUserById.set(serverUser.id, serverUser); } // Track server users that need to be created and their session indices const serverUsersToCreate: { externalId: string; username: string; thumbUrl: string | null; sessionIndex: number }[] = []; // First pass: identify server users and resolve from cache or mark for creation const sessionServerUserIds: (string | null)[] = []; for (let i = 0; i < processedSessions.length; i++) { const processed = processedSessions[i]!; const existingServerUser = serverUserByExternalId.get(processed.externalUserId); if (existingServerUser) { // Check if server user data needs update const needsUpdate = existingServerUser.username !== processed.username || (processed.userThumb && existingServerUser.thumbUrl !== processed.userThumb); if (needsUpdate) { await db .update(serverUsers) .set({ username: processed.username, thumbUrl: processed.userThumb || existingServerUser.thumbUrl, updatedAt: new Date(), }) .where(eq(serverUsers.id, existingServerUser.id)); // Update cache existingServerUser.username = processed.username; if (processed.userThumb) existingServerUser.thumbUrl = processed.userThumb; } sessionServerUserIds.push(existingServerUser.id); } else { // Need to create server user - mark for batch creation serverUsersToCreate.push({ externalId: processed.externalUserId, username: processed.username, thumbUrl: processed.userThumb || null, sessionIndex: i, }); sessionServerUserIds.push(null); // Will be filled after creation } } // Batch create new server users (and their identity users) if (serverUsersToCreate.length > 0) { // First, create identity users for each new server user const newIdentityUsers = await db .insert(users) .values(serverUsersToCreate.map(u => ({ username: u.username, // Login identifier name: u.username, // Use username as initial display name thumbnail: u.thumbUrl, }))) .returning(); // Then create server users linked to the identity users const newServerUsers = await db .insert(serverUsers) .values(serverUsersToCreate.map((u, idx) => ({ userId: newIdentityUsers[idx]!.id, serverId: server.id, externalId: u.externalId, username: u.username, thumbUrl: u.thumbUrl, }))) .returning(); // Update sessionServerUserIds with newly created server user IDs for (let i = 0; i < serverUsersToCreate.length; i++) { const serverUserToCreate = serverUsersToCreate[i]!; const newServerUser = newServerUsers[i]; const newIdentityUser = newIdentityUsers[i]; if (newServerUser && newIdentityUser) { sessionServerUserIds[serverUserToCreate.sessionIndex] = newServerUser.id; // Add to cache with identityName from the identity user const serverUserWithIdentity = { ...newServerUser, identityName: newIdentityUser.name, }; serverUserById.set(newServerUser.id, serverUserWithIdentity); serverUserByExternalId.set(serverUserToCreate.externalId, serverUserWithIdentity); } } } // OPTIMIZATION: Batch load recent sessions for rule evaluation // Only load for server users with new sessions (not cached) const serverUsersWithNewSessions = new Set(); for (let i = 0; i < processedSessions.length; i++) { const processed = processedSessions[i]!; const sessionKey = `${server.id}:${processed.sessionKey}`; const isNew = !cachedSessionKeys.has(sessionKey); const serverUserId = sessionServerUserIds[i]; if (isNew && serverUserId) { serverUsersWithNewSessions.add(serverUserId); } } const recentSessionsMap = await batchGetRecentUserSessions([...serverUsersWithNewSessions]); // Process each session for (let i = 0; i < processedSessions.length; i++) { const processed = processedSessions[i]!; const sessionKey = `${server.id}:${processed.sessionKey}`; currentSessionKeys.add(sessionKey); const serverUserId = sessionServerUserIds[i]; if (!serverUserId) { console.error('Failed to get/create server user for session'); continue; } // Get server user details from cache const serverUserFromCache = serverUserById.get(serverUserId); const userDetail = serverUserFromCache ? { id: serverUserFromCache.id, username: serverUserFromCache.username, thumbUrl: serverUserFromCache.thumbUrl, identityName: serverUserFromCache.identityName, } : { id: serverUserId, username: 'Unknown', thumbUrl: null, identityName: null }; // Get GeoIP location const geo: GeoLocation = geoipService.lookup(processed.ipAddress); const isNew = !cachedSessionKeys.has(sessionKey); if (isNew) { // RACE CONDITION CHECK: Verify no active session exists with this sessionKey // This can happen when SSE and poller both try to create a session simultaneously const existingWithSameKey = await db .select({ id: sessions.id }) .from(sessions) .where( and( eq(sessions.serverId, server.id), eq(sessions.sessionKey, processed.sessionKey), isNull(sessions.stoppedAt) ) ) .limit(1); if (existingWithSameKey.length > 0) { // Session already exists (likely created by SSE), skip insert // Add to cache so we don't check again next poll cachedSessionKeys.add(sessionKey); console.log(`[Poller] Active session already exists for ${processed.sessionKey}, skipping create`); continue; } // Check for session grouping - find recent unfinished session with same serverUser+ratingKey let referenceId: string | null = null; // FIRST: Check if there's an ACTIVE session for the same user+content // This handles quality changes mid-stream where Plex assigns a new sessionKey if (processed.ratingKey) { const activeSameContent = await db .select() .from(sessions) .where( and( eq(sessions.serverUserId, serverUserId), eq(sessions.ratingKey, processed.ratingKey), isNull(sessions.stoppedAt) // Still active ) ) .orderBy(desc(sessions.startedAt)) .limit(1); const existingActiveSession = activeSameContent[0]; if (existingActiveSession) { // This is a quality/resolution change during playback // Stop the old session and link the new one const now = new Date(); const { durationMs, finalPausedDurationMs } = calculateStopDuration( { startedAt: existingActiveSession.startedAt, lastPausedAt: existingActiveSession.lastPausedAt, pausedDurationMs: existingActiveSession.pausedDurationMs || 0, }, now ); await db .update(sessions) .set({ state: 'stopped', stoppedAt: now, durationMs, pausedDurationMs: finalPausedDurationMs, lastPausedAt: null, // Keep watched=false since playback is continuing }) .where(eq(sessions.id, existingActiveSession.id)); // Remove from cache if (cacheService) { await cacheService.deleteSessionById(existingActiveSession.id); await cacheService.removeUserSession(existingActiveSession.serverUserId, existingActiveSession.id); } // Publish stop event for the old session if (pubSubService) { await pubSubService.publish('session:stopped', existingActiveSession.id); } // Remove from cached session keys to prevent "stale" detection for this server cachedSessionKeys.delete(`${server.id}:${existingActiveSession.sessionKey}`); // Link to the original session chain referenceId = existingActiveSession.referenceId || existingActiveSession.id; console.log(`[Poller] Quality change detected for user ${serverUserId}, content ${processed.ratingKey}. Old session ${existingActiveSession.id} stopped, linking new session.`); } } // SECOND: Check for recently stopped sessions (resume tracking) if (!referenceId && processed.ratingKey) { const oneDayAgo = new Date(Date.now() - TIME_MS.DAY); const recentSameContent = await db .select() .from(sessions) .where( and( eq(sessions.serverUserId, serverUserId), eq(sessions.ratingKey, processed.ratingKey), gte(sessions.stoppedAt, oneDayAgo), eq(sessions.watched, false) // Not fully watched ) ) .orderBy(desc(sessions.stoppedAt)) .limit(1); const previousSession = recentSameContent[0]; // If user is resuming (progress >= previous), link to the original session if (previousSession && processed.progressMs !== undefined) { const prevProgress = previousSession.progressMs || 0; if (processed.progressMs >= prevProgress) { // This is a "resume" - link to the first session in the chain referenceId = previousSession.referenceId || previousSession.id; } } } // Use transaction to ensure session insert + rule evaluation + violation creation are atomic // This prevents orphaned sessions without violations on crash const recentSessions = recentSessionsMap.get(serverUserId) ?? []; const { insertedSession, violationResults } = await db.transaction(async (tx) => { // Insert new session with pause tracking fields const insertedRows = await tx .insert(sessions) .values({ serverId: server.id, serverUserId, sessionKey: processed.sessionKey, plexSessionId: processed.plexSessionId || null, ratingKey: processed.ratingKey || null, state: processed.state, mediaType: processed.mediaType, mediaTitle: processed.mediaTitle, // Enhanced media metadata grandparentTitle: processed.grandparentTitle || null, seasonNumber: processed.seasonNumber || null, episodeNumber: processed.episodeNumber || null, year: processed.year || null, thumbPath: processed.thumbPath || null, startedAt: new Date(), lastSeenAt: new Date(), // Track when we first saw this session totalDurationMs: processed.totalDurationMs || null, progressMs: processed.progressMs || null, // Pause tracking - use Jellyfin's precise timestamp if available, otherwise infer from state lastPausedAt: processed.lastPausedDate ?? (processed.state === 'paused' ? new Date() : null), pausedDurationMs: 0, // Session grouping referenceId, watched: false, // Network/device info ipAddress: processed.ipAddress, geoCity: geo.city, geoRegion: geo.region, geoCountry: geo.country, geoLat: geo.lat, geoLon: geo.lon, playerName: processed.playerName, deviceId: processed.deviceId || null, product: processed.product || null, device: processed.device || null, platform: processed.platform, quality: processed.quality, isTranscode: processed.isTranscode, bitrate: processed.bitrate, }) .returning(); const inserted = insertedRows[0]; if (!inserted) { throw new Error('Failed to insert session'); } // Evaluate rules within same transaction const session = { id: inserted.id, serverId: server.id, serverUserId, sessionKey: processed.sessionKey, state: processed.state, mediaType: processed.mediaType, mediaTitle: processed.mediaTitle, grandparentTitle: processed.grandparentTitle || null, seasonNumber: processed.seasonNumber || null, episodeNumber: processed.episodeNumber || null, year: processed.year || null, thumbPath: processed.thumbPath || null, ratingKey: processed.ratingKey || null, externalSessionId: null, startedAt: inserted.startedAt, stoppedAt: null, durationMs: null, totalDurationMs: processed.totalDurationMs || null, progressMs: processed.progressMs || null, lastPausedAt: inserted.lastPausedAt, pausedDurationMs: inserted.pausedDurationMs, referenceId: inserted.referenceId, watched: inserted.watched, ipAddress: processed.ipAddress, geoCity: geo.city, geoRegion: geo.region, geoCountry: geo.country, geoLat: geo.lat, geoLon: geo.lon, playerName: processed.playerName, deviceId: processed.deviceId || null, product: processed.product || null, device: processed.device || null, platform: processed.platform, quality: processed.quality, isTranscode: processed.isTranscode, bitrate: processed.bitrate, }; const ruleResults = await ruleEngine.evaluateSession(session, activeRules, recentSessions); // Create violations within same transaction const createdViolations: ViolationInsertResult[] = []; for (const result of ruleResults) { if (result.violated) { const matchingRule = activeRules.find( (r) => doesRuleApplyToUser(r, serverUserId) ); if (matchingRule) { // Check for duplicate violations before creating // This prevents multiple violations when sessions start simultaneously const relatedSessionIds = (result.data?.relatedSessionIds as string[]) || []; const isDuplicate = await isDuplicateViolation( serverUserId, matchingRule.type, inserted.id, relatedSessionIds ); if (isDuplicate) { continue; // Skip creating duplicate violation } const violationResult = await createViolationInTransaction( tx, matchingRule.id, serverUserId, inserted.id, result, matchingRule ); createdViolations.push(violationResult); } } } return { insertedSession: inserted, violationResults: createdViolations }; }); // Build active session for cache/broadcast (outside transaction - read only) const activeSession: ActiveSession = { id: insertedSession.id, serverId: server.id, serverUserId, sessionKey: processed.sessionKey, state: processed.state, mediaType: processed.mediaType, mediaTitle: processed.mediaTitle, // Enhanced media metadata grandparentTitle: processed.grandparentTitle || null, seasonNumber: processed.seasonNumber || null, episodeNumber: processed.episodeNumber || null, year: processed.year || null, thumbPath: processed.thumbPath || null, ratingKey: processed.ratingKey || null, externalSessionId: null, startedAt: insertedSession.startedAt, stoppedAt: null, durationMs: null, totalDurationMs: processed.totalDurationMs || null, progressMs: processed.progressMs || null, // Pause tracking lastPausedAt: insertedSession.lastPausedAt, pausedDurationMs: insertedSession.pausedDurationMs, referenceId: insertedSession.referenceId, watched: insertedSession.watched, // Network/device info ipAddress: processed.ipAddress, geoCity: geo.city, geoRegion: geo.region, geoCountry: geo.country, geoLat: geo.lat, geoLon: geo.lon, playerName: processed.playerName, deviceId: processed.deviceId || null, product: processed.product || null, device: processed.device || null, platform: processed.platform, quality: processed.quality, isTranscode: processed.isTranscode, bitrate: processed.bitrate, user: userDetail, server: { id: server.id, name: server.name, type: server.type }, }; newSessions.push(activeSession); // Broadcast violations AFTER transaction commits (outside transaction) // Wrapped in try-catch to prevent broadcast failures from crashing the poller try { await broadcastViolations(violationResults, insertedSession.id, pubSubService); } catch (err) { console.error('[Poller] Failed to broadcast violations:', err); // Violations are already persisted in DB, broadcast failure is non-fatal } } else { // Get existing ACTIVE session to check for state changes const existingRows = await db .select() .from(sessions) .where( and( eq(sessions.serverId, server.id), eq(sessions.sessionKey, processed.sessionKey), isNull(sessions.stoppedAt) ) ) .limit(1); const existingSession = existingRows[0]; if (!existingSession) continue; const previousState = existingSession.state; const newState = processed.state; const now = new Date(); // Build update payload with pause tracking const updatePayload: { state: 'playing' | 'paused'; quality: string; bitrate: number; progressMs: number | null; lastSeenAt: Date; plexSessionId?: string | null; lastPausedAt?: Date | null; pausedDurationMs?: number; watched?: boolean; } = { state: newState, quality: processed.quality, bitrate: processed.bitrate, progressMs: processed.progressMs || null, lastSeenAt: now, // Track when we last saw this session (for stale detection) // Always update plexSessionId (backfills sessions created before migration 0012) plexSessionId: processed.plexSessionId || null, }; // Handle state transitions for pause tracking const pauseResult = calculatePauseAccumulation( previousState as SessionState, newState, { lastPausedAt: existingSession.lastPausedAt, pausedDurationMs: existingSession.pausedDurationMs || 0 }, now ); updatePayload.lastPausedAt = pauseResult.lastPausedAt; updatePayload.pausedDurationMs = pauseResult.pausedDurationMs; // Check for watch completion (80% threshold) if (!existingSession.watched && checkWatchCompletion(processed.progressMs, processed.totalDurationMs)) { updatePayload.watched = true; } // Update existing session with state changes and pause tracking await db .update(sessions) .set(updatePayload) .where(eq(sessions.id, existingSession.id)); // Build active session for cache/broadcast (with updated pause tracking values) const activeSession: ActiveSession = { id: existingSession.id, serverId: server.id, serverUserId, sessionKey: processed.sessionKey, state: newState, mediaType: processed.mediaType, mediaTitle: processed.mediaTitle, // Enhanced media metadata grandparentTitle: processed.grandparentTitle || null, seasonNumber: processed.seasonNumber || null, episodeNumber: processed.episodeNumber || null, year: processed.year || null, thumbPath: processed.thumbPath || null, ratingKey: processed.ratingKey || null, externalSessionId: existingSession.externalSessionId, startedAt: existingSession.startedAt, stoppedAt: null, durationMs: null, totalDurationMs: processed.totalDurationMs || null, progressMs: processed.progressMs || null, // Pause tracking - use updated values lastPausedAt: updatePayload.lastPausedAt ?? existingSession.lastPausedAt, pausedDurationMs: updatePayload.pausedDurationMs ?? existingSession.pausedDurationMs ?? 0, referenceId: existingSession.referenceId, watched: updatePayload.watched ?? existingSession.watched ?? false, // Network/device info ipAddress: processed.ipAddress, geoCity: geo.city, geoRegion: geo.region, geoCountry: geo.country, geoLat: geo.lat, geoLon: geo.lon, playerName: processed.playerName, deviceId: processed.deviceId || null, product: processed.product || null, device: processed.device || null, platform: processed.platform, quality: processed.quality, isTranscode: processed.isTranscode, bitrate: processed.bitrate, user: userDetail, server: { id: server.id, name: server.name, type: server.type }, }; updatedSessions.push(activeSession); } } // Find stopped sessions const stoppedSessionKeys: string[] = []; for (const cachedKey of cachedSessionKeys) { if (cachedKey.startsWith(`${server.id}:`) && !currentSessionKeys.has(cachedKey)) { stoppedSessionKeys.push(cachedKey); // Mark session as stopped in database const sessionKey = cachedKey.replace(`${server.id}:`, ''); const stoppedRows = await db .select() .from(sessions) .where( and( eq(sessions.serverId, server.id), eq(sessions.sessionKey, sessionKey), isNull(sessions.stoppedAt) ) ) .limit(1); const stoppedSession = stoppedRows[0]; if (stoppedSession) { const stoppedAt = new Date(); // Calculate final duration const { durationMs, finalPausedDurationMs } = calculateStopDuration( { startedAt: stoppedSession.startedAt, lastPausedAt: stoppedSession.lastPausedAt, pausedDurationMs: stoppedSession.pausedDurationMs || 0, }, stoppedAt ); // Check for watch completion const watched = stoppedSession.watched || checkWatchCompletion( stoppedSession.progressMs, stoppedSession.totalDurationMs ); // Check if session meets minimum play time threshold (default 120s) // Short sessions are recorded but can be filtered from stats const shortSession = !shouldRecordSession(durationMs); await db .update(sessions) .set({ state: 'stopped', stoppedAt, durationMs, pausedDurationMs: finalPausedDurationMs, lastPausedAt: null, // Clear the pause timestamp watched, shortSession, }) .where(eq(sessions.id, stoppedSession.id)); } } } return { success: true, newSessions, stoppedSessionKeys, updatedSessions }; } catch (error) { console.error(`Error polling server ${server.name}:`, error); return { success: false, newSessions: [], stoppedSessionKeys: [], updatedSessions: [] }; } } // ============================================================================ // Main Polling Orchestration // ============================================================================ /** * Poll all connected servers for active sessions * * With SSE integration: * - Plex servers with active SSE connections are skipped (handled by SSE) * - Plex servers in fallback mode are polled * - Jellyfin/Emby servers are always polled (no SSE support) */ async function pollServers(): Promise { try { // Get all connected servers const allServers = await db.select().from(servers); if (allServers.length === 0) { return; } // Filter to only servers that need polling // SSE-connected Plex servers are handled by SSE, not polling const serversNeedingPoll = allServers.filter((server) => { // Non-Plex servers always need polling (Jellyfin/Emby don't support SSE yet) if (server.type !== 'plex') { return true; } // Plex servers in fallback mode need polling return sseManager.isInFallback(server.id); }); if (serversNeedingPoll.length === 0) { // All Plex servers are connected via SSE, no polling needed return; } // Get cached session keys from atomic SET-based cache const cachedSessions = cacheService ? await cacheService.getAllActiveSessions() : []; const cachedSessionKeys = new Set( cachedSessions.map((s) => `${s.serverId}:${s.sessionKey}`) ); // Get active rules const activeRules = await getActiveRules(); // Collect results from all servers const allNewSessions: ActiveSession[] = []; const allStoppedKeys: string[] = []; const allUpdatedSessions: ActiveSession[] = []; // Process each server with health tracking for (const server of serversNeedingPoll) { const serverWithToken = server as ServerWithToken; // Get previous health state for transition detection const wasHealthy = cacheService ? await cacheService.getServerHealth(server.id) : null; const { success, newSessions, stoppedSessionKeys, updatedSessions } = await processServerSessions( serverWithToken, activeRules, cachedSessionKeys ); // Track health state and notify on transitions if (cacheService) { await cacheService.setServerHealth(server.id, success); // Detect health state transitions if (wasHealthy === true && !success) { // Server went down - notify console.log(`[Poller] Server ${server.name} is DOWN`); await enqueueNotification({ type: 'server_down', payload: { serverName: server.name, serverId: server.id }, }); } else if (wasHealthy === false && success) { // Server came back up - notify console.log(`[Poller] Server ${server.name} is back UP`); await enqueueNotification({ type: 'server_up', payload: { serverName: server.name, serverId: server.id }, }); } // wasHealthy === null means first poll, don't notify } allNewSessions.push(...newSessions); allStoppedKeys.push(...stoppedSessionKeys); allUpdatedSessions.push(...updatedSessions); } // Update cache incrementally (preserves concurrent SSE operations) if (cacheService) { // Extract stopped session IDs from the key format "serverId:sessionKey" const stoppedSessionIds: string[] = []; for (const key of allStoppedKeys) { const parts = key.split(':'); if (parts.length >= 2) { const serverId = parts[0]; const sessionKey = parts.slice(1).join(':'); const stoppedSession = cachedSessions.find( (s) => s.serverId === serverId && s.sessionKey === sessionKey ); if (stoppedSession) { stoppedSessionIds.push(stoppedSession.id); } } } // Incremental sync: adds new, removes stopped, updates existing await cacheService.incrementalSyncActiveSessions( allNewSessions, stoppedSessionIds, allUpdatedSessions ); // Update user session sets for new sessions for (const session of allNewSessions) { await cacheService.addUserSession(session.serverUserId, session.id); } // Remove stopped sessions from user session sets for (const key of allStoppedKeys) { const parts = key.split(':'); if (parts.length >= 2) { const serverId = parts[0]; const sessionKey = parts.slice(1).join(':'); const stoppedSession = cachedSessions.find( (s) => s.serverId === serverId && s.sessionKey === sessionKey ); if (stoppedSession) { await cacheService.removeUserSession(stoppedSession.serverUserId, stoppedSession.id); } } } } // Publish events via pub/sub if (pubSubService) { for (const session of allNewSessions) { await pubSubService.publish('session:started', session); // Enqueue notification for async dispatch await enqueueNotification({ type: 'session_started', payload: session }); } for (const session of allUpdatedSessions) { await pubSubService.publish('session:updated', session); } for (const key of allStoppedKeys) { const parts = key.split(':'); if (parts.length >= 2) { const serverId = parts[0]; const sessionKey = parts.slice(1).join(':'); const stoppedSession = cachedSessions?.find( (s) => s.serverId === serverId && s.sessionKey === sessionKey ); if (stoppedSession) { await pubSubService.publish('session:stopped', stoppedSession.id); // Enqueue notification for async dispatch await enqueueNotification({ type: 'session_stopped', payload: stoppedSession }); } } } } if (allNewSessions.length > 0 || allStoppedKeys.length > 0) { console.log( `Poll complete: ${allNewSessions.length} new, ${allUpdatedSessions.length} updated, ${allStoppedKeys.length} stopped` ); } // Sweep for stale sessions that haven't been seen in a while // This catches sessions where server went down or SSE missed the stop event await sweepStaleSessions(); } catch (error) { console.error('Polling error:', error); } } // ============================================================================ // Stale Session Detection // ============================================================================ /** * Sweep for stale sessions and force-stop them * * A session is considered stale when: * - It hasn't been stopped (stoppedAt IS NULL) * - It hasn't been seen in a poll for > STALE_SESSION_TIMEOUT_SECONDS (default 5 min) * * This catches sessions where: * - Server became unreachable during playback * - SSE connection dropped and we missed the stop event * - The session hung on the media server side * * Stale sessions are marked with forceStopped = true to distinguish from normal stops. * Sessions with insufficient play time (< MIN_PLAY_TIME_MS) are still recorded for * audit purposes but can be filtered from stats queries. */ export async function sweepStaleSessions(): Promise { try { // Calculate the stale threshold (sessions not seen in last 5 minutes) const staleThreshold = new Date( Date.now() - SESSION_LIMITS.STALE_SESSION_TIMEOUT_SECONDS * 1000 ); // Find all active sessions that haven't been seen recently const staleSessions = await db .select() .from(sessions) .where( and( isNull(sessions.stoppedAt), // Still active lte(sessions.lastSeenAt, staleThreshold) // Not seen recently ) ); if (staleSessions.length === 0) { return 0; } console.log(`[Poller] Force-stopping ${staleSessions.length} stale session(s)`); const now = new Date(); for (const staleSession of staleSessions) { // Check if session should be force-stopped (using the stateTracker function) if (!shouldForceStopStaleSession(staleSession.lastSeenAt)) { // Shouldn't happen since we already filtered, but double-check continue; } // Calculate final duration const { durationMs, finalPausedDurationMs } = calculateStopDuration( { startedAt: staleSession.startedAt, lastPausedAt: staleSession.lastPausedAt, pausedDurationMs: staleSession.pausedDurationMs || 0, }, now ); // Check for watch completion const watched = staleSession.watched || checkWatchCompletion(staleSession.progressMs, staleSession.totalDurationMs); // Check if session meets minimum play time threshold const shortSession = !shouldRecordSession(durationMs); // Force-stop the session await db .update(sessions) .set({ state: 'stopped', stoppedAt: now, durationMs, pausedDurationMs: finalPausedDurationMs, lastPausedAt: null, watched, forceStopped: true, // Mark as force-stopped shortSession, }) .where(eq(sessions.id, staleSession.id)); // Remove from cache if cached if (cacheService) { await cacheService.deleteSessionById(staleSession.id); await cacheService.removeUserSession(staleSession.serverUserId, staleSession.id); } // Publish stop event if (pubSubService) { await pubSubService.publish('session:stopped', staleSession.id); } } // Invalidate dashboard stats after force-stopping sessions if (redisClient) { await redisClient.del(REDIS_KEYS.DASHBOARD_STATS); } return staleSessions.length; } catch (error) { console.error('[Poller] Error sweeping stale sessions:', error); return 0; } } // ============================================================================ // Lifecycle Management // ============================================================================ /** * Initialize the poller with cache services and Redis client */ export function initializePoller(cache: CacheService, pubSub: PubSubService, redis: Redis): void { cacheService = cache; pubSubService = pubSub; redisClient = redis; } /** * Start the polling job */ export function startPoller(config: Partial = {}): void { const mergedConfig = { ...defaultConfig, ...config }; if (!mergedConfig.enabled) { console.log('Session poller disabled'); return; } if (pollingInterval) { console.log('Poller already running'); return; } console.log(`Starting session poller with ${mergedConfig.intervalMs}ms interval`); // Run immediately on start void pollServers(); // Then run on interval pollingInterval = setInterval(() => void pollServers(), mergedConfig.intervalMs); // Start stale session sweep (runs every 60 seconds to detect abandoned sessions) if (!staleSweepInterval) { console.log(`Starting stale session sweep with ${SESSION_LIMITS.STALE_SWEEP_INTERVAL_MS}ms interval`); staleSweepInterval = setInterval(() => void sweepStaleSessions(), SESSION_LIMITS.STALE_SWEEP_INTERVAL_MS); } } /** * Stop the polling job */ export function stopPoller(): void { if (pollingInterval) { clearInterval(pollingInterval); pollingInterval = null; console.log('Session poller stopped'); } if (staleSweepInterval) { clearInterval(staleSweepInterval); staleSweepInterval = null; console.log('Stale session sweep stopped'); } } /** * Force an immediate poll */ export async function triggerPoll(): Promise { await pollServers(); } /** * Reconciliation poll for SSE-connected servers * * This is a lighter poll that runs periodically to catch any events * that might have been missed by SSE. Only polls Plex servers that * have active SSE connections (not in fallback mode). * * Unlike the main poller, this processes results and updates the cache * to sync any sessions that SSE may have missed. */ export async function triggerReconciliationPoll(): Promise { try { // Get all Plex servers with active SSE connections const allServers = await db.select().from(servers); const sseServers = allServers.filter( (server) => server.type === 'plex' && !sseManager.isInFallback(server.id) ); if (sseServers.length === 0) { return; } console.log(`[Poller] Running reconciliation poll for ${sseServers.length} SSE-connected server(s)`); // Get cached session keys from atomic SET-based cache const cachedSessions = cacheService ? await cacheService.getAllActiveSessions() : []; const cachedSessionKeys = new Set( cachedSessions.map((s) => `${s.serverId}:${s.sessionKey}`) ); // Get active rules const activeRules = await getActiveRules(); // Collect results from all SSE servers const allNewSessions: ActiveSession[] = []; const allStoppedKeys: string[] = []; const allUpdatedSessions: ActiveSession[] = []; // Process each SSE server and collect results for (const server of sseServers) { const serverWithToken = server as ServerWithToken; const { newSessions, stoppedSessionKeys, updatedSessions } = await processServerSessions( serverWithToken, activeRules, cachedSessionKeys ); allNewSessions.push(...newSessions); allStoppedKeys.push(...stoppedSessionKeys); allUpdatedSessions.push(...updatedSessions); } // Update cache incrementally if there were any changes if (cacheService && (allNewSessions.length > 0 || allStoppedKeys.length > 0 || allUpdatedSessions.length > 0)) { // Extract stopped session IDs from the key format "serverId:sessionKey" const stoppedSessionIds: string[] = []; for (const key of allStoppedKeys) { const parts = key.split(':'); if (parts.length >= 2) { const serverId = parts[0]; const sessionKey = parts.slice(1).join(':'); const stoppedSession = cachedSessions.find( (s) => s.serverId === serverId && s.sessionKey === sessionKey ); if (stoppedSession) { stoppedSessionIds.push(stoppedSession.id); } } } // Incremental sync: adds new, removes stopped, updates existing await cacheService.incrementalSyncActiveSessions( allNewSessions, stoppedSessionIds, allUpdatedSessions ); // Update user session sets for new sessions for (const session of allNewSessions) { await cacheService.addUserSession(session.serverUserId, session.id); } // Remove stopped sessions from user session sets for (const key of allStoppedKeys) { const parts = key.split(':'); if (parts.length >= 2) { const serverId = parts[0]; const sessionKey = parts.slice(1).join(':'); const stoppedSession = cachedSessions.find( (s) => s.serverId === serverId && s.sessionKey === sessionKey ); if (stoppedSession) { await cacheService.removeUserSession(stoppedSession.serverUserId, stoppedSession.id); } } } console.log( `[Poller] Reconciliation complete: ${allNewSessions.length} new, ${allUpdatedSessions.length} updated, ${allStoppedKeys.length} stopped` ); } // Publish events via pub/sub for any changes if (pubSubService) { for (const session of allNewSessions) { await pubSubService.publish('session:started', session); await enqueueNotification({ type: 'session_started', payload: session }); } for (const session of allUpdatedSessions) { await pubSubService.publish('session:updated', session); } for (const key of allStoppedKeys) { const parts = key.split(':'); if (parts.length >= 2) { const serverId = parts[0]; const sessionKey = parts.slice(1).join(':'); const stoppedSession = cachedSessions.find( (s) => s.serverId === serverId && s.sessionKey === sessionKey ); if (stoppedSession) { await pubSubService.publish('session:stopped', stoppedSession.id); await enqueueNotification({ type: 'session_stopped', payload: stoppedSession }); } } } } } catch (error) { console.error('[Poller] Reconciliation poll error:', error); } }