diff --git a/routes/prgs.py b/routes/prgs.py index d6a2745..9ac9c1a 100755 --- a/routes/prgs.py +++ b/routes/prgs.py @@ -31,6 +31,15 @@ ACTIVE_TASK_STATES = { "real-time", # "real-time" - real-time download progress (hyphenated version) } +# Define terminal task states that should be included when recently completed +TERMINAL_TASK_STATES = { + ProgressState.COMPLETE, # "complete" - task completed successfully + ProgressState.DONE, # "done" - task finished processing + ProgressState.ERROR, # "error" - task failed + ProgressState.CANCELLED, # "cancelled" - task was cancelled + ProgressState.SKIPPED, # "skipped" - task was skipped +} + def get_task_status_from_last_status(last_status): """ Extract the task status from last_status, checking both possible locations. @@ -506,7 +515,9 @@ async def get_task_updates(request: Request): task_counts["active"] += 1 # Always include active tasks in updates, apply filtering to others - should_include = is_active_task or (task_timestamp > since_timestamp and not active_only) + # Also include recently completed/terminal tasks to ensure "done" status gets sent + is_recently_terminal = task_status in TERMINAL_TASK_STATES and task_timestamp > since_timestamp + should_include = is_active_task or (task_timestamp > since_timestamp and not active_only) or is_recently_terminal if should_include: # Construct the same detailed task object as in list_tasks() @@ -661,21 +672,40 @@ async def stream_task_updates(request: Request): active_only = request.query_params.get('active_only', '').lower() == 'true' async def event_generator(): - # Track last update timestamp for this client connection + # Track last known state of each task to detect actual changes + last_task_states = {} # task_id -> {"status": str, "timestamp": float, "status_count": int} last_update_timestamp = time.time() + last_heartbeat = time.time() + heartbeat_interval = 10.0 # Reduced from 30s to 10s for faster connection monitoring + burst_mode_until = 0 # Timestamp until which we stay in burst mode try: # Send initial data immediately upon connection - yield await generate_task_update_event(last_update_timestamp, active_only, request) + initial_data = await generate_task_update_event(last_update_timestamp, active_only, request) + yield initial_data + + # Initialize task states from initial data + try: + initial_json = json.loads(initial_data.replace("data: ", "").strip()) + for task in initial_json.get("tasks", []): + task_id = task.get("task_id") + if task_id: + last_task_states[task_id] = { + "status": get_task_status_from_last_status(task.get("last_line")), + "timestamp": task.get("timestamp", last_update_timestamp), + "status_count": task.get("status_count", 0) + } + except: + pass # Continue if initial state parsing fails + last_update_timestamp = time.time() - # Continuous monitoring loop + # Optimized monitoring loop - only send when changes detected while True: try: - # Check for updates since last timestamp current_time = time.time() - # Get all tasks and check for updates + # Get all tasks and detect actual changes all_tasks = get_all_tasks() updated_tasks = [] active_tasks = [] @@ -691,27 +721,26 @@ async def stream_task_updates(request: Request): "skipped": 0 } - has_updates = False + has_actual_changes = False + current_task_ids = set() for task_summary in all_tasks: task_id = task_summary.get("task_id") if not task_id: continue + current_task_ids.add(task_id) task_info = get_task_info(task_id) if not task_info: continue last_status = get_last_task_status(task_id) - - # Check if task has been updated since the given timestamp task_timestamp = last_status.get("timestamp") if last_status else task_info.get("created_at", 0) - - # Determine task status and categorize task_status = get_task_status_from_last_status(last_status) is_active_task = is_task_active(task_status) + status_count = len(get_task_status(task_id)) - # Categorize tasks by status using ProgressState constants + # Categorize tasks by status if task_status == ProgressState.RETRYING: task_counts["retrying"] += 1 elif task_status in {ProgressState.QUEUED, "pending"}: @@ -727,22 +756,102 @@ async def stream_task_updates(request: Request): elif is_active_task: task_counts["active"] += 1 - # Always include active tasks in updates, apply filtering to others - should_include = is_active_task or (task_timestamp > last_update_timestamp and not active_only) + # Check if this task has actually changed + previous_state = last_task_states.get(task_id) + + # Determine if task has meaningful changes + task_changed = False + is_new_task = previous_state is None + just_became_terminal = False + + if is_new_task: + # Include new tasks if they're active OR if they're recently terminal + # (avoid sending old completed/cancelled tasks on connection) + if not (task_status in TERMINAL_TASK_STATES): + task_changed = True + # Trigger burst mode for new active tasks to catch rapid completions + burst_mode_until = current_time + 10.0 # 10 seconds of frequent polling + logger.debug(f"SSE: New active task detected: {task_id} - entering burst mode") + else: + # Check if terminal task is recent (completed within last 30 seconds) + is_recently_terminal = (current_time - task_timestamp) <= 30.0 + if is_recently_terminal: + task_changed = True + logger.info(f"SSE: New recently terminal task detected: {task_id} (status: {task_status}, age: {current_time - task_timestamp:.1f}s)") + else: + logger.debug(f"SSE: Skipping old terminal task: {task_id} (status: {task_status}, age: {current_time - task_timestamp:.1f}s)") + else: + # Check for status changes + status_changed = previous_state["status"] != task_status + # Check for new status updates (more detailed progress) + status_count_changed = previous_state["status_count"] != status_count + # Check for significant timestamp changes (new activity) + significant_timestamp_change = task_timestamp > previous_state["timestamp"] + + if status_changed: + task_changed = True + # Check if this is a transition TO terminal state + was_terminal = previous_state["status"] in TERMINAL_TASK_STATES + is_now_terminal = task_status in TERMINAL_TASK_STATES + just_became_terminal = not was_terminal and is_now_terminal + + # Extend burst mode on significant status changes + if not is_now_terminal: + burst_mode_until = max(burst_mode_until, current_time + 5.0) # 5 more seconds + + logger.debug(f"SSE: Status changed for {task_id}: {previous_state['status']} -> {task_status}") + if just_became_terminal: + logger.debug(f"SSE: Task {task_id} just became terminal") + elif status_count_changed and significant_timestamp_change and not (task_status in TERMINAL_TASK_STATES): + # Only track progress updates for non-terminal tasks + task_changed = True + logger.debug(f"SSE: Progress update for {task_id}: status_count {previous_state['status_count']} -> {status_count}") + + # Include task if it changed and meets criteria + should_include = False + if task_changed: + # For terminal state tasks, only include if they just became terminal + if task_status in TERMINAL_TASK_STATES: + if just_became_terminal: + should_include = True + has_actual_changes = True + logger.debug(f"SSE: Including terminal task {task_id} (just transitioned)") + # Note: we don't include new terminal tasks (handled above) + else: + # Non-terminal tasks are always included when they change + should_include = True + has_actual_changes = True + elif is_active_task and not active_only: + # For non-active_only streams, include active tasks periodically for frontend state sync + # But only if significant time has passed since last update + if current_time - last_update_timestamp > 10.0: # Every 10 seconds max + should_include = True if should_include: - has_updates = True - # Construct the same detailed task object as in updates endpoint + # Update our tracked state + last_task_states[task_id] = { + "status": task_status, + "timestamp": task_timestamp, + "status_count": status_count + } + + # Build response task_response = _build_task_response(task_info, last_status, task_id, current_time, request) if is_active_task: active_tasks.append(task_response) else: updated_tasks.append(task_response) + + # Clean up states for tasks that no longer exist + removed_tasks = set(last_task_states.keys()) - current_task_ids + for removed_task_id in removed_tasks: + del last_task_states[removed_task_id] + has_actual_changes = True + logger.debug(f"SSE: Task removed: {removed_task_id}") - # Only send update if there are changes - if has_updates: - # Combine active tasks (always shown) with updated tasks + # Send update only if there are actual changes + if has_actual_changes: all_returned_tasks = active_tasks + updated_tasks # Sort by priority (active first, then by creation time) @@ -760,24 +869,55 @@ async def stream_task_updates(request: Request): "active_tasks": len(active_tasks), "updated_count": len(updated_tasks), "since_timestamp": last_update_timestamp, + "change_type": "update" } # Send SSE event with update data event_data = json.dumps(update_data) yield f"data: {event_data}\n\n" - logger.debug(f"SSE: Sent {len(active_tasks)} active + {len(updated_tasks)} updated tasks") + # Log details about what was sent + task_statuses = [f"{task.get('task_id', 'unknown')}:{get_task_status_from_last_status(task.get('last_line'))}" for task in all_returned_tasks] + logger.info(f"SSE: Sent {len(active_tasks)} active + {len(updated_tasks)} updated tasks: {task_statuses}") - # Update last timestamp last_update_timestamp = current_time + last_heartbeat = current_time + + # Send heartbeat if no updates for a while (keeps connection alive) + elif current_time - last_heartbeat > heartbeat_interval: + heartbeat_data = { + "current_timestamp": current_time, + "total_tasks": task_counts["active"] + task_counts["retrying"], + "task_counts": task_counts, + "change_type": "heartbeat" + } + + event_data = json.dumps(heartbeat_data) + yield f"data: {event_data}\n\n" + + last_heartbeat = current_time + logger.debug("SSE: Sent heartbeat") - # Wait before next check (much shorter than polling interval) - await asyncio.sleep(0.5) # Check every 500ms for real-time feel + # Responsive polling - much faster for real-time updates + active_task_count = task_counts["active"] + task_counts["retrying"] + + if current_time < burst_mode_until: + # Burst mode: poll every 100ms to catch rapid task completions + await asyncio.sleep(0.1) + elif has_actual_changes or active_task_count > 0: + # When there are changes or active tasks, poll very frequently + await asyncio.sleep(0.2) # 200ms for immediate responsiveness + elif current_time - last_update_timestamp < 30.0: + # For 30 seconds after last update, poll more frequently to catch fast completions + await asyncio.sleep(0.5) # 500ms to catch fast transitions + else: + # Only when truly idle for >30s, use longer interval + await asyncio.sleep(2.0) # 2 seconds max when completely idle except Exception as e: logger.error(f"Error in SSE event generation: {e}", exc_info=True) # Send error event and continue - error_data = json.dumps({"error": "Internal server error", "timestamp": time.time()}) + error_data = json.dumps({"error": "Internal server error", "timestamp": time.time(), "change_type": "error"}) yield f"data: {error_data}\n\n" await asyncio.sleep(1) # Wait longer on error @@ -859,7 +999,9 @@ async def generate_task_update_event(since_timestamp: float, active_only: bool, task_counts["active"] += 1 # Always include active tasks in updates, apply filtering to others - should_include = is_active_task or (task_timestamp > since_timestamp and not active_only) + # Also include recently completed/terminal tasks to ensure "done" status gets sent + is_recently_terminal = task_status in TERMINAL_TASK_STATES and task_timestamp > since_timestamp + should_include = is_active_task or (task_timestamp > since_timestamp and not active_only) or is_recently_terminal if should_include: # Construct the same detailed task object as in updates endpoint diff --git a/routes/utils/celery_tasks.py b/routes/utils/celery_tasks.py index c8811c7..a6268ce 100644 --- a/routes/utils/celery_tasks.py +++ b/routes/utils/celery_tasks.py @@ -235,12 +235,12 @@ def cancel_task(task_id): # Try to revoke the Celery task if it hasn't started yet celery_app.control.revoke(task_id, terminate=True, signal="SIGTERM") - # Schedule deletion of task data after 30 seconds + # Schedule deletion of task data after 3 seconds delayed_delete_task_data.apply_async( - args=[task_id, "Task cancelled by user and auto-cleaned."], countdown=30 + args=[task_id, "Task cancelled by user and auto-cleaned."], countdown=3 ) logger.info( - f"Task {task_id} cancelled by user. Data scheduled for deletion in 30s." + f"Task {task_id} cancelled by user. Data scheduled for deletion in 3s." ) return {"status": "cancelled", "task_id": task_id} @@ -917,7 +917,7 @@ class ProgressTrackingTask(Task): # Schedule deletion for completed multi-track downloads delayed_delete_task_data.apply_async( args=[task_id, "Task completed successfully and auto-cleaned."], - countdown=30, # Delay in seconds + countdown=3, # Delay in seconds ) # If from playlist_watch and successful, add track to DB @@ -1055,7 +1055,7 @@ def task_postrun_handler( ): # Applies to single track downloads and tracks from playlists/albums delayed_delete_task_data.apply_async( args=[task_id, "Task completed successfully and auto-cleaned."], - countdown=30, + countdown=3, ) original_request = task_info.get("original_request", {}) @@ -1175,14 +1175,14 @@ def task_failure_handler( else: # If task cannot be retried, schedule its data for deletion logger.info( - f"Task {task_id} failed and cannot be retried. Data scheduled for deletion in 30s." + f"Task {task_id} failed and cannot be retried. Data scheduled for deletion in 3s." ) delayed_delete_task_data.apply_async( args=[ task_id, f"Task failed ({str(exception)}) and max retries reached. Auto-cleaned.", ], - countdown=30, + countdown=3, ) except Exception as e: diff --git a/spotizerr-ui/src/components/Queue.tsx b/spotizerr-ui/src/components/Queue.tsx index 94551d0..356e191 100644 --- a/spotizerr-ui/src/components/Queue.tsx +++ b/spotizerr-ui/src/components/Queue.tsx @@ -1,179 +1,26 @@ import { useContext, useState, useRef, useEffect } from "react"; -import { - FaTimes, - FaSync, - FaCheckCircle, - FaExclamationCircle, - FaHourglassHalf, - FaMusic, - FaCompactDisc, -} from "react-icons/fa"; -import { QueueContext, type QueueItem, type QueueStatus, isActiveTaskStatus } from "@/contexts/queue-context"; - -const isTerminalStatus = (status: QueueStatus) => - ["completed", "error", "cancelled", "skipped", "done"].includes(status); - -const statusStyles: Record< - QueueStatus, - { icon: React.ReactNode; color: string; bgColor: string; borderColor: string; name: string } -> = { - queued: { - icon: , - color: "text-content-muted dark:text-content-muted-dark", - bgColor: "bg-gradient-to-r from-surface-muted to-surface-accent dark:from-surface-muted-dark dark:to-surface-accent-dark", - borderColor: "border-border dark:border-border-dark", - name: "Queued", - }, - initializing: { - icon: , - color: "text-info", - bgColor: "bg-gradient-to-r from-blue-50 to-blue-100 dark:from-blue-900/20 dark:to-blue-800/30", - borderColor: "border-info/30 dark:border-info/40", - name: "Initializing", - }, - downloading: { - icon: , - color: "text-info", - bgColor: "bg-gradient-to-r from-blue-50 to-blue-100 dark:from-blue-900/20 dark:to-blue-800/30", - borderColor: "border-info/30 dark:border-info/40", - name: "Downloading", - }, - processing: { - icon: , - color: "text-processing", - bgColor: "bg-gradient-to-r from-purple-50 to-purple-100 dark:from-purple-900/20 dark:to-purple-800/30", - borderColor: "border-processing/30 dark:border-processing/40", - name: "Processing", - }, - retrying: { - icon: , - color: "text-warning", - bgColor: "bg-gradient-to-r from-orange-50 to-orange-100 dark:from-orange-900/20 dark:to-orange-800/30", - borderColor: "border-warning/30 dark:border-warning/40", - name: "Retrying", - }, - completed: { - icon: , - color: "text-success", - bgColor: "bg-gradient-to-r from-green-50 to-green-100 dark:from-green-900/20 dark:to-green-800/30", - borderColor: "border-success/30 dark:border-success/40", - name: "Completed", - }, - done: { - icon: , - color: "text-success", - bgColor: "bg-gradient-to-r from-green-50 to-green-100 dark:from-green-900/20 dark:to-green-800/30", - borderColor: "border-success/30 dark:border-success/40", - name: "Done", - }, - error: { - icon: , - color: "text-error", - bgColor: "bg-gradient-to-r from-red-50 to-red-100 dark:from-red-900/20 dark:to-red-800/30", - borderColor: "border-error/30 dark:border-error/40", - name: "Error", - }, - cancelled: { - icon: , - color: "text-warning", - bgColor: "bg-gradient-to-r from-orange-50 to-orange-100 dark:from-orange-900/20 dark:to-orange-800/30", - borderColor: "border-warning/30 dark:border-warning/40", - name: "Cancelled", - }, - skipped: { - icon: , - color: "text-content-muted dark:text-content-muted-dark", - bgColor: "bg-gradient-to-r from-surface-muted to-surface-accent dark:from-surface-muted-dark dark:to-surface-accent-dark", - borderColor: "border-border dark:border-border-dark", - name: "Skipped", - }, - pending: { - icon: , - color: "text-content-muted dark:text-content-muted-dark", - bgColor: "bg-gradient-to-r from-surface-muted to-surface-accent dark:from-surface-muted-dark dark:to-surface-accent-dark", - borderColor: "border-border dark:border-border-dark", - name: "Pending", - }, - "real-time": { - icon: , - color: "text-info", - bgColor: "bg-gradient-to-r from-blue-50 to-blue-100 dark:from-blue-900/20 dark:to-blue-800/30", - borderColor: "border-info/30 dark:border-info/40", - name: "Real-time Download", - }, - progress: { - icon: , - color: "text-info", - bgColor: "bg-gradient-to-r from-blue-50 to-blue-100 dark:from-blue-900/20 dark:to-blue-800/30", - borderColor: "border-info/30 dark:border-info/40", - name: "Progress", - }, - track_progress: { - icon: , - color: "text-info", - bgColor: "bg-gradient-to-r from-blue-50 to-blue-100 dark:from-blue-900/20 dark:to-blue-800/30", - borderColor: "border-info/30 dark:border-info/40", - name: "Track Progress", - }, -}; +import { FaTimes, FaSync, FaCheckCircle, FaExclamationCircle, FaHourglassHalf, FaMusic, FaCompactDisc } from "react-icons/fa"; +import { QueueContext, type QueueItem, getStatus, getProgress, getCurrentTrackInfo, isActiveStatus, isTerminalStatus } from "@/contexts/queue-context"; // Circular Progress Component const CircularProgress = ({ progress, isCompleted = false, - isRealProgress = false, size = 60, - strokeWidth = 6, - className = "" + strokeWidth = 6 }: { progress: number; isCompleted?: boolean; - isRealProgress?: boolean; size?: number; strokeWidth?: number; - className?: string; }) => { - // Apply a logarithmic curve to make progress slower near the end - ONLY for fake progress - const getAdjustedProgress = (rawProgress: number) => { - if (isCompleted) return 100; - if (rawProgress <= 0) return 0; - - // If this is real progress data, show it as-is without any artificial manipulation - if (isRealProgress) { - return Math.min(Math.max(rawProgress, 0), 100); - } - - // Only apply logarithmic curve for fake/simulated progress - // Use a logarithmic curve that slows down significantly near 100% - // This creates the effect of filling more slowly as it approaches completion - const normalized = Math.min(Math.max(rawProgress, 0), 100) / 100; - - // Apply easing function that slows down dramatically near the end - const eased = 1 - Math.pow(1 - normalized, 3); // Cubic ease-out - const logarithmic = Math.log(normalized * 9 + 1) / Math.log(10); // Logarithmic scaling - - // Combine both for a very slow approach to 100% - const combined = (eased * 0.7 + logarithmic * 0.3) * 95; // Cap at 95% during download - - // Ensure minimum visibility for any progress > 0 - const minVisible = rawProgress > 0 ? Math.max(combined, 8) : 0; - - return Math.min(minVisible, 95); // Never quite reach 100% during download - }; - - const adjustedProgress = getAdjustedProgress(progress); const radius = (size - strokeWidth) / 2; const circumference = radius * 2 * Math.PI; - const strokeDasharray = circumference; - const strokeDashoffset = circumference - (adjustedProgress / 100) * circumference; + const strokeDashoffset = circumference - (progress / 100) * circumference; return ( -
- +
+ {/* Background circle */} {/* Center content */} @@ -219,192 +61,235 @@ const CircularProgress = ({ ); }; -const QueueItemCard = ({ item }: { item: QueueItem }) => { - const { removeItem, retryItem, cancelItem } = useContext(QueueContext) || {}; +// Status styling configuration +const statusStyles = { + initializing: { + icon: , + color: "text-info", + bgColor: "bg-gradient-to-r from-blue-50 to-blue-100 dark:from-blue-900/20 dark:to-blue-800/30", + borderColor: "border-info/30 dark:border-info/40", + name: "Initializing", + }, + processing: { + icon: , + color: "text-processing", + bgColor: "bg-gradient-to-r from-purple-50 to-purple-100 dark:from-purple-900/20 dark:to-purple-800/30", + borderColor: "border-processing/30 dark:border-processing/40", + name: "Processing", + }, + downloading: { + icon: , + color: "text-info", + bgColor: "bg-gradient-to-r from-blue-50 to-blue-100 dark:from-blue-900/20 dark:to-blue-800/30", + borderColor: "border-info/30 dark:border-info/40", + name: "Downloading", + }, + "real-time": { + icon: , + color: "text-info", + bgColor: "bg-gradient-to-r from-blue-50 to-blue-100 dark:from-blue-900/20 dark:to-blue-800/30", + borderColor: "border-info/30 dark:border-info/40", + name: "Downloading", + }, + done: { + icon: , + color: "text-success", + bgColor: "bg-gradient-to-r from-green-50 to-green-100 dark:from-green-900/20 dark:to-green-800/30", + borderColor: "border-success/30 dark:border-success/40", + name: "Done", + }, + completed: { + icon: , + color: "text-success", + bgColor: "bg-gradient-to-r from-green-50 to-green-100 dark:from-green-900/20 dark:to-green-800/30", + borderColor: "border-success/30 dark:border-success/40", + name: "Completed", + }, + error: { + icon: , + color: "text-error", + bgColor: "bg-gradient-to-r from-red-50 to-red-100 dark:from-red-900/20 dark:to-red-800/30", + borderColor: "border-error/30 dark:border-error/40", + name: "Error", + }, + cancelled: { + icon: , + color: "text-warning", + bgColor: "bg-gradient-to-r from-orange-50 to-orange-100 dark:from-orange-900/20 dark:to-orange-800/30", + borderColor: "border-warning/30 dark:border-warning/40", + name: "Cancelled", + }, + queued: { + icon: , + color: "text-content-muted dark:text-content-muted-dark", + bgColor: "bg-gradient-to-r from-surface-muted to-surface-accent dark:from-surface-muted-dark dark:to-surface-accent-dark", + borderColor: "border-border dark:border-border-dark", + name: "Queued", + }, + retrying: { + icon: , + color: "text-warning", + bgColor: "bg-gradient-to-r from-orange-50 to-orange-100 dark:from-orange-900/20 dark:to-orange-800/30", + borderColor: "border-warning/30 dark:border-warning/40", + name: "Retrying", + }, +} as const; + +// Cancelled Task Component +const CancelledTaskCard = ({ item }: { item: QueueItem }) => { + const { removeItem } = useContext(QueueContext) || {}; - // Extract the actual status - prioritize status_info.status, then last_line.status, then item.status - const actualStatus = (item.last_line?.status_info?.status as QueueStatus) || - (item.last_line?.status as QueueStatus) || - item.status; - const statusInfo = statusStyles[actualStatus] || statusStyles.queued; - const isTerminal = isTerminalStatus(actualStatus); - - const getProgressText = () => { - const { type, progress, totalTracks, summary, last_line } = item; - - // Handle real-time downloads - if (actualStatus === "real-time") { - const realTimeProgress = last_line?.status_info?.progress; - if (type === "track" && realTimeProgress !== undefined) { - return `${realTimeProgress.toFixed(0)}%`; - } - return null; - } - - if (actualStatus === "downloading" || actualStatus === "processing" || actualStatus === "progress" || actualStatus === "track_progress") { - if (type === "track") { - return progress !== undefined ? `${progress.toFixed(0)}%` : null; - } - // For albums/playlists, detailed progress is in the main body - return null; - } - - if ((actualStatus === "completed" || actualStatus === "done") && summary) { - if (type === "track") { - // For single tracks, don't show redundant text since status badge already shows "Done" - return null; - } - return `${summary.total_successful}/${totalTracks} tracks`; - } - - return null; - }; - - const progressText = getProgressText(); + const trackInfo = getCurrentTrackInfo(item); + const TypeIcon = item.downloadType === "album" ? FaCompactDisc : FaMusic; return ( -
- {/* Mobile-first layout: stack status and actions on mobile, inline on desktop */} +
+ + {/* Main content */}
-
- {statusInfo.icon} +
+
+
- {item.type === "track" && ( - <> -
- -

- {item.name} -

-
-

- {item.artist} -

- {item.albumName && ( -

- {item.albumName} -

- )} - - )} - {item.type === "album" && ( - <> -
- -

- {item.name} -

-
-

- {item.artist} -

- {item.currentTrackTitle && ( -

- {item.currentTrackNumber}/{item.totalTracks}: {item.currentTrackTitle} -

- )} - - )} - {item.type === "playlist" && ( - <> -
- -

- {item.name} -

-
-

- {item.playlistOwner} -

- {item.currentTrackTitle && ( -

- {item.currentTrackNumber}/{item.totalTracks}: {item.currentTrackTitle} -

- )} - +
+ +

+ {item.name} +

+
+ +

+ {item.artist} +

+ + {/* Show current track info for parent downloads */} + {(item.downloadType === "album" || item.downloadType === "playlist") && trackInfo.title && ( +

+ {trackInfo.current}/{trackInfo.total}: {trackInfo.title} +

)}
- {/* Status and actions - stacked on mobile, inline on desktop */} + {/* Status and actions */}
-
- {statusInfo.name} +
+ Cancelled
- {(() => { - // Only show text progress if we're not showing circular progress - const hasCircularProgress = item.type === "track" && !isTerminal && - (item.last_line?.status_info?.progress !== undefined || item.progress !== undefined); - - return !hasCircularProgress && progressText && ( -

{progressText}

- ); - })()}
- {/* Add circular progress for downloading tracks */} - {(() => { - // Calculate progress based on item type and data availability - let currentProgress: number | undefined; - let isRealProgress = false; - - if (item.type === "track") { - // For tracks, use direct progress - const realTimeProgress = item.last_line?.status_info?.progress; - const fallbackProgress = item.progress; - currentProgress = realTimeProgress ?? fallbackProgress; - isRealProgress = realTimeProgress !== undefined; - } else if ((item.type === "album" || item.type === "playlist") && item.last_line?.status_info?.progress !== undefined) { - // For albums/playlists with real-time data, calculate overall progress - const trackProgress = item.last_line.status_info.progress; - const currentTrack = item.last_line.current_track || 1; - const totalTracks = item.last_line.total_tracks || item.totalTracks || 1; - - // Formula: ((completed_tracks + current_track_progress/100) / total_tracks) * 100 - const completedTracks = currentTrack - 1; // current_track is 1-indexed - currentProgress = ((completedTracks + (trackProgress / 100)) / totalTracks) * 100; - isRealProgress = true; - } else if ((item.type === "album" || item.type === "playlist") && item.progress !== undefined) { - // Fallback for albums/playlists without real-time data - currentProgress = item.progress; - isRealProgress = false; - } - - // Show circular progress for items that are not in terminal state - const shouldShowProgress = !isTerminal && currentProgress !== undefined; - - return shouldShowProgress && ( -
- -
- ); - })()} + {/* Remove button */} +
+ +
+
+
+ + {/* Cancellation reason */} + {item.error && ( +
+

+ Cancelled: {item.error} +

+
+ )} +
+ ); +}; + +const QueueItemCard = ({ item }: { item: QueueItem }) => { + const { removeItem, cancelItem } = useContext(QueueContext) || {}; + + const status = getStatus(item); + const progress = getProgress(item); + const trackInfo = getCurrentTrackInfo(item); + const styleInfo = statusStyles[status as keyof typeof statusStyles] || statusStyles.queued; + const isTerminal = isTerminalStatus(status); + const isActive = isActiveStatus(status); + + // Get type icon + const TypeIcon = item.downloadType === "album" ? FaCompactDisc : FaMusic; + + return ( +
+
+ + {/* Main content */} +
+
+ {styleInfo.icon} +
- {/* Show completed circular progress for completed tracks */} - {(actualStatus === "completed" || actualStatus === "done") && - item.type === "track" && ( +
+
+ +

+ {item.name} +

+
+ +

+ {item.artist} +

+ + {/* Show current track info for parent downloads */} + {(item.downloadType === "album" || item.downloadType === "playlist") && trackInfo.title && ( +

+ {trackInfo.current}/{trackInfo.total}: {trackInfo.title} +

+ )} +
+
+ + {/* Status and progress */} +
+
+
+ {styleInfo.name} +
+ + {/* Summary info for completed downloads */} + {isTerminal && item.summary && item.downloadType !== "track" && ( +

+ {item.summary.total_successful}/{trackInfo.total || item.summary.total_successful + item.summary.total_failed + item.summary.total_skipped} tracks +

+ )} +
+ + {/* Circular progress for active downloads */} + {isActive && progress !== undefined && (
)} + {/* Completed progress for finished downloads */} + {isTerminal && status === "done" && item.downloadType === "track" && ( +
+ +
+ )} + + {/* Action buttons */}
{isTerminal ? ( )} - {item.canRetry && ( - - )}
- {(actualStatus === "error" || actualStatus === "retrying" || actualStatus === "cancelled") && (item.error || item.last_line?.error || item.last_line?.status_info?.error) && ( + + {/* Error message */} + {item.error && (

- {actualStatus === "cancelled" ? "Cancelled: " : "Error: "} - {item.last_line?.status_info?.error || item.last_line?.error || item.error} + {status === "cancelled" ? "Cancelled: " : "Error: "} + {item.error}

)} + + {/* Summary for failed/skipped tracks */} {isTerminal && item.summary && (item.summary.total_failed > 0 || item.summary.total_skipped > 0) && (
@@ -468,7 +348,6 @@ const QueueItemCard = ({ item }: { item: QueueItem }) => { export const Queue = () => { const context = useContext(QueueContext); const [startY, setStartY] = useState(null); - const [currentY, setCurrentY] = useState(null); const [isDragging, setIsDragging] = useState(false); const [dragDistance, setDragDistance] = useState(0); const queueRef = useRef(null); @@ -476,7 +355,6 @@ export const Queue = () => { const headerRef = useRef(null); const [canDrag, setCanDrag] = useState(false); - // Extract values from context (with defaults to avoid crashes) const { items = [], isVisible = false, @@ -489,18 +367,15 @@ export const Queue = () => { totalTasks = 0 } = context || {}; - // Infinite scroll effect - MUST be called before any conditional returns + // Infinite scroll useEffect(() => { - if (!isVisible) return; // Early return if not visible - + if (!isVisible) return; const scrollContainer = scrollContainerRef.current; if (!scrollContainer) return; const handleScroll = () => { const { scrollTop, scrollHeight, clientHeight } = scrollContainer; const scrollPercentage = (scrollTop + clientHeight) / scrollHeight; - - // Load more when user has scrolled 80% of the way down if (scrollPercentage > 0.8 && hasMore && !isLoadingMore) { loadMoreTasks(); } @@ -510,72 +385,18 @@ export const Queue = () => { return () => scrollContainer.removeEventListener('scroll', handleScroll); }, [isVisible, hasMore, isLoadingMore, loadMoreTasks]); - // Reset drag state when queue visibility changes - useEffect(() => { - if (!isVisible) { - setStartY(null); - setCurrentY(null); - setIsDragging(false); - setDragDistance(0); - setCanDrag(false); - } - }, [isVisible]); - - // Prevent body scroll when queue is visible on mobile - useEffect(() => { - if (!isVisible) return; - - // Only apply on mobile (when queue covers full screen) - const isMobile = window.innerWidth < 768; // md breakpoint - if (!isMobile) return; - - // Store original styles - const originalOverflow = document.body.style.overflow; - const originalTouchAction = document.body.style.touchAction; - const originalPosition = document.body.style.position; - - // Prevent body scroll and interactions - document.body.style.overflow = 'hidden'; - document.body.style.touchAction = 'none'; - document.body.style.position = 'fixed'; - document.body.style.width = '100%'; - - // Cleanup function - return () => { - document.body.style.overflow = originalOverflow; - document.body.style.touchAction = originalTouchAction; - document.body.style.position = originalPosition; - document.body.style.width = ''; - }; - }, [isVisible]); - - // Early returns after all hooks - if (!context) return null; - if (!isVisible) return null; - - const hasActive = items.some((item) => { - // Check for status in both possible locations (nested status_info for real-time, or top-level for others) - const actualStatus = (item.last_line?.status_info?.status as QueueStatus) || - (item.last_line?.status as QueueStatus) || - item.status; - return isActiveTaskStatus(actualStatus); - }); - const hasFinished = items.some((item) => isTerminalStatus(item.status)); - - // Enhanced mobile touch handling for drag-to-dismiss + // Mobile drag-to-dismiss const handleTouchStart = (e: React.TouchEvent) => { const touch = e.touches[0]; const scrollContainer = scrollContainerRef.current; const headerElement = headerRef.current; - // Only allow dragging if touch starts on header or if scroll is at top const touchedHeader = headerElement?.contains(e.target as Node); const scrollAtTop = scrollContainer ? scrollContainer.scrollTop <= 5 : true; if (touchedHeader || scrollAtTop) { setCanDrag(true); setStartY(touch.clientY); - setCurrentY(touch.clientY); setIsDragging(false); setDragDistance(0); } else { @@ -587,17 +408,11 @@ export const Queue = () => { if (!canDrag || startY === null) return; const touch = e.touches[0]; - const currentTouchY = touch.clientY; - const deltaY = currentTouchY - startY; + const deltaY = touch.clientY - startY; - setCurrentY(currentTouchY); - - // Only handle downward swipes (positive deltaY) if (deltaY > 0) { - // Start dragging if moved more than 10px down if (!isDragging && deltaY > 10) { setIsDragging(true); - // Prevent scrolling when dragging starts e.preventDefault(); } @@ -605,11 +420,10 @@ export const Queue = () => { e.preventDefault(); e.stopPropagation(); - const clampedDelta = Math.min(deltaY, 200); // Max drag distance + const clampedDelta = Math.min(deltaY, 200); setDragDistance(clampedDelta); if (queueRef.current) { - // Apply transform with resistance curve const resistance = Math.pow(clampedDelta / 200, 0.7); const transformY = clampedDelta * resistance; const opacity = Math.max(0.3, 1 - (clampedDelta / 300)); @@ -618,58 +432,28 @@ export const Queue = () => { queueRef.current.style.opacity = `${opacity}`; queueRef.current.style.transition = 'none'; } - - // Add haptic feedback on certain thresholds - if (clampedDelta > 80 && clampedDelta < 85) { - // Light haptic feedback when reaching dismiss threshold - if ('vibrate' in navigator) { - navigator.vibrate(10); - } - } - } - } else { - // If dragging upward, reset drag state - if (isDragging) { - setIsDragging(false); - setDragDistance(0); - if (queueRef.current) { - queueRef.current.style.transform = ''; - queueRef.current.style.opacity = ''; - queueRef.current.style.transition = ''; - } } } }; - const handleTouchEnd = (e: React.TouchEvent) => { - if (!canDrag || startY === null || currentY === null) { + const handleTouchEnd = () => { + if (!canDrag || startY === null) { resetDragState(); return; } - const deltaY = currentY - startY; - const wasScrolling = !isDragging && Math.abs(deltaY) > 0; - if (queueRef.current) { queueRef.current.style.transition = 'transform 0.3s cubic-bezier(0.4, 0, 0.2, 1), opacity 0.3s cubic-bezier(0.4, 0, 0.2, 1)'; - // Dismiss if dragged down more than 80px or with sufficient velocity if (isDragging && dragDistance > 80) { - // Animate out before closing queueRef.current.style.transform = 'translateY(100%)'; queueRef.current.style.opacity = '0'; - // Provide haptic feedback for successful dismiss - if ('vibrate' in navigator) { - navigator.vibrate(20); - } - setTimeout(() => { toggleVisibility(); resetDragState(); }, 300); } else { - // Spring back to original position queueRef.current.style.transform = ''; queueRef.current.style.opacity = ''; @@ -688,86 +472,90 @@ export const Queue = () => { const resetDragState = () => { setStartY(null); - setCurrentY(null); setIsDragging(false); setDragDistance(0); setCanDrag(false); }; - // Handle backdrop click - prevent when dragging - const handleBackdropClick = (e: React.MouseEvent) => { - if (!isDragging) { - toggleVisibility(); - } - }; + // Prevent body scroll on mobile + useEffect(() => { + if (!isVisible) return; + const isMobile = window.innerWidth < 768; + if (!isMobile) return; + + const originalOverflow = document.body.style.overflow; + const originalTouchAction = document.body.style.touchAction; + + document.body.style.overflow = 'hidden'; + document.body.style.touchAction = 'none'; + + return () => { + document.body.style.overflow = originalOverflow; + document.body.style.touchAction = originalTouchAction; + }; + }, [isVisible]); + + if (!context || !isVisible) return null; + + const hasActive = items.some(item => isActiveStatus(getStatus(item))); + const hasFinished = items.some(item => isTerminalStatus(getStatus(item))); + + // Sort items by priority + const sortedItems = [...items].sort((a, b) => { + const statusA = getStatus(a); + const statusB = getStatus(b); + + const getPriority = (status: string) => { + const priorities = { + "real-time": 1, downloading: 2, processing: 3, initializing: 4, + retrying: 5, queued: 6, done: 7, completed: 7, error: 8, cancelled: 9 + }; + return priorities[status as keyof typeof priorities] || 10; + }; + + return getPriority(statusA) - getPriority(statusB); + }); return ( <> - {/* Mobile backdrop overlay - improved isolation */} + {/* Mobile backdrop */}
{ - e.preventDefault(); - e.stopPropagation(); - }} - onTouchMove={(e) => { - e.preventDefault(); - e.stopPropagation(); - }} - onTouchEnd={(e) => { - e.preventDefault(); - e.stopPropagation(); - }} - style={{ - touchAction: 'none', // Prevent all default touch behaviors - overflowY: 'hidden', // Prevent scrolling on backdrop - }} + onClick={!isDragging ? toggleVisibility : undefined} + onTouchStart={(e) => e.preventDefault()} + onTouchMove={(e) => e.preventDefault()} + onTouchEnd={(e) => e.preventDefault()} + style={{ touchAction: 'none', overflowY: 'hidden' }} />
{ - handleTouchStart(e); - // Ensure events don't propagate beyond the queue - e.stopPropagation(); - }} - onTouchMove={(e) => { - handleTouchMove(e); - // Always prevent propagation during move to avoid affecting background - e.stopPropagation(); - }} - onTouchEnd={(e) => { - handleTouchEnd(e); - e.stopPropagation(); - }} + className="fixed inset-x-0 bottom-0 md:bottom-4 md:right-4 md:inset-x-auto w-full md:max-w-md bg-surface dark:bg-surface-dark md:rounded-xl shadow-2xl border-t md:border border-border dark:border-border-dark z-50 backdrop-blur-sm" + onTouchStart={handleTouchStart} + onTouchMove={handleTouchMove} + onTouchEnd={handleTouchEnd} style={{ - touchAction: isDragging ? 'none' : 'auto', // Prevent scrolling when dragging - willChange: isDragging ? 'transform, opacity' : 'auto', // Optimize for animations - isolation: 'isolate', // Create a new stacking context + touchAction: isDragging ? 'none' : 'auto', + willChange: isDragging ? 'transform, opacity' : 'auto', + isolation: 'isolate', }} >
- {/* Enhanced drag indicator for mobile */} -
-

+ {/* Drag indicator for mobile */} +
+ +

Download Queue ({totalTasks})

+
@@ -775,110 +563,70 @@ export const Queue = () => { onClick={clearCompleted} className="text-xs md:text-sm text-content-muted dark:text-content-muted-dark hover:text-success transition-colors px-3 py-2 md:px-2 md:py-1 rounded-md hover:bg-success/10 min-h-[44px] md:min-h-auto" disabled={!hasFinished} - aria-label="Clear all finished downloads" > Clear Finished

+
- {items.length === 0 ? ( -
-
- + {(() => { + const visibleItems = sortedItems.filter(item => !isTerminalStatus(getStatus(item)) || (item.lastCallback && 'timestamp' in item.lastCallback)); + + return visibleItems.length === 0 ? ( +
+
+ +
+

The queue is empty.

+

Downloads will appear here

-

The queue is empty.

-

Downloads will appear here

-
- ) : ( - (() => { - // Sort items by priority hierarchy - const sortedItems = [...items].sort((a, b) => { - // Extract actual status for both items - const statusA = (a.last_line?.status_info?.status as QueueStatus) || - (a.last_line?.status as QueueStatus) || - a.status; - const statusB = (b.last_line?.status_info?.status as QueueStatus) || - (b.last_line?.status as QueueStatus) || - b.status; - - // Define priority groups (lower number = higher priority) - const getPriority = (status: QueueStatus) => { - switch (status) { - case "real-time": return 1; - case "downloading": return 2; - case "processing": return 3; - case "initializing": return 4; - case "retrying": return 5; - case "queued": return 6; - case "pending": return 7; - case "completed": - case "done": return 8; - case "error": return 9; - case "cancelled": return 10; - case "skipped": return 11; - default: return 12; + ) : ( + <> + {visibleItems.map(item => { + if (getStatus(item) === "cancelled") { + return ; } - }; - - const priorityA = getPriority(statusA); - const priorityB = getPriority(statusB); - - // First sort by priority - if (priorityA !== priorityB) { - return priorityA - priorityB; - } - - // Within same priority group, maintain original order (FIFO) - // Assuming items have some sort of timestamp or creation order - return 0; - }); - - return ( - <> - {sortedItems.map((item) => )} - - {/* Loading indicator for infinite scroll */} - {isLoadingMore && ( -
-
-
- Loading more tasks... -
+ return ; + })} + + {/* Loading indicator */} + {isLoadingMore && ( +
+
+
+ Loading more tasks...
- )} - - {/* Load More Button (fallback for manual loading) */} - {hasMore && !isLoadingMore && ( -
- -
- )} - - ); - })() - )} +
+ )} + + {/* Load more button */} + {hasMore && !isLoadingMore && ( +
+ +
+ )} + + ); + })()}
diff --git a/spotizerr-ui/src/contexts/QueueProvider.tsx b/spotizerr-ui/src/contexts/QueueProvider.tsx index d687c44..110a01f 100644 --- a/spotizerr-ui/src/contexts/QueueProvider.tsx +++ b/spotizerr-ui/src/contexts/QueueProvider.tsx @@ -4,340 +4,384 @@ import { QueueContext, type QueueItem, type DownloadType, - type QueueStatus, - isActiveTaskStatus, + getStatus, + isActiveStatus, + isTerminalStatus, } from "./queue-context"; import { toast } from "sonner"; import { v4 as uuidv4 } from "uuid"; -import type { - CallbackObject, - SummaryObject, - ProcessingCallbackObject, - TrackCallbackObject, - AlbumCallbackObject, - PlaylistCallbackObject, -} from "@/types/callbacks"; - -const isTerminalStatus = (status: QueueStatus) => - ["completed", "error", "cancelled", "skipped", "done"].includes(status); - -function isProcessingCallback(obj: CallbackObject): obj is ProcessingCallbackObject { - return obj && "status" in obj && obj.status === "processing"; -} - -function isTrackCallback(obj: any): obj is TrackCallbackObject { - return obj && "track" in obj && "status_info" in obj; -} - -function isAlbumCallback(obj: any): obj is AlbumCallbackObject { - return obj && "album" in obj && "status_info" in obj; -} - -function isPlaylistCallback(obj: any): obj is PlaylistCallbackObject { - return obj && "playlist" in obj && "status_info" in obj; -} +import type { CallbackObject } from "@/types/callbacks"; export function QueueProvider({ children }: { children: ReactNode }) { const [items, setItems] = useState([]); const [isVisible, setIsVisible] = useState(false); - const pollingIntervals = useRef>({}); - const cancelledRemovalTimers = useRef>({}); - - // SSE connection state - const sseConnection = useRef(null); - const isInitialized = useRef(false); - const reconnectTimeoutRef = useRef(null); - const maxReconnectAttempts = 5; - const reconnectAttempts = useRef(0); - - // Pagination state - const [currentPage, setCurrentPage] = useState(1); + const [totalTasks, setTotalTasks] = useState(0); const [hasMore, setHasMore] = useState(true); const [isLoadingMore, setIsLoadingMore] = useState(false); - const [totalTasks, setTotalTasks] = useState(0); - const pageSize = 20; // Number of non-active tasks per page + const [currentPage, setCurrentPage] = useState(1); + + // SSE connection + const sseConnection = useRef(null); + const reconnectTimeoutRef = useRef(null); + const reconnectAttempts = useRef(0); + const maxReconnectAttempts = 5; + const pageSize = 20; + + // Health check for SSE connection + const lastHeartbeat = useRef(Date.now()); + const healthCheckInterval = useRef(null); + + // Auto-removal timers for completed tasks + const removalTimers = useRef>({}); + + // Track pending downloads to prevent duplicates + const pendingDownloads = useRef>(new Set()); - // Calculate active downloads count (active + queued) const activeCount = useMemo(() => { - return items.filter(item => { - // Check for status in both possible locations (nested status_info for real-time, or top-level for others) - const actualStatus = (item.last_line?.status_info?.status as QueueStatus) || - (item.last_line?.status as QueueStatus) || - item.status; - return isActiveTaskStatus(actualStatus); - }).length; + return items.filter(item => isActiveStatus(getStatus(item))).length; }, [items]); - const stopPolling = useCallback((internalId: string) => { - if (pollingIntervals.current[internalId]) { - clearInterval(pollingIntervals.current[internalId]); - delete pollingIntervals.current[internalId]; - } + // Improved deduplication - check both id and taskId fields + const itemExists = useCallback((taskId: string, items: QueueItem[]): boolean => { + return items.some(item => + item.id === taskId || + item.taskId === taskId || + // Also check spotify ID to prevent same track being added multiple times + (item.spotifyId && item.spotifyId === taskId) + ); }, []); - const scheduleCancelledTaskRemoval = useCallback((taskId: string) => { - // Clear any existing timer for this task - if (cancelledRemovalTimers.current[taskId]) { - clearTimeout(cancelledRemovalTimers.current[taskId]); - } - - // Schedule removal after 5 seconds - cancelledRemovalTimers.current[taskId] = window.setTimeout(() => { - setItems(prevItems => prevItems.filter(item => item.id !== taskId)); - delete cancelledRemovalTimers.current[taskId]; - }, 5000); - }, []); - - const updateItemFromPrgs = useCallback((item: QueueItem, prgsData: any): QueueItem => { - const updatedItem: QueueItem = { ...item }; - const { last_line, summary, status, name, artist, download_type } = prgsData; - - if (status) updatedItem.status = status as QueueStatus; - if (summary) updatedItem.summary = summary; - if (name) updatedItem.name = name; - if (artist) updatedItem.artist = artist; - if (download_type) updatedItem.type = download_type; + // Convert SSE task data to QueueItem + const createQueueItemFromTask = useCallback((task: any): QueueItem => { + const spotifyId = task.original_url?.split("/").pop() || ""; - // Preserve the last_line object for progress tracking - if (last_line) updatedItem.last_line = last_line; - - // Check if task is cancelled and schedule removal - const actualStatus = last_line?.status_info?.status || last_line?.status || status; - if (actualStatus === "cancelled") { - scheduleCancelledTaskRemoval(updatedItem.id); - } - - if (last_line) { - if (isProcessingCallback(last_line)) { - updatedItem.status = "processing"; - } else if (isTrackCallback(last_line)) { - const { status_info, track, current_track, total_tracks, parent } = last_line; - updatedItem.currentTrackTitle = track.title; - if (current_track) updatedItem.currentTrackNumber = current_track; - if (total_tracks) updatedItem.totalTracks = total_tracks; - updatedItem.status = (parent && ["done", "skipped"].includes(status_info.status)) ? "downloading" : status_info.status as QueueStatus; - if (status_info.status === "skipped") { - updatedItem.error = status_info.reason; - } else if (status_info.status === "error" || status_info.status === "retrying") { - updatedItem.error = status_info.error; - } - if (!parent && status_info.status === "done" && status_info.summary) updatedItem.summary = status_info.summary; - } else if (isAlbumCallback(last_line)) { - const { status_info, album } = last_line; - updatedItem.status = status_info.status as QueueStatus; - updatedItem.name = album.title; - updatedItem.artist = album.artists.map(a => a.name).join(", "); - updatedItem.totalTracks = album.total_tracks; - if (status_info.status === "done") { - if (status_info.summary) updatedItem.summary = status_info.summary; - updatedItem.currentTrackTitle = undefined; - } else if (status_info.status === "error") { - updatedItem.error = status_info.error; - } - } else if (isPlaylistCallback(last_line)) { - const { status_info, playlist } = last_line; - updatedItem.status = status_info.status as QueueStatus; - updatedItem.name = playlist.title; - updatedItem.playlistOwner = playlist.owner.name; - if (status_info.status === "done") { - if (status_info.summary) updatedItem.summary = status_info.summary; - updatedItem.currentTrackTitle = undefined; - } else if (status_info.status === "error") { - updatedItem.error = status_info.error; - } - } - } - - return updatedItem; - }, [scheduleCancelledTaskRemoval]); - - const startSmartPolling = useCallback(() => { - if (sseConnection.current) return; // Already connected - - console.log("Starting SSE connection"); + // Extract display info from callback + let name = task.name || "Unknown"; + let artist = task.artist || ""; - const connectSSE = () => { + // Handle different callback structures + if (task.last_line) { try { - // Create SSE connection - const eventSource = new EventSource(`/api/prgs/stream?active_only=true`); + if ("track" in task.last_line) { + name = task.last_line.track.title || name; + artist = task.last_line.track.artists?.[0]?.name || artist; + } else if ("album" in task.last_line) { + name = task.last_line.album.title || name; + artist = task.last_line.album.artists?.map((a: any) => a.name).join(", ") || artist; + } else if ("playlist" in task.last_line) { + name = task.last_line.playlist.title || name; + artist = task.last_line.playlist.owner?.name || artist; + } + } catch (error) { + console.warn(`createQueueItemFromTask: Error parsing callback for task ${task.task_id}:`, error); + } + } + + const queueItem: QueueItem = { + id: task.task_id, + taskId: task.task_id, + downloadType: task.download_type || task.type || "track", + spotifyId, + lastCallback: task.last_line as CallbackObject, + name, + artist, + summary: task.summary, + error: task.error, + }; + + // Debug log for status detection issues + const status = getStatus(queueItem); + if (status === "unknown" || !status) { + console.warn(`createQueueItemFromTask: Created item ${task.task_id} with problematic status "${status}", type: ${queueItem.downloadType}`); + } + + return queueItem; + }, []); + + // Schedule auto-removal for completed tasks + const scheduleRemoval = useCallback((taskId: string, delay: number = 10000) => { + if (removalTimers.current[taskId]) { + clearTimeout(removalTimers.current[taskId]); + } + + removalTimers.current[taskId] = window.setTimeout(() => { + setItems(prev => prev.filter(item => item.id !== taskId)); + delete removalTimers.current[taskId]; + }, delay); + }, []); + + // SSE Health Check - detects stuck connections + const startHealthCheck = useCallback(() => { + if (healthCheckInterval.current) return; + + healthCheckInterval.current = window.setInterval(() => { + const timeSinceLastHeartbeat = Date.now() - lastHeartbeat.current; + const maxSilentTime = 60000; // 60 seconds without any message + + if (timeSinceLastHeartbeat > maxSilentTime) { + console.warn(`SSE: No heartbeat for ${timeSinceLastHeartbeat}ms, forcing reconnection`); + disconnectSSE(); + setTimeout(() => connectSSE(), 1000); + } + }, 30000); // Check every 30 seconds + }, []); + + const stopHealthCheck = useCallback(() => { + if (healthCheckInterval.current) { + clearInterval(healthCheckInterval.current); + healthCheckInterval.current = null; + } + }, []); + + // SSE Connection Management + const connectSSE = useCallback(() => { + if (sseConnection.current) return; + + try { + const eventSource = new EventSource("/api/prgs/stream"); sseConnection.current = eventSource; eventSource.onopen = () => { - console.log("SSE connection established"); - reconnectAttempts.current = 0; // Reset reconnect attempts on successful connection + console.log("SSE connected successfully"); + reconnectAttempts.current = 0; + lastHeartbeat.current = Date.now(); + startHealthCheck(); + // Clear any existing reconnect timeout since we're now connected + if (reconnectTimeoutRef.current) { + clearTimeout(reconnectTimeoutRef.current); + reconnectTimeoutRef.current = null; + } }; eventSource.onmessage = (event) => { try { const data = JSON.parse(event.data); - // Handle error events if (data.error) { - console.error("SSE error event:", data.error); - toast.error("Connection error: " + data.error); + console.error("SSE error:", data.error); + toast.error("Connection error"); return; } - const { tasks: updatedTasks, current_timestamp, total_tasks, task_counts } = data; + // Handle different message types from optimized backend + const changeType = data.change_type || "update"; - // Update total tasks count - use active + queued if task_counts available + if (changeType === "heartbeat") { + // Heartbeat - just update counts, no task processing + const { total_tasks, task_counts } = data; const calculatedTotal = task_counts ? (task_counts.active + task_counts.queued) : (total_tasks || 0); setTotalTasks(calculatedTotal); + lastHeartbeat.current = Date.now(); + // Reduce heartbeat logging noise - only log every 10th heartbeat + if (Math.random() < 0.1) { + console.log("SSE: Connection active (heartbeat)"); + } + return; + } + + if (changeType === "error") { + console.error("SSE backend error:", data.error); + return; + } - if (updatedTasks && updatedTasks.length > 0) { - console.log(`SSE: ${updatedTasks.length} tasks updated (${data.active_tasks} active) out of ${data.total_tasks} total`); + // Process actual updates - also counts as heartbeat + lastHeartbeat.current = Date.now(); + const { tasks: updatedTasks, total_tasks, task_counts } = data; + + // Update total count + const calculatedTotal = task_counts ? + (task_counts.active + task_counts.queued) : + (total_tasks || 0); + setTotalTasks(calculatedTotal); + + if (updatedTasks?.length > 0) { + console.log(`SSE: Processing ${updatedTasks.length} task updates`); + + setItems(prev => { + // Create improved deduplication maps + const existingTaskIds = new Set(); + const existingSpotifyIds = new Set(); + const existingItemsMap = new Map(); - // Create a map of updated tasks by task_id for efficient lookup - const updatedTasksMap = new Map(updatedTasks.map((task: any) => [task.task_id, task])); - - setItems(prev => { - // Update existing items with new data, and add any new active tasks - const updatedItems = prev.map(item => { - const updatedTaskData = updatedTasksMap.get(item.taskId || item.id); - if (updatedTaskData) { - return updateItemFromPrgs(item, updatedTaskData); - } - return item; - }); - - // Only add new active tasks that aren't in our current items and aren't in terminal state - const currentTaskIds = new Set(prev.map(item => item.taskId || item.id)); - const newActiveTasks = updatedTasks - .filter((task: any) => { - const isNew = !currentTaskIds.has(task.task_id); - const status = task.last_line?.status_info?.status || task.last_line?.status || "unknown"; - const isActive = isActiveTaskStatus(status); - const isTerminal = ["completed", "error", "cancelled", "skipped", "done"].includes(status); - return isNew && isActive && !isTerminal; - }) - .map((task: any) => { - const spotifyId = task.original_url?.split("/").pop() || ""; - const baseItem: QueueItem = { - id: task.task_id, - taskId: task.task_id, - name: task.name || "Unknown", - type: task.download_type || "track", - spotifyId: spotifyId, - status: "initializing", - artist: task.artist, - }; - return updateItemFromPrgs(baseItem, task); - }); - - return newActiveTasks.length > 0 ? [...newActiveTasks, ...updatedItems] : updatedItems; + prev.forEach(item => { + if (item.id) { + existingTaskIds.add(item.id); + existingItemsMap.set(item.id, item); + } + if (item.taskId) { + existingTaskIds.add(item.taskId); + existingItemsMap.set(item.taskId, item); + } + if (item.spotifyId) existingSpotifyIds.add(item.spotifyId); }); + + // Process each updated task + const processedTaskIds = new Set(); + const updatedItems: QueueItem[] = []; + const newTasksToAdd: QueueItem[] = []; + + for (const task of updatedTasks) { + const taskId = task.task_id; + const spotifyId = task.original_url?.split("/").pop(); + + // Skip if already processed (shouldn't happen but safety check) + if (processedTaskIds.has(taskId)) continue; + processedTaskIds.add(taskId); + + // Check if this task exists in current queue + const existingItem = existingItemsMap.get(taskId) || + Array.from(existingItemsMap.values()).find(item => + item.spotifyId === spotifyId + ); + + if (existingItem) { + // Skip SSE updates for items that are already cancelled by user action + const existingStatus = getStatus(existingItem); + if (existingStatus === "cancelled" && existingItem.error === "Cancelled by user") { + console.log(`SSE: Skipping update for user-cancelled task ${taskId}`); + continue; + } + + // Update existing item + const updatedItem = createQueueItemFromTask(task); + const status = getStatus(updatedItem); + const previousStatus = getStatus(existingItem); + + // Only log significant status changes + if (previousStatus !== status) { + console.log(`SSE: Status change ${taskId}: ${previousStatus} → ${status}`); + } + + // Schedule removal for terminal states + if (isTerminalStatus(status)) { + const delay = status === "cancelled" ? 5000 : 10000; + scheduleRemoval(existingItem.id, delay); + console.log(`SSE: Scheduling removal for terminal task ${taskId} (${status}) in ${delay}ms`); + } + + updatedItems.push(updatedItem); + } else { + // This is a new task from SSE + const newItem = createQueueItemFromTask(task); + const status = getStatus(newItem); + + // Check for duplicates by spotify ID + if (spotifyId && existingSpotifyIds.has(spotifyId)) { + console.log(`SSE: Skipping duplicate by spotify ID: ${spotifyId}`); + continue; + } + + // Check if this is a pending download + if (pendingDownloads.current.has(spotifyId || taskId)) { + console.log(`SSE: Skipping pending download: ${taskId}`); + continue; + } + + // For terminal tasks from SSE, these should be tasks that just transitioned + // (backend now filters out already-terminal tasks) + if (isTerminalStatus(status)) { + console.log(`SSE: Adding recently completed task: ${taskId} (${status})`); + // Schedule immediate removal for terminal tasks + const delay = status === "cancelled" ? 5000 : 10000; + scheduleRemoval(newItem.id, delay); + } else if (isActiveStatus(status)) { + console.log(`SSE: Adding new active task: ${taskId} (${status})`); + } else { + console.warn(`SSE: Skipping task with unknown status: ${taskId} (${status})`); + continue; + } + + newTasksToAdd.push(newItem); + } + } + + // Update existing items that weren't in the update + const finalItems = prev.map(item => { + const updated = updatedItems.find(u => + u.id === item.id || u.taskId === item.id || + u.id === item.taskId || u.taskId === item.taskId + ); + return updated || item; + }); + + // Add new tasks + return newTasksToAdd.length > 0 ? [...newTasksToAdd, ...finalItems] : finalItems; + }); + } else if (changeType === "update") { + // Update received but no tasks - might be count updates only + console.log("SSE: Received update with count changes only"); } } catch (error) { - console.error("Failed to parse SSE message:", error); + console.error("Failed to parse SSE message:", error, event.data); } }; - eventSource.onerror = (error) => { + eventSource.onerror = (error) => { console.error("SSE connection error:", error); - - // Close the connection eventSource.close(); sseConnection.current = null; - // Attempt to reconnect with exponential backoff if (reconnectAttempts.current < maxReconnectAttempts) { reconnectAttempts.current++; - const delay = Math.min(1000 * Math.pow(2, reconnectAttempts.current - 1), 30000); // Max 30 seconds + const delay = Math.min(1000 * Math.pow(2, reconnectAttempts.current - 1), 30000); - console.log(`SSE reconnecting in ${delay}ms (attempt ${reconnectAttempts.current}/${maxReconnectAttempts})`); + console.log(`SSE: Reconnecting in ${delay}ms (attempt ${reconnectAttempts.current}/${maxReconnectAttempts})`); reconnectTimeoutRef.current = window.setTimeout(() => { + console.log("SSE: Attempting to reconnect..."); connectSSE(); }, delay); } else { - console.error("SSE max reconnection attempts reached"); + console.error("SSE: Max reconnection attempts reached"); toast.error("Connection lost. Please refresh the page."); } }; } catch (error) { console.error("Failed to create SSE connection:", error); - toast.error("Failed to establish real-time connection"); + toast.error("Failed to establish connection"); } - }; + }, [createQueueItemFromTask, scheduleRemoval, startHealthCheck]); - connectSSE(); - }, [updateItemFromPrgs]); - - const stopSmartPolling = useCallback(() => { + const disconnectSSE = useCallback(() => { if (sseConnection.current) { - console.log("Closing SSE connection"); sseConnection.current.close(); sseConnection.current = null; } - - // Clear any pending reconnection timeout if (reconnectTimeoutRef.current) { clearTimeout(reconnectTimeoutRef.current); reconnectTimeoutRef.current = null; } - - // Reset reconnection attempts reconnectAttempts.current = 0; - }, []); + stopHealthCheck(); + }, [stopHealthCheck]); + // Load more tasks for pagination const loadMoreTasks = useCallback(async () => { if (!hasMore || isLoadingMore) return; setIsLoadingMore(true); try { const nextPage = currentPage + 1; - const response = await apiClient.get<{ - tasks: any[]; - pagination: { - has_more: boolean; - }; - task_counts?: { - active: number; - queued: number; - retrying: number; - completed: number; - error: number; - cancelled: number; - skipped: number; - }; - }>(`/prgs/list?page=${nextPage}&limit=${pageSize}`); - + const response = await apiClient.get(`/prgs/list?page=${nextPage}&limit=${pageSize}`); const { tasks: newTasks, pagination } = response.data; if (newTasks.length > 0) { - // Add new tasks to the end of the list (avoiding duplicates and filtering out terminal state tasks) setItems(prev => { - const existingTaskIds = new Set(prev.map(item => item.taskId || item.id)); const uniqueNewTasks = newTasks - .filter(task => { - // Skip if already exists - if (existingTaskIds.has(task.task_id)) return false; - - // Filter out terminal state tasks - const status = task.last_line?.status_info?.status || task.last_line?.status || "unknown"; - const isTerminal = ["completed", "error", "cancelled", "skipped", "done"].includes(status); - return !isTerminal; + .filter((task: any) => !itemExists(task.task_id, prev)) + .filter((task: any) => { + const tempItem = createQueueItemFromTask(task); + const status = getStatus(tempItem); + // Consistent filtering - exclude all terminal state tasks in pagination too + return !isTerminalStatus(status); }) - .map(task => { - const spotifyId = task.original_url?.split("/").pop() || ""; - const baseItem: QueueItem = { - id: task.task_id, - taskId: task.task_id, - name: task.name || "Unknown", - type: task.download_type || "track", - spotifyId: spotifyId, - status: "initializing", - artist: task.artist, - }; - return updateItemFromPrgs(baseItem, task); - }); + .map((task: any) => createQueueItemFromTask(task)); return [...prev, ...uniqueNewTasks]; }); - setCurrentPage(nextPage); } @@ -348,306 +392,239 @@ export function QueueProvider({ children }: { children: ReactNode }) { } finally { setIsLoadingMore(false); } - }, [hasMore, isLoadingMore, currentPage, pageSize, updateItemFromPrgs]); - - const startPolling = useCallback( - (taskId: string) => { - // Legacy function - now just ensures SSE connection is active - startSmartPolling(); - }, - [startSmartPolling], - ); + }, [hasMore, isLoadingMore, currentPage, createQueueItemFromTask, itemExists]); + // Initialize queue on mount useEffect(() => { - const fetchQueue = async () => { + const initializeQueue = async () => { try { - console.log("Fetching initial queue with pagination"); - const response = await apiClient.get<{ - tasks: any[]; - pagination: { - has_more: boolean; - }; - total_tasks: number; - timestamp: number; - task_counts?: { - active: number; - queued: number; - retrying: number; - completed: number; - error: number; - cancelled: number; - skipped: number; - }; - }>(`/prgs/list?page=1&limit=${pageSize}`); + const response = await apiClient.get(`/prgs/list?page=1&limit=${pageSize}`); + const { tasks, pagination, total_tasks, task_counts } = response.data; - const { tasks, pagination, total_tasks, timestamp, task_counts } = response.data; - - const backendItems = tasks + const queueItems = tasks .filter((task: any) => { - // Filter out terminal state tasks on initial fetch - const status = task.last_line?.status_info?.status || task.last_line?.status || task.status; - const isTerminal = ["completed", "error", "cancelled", "skipped", "done"].includes(status); - return !isTerminal; + const tempItem = createQueueItemFromTask(task); + const status = getStatus(tempItem); + // On refresh, exclude all terminal state tasks to start with a clean queue + return !isTerminalStatus(status); }) - .map((task: any) => { - const spotifyId = task.original_url?.split("/").pop() || ""; - const baseItem: QueueItem = { - id: task.task_id, - taskId: task.task_id, - name: task.name || "Unknown", - type: task.download_type || "track", - spotifyId: spotifyId, - status: "initializing", - artist: task.artist, - }; - return updateItemFromPrgs(baseItem, task); - }); + .map((task: any) => createQueueItemFromTask(task)); - setItems(backendItems); + console.log(`Queue initialized: ${queueItems.length} items (filtered out terminal state tasks)`); + setItems(queueItems); setHasMore(pagination.has_more); - // Update total tasks count - use active + queued if task_counts available const calculatedTotal = task_counts ? (task_counts.active + task_counts.queued) : (total_tasks || 0); setTotalTasks(calculatedTotal); - // Set initial timestamp to current time - isInitialized.current = true; - - // Start SSE connection for real-time updates - startSmartPolling(); + connectSSE(); } catch (error) { - console.error("Failed to fetch queue from backend:", error); - toast.error("Could not load queue. Please refresh the page."); + console.error("Failed to initialize queue:", error); + toast.error("Could not load queue"); } }; - fetchQueue(); + initializeQueue(); - // Cleanup function to stop SSE connection when component unmounts return () => { - stopSmartPolling(); - // Clean up any remaining individual polling intervals (legacy cleanup) - Object.values(pollingIntervals.current).forEach(clearInterval); - pollingIntervals.current = {}; - // Clean up removal timers - Object.values(cancelledRemovalTimers.current).forEach(clearTimeout); - cancelledRemovalTimers.current = {}; + disconnectSSE(); + stopHealthCheck(); + Object.values(removalTimers.current).forEach(clearTimeout); + removalTimers.current = {}; }; - // eslint-disable-next-line react-hooks/exhaustive-deps - }, []); + }, [connectSSE, disconnectSSE, createQueueItemFromTask, stopHealthCheck]); - const addItem = useCallback( - async (item: { name: string; type: DownloadType; spotifyId: string; artist?: string }) => { - const internalId = uuidv4(); + // Queue actions + const addItem = useCallback(async (item: { name: string; type: DownloadType; spotifyId: string; artist?: string }) => { + // Prevent duplicate downloads + if (pendingDownloads.current.has(item.spotifyId)) { + toast.info("Download already in progress"); + return; + } + + // Check if item already exists in queue + if (itemExists(item.spotifyId, items)) { + toast.info("Item already in queue"); + return; + } + + const tempId = uuidv4(); + pendingDownloads.current.add(item.spotifyId); + const newItem: QueueItem = { - ...item, - id: internalId, - status: "initializing", - }; + id: tempId, + downloadType: item.type, + spotifyId: item.spotifyId, + name: item.name, + artist: item.artist || "", + }; + setItems(prev => [newItem, ...prev]); try { - const response = await apiClient.get<{ task_id: string }>( - `/${item.type}/download/${item.spotifyId}`, - ); + const response = await apiClient.get(`/${item.type}/download/${item.spotifyId}`); const { task_id: taskId } = response.data; setItems(prev => prev.map(i => - i.id === internalId - ? { ...i, id: taskId, taskId, status: "queued" } - : i, - ), - ); - - // Ensure smart polling is active for the new task - startSmartPolling(); + i.id === tempId ? { ...i, id: taskId, taskId } : i + ) + ); + + // Remove from pending after successful API call + pendingDownloads.current.delete(item.spotifyId); + + connectSSE(); // Ensure connection is active } catch (error: any) { - console.error(`Failed to start download for ${item.name}:`, error); + console.error(`Failed to start download:`, error); toast.error(`Failed to start download for ${item.name}`); - setItems(prev => - prev.map(i => - i.id === internalId - ? { - ...i, - status: "error", - error: "Failed to start download task.", - } - : i, - ), - ); - } - }, - [isVisible, startSmartPolling], - ); + + // Remove failed item and clear from pending + setItems(prev => prev.filter(i => i.id !== tempId)); + pendingDownloads.current.delete(item.spotifyId); + } + }, [connectSSE, itemExists, items]); const removeItem = useCallback((id: string) => { const item = items.find(i => i.id === id); - if (item && item.taskId) { - stopPolling(item.taskId); - apiClient.delete(`/prgs/delete/${item.taskId}`).catch(err => { - console.error(`Failed to delete task ${item.taskId} from backend`, err); - // Proceed with frontend removal anyway - }); + if (item?.taskId) { + apiClient.delete(`/prgs/delete/${item.taskId}`).catch(console.error); } setItems(prev => prev.filter(i => i.id !== id)); - }, [items, stopPolling]); + + if (removalTimers.current[id]) { + clearTimeout(removalTimers.current[id]); + delete removalTimers.current[id]; + } + + // Clear from pending downloads if it was pending + if (item?.spotifyId) { + pendingDownloads.current.delete(item.spotifyId); + } + }, [items]); - const cancelItem = useCallback( - async (id: string) => { + const cancelItem = useCallback(async (id: string) => { const item = items.find(i => i.id === id); - if (!item || !item.taskId) return; + if (!item?.taskId) return; try { await apiClient.post(`/prgs/cancel/${item.taskId}`); - stopPolling(item.taskId); - // Immediately update UI to show cancelled status setItems(prev => prev.map(i => - i.id === id - ? { - ...i, - status: "cancelled", - last_line: { - ...i.last_line, - status_info: { - ...i.last_line?.status_info, - status: "cancelled", - error: "Task cancelled by user", - timestamp: Date.now() / 1000, - } - } - } - : i, - ), - ); - - // Schedule removal after 5 seconds - scheduleCancelledTaskRemoval(id); - - toast.info(`Cancelled download: ${item.name}`); + i.id === id ? { + ...i, + error: "Cancelled by user", + lastCallback: { + status: "cancelled", + timestamp: Date.now() / 1000, + type: item.downloadType, + name: item.name, + artist: item.artist + } as unknown as CallbackObject + } : i + ) + ); + + // Remove immediately after showing cancelled state briefly + setTimeout(() => { + setItems(prev => prev.filter(i => i.id !== id)); + // Clean up any existing removal timer + if (removalTimers.current[id]) { + clearTimeout(removalTimers.current[id]); + delete removalTimers.current[id]; + } + }, 500); + + toast.info(`Cancelled: ${item.name}`); } catch (error) { - console.error(`Failed to cancel task ${item.taskId}:`, error); - toast.error(`Failed to cancel download: ${item.name}`); - } - }, - [items, stopPolling, scheduleCancelledTaskRemoval], - ); - - const retryItem = useCallback( - (id: string) => { - const item = items.find((i) => i.id === id); - if (item && item.taskId) { - setItems((prev) => - prev.map((i) => - i.id === id - ? { - ...i, - status: "pending", - error: undefined, - } - : i, - ), - ); - // Ensure smart polling is active for the retry - startSmartPolling(); - toast.info(`Retrying download: ${item.name}`); - } - }, - [items, startSmartPolling], - ); - - const toggleVisibility = useCallback(() => { - setIsVisible((prev) => !prev); - }, []); - - const clearCompleted = useCallback(() => { - setItems((prev) => prev.filter((item) => !isTerminalStatus(item.status) || item.status === "error")); - }, []); + console.error("Failed to cancel task:", error); + toast.error(`Failed to cancel: ${item.name}`); + } + }, [items, scheduleRemoval]); const cancelAll = useCallback(async () => { - const activeItems = items.filter((item) => { - if (!item.taskId) return false; - // Check for status in both possible locations (nested status_info for real-time, or top-level for others) - const actualStatus = (item.last_line?.status_info?.status as QueueStatus) || - (item.last_line?.status as QueueStatus) || - item.status; - return isActiveTaskStatus(actualStatus); + const activeItems = items.filter(item => { + const status = getStatus(item); + return isActiveStatus(status) && item.taskId; }); + if (activeItems.length === 0) { - toast.info("No active downloads to cancel."); + toast.info("No active downloads to cancel"); return; } try { - const taskIds = activeItems.map((item) => item.taskId!); - const response = await apiClient.post("/prgs/cancel/all", { task_ids: taskIds }); + await apiClient.post("/prgs/cancel/all"); - // Get the list of successfully cancelled task IDs from response - const cancelledTaskIds = response.data.task_ids || taskIds; // Fallback to all if response is different - - activeItems.forEach((item) => { - if (cancelledTaskIds.includes(item.taskId)) { - stopPolling(item.taskId!); - } - }); - - // Immediately update UI to show cancelled status for all cancelled tasks - setItems((prev) => - prev.map((item) => - cancelledTaskIds.includes(item.taskId!) - ? { - ...item, + activeItems.forEach(item => { + setItems(prev => + prev.map(i => + i.id === item.id ? { + ...i, + error: "Cancelled by user", + lastCallback: { status: "cancelled", - last_line: { - ...item.last_line, - status_info: { - ...item.last_line?.status_info, - status: "cancelled", - error: "Task cancelled by user", - timestamp: Date.now() / 1000, - } - } - } - : item, - ), - ); - - // Schedule removal for all cancelled tasks - activeItems.forEach((item) => { - if (cancelledTaskIds.includes(item.taskId)) { - scheduleCancelledTaskRemoval(item.id); - } + timestamp: Date.now() / 1000, + type: item.downloadType, + name: item.name, + artist: item.artist + } as unknown as CallbackObject + } : i + ) + ); + // Remove immediately after showing cancelled state briefly + setTimeout(() => { + setItems(prev => prev.filter(i => i.id !== item.id)); + // Clean up any existing removal timer + if (removalTimers.current[item.id]) { + clearTimeout(removalTimers.current[item.id]); + delete removalTimers.current[item.id]; + } + }, 500); }); - - toast.info(`Cancelled ${cancelledTaskIds.length} active downloads.`); + + toast.info(`Cancelled ${activeItems.length} downloads`); } catch (error) { - console.error("Failed to cancel all tasks:", error); - toast.error("Failed to cancel all downloads."); + console.error("Failed to cancel all:", error); + toast.error("Failed to cancel downloads"); } - }, [items, stopPolling, scheduleCancelledTaskRemoval]); + }, [items, scheduleRemoval]); + const clearCompleted = useCallback(() => { + setItems(prev => prev.filter(item => { + const status = getStatus(item); + const shouldKeep = !isTerminalStatus(status) || status === "error"; + + if (!shouldKeep && removalTimers.current[item.id]) { + clearTimeout(removalTimers.current[item.id]); + delete removalTimers.current[item.id]; + } + + return shouldKeep; + })); + }, []); + const toggleVisibility = useCallback(() => { + setIsVisible(prev => !prev); + }, []); const value = { items, isVisible, activeCount, + totalTasks, + hasMore, + isLoadingMore, addItem, removeItem, - retryItem, + cancelItem, toggleVisibility, clearCompleted, cancelAll, - cancelItem, - // Pagination - hasMore, - isLoadingMore, loadMoreTasks, - totalTasks, }; return {children}; diff --git a/spotizerr-ui/src/contexts/queue-context.ts b/spotizerr-ui/src/contexts/queue-context.ts index f443f2d..00d4e5a 100644 --- a/spotizerr-ui/src/contexts/queue-context.ts +++ b/spotizerr-ui/src/contexts/queue-context.ts @@ -1,113 +1,145 @@ import { createContext, useContext } from "react"; -import type { SummaryObject } from "@/types/callbacks"; +import type { SummaryObject, CallbackObject, TrackCallbackObject, AlbumCallbackObject, PlaylistCallbackObject, ProcessingCallbackObject } from "@/types/callbacks"; -export type DownloadType = "track" | "album" | "artist" | "playlist"; -export type QueueStatus = - | "initializing" - | "pending" - | "downloading" - | "processing" - | "completed" - | "error" - | "skipped" - | "cancelled" - | "done" - | "queued" - | "retrying" - | "real-time" - | "progress" - | "track_progress"; +export type DownloadType = "track" | "album" | "playlist"; -// Active task statuses - tasks that are currently working/processing or queued -// This matches the ACTIVE_TASK_STATES constant in the backend plus queued tasks -export const ACTIVE_TASK_STATUSES: Set = new Set([ - "initializing", // task is starting up - "processing", // task is being processed - "downloading", // actively downloading - "progress", // album/playlist progress updates - "track_progress", // real-time track progress - "real-time", // real-time download progress - "retrying", // task is retrying after error - "queued", // task is queued and waiting to start -]); +// Type guards for callback objects +const isProcessingCallback = (obj: CallbackObject): obj is ProcessingCallbackObject => { + return "status" in obj && typeof obj.status === "string"; +}; -/** - * Determine if a task status represents an active (working/processing) task - */ -export function isActiveTaskStatus(status: string): boolean { - return ACTIVE_TASK_STATUSES.has(status as QueueStatus); -} +const isTrackCallback = (obj: CallbackObject): obj is TrackCallbackObject => { + return "track" in obj && "status_info" in obj; +}; +const isAlbumCallback = (obj: CallbackObject): obj is AlbumCallbackObject => { + return "album" in obj && "status_info" in obj; +}; + +const isPlaylistCallback = (obj: CallbackObject): obj is PlaylistCallbackObject => { + return "playlist" in obj && "status_info" in obj; +}; + +// Simplified queue item that works directly with callback objects export interface QueueItem { - id: string; - name: string; - type: DownloadType; - spotifyId: string; - - // Display Info - artist?: string; - albumName?: string; - playlistOwner?: string; - currentTrackTitle?: string; - - // Status and Progress - status: QueueStatus; - taskId?: string; - error?: string; - canRetry?: boolean; - progress?: number; - speed?: string; - size?: string; - eta?: string; - currentTrackNumber?: number; - totalTracks?: number; - summary?: SummaryObject; - - // Real-time download data - last_line?: { - // Direct status and error fields - status?: string; - error?: string; - id?: number; - timestamp?: number; - - // Album/playlist progress fields - current_track?: number; - total_tracks?: number; - parent?: any; // Parent album/playlist information - - // Real-time progress data (when status is "real-time") - status_info?: { - progress?: number; - status?: string; - time_elapsed?: number; - error?: string; - timestamp?: number; - ids?: { - isrc?: string; - spotify?: string; - }; - }; - track?: any; // Contains detailed track information - }; + id: string; + taskId?: string; + downloadType: DownloadType; + spotifyId: string; + + // Current callback data - this is the source of truth + lastCallback?: CallbackObject; + + // Derived display properties (computed from callback) + name: string; + artist: string; + + // Summary data for completed downloads + summary?: SummaryObject; + + // Error state + error?: string; } +// Status extraction utilities +export const getStatus = (item: QueueItem): string => { + if (!item.lastCallback) { + // Only log if this seems problematic (task has been around for a while) + return "initializing"; + } + + if (isProcessingCallback(item.lastCallback)) { + return item.lastCallback.status; + } + + if (isTrackCallback(item.lastCallback)) { + // For parent downloads, if we're getting track callbacks, the parent is "downloading" + if (item.downloadType === "album" || item.downloadType === "playlist") { + return item.lastCallback.status_info.status === "done" ? "downloading" : "downloading"; + } + return item.lastCallback.status_info.status; + } + + if (isAlbumCallback(item.lastCallback)) { + return item.lastCallback.status_info.status; + } + + if (isPlaylistCallback(item.lastCallback)) { + return item.lastCallback.status_info.status; + } + + console.warn(`getStatus: Unknown callback type for item ${item.id}:`, item.lastCallback); + return "unknown"; +}; + +export const isActiveStatus = (status: string): boolean => { + return ["initializing", "processing", "downloading", "real-time", "progress", "track_progress", "retrying", "queued"].includes(status); +}; + +export const isTerminalStatus = (status: string): boolean => { + // Handle both "complete" (backend) and "completed" (frontend) for compatibility + return ["completed", "complete", "done", "error", "cancelled", "skipped"].includes(status); +}; + +// Progress calculation utilities +export const getProgress = (item: QueueItem): number | undefined => { + if (!item.lastCallback) return undefined; + + // For individual tracks + if (item.downloadType === "track" && isTrackCallback(item.lastCallback)) { + if (item.lastCallback.status_info.status === "real-time" && "progress" in item.lastCallback.status_info) { + return item.lastCallback.status_info.progress; + } + return undefined; + } + + // For parent downloads (albums/playlists) - calculate based on track callbacks + if ((item.downloadType === "album" || item.downloadType === "playlist") && isTrackCallback(item.lastCallback)) { + const callback = item.lastCallback; + const currentTrack = callback.current_track || 1; + const totalTracks = callback.total_tracks || 1; + const trackProgress = (callback.status_info.status === "real-time" && "progress" in callback.status_info) + ? callback.status_info.progress : 0; + + // Formula: ((completed tracks) + (current track progress / 100)) / total tracks * 100 + const completedTracks = currentTrack - 1; + return ((completedTracks + (trackProgress / 100)) / totalTracks) * 100; + } + + return undefined; +}; + +// Display info extraction +export const getCurrentTrackInfo = (item: QueueItem): { current?: number; total?: number; title?: string } => { + if (!item.lastCallback) return {}; + + if (isTrackCallback(item.lastCallback)) { + return { + current: item.lastCallback.current_track, + total: item.lastCallback.total_tracks, + title: item.lastCallback.track.title + }; + } + + return {}; +}; + export interface QueueContextType { items: QueueItem[]; isVisible: boolean; activeCount: number; + totalTasks: number; + hasMore: boolean; + isLoadingMore: boolean; + + // Actions addItem: (item: { name: string; type: DownloadType; spotifyId: string; artist?: string }) => void; removeItem: (id: string) => void; - retryItem: (id: string) => void; + cancelItem: (id: string) => void; toggleVisibility: () => void; clearCompleted: () => void; cancelAll: () => void; - cancelItem: (id: string) => void; - // Pagination - hasMore: boolean; - isLoadingMore: boolean; loadMoreTasks: () => void; - totalTasks: number; } export const QueueContext = createContext(undefined);