Merge pull request #342 from arabcoders/dev

Cleaning up the event dispatcher system.

app/library/BackgroundWorker.py

@@ -6,10 +6,16 @@ from queue import Empty, Queue
 
 from .Singleton import Singleton
 
-LOG = logging.getLogger(__name__)
+LOG: logging.Logger = logging.getLogger(__name__)
 
 
 class BackgroundWorker(metaclass=Singleton):
+    """
+    Background worker to run tasks in a separate thread.
+    This class uses a queue to submit tasks that will be executed in the background.
+    It is designed to run in a separate thread and uses asyncio to handle asynchronous tasks.
+    """
+
     _instance = None
     """The instance of the Notification class."""
 
@@ -37,19 +43,18 @@ class BackgroundWorker(metaclass=Singleton):
             except Exception as e:
                 LOG.exception("Loop error: %s", e)
 
-        threading.Thread(target=_loop_runner, daemon=True).start()
+        threading.Thread(target=_loop_runner, daemon=True, name="Background Runner").start()
 
         while self.running:
             try:
                 fn, args, kwargs = self.queue.get(timeout=1)
                 try:
-                    LOG.debug("Running background task: %s", fn.__name__)
                     result = fn(*args, **kwargs)
                     if inspect.iscoroutine(result):
                         loop.call_soon_threadsafe(loop.create_task, result)
                 except Exception as e:
                     LOG.exception(e)
-                    LOG.error(f"Error in background task: {fn.__name__}")
+                    LOG.error(f"Failed to run '{fn.__name__}'. {e!s}")
             except Empty:
                 continue
 

app/library/Download.py

@@ -492,10 +492,10 @@ class Download:
             if "error" == self.info.status and "error" in status:
                 self.info.error = status.get("error")
                 await self._notify.emit(
-                    Events.ERROR,
+                    Events.LOG_ERROR,
-                    data={"message": self.info.error, "data": self.info},
+                    data=self.info,
                     title="Download Error",
-                    message=f"Error in download task '{self.info.title}': {self.info.error}",
+                    message=f"'{self.info.title}' failed to download: {self.info.error}",
                 )
 
             if "downloaded_bytes" in status and status.get("downloaded_bytes") > 0:

app/library/DownloadQueue.py

@@ -19,8 +19,6 @@ from .config import Config
 from .DataStore import DataStore, StoreType
 from .Download import Download
 from .Events import EventBus, Events
-from .Events import info as event_info
-from .Events import warning as event_warning
 from .ItemDTO import Item, ItemDTO
 from .Presets import Presets
 from .Scheduler import Scheduler
@@ -452,39 +450,48 @@ class DownloadQueue(metaclass=Singleton):
 
         try:
             dlInfo: Download = Download(info=dl, info_dict=entry if item.auto_start else None, logs=logs)
+            notifyTitle: str | None = None
+            notifyMessage: str | None = None
 
             text_logs: str = ""
             if filtered_logs := extract_ytdlp_logs(logs):
                 text_logs = f" Logs: {', '.join(filtered_logs)}"
 
             if "is_upcoming" == entry.get("live_status"):
-                NotifyEvent = Events.COMPLETED
+                notifyEvent = Events.COMPLETED
+                notifyTitle = "Upcoming Premiere" if is_premiere else "Upcoming Live Stream"
+                notifyMessage = f"{'Premiere video' if is_premiere else 'Stream' } '{dlInfo.info.title}' is not available yet. {text_logs}"
+
                 dlInfo.info.status = "not_live"
-                dlInfo.info.msg = f"{'Premiere video' if is_premiere else 'Stream' } is not available yet." + text_logs
+                dlInfo.info.msg = notifyMessage.replace(f" '{dlInfo.info.title}'", "")
+
                 await self._notify.emit(
                     Events.LOG_INFO,
-                    data=event_info(dlInfo.info.msg, {"lowPriority": True}),
+                    data={"lowPriority": True},
-                    title="Premiere video" if is_premiere else "Live Stream",
+                    title=notifyTitle,
-                    message=f"Item '{dlInfo.info.title}' is not available yet. {dlInfo.info.msg}",
+                    message=notifyMessage,
                 )
+
                 itemDownload: Download = self.done.put(dlInfo)
             elif len(entry.get("formats", [])) < 1:
-                availability: str = entry.get("availability", "public")
+                ava: str = entry.get("availability", "public")
-                msg: str = "No formats found."
+                notifyTitle = "Download Error"
-                if availability and availability not in ("public",):
+                notifyMessage: str = f"No formats for '{dl.title}'."
-                    msg += f" Availability is set for '{availability}'."
+                if ava and ava not in ("public",):
+                    notifyMessage += f" Availability is set for '{ava}'."
 
-                dlInfo.info.error = msg + text_logs
+                dlInfo.info.error = notifyMessage.replace(f" for '{dl.title}'.", ".") + text_logs
                 dlInfo.info.status = "error"
                 itemDownload = self.done.put(dlInfo)
-                NotifyEvent = Events.COMPLETED
+                notifyEvent = Events.COMPLETED
                 await self._notify.emit(
                     Events.LOG_WARNING,
-                    data=event_warning(f"No formats found for '{dl.title}'."),
+                    data={"logs": text_logs},
-                    title="No Formats Found",
+                    title=notifyTitle,
-                    message=f"No formats found for '{dl.title}'. {dlInfo.info.error}",
+                    message=notifyMessage,
                 )
             elif is_premiere and self.config.prevent_live_premiere:
+                notifyTitle = "Premiere Video"
                 dlInfo.info.error = "Premiering right now."
 
                 _requeue = True
@@ -496,6 +503,7 @@ class DownloadQueue(metaclass=Singleton):
                         )
                         starts_in = starts_in + timedelta(minutes=5, seconds=dl.extras.get("duration", 0))
                         dlInfo.info.error += f" Download will start at {starts_in.astimezone().isoformat()}."
+                        notifyMessage = dlInfo.info.error.strip()
                         _requeue = False
                     except Exception as e:
                         LOG.error(f"Failed to parse live_in date '{release_in}'. {e!s}")
@@ -503,25 +511,24 @@ class DownloadQueue(metaclass=Singleton):
                 else:
                     dlInfo.info.error += f" Delaying download by '{300+dl.extras.get('duration',0)}' seconds."
 
+                notifyMessage = dlInfo.info.error.strip()
+
                 if _requeue:
-                    NotifyEvent = Events.ADDED
+                    notifyEvent = Events.ADDED
                     itemDownload = self.queue.put(dlInfo)
                     if item.auto_start:
                         self.event.set()
-                    else:
-                        LOG.debug(f"Item {itemDownload.info.name()} is not set to auto-start.")
                 else:
                     dlInfo.info.status = "not_live"
                     itemDownload = self.done.put(dlInfo)
-                    NotifyEvent = Events.COMPLETED
+                    notifyEvent = Events.COMPLETED
-                    await self._notify.emit(
+                    notifyTitle = "Item Not Live"
-                        Events.LOG_INFO,
+                    notifyMessage = f"Item '{dlInfo.info.title}' is not live."
-                        data=event_info(f"'{dl.title}' is {dlInfo.info.error}.", {"lowPriority": True}),
+                    await self._notify.emit(Events.LOG_INFO, title=notifyTitle, message=notifyMessage)
-                        title="Item Not Live",
-                        message=f"Item '{dl.title}' is not live. {dlInfo.info.error}",
-                    )
             else:
-                NotifyEvent = Events.ADDED
+                notifyEvent = Events.ADDED
+                notifyTitle = "Item Added"
+                notifyMessage = f"Item '{dlInfo.info.title}' has been added to the download queue."
                 itemDownload = self.queue.put(dlInfo)
                 if item.auto_start:
                     self.event.set()
@@ -529,10 +536,10 @@ class DownloadQueue(metaclass=Singleton):
                     LOG.debug(f"Item {itemDownload.info.name()} is not set to auto-start.")
 
             await self._notify.emit(
-                NotifyEvent,
+                notifyEvent,
                 data=itemDownload.info.serialize(),
-                title="Item Added" if Events.ADDED == NotifyEvent else None,
+                title=notifyTitle,
-                message=f"Added '{itemDownload.info.title}'." if NotifyEvent == Events.ADDED else None,
+                message=notifyMessage,
             )
 
             return {"status": "ok"}
@@ -740,14 +747,18 @@ class DownloadQueue(metaclass=Singleton):
                 self.queue.delete(id)
                 await self._notify.emit(
                     Events.CANCELLED,
-                    data=item.info.serialize(),
+                    data=item.info,
                     title="Download Cancelled",
                     message=f"Download '{item.info.title}' has been cancelled.",
                 )
                 item.info.status = "cancelled"
-                # item.info.error = "Cancelled by user."
                 self.done.put(item)
-                await self._notify.emit(Events.COMPLETED, data=item.info.serialize())
+                await self._notify.emit(
+                    Events.COMPLETED,
+                    data=item.info,
+                    title="Download Cancelled",
+                    message=f"Download '{item.info.title}' has been cancelled.",
+                )
                 LOG.info(f"Deleted from queue {item_ref}")
 
             status[id] = "ok"
@@ -815,11 +826,13 @@ class DownloadQueue(metaclass=Singleton):
                     LOG.error(f"Unable to remove '{itemRef}' local file '{filename}'. {e!s}")
 
             self.done.delete(id)
+
+            _status: str = "Removed" if removed_files > 0 else "Cleared"
             await self._notify.emit(
                 Events.CLEARED,
-                data=item.info.serialize(),
+                data=item.info,
-                title="Download Cleared",
+                title=f"Download {_status}",
-                message=f"Cleared download '{item.info.title}' from history.",
+                message=f"{_status} '{item.info.title}' from history.",
             )
 
             msg = f"Deleted completed download '{itemRef}'."
@@ -857,6 +870,27 @@ class DownloadQueue(metaclass=Singleton):
 
         return items
 
+    def get_item(self, id: str) -> Download | None:
+        """
+        Get a specific item from the download queue or history.
+
+        Args:
+            id (str): The ID of the item to retrieve.
+
+        Returns:
+            Download | None: The requested item if found, otherwise None.
+
+        """
+        try:
+            return self.queue.get(key=id)
+        except KeyError:
+            pass
+
+        try:
+            return self.done.get(key=id)
+        except KeyError:
+            return None
+
     async def _download_pool(self) -> None:
         """
         Create a pool of workers to download the files.
@@ -931,25 +965,25 @@ class DownloadQueue(metaclass=Singleton):
             LOG.debug(f"Download Task '{id}' is completed. Removing from queue.")
             self.queue.delete(key=id)
 
+            notifyTitle: str | None = None
+            notifyMessage: str | None = None
+
             if entry.is_cancelled() is True:
                 await self._notify.emit(
                     Events.CANCELLED,
-                    data=entry.info.serialize(),
+                    data=entry.info,
                     title="Download Cancelled",
                     message=f"Download '{entry.info.title}' has been cancelled.",
                 )
                 entry.info.status = "cancelled"
-                # entry.info.error = "Cancelled by user."
+                notifyTitle = "Download Cancelled"
+                notifyMessage = f"Download '{entry.info.title}' has been cancelled."
+            elif entry.info.status == "finished":
+                notifyTitle = "Download Completed"
+                notifyMessage = f"Download '{entry.info.title}' has been completed."
 
             self.done.put(value=entry)
-            await self._notify.emit(
+            await self._notify.emit(Events.COMPLETED, data=entry.info, title=notifyTitle, message=notifyMessage)
-                Events.COMPLETED,
-                data=entry.info.serialize(),
-                title="Download Completed" if entry.info.status == "finished" else None,
-                message=f"Download '{entry.info.title}' has been completed."
-                if entry.info.status == "finished"
-                else None,
-            )
         else:
             LOG.warning(f"Download '{id}' not found in queue.")
 

app/library/Events.py

@@ -6,85 +6,10 @@ from collections.abc import Awaitable
 from dataclasses import dataclass, field
 from typing import Any
 
+from .BackgroundWorker import BackgroundWorker
 from .Singleton import Singleton
 
-LOG = logging.getLogger("Events")
+LOG: logging.Logger = logging.getLogger("Events")
-
-
-def error(msg: str, data: dict | None = None) -> dict:
-    """
-    Create an error message.
-
-    Args:
-        msg (str): The message.
-        data (dict|None): The data to include in the message.
-
-    Returns:
-        dict : The message wrapped in a dictionary.
-
-    """
-    return message("error", msg, data)
-
-
-def warning(msg: str, data: dict | None = None) -> dict:
-    """
-    Create an error message.
-
-    Args:
-        msg (str): The message.
-        data (dict|None): The data to include in the message.
-
-    Returns:
-        dict : The message wrapped in a dictionary.
-
-    """
-    return message("warning", msg, data)
-
-
-def info(msg: str, data: dict | None = None) -> dict:
-    """
-    Create an info message.
-
-    Args:
-        msg (str): The message.
-        data (dict|None): The data to include in the message.
-
-    Returns:
-        dict : The message wrapped in a dictionary.
-
-    """
-    return message("info", msg, data)
-
-
-def success(msg: str, data: dict | None = None) -> dict:
-    """
-    Create a success message.
-
-    Args:
-        msg (str): The message.
-        data (dict|None): The data to include in the message.
-
-    Returns:
-        dict : The message wrapped in a dictionary.
-
-    """
-    return message("success", msg, data)
-
-
-def message(type: str, message: str, data: dict | None = None) -> dict:
-    """
-    Create a message.
-
-    Args:
-        type (str): The type of the message.
-        message (str): The message.
-        data (dict|None): The data to include in the message.
-
-    Returns:
-        dict : The message wrapped in a dictionary.
-
-    """
-    return {"type": type, "message": message, "data": data if data else {}}
 
 
 class Events:
@@ -98,30 +23,32 @@ class Events:
     SHUTDOWN = "shutdown"
 
     ADDED = "added"
+    UPDATE = "update"
     UPDATED = "updated"
     COMPLETED = "completed"
     CANCELLED = "cancelled"
     CLEARED = "cleared"
-    ERROR = "error"
+    CONNECTED = "connected"
+    STATUS = "status"
+
     LOG_INFO = "log_info"
     LOG_WARNING = "log_warning"
     LOG_ERROR = "log_error"
     LOG_SUCCESS = "log_success"
 
-    INITIAL_DATA = "initial_data"
     ITEM_DELETE = "item_delete"
     ITEM_CANCEL = "item_cancel"
     ITEM_ERROR = "item_error"
-    STATUS = "status"
+
-    CLI_CLOSE = "cli_close"
-    CLI_OUTPUT = "cli_output"
-    UPDATE = "update"
     TEST = "test"
     ADD_URL = "add_url"
 
-    CLI_POST = "cli_post"
     PAUSED = "paused"
 
+    CLI_POST = "cli_post"
+    CLI_CLOSE = "cli_close"
+    CLI_OUTPUT = "cli_output"
+
     TASKS_ADD = "task_add"
     TASK_DISPATCHED = "task_dispatched"
     TASK_FINISHED = "task_finished"
@@ -129,6 +56,7 @@ class Events:
 
     PRESETS_ADD = "presets_add"
     PRESETS_UPDATE = "presets_update"
+
     SCHEDULE_ADD = "schedule_add"
 
     CONDITIONS_ADD = "conditions_add"
@@ -158,9 +86,8 @@ class Events:
 
         """
         return [
-            Events.INITIAL_DATA,
+            Events.CONNECTED,
             Events.ADDED,
-            Events.ERROR,
             Events.LOG_INFO,
             Events.LOG_WARNING,
             Events.LOG_ERROR,
@@ -185,7 +112,7 @@ class Events:
             list: The list of debug events.
 
         """
-        return [Events.UPDATED]
+        return [Events.UPDATED, Events.CLI_OUTPUT]
 
 
 @dataclass(kw_only=True)
@@ -209,7 +136,7 @@ class Event:
     message: str | None = None
     """The message of the event, if any."""
 
-    data: any
+    data: Any
     """The data that was passed to the event."""
 
     def serialize(self) -> dict:
@@ -277,12 +204,16 @@ class EventBus(metaclass=Singleton):
     debug: bool = False
     """Whether to log debug messages or not."""
 
+    _offload: BackgroundWorker
+    """The background worker to offload tasks to."""
+
     def __init__(self):
         EventBus._instance = self
 
         from .config import Config
 
         self.debug = Config.get_instance().debug
+        self._offload = BackgroundWorker.get_instance()
 
     @staticmethod
     def get_instance() -> "EventBus":
@@ -368,13 +299,24 @@ class EventBus(metaclass=Singleton):
 
         return self
 
-    def sync_emit(self, event: str, data: Any, loop=None, wait: bool = True, **kwargs):
+    def sync_emit(
+        self,
+        event: str,
+        data: Any | None = None,
+        title: str | None = None,
+        message: str | None = None,
+        loop=None,
+        wait: bool = True,
+        **kwargs,
+    ):
         """
         Emit event and (optionally) wait for results.
 
         Args:
             event (str): The event to emit.
-            data (Any): The data to pass to the event.
+            data (Any|None): The data to pass to the event.
+            title (str | None): The title of the event, if any.
+            message (str | None): The message of the event, if any.
             loop (asyncio.AbstractEventLoop | None): The event loop to use. If None, the current running loop is used.
             wait (bool): Whether to wait for the results of the event handlers. Defaults to True.
             **kwargs: Additional keyword arguments to pass to the event
@@ -389,8 +331,11 @@ class EventBus(metaclass=Singleton):
         if event not in self._listeners:
             return []
 
+        if not data:
+            data = {}
+
         async def emit_all():
-            ev = Event(event=event, data=data)
+            ev = Event(event=event, title=title, message=message, data=data)
             LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data})
 
             res: list = []
@@ -428,14 +373,14 @@ class EventBus(metaclass=Singleton):
         return fut.result() if wait else fut
 
     async def emit(
-        self, event: str, data: Any, title: str | None = None, message: str | None = None, **kwargs
+        self, event: str, data: Any | None = None, title: str | None = None, message: str | None = None, **kwargs
     ) -> Awaitable:
         """
         Emit an event.
 
         Args:
             event (str): The event to emit.
-            data (Any): The data to pass to the event.
+            data (Any|None): The data to pass to the event.
             title (str | None): The title of the event, if any.
             message (str | None): The message of the event, if any.
             **kwargs: The keyword arguments to pass to the event.
@@ -447,7 +392,10 @@ class EventBus(metaclass=Singleton):
         if event not in self._listeners:
             return []
 
-        ev = Event(event=event, data=data, title=title, message=message)
+        if not data:
+            data = {}
+
+        ev = Event(event=event, title=title, message=message, data=data)
 
         if self.debug or event not in Events.only_debug():
             LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data})
@@ -461,3 +409,35 @@ class EventBus(metaclass=Singleton):
                 LOG.error(f"Failed to emit event '{ev.event}' to '{handler.name}'. Error message '{e!s}'.")
 
         return asyncio.gather(*tasks)
+
+    def offload(
+        self, event: str, title: str | None = None, message: str | None = None, data: Any | None = None, **kwargs
+    ) -> None:
+        """
+        Offload an dispatching event to a background worker.
+
+        Args:
+            event (str): The event to offload.
+            data (Any|None): The data to pass to the event.
+            title (str | None): The title of the event, if any.
+            message (str | None): The message of the event, if any.
+            **kwargs: Additional keyword arguments to pass to the event.
+
+        """
+        if event not in self._listeners:
+            return
+
+        if not data:
+            data = {}
+
+        ev = Event(event=event, title=title, message=message, data=data)
+
+        if self.debug or event not in Events.only_debug():
+            LOG.debug(f"Offloading event '{ev.id}: {ev.event}'.", extra={"data": data})
+
+        for handler in self._listeners[event].values():
+            try:
+                self._offload.submit(handler.handle, ev, **kwargs)
+            except Exception as e:
+                LOG.exception(e)
+                LOG.error(f"Failed to offload event '{ev.event}' to '{handler.name}'. Error message '{e!s}'.")

app/library/HttpSocket.py

@@ -55,7 +55,7 @@ class HttpSocket:
         self.rootPath = root_path
 
         def emit(e: Event, _, **kwargs):
-            return self.sio.emit(event=e.event, data=encoder.encode(e.data), **kwargs)
+            return self.sio.emit(event=e.event, data=encoder.encode(e), **kwargs)
 
         services = Services.get_instance()
         services.add_all(

app/library/Notifications.py

@@ -92,7 +92,6 @@ class Target:
 class NotificationEvents:
     ADDED = Events.ADDED
     COMPLETED = Events.COMPLETED
-    ERROR = Events.ERROR
     CANCELLED = Events.CANCELLED
     CLEARED = Events.CLEARED
     LOG_INFO = Events.LOG_INFO
@@ -308,10 +307,17 @@ class Notification(metaclass=Singleton):
                 msg = "Invalid notification target. Invalid 'on' event list found."
                 raise ValueError(msg)
 
+            removed_events = []
+            all_events = NotificationEvents.get_events().values()
             for e in target["on"]:
-                if e not in NotificationEvents.get_events().values():
+                if e not in all_events:
-                    msg = f"Invalid notification target. Invalid event '{e}' found."
+                    removed_events.append(e)
-                    raise ValueError(msg)
+                    target["on"].remove(e)
+                    continue
+
+            if len(removed_events) > 0 and len(target["on"]) < 1:
+                msg: str = f"Invalid notification target. Invalid events '{', '.join(removed_events)}' found."
+                raise ValueError(msg)
 
         if "headers" in target["request"]:
             if not isinstance(target["request"]["headers"], list):

app/library/Tasks.py

@@ -15,7 +15,7 @@ from app.library.Services import Services
 
 from .config import Config
 from .encoder import Encoder
-from .Events import EventBus, Events, error, success
+from .Events import EventBus, Events
 from .Scheduler import Scheduler
 from .Singleton import Singleton
 from .Utils import init_class, validate_url
@@ -325,7 +325,7 @@ class Tasks(metaclass=Singleton):
                     "template": template,
                     "cli": cli,
                 },
-                title=f"Task '{task.name}' started",
+                title="Tasks",
                 message=f"Task '{task.name}' started at '{timeNow}'",
                 id=task.id,
             )
@@ -337,19 +337,16 @@ class Tasks(metaclass=Singleton):
 
             await self._notify.emit(
                 Events.LOG_SUCCESS,
-                data=success(
+                data={"lowPriority": True},
-                    f"Task '{task.name}' completed in '{ended - started:.2f}' seconds.", data={"lowPriority": True}
+                title="Task completed",
-                ),
+                message=f"Task '{task.name}' completed in '{ended - started:.2f}'.",
-                title=f"Task '{task.name}' completed",
-                message=f"Task '{task.name}' completed at '{timeNow}'",
             )
         except Exception as e:
             LOG.error(f"Failed to execute '{task.name}' at '{timeNow}'. '{e!s}'.")
             await self._notify.emit(
-                Events.ERROR,
+                Events.LOG_ERROR,
-                data=error(f"Failed to execute '{task.name}' at '{timeNow}'. '{e!s}'."),
+                title="Task failed",
-                title=f"Task '{task.name}' failed",
+                message=f"Failed to execute '{task.name}'. '{e!s}'",
-                message=str(e),
             )
 
 

app/main.py

@@ -111,7 +111,12 @@ class Main:
         host = host or self._config.host
         port = port or self._config.port
 
-        EventBus.get_instance().sync_emit(Events.STARTUP, data={"app": self._app})
+        EventBus.get_instance().sync_emit(
+            Events.STARTUP,
+            data={"app": self._app},
+            title="Application Startup",
+            message="The application is starting up.",
+        )
         Scheduler.get_instance().attach(self._app)
 
         self._socket.attach(self._app)
@@ -123,7 +128,12 @@ class Main:
         Notification.get_instance().attach(self._app)
         Conditions.get_instance().attach(self._app)
 
-        EventBus.get_instance().sync_emit(Events.LOADED, data={"app": self._app})
+        EventBus.get_instance().sync_emit(
+            Events.LOADED,
+            data={"app": self._app},
+            title="Application Loaded",
+            message="The application has loaded all components.",
+        )
 
         def started(_):
             LOG.info("=" * 40)
@@ -132,7 +142,14 @@ class Main:
 
             loop = asyncio.get_event_loop()
 
-            EventBus.get_instance().sync_emit(Events.STARTED, data={"app": self._app}, loop=loop, wait=False)
+            EventBus.get_instance().sync_emit(
+                Events.STARTED,
+                data={"app": self._app},
+                title="Application Started",
+                message="The application has started successfully.",
+                loop=loop,
+                wait=False,
+            )
 
             if loop and self._config.debug:
                 loop.set_debug(True)

app/routes/api/notifications.py

@@ -5,7 +5,7 @@ from aiohttp import web
 from aiohttp.web import Request, Response
 
 from app.library.encoder import Encoder
-from app.library.Events import EventBus, Events, message
+from app.library.Events import EventBus, Events
 from app.library.Notifications import Notification, NotificationEvents
 from app.library.router import route
 from app.library.Utils import validate_uuid
@@ -104,8 +104,6 @@ async def notification_test(encoder: Encoder, notify: EventBus) -> Response:
         Response: The response object.
 
     """
-    data = message("test", "This is a test notification.")
+    await notify.emit(Events.TEST, title="Test Notification", message="This is a test notification.")
 
-    await notify.emit(Events.TEST, data=data, title="Test Notification", message="This is a test notification.")
+    return web.json_response(data={}, status=web.HTTPOk.status_code, dumps=encoder.encode)
-
-    return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=encoder.encode)

app/routes/socket/connection.py

@@ -1,12 +1,13 @@
 import asyncio
 import logging
 from pathlib import Path
+from typing import Any
 
 import socketio
 
 from app.library.config import Config
 from app.library.DownloadQueue import DownloadQueue
-from app.library.Events import EventBus, Events, error
+from app.library.Events import EventBus, Events
 from app.library.Presets import Presets
 from app.library.router import RouteType, route
 from app.library.Utils import tail_log
@@ -30,7 +31,13 @@ async def connect(config: Config, queue: DownloadQueue, notify: EventBus, sid: s
 
     data["folders"] = [folder.name for folder in Path(config.download_path).iterdir() if folder.is_dir()]
 
-    await notify.emit(Events.INITIAL_DATA, data=data, to=sid)
+    await notify.emit(
+        Events.CONNECTED,
+        data=data,
+        title="Client connected",
+        message=f"Client '{sid}' connected.",
+        to=sid,
+    )
 
 
 @route(RouteType.SOCKET, "disconnect", "socket_disconnect")
@@ -52,7 +59,7 @@ async def disconnect(sio: socketio.AsyncServer, sid: str, data: str = None):
 
 
 @route(RouteType.SOCKET, "subscribe", "socket_subscribe")
-async def subscribe(config: Config, notify: EventBus, sio: socketio.AsyncServer, sid: str, data: str):
+async def subscribe(config: Config, notify: EventBus, sio: socketio.AsyncServer, sid: str, data: str | Any):
     """
     Subscribe to a specific event.
 
@@ -64,8 +71,13 @@ async def subscribe(config: Config, notify: EventBus, sio: socketio.AsyncServer,
         data (str): The event to subscribe to.
 
     """
-    if not isinstance(data, str):
+    if not isinstance(data, str) or not data:
-        await notify.emit(Events.ERROR, data=error("Invalid event."), to=sid)
+        await notify.emit(
+            Events.LOG_ERROR,
+            title="Subscription Error",
+            message="Invalid event type was expecting a string.",
+            to=sid,
+        )
         return
 
     if data not in _Data.subscribers:

app/routes/socket/history.py

@@ -7,7 +7,7 @@ import anyio
 
 from app.library.config import Config
 from app.library.DownloadQueue import DownloadQueue
-from app.library.Events import EventBus, Events, error
+from app.library.Events import EventBus, Events
 from app.library.ItemDTO import Item
 from app.library.router import RouteType, route
 from app.library.Utils import is_downloaded
@@ -43,43 +43,77 @@ async def add_url(queue: DownloadQueue, notify: EventBus, sid: str, data: dict):
     url: str | None = data.get("url")
 
     if not url:
-        await notify.emit(Events.ERROR, data=error("No URL provided.", data={"unlock": True}), to=sid)
+        await notify.emit(
+            Events.LOG_ERROR,
+            title="Invalid URL",
+            message="Please provide a valid URL to add to the download queue.",
+            to=sid,
+        )
         return
 
     try:
+        status = await queue.add(item=Item.format(data))
         await notify.emit(
             event=Events.STATUS,
-            data=await queue.add(item=Item.format(data)),
+            title="Adding URL",
+            message=f"Adding URL '{url}' to the download queue.",
+            data=status,
             to=sid,
         )
     except ValueError as e:
         LOG.exception(e)
-        await notify.emit(Events.ERROR, data=error(str(e)), to=sid)
+        await notify.emit(Events.LOG_ERROR, title="Error Adding URL", message=str(e), to=sid)
-        return
 
 
 @route(RouteType.SOCKET, "item_cancel", "item_cancel")
 async def item_cancel(queue: DownloadQueue, notify: EventBus, sid: str, data: str):
     if not data:
-        await notify.emit(Events.ERROR, data=error("Invalid request."), to=sid)
+        await notify.emit(
+            Events.LOG_ERROR,
+            title="Invalid Request",
+            message="No item ID provided to cancel.",
+            to=sid,
+        )
         return
 
     status: dict[str, str] = {}
     status = await queue.cancel([data])
     status.update({"identifier": data})
 
-    await notify.emit(Events.ITEM_CANCEL, data=status)
+    await notify.emit(
+        Events.ITEM_CANCEL, data=status, title="Item Cancelled", message=f"Item '{data}': has been cancelled."
+    )
 
 
 @route(RouteType.SOCKET, "item_delete", "item_delete")
 async def item_delete(queue: DownloadQueue, notify: EventBus, sid: str, data: dict):
     if not data:
-        await notify.emit(Events.ERROR, data=error("Invalid request."), to=sid)
+        await notify.emit(
+            Events.LOG_ERROR,
+            title="Invalid Request",
+            message="No item ID provided to delete.",
+            to=sid,
+        )
         return
 
     id: str | None = data.get("id")
     if not id:
-        await notify.emit(Events.ERROR, data=error("Invalid request."), to=sid)
+        await notify.emit(
+            Events.LOG_ERROR,
+            title="Invalid Request",
+            message="No item ID provided to delete.",
+            to=sid,
+        )
+        return
+
+    item = queue.get_item(id=id)
+    if not item:
+        await notify.emit(
+            Events.LOG_ERROR,
+            title="Item Not Found",
+            message=f"Item with ID '{id}' not found.",
+            to=sid,
+        )
         return
 
     status: dict[str, str] = {}
@@ -87,7 +121,7 @@ async def item_delete(queue: DownloadQueue, notify: EventBus, sid: str, data: di
     status.update({"identifier": id})
 
     await notify.emit(
-        Events.ITEM_DELETE, data=status, title="Item Deleted", message=f"Item with ID '{id}' has been deleted."
+        Events.ITEM_DELETE, data=status, title="Item Deleted", message=f"Item '{item.info.title}': has been deleted."
     )
 
 
@@ -145,7 +179,12 @@ async def archive_item(config: Config, data: dict):
 @route(RouteType.SOCKET, "item_start", "item_start")
 async def item_start(queue: DownloadQueue, notify: EventBus, sid: str, data: list | str) -> None:
     if not data:
-        await notify.emit(Events.ERROR, data=error("Invalid request."), to=sid)
+        await notify.emit(
+            Events.LOG_ERROR,
+            title="Invalid Request",
+            message="No items provided to start.",
+            to=sid,
+        )
         return
 
     if isinstance(data, str):
@@ -157,7 +196,12 @@ async def item_start(queue: DownloadQueue, notify: EventBus, sid: str, data: lis
 @route(RouteType.SOCKET, "item_pause", "item_pause")
 async def item_pause(queue: DownloadQueue, notify: EventBus, sid: str, data: list | str) -> None:
     if not data:
-        await notify.emit(Events.ERROR, data=error("Invalid request."), to=sid)
+        await notify.emit(
+            Events.LOG_ERROR,
+            title="Invalid Request",
+            message="No items provided to pause.",
+            to=sid,
+        )
         return
 
     if isinstance(data, str):

app/routes/socket/terminal.py

@@ -3,7 +3,7 @@ import os
 from typing import TYPE_CHECKING
 
 from app.library.config import Config
-from app.library.Events import EventBus, Events, error
+from app.library.Events import EventBus, Events
 from app.library.router import RouteType, route
 
 if TYPE_CHECKING:
@@ -17,7 +17,12 @@ LOG: logging.Logger = logging.getLogger(__name__)
 @route(RouteType.SOCKET, "cli_post", "socket_cli_post")
 async def cli_post(config: Config, notify: EventBus, sid: str, data: str):
     if not config.console_enabled:
-        await notify.emit(Events.ERROR, data=error("Console is disabled."), to=sid)
+        await notify.emit(
+            Events.LOG_ERROR,
+            title="Feature disabled",
+            message="Console feature is disabled.",
+            to=sid,
+        )
         return
 
     if not data:
@@ -29,6 +34,7 @@ async def cli_post(config: Config, notify: EventBus, sid: str, data: str):
     import shlex
     import subprocess  # ignore
 
+    returncode: int = -1
     try:
         LOG.info(f"Cli command from client '{sid}'. '{data}'")
 
@@ -36,16 +42,22 @@ async def cli_post(config: Config, notify: EventBus, sid: str, data: str):
         _env: dict[str, str] = os.environ.copy()
         _env.update(
             {
-                "TERM": "xterm-256color",
-                "LANG": "en_US.UTF-8",
-                "SHELL": "/bin/bash",
-                "LC_ALL": "en_US.UTF-8",
                 "PWD": config.download_path,
                 "FORCE_COLOR": "1",
                 "PYTHONUNBUFFERED": "1",
             }
         )
 
+        if "nt" != os.name:
+            _env.update(
+                {
+                    "TERM": "xterm-256color",
+                    "LANG": "en_US.UTF-8",
+                    "LC_ALL": "en_US.UTF-8",
+                    "SHELL": "/bin/bash",
+                }
+            )
+
         try:
             import pty
 
@@ -126,10 +138,9 @@ async def cli_post(config: Config, notify: EventBus, sid: str, data: str):
         returncode: int = await proc.wait()
 
         await read_task
-
-        await notify.emit(Events.CLI_CLOSE, data={"exitcode": returncode}, to=sid)
     except Exception as e:
         LOG.error(f"CLI execute exception was thrown for client '{sid}'.")
         LOG.exception(e)
         await notify.emit(Events.CLI_OUTPUT, data={"type": "stderr", "line": str(e)}, to=sid)
-        await notify.emit(Events.CLI_CLOSE, data={"exitcode": -1}, to=sid)
+    finally:
+        await notify.emit(Events.CLI_CLOSE, data={"exitcode": returncode}, to=sid)

ui/assets/css/style.css

@@ -336,3 +336,7 @@ hr {
 .fa-spin-10 {
   --fa-animation-iteration-count: 10;
 }
+
+.Vue-Toastification__toast-body {
+  user-select: none;
+}

ui/components/ConditionForm.vue

@@ -209,7 +209,7 @@
 <script setup lang="ts">
 import { useStorage } from '@vueuse/core'
 import { decode } from '~/utils/importer'
-import type { ConditionItem, ImportedConditionItem } from '~/@types/conditions'
+import type { ConditionItem, ImportedConditionItem } from '~/types/conditions'
 
 const emitter = defineEmits<{
   (e: 'cancel'): void

ui/components/ConfirmDialog.vue

@@ -23,7 +23,7 @@
       <footer class="modal-card-foot p-5">
         <div class="field is-grouped" style="width:100%">
           <div class="control is-expanded">
-            <button class="button is-fullwidth" :class="confirm_button_color" @click="handleConfirm">
+            <button ref="confirmBtn" class="button is-fullwidth" :class="confirm_button_color" @click="handleConfirm">
               {{ confirm_button_label }}
             </button>
           </div>
@@ -83,21 +83,25 @@ const emit = defineEmits<{
   (e: 'cancel'): void
 }>()
 
+const confirmBtn = ref<HTMLButtonElement | null>(null)
 const selected = reactive<Record<string, boolean>>({})
 
-watch(() => props.visible, visible => {
+watch(() => props.visible, async visible => {
-  if (visible && props.options) {
+  if (!visible) {
+    return
+  }
+
+  if (props.options) {
     for (const opt of props.options) {
       selected[opt.key] ??= false
     }
   }
-})
 
-function handleConfirm() {
+  await nextTick()
-  emit('confirm', { ...selected })
+  confirmBtn.value?.focus()
-}
+}, { immediate: true })
 
-function cancel() {
+
-  emit('cancel')
+const handleConfirm = () => emit('confirm', { ...selected })
-}
+const cancel = () => emit('cancel')
 </script>

ui/components/NewDownload.vue

@@ -231,7 +231,7 @@
 <script setup lang="ts">
 import 'assets/css/bulma-switch.css'
 import { useStorage } from '@vueuse/core'
-import type { item_request } from '~/@types/item'
+import type { item_request } from '~/types/item'
 import { getSeparatorsName, separators } from '~/utils/utils'
 
 const props = defineProps<{ item?: Partial<item_request> }>()

ui/pages/changelog.vue

@@ -72,7 +72,7 @@ hr {
 
 <script setup lang="ts">
 import moment from 'moment'
-import type { changelogs, changeset } from '~/@types/changelogs'
+import type { changelogs, changeset } from '~/types/changelogs'
 
 const toast = useNotification()
 const config = useConfigStore()

ui/pages/conditions.vue

@@ -117,7 +117,7 @@
 <script setup lang="ts">
 import { request } from '~/utils/index'
 import { encode } from '~/utils/importer'
-import type { ConditionItem, ImportedConditionItem } from '~/@types/conditions'
+import type { ConditionItem, ImportedConditionItem } from '~/types/conditions'
 
 const toast = useNotification()
 const config = useConfigStore()

ui/pages/console.vue

@@ -63,6 +63,7 @@ import { useStorage } from '@vueuse/core'
 
 const config = useConfigStore()
 const socket = useSocketStore()
+const toast = useNotification()
 
 const bg_enable = useStorage<boolean>('random_bg', true)
 const bg_opacity = useStorage<number>('random_bg_opacity', 0.95)
@@ -88,14 +89,15 @@ watch(() => config.app.basic_mode, async () => {
     return
   }
   await navigateTo('/')
-})
+}, { immediate: true })
 
 watch(() => config.app.console_enabled, async () => {
   if (config.app.console_enabled) {
     return
   }
+  toast.error('Console is disabled in the configuration. Please enable it to use this feature.')
   await navigateTo('/')
-})
+}, { immediate: true })
 
 const handle_event = () => {
   if (!terminal.value) {
@@ -171,9 +173,9 @@ const writer = (s: string) => {
     return
   }
 
-  const data = JSON.parse(s)
+  const json = JSON.parse(s)
 
-  terminal.value.writeln(data.line)
+  terminal.value.writeln(json.data.line)
 }
 
 const loader = () => isLoading.value = false

ui/pages/index.vue

@@ -80,7 +80,7 @@
 
 <script setup lang="ts">
 import { useStorage } from '@vueuse/core'
-import type { item_request } from '~/@types/item'
+import type { item_request } from '~/types/item'
 
 const config = useConfigStore()
 const stateStore = useStateStore()

ui/pages/logs.vue

@@ -114,7 +114,7 @@ import moment from 'moment'
 import { request } from '~/utils/index'
 import { ref, onMounted, nextTick } from 'vue'
 import { useStorage } from '@vueuse/core'
-import type { log_line } from '~/@types/logs'
+import type { log_line } from '~/types/logs'
 
 let scrollTimeout: NodeJS.Timeout | null = null
 

ui/pages/tasks.vue

@@ -330,7 +330,7 @@ import { useStorage } from '@vueuse/core'
 import { CronExpressionParser } from 'cron-parser'
 import { request, sleep } from '~/utils/index'
 import { encode } from '~/utils/importer'
-import type { task_item, exported_task, error_response } from '~/@types/tasks'
+import type { task_item, exported_task, error_response } from '~/types/tasks'
 
 const box = useConfirm()
 const toast = useNotification()
@@ -686,7 +686,8 @@ const runNow = async (item: task_item, mass: boolean = false) => {
 onBeforeUnmount(() => socket.off('status', statusHandler))
 
 const statusHandler = async (stream: string) => {
-  const { status, msg } = JSON.parse(stream)
+  const json = JSON.parse(stream)
+  const { status, msg } = json.data
 
   if ('error' === status) {
     toast.error(msg)

ui/stores/ConfigStore.ts

@@ -1,5 +1,5 @@
 import { useStorage } from '@vueuse/core'
-import type { ConfigState } from '~/@types/config';
+import type { ConfigState } from '~/types/config';
 
 export const useConfigStore = defineStore('config', () => {
   const state = reactive<ConfigState>({

ui/stores/SocketStore.js

@@ -1,163 +0,0 @@
-import { io } from "socket.io-client";
-import { ag } from "~/utils/index"
-
-export const useSocketStore = defineStore('socket', () => {
-  const runtimeConfig = useRuntimeConfig()
-  const config = useConfigStore()
-  const stateStore = useStateStore()
-  const toast = useNotification()
-
-  const socket = ref(null)
-  const isConnected = ref(false)
-
-  const connect = () => {
-    let opts = {
-      transports: ['websocket', 'polling'],
-      withCredentials: true,
-    }
-
-    let url = runtimeConfig.public.wss
-
-    if ('development' !== runtimeConfig.public?.APP_ENV) {
-      url = window.origin;
-      opts.path = `${runtimeConfig.app.baseURL.replace(/\/$/, '')}/socket.io`;
-    }
-
-    socket.value = io(url, opts)
-
-    socket.value.on('connect', () => isConnected.value = true);
-    socket.value.on('disconnect', () => isConnected.value = false);
-
-    socket.value.on('initial_data', stream => {
-      const initialData = JSON.parse(stream)
-
-      config.setAll({
-        app: initialData['config'],
-        tasks: initialData['tasks'],
-        folders: initialData['folders'],
-        presets: initialData['presets'],
-        paused: Boolean(initialData['paused'])
-      })
-
-      stateStore.addAll('queue', initialData['queue'] ?? {})
-      stateStore.addAll('history', initialData['done'] ?? {})
-    })
-
-    socket.value.on('added', stream => {
-      const item = JSON.parse(stream);
-      stateStore.add('queue', item._id, item);
-      toast.success(`Item queued: ${ag(stateStore.get('queue', item._id, {}), 'title')}`);
-    });
-
-    socket.value.on('error', stream => {
-      const json = JSON.parse(stream);
-      toast.error(`${json.data?.id ?? json?.type}: ${json?.message}`, json.data || {});
-    });
-
-    socket.value.on('log_info', stream => {
-      const json = JSON.parse(stream);
-      toast.info(json?.message, json.data || {});
-    });
-
-    socket.value.on('log_success', stream => {
-      const json = JSON.parse(stream);
-      toast.success(json?.message, json.data || {});
-    });
-
-    socket.value.on('log_warning', stream => {
-      const json = JSON.parse(stream);
-      toast.warning(json?.message, json.data || {});
-    });
-
-    socket.value.on('log_error', stream => {
-      const json = JSON.parse(stream);
-      toast.error(json?.message, json.data || {});
-    });
-
-    socket.value.on('completed', stream => {
-      const item = JSON.parse(stream);
-
-      if (true === stateStore.has('queue', item._id)) {
-        stateStore.remove('queue', item._id);
-      }
-
-      stateStore.add('history', item._id, item);
-    });
-
-    socket.value.on('cancelled', stream => {
-      const item = JSON.parse(stream);
-      const id = item._id
-
-      if (true !== stateStore.has('queue', id)) {
-        return
-      }
-
-      toast.warning(`Download cancelled: ${ag(stateStore.get('queue', id, {}), 'title')}`);
-
-      if (true === stateStore.has('queue', id)) {
-        stateStore.remove('queue', id);
-      }
-    });
-
-    socket.value.on('cleared', stream => {
-      const item = JSON.parse(stream);
-      const id = item._id
-
-      if (true !== stateStore.has('history', id)) {
-        return
-      }
-
-      stateStore.remove('history', id);
-    });
-
-    socket.value.on("updated", stream => {
-      const data = JSON.parse(stream);
-
-      if (true === stateStore.has('history', data._id)) {
-        stateStore.update('history', data._id, data);
-        return;
-      }
-
-      let dl = stateStore.get('queue', data._id, {});
-      data.deleting = dl?.deleting;
-      stateStore.update('queue', data._id, data);
-    });
-
-    socket.value.on("update", stream => {
-      const data = JSON.parse(stream);
-      if (true === stateStore.has('history', data._id)) {
-        stateStore.update('history', data._id, data);
-        return;
-      }
-    });
-
-    socket.value.on('paused', data => {
-      const json = JSON.parse(data);
-      const pausedState = Boolean(json.paused);
-      config.update('paused', pausedState);
-
-      if (false === pausedState) {
-        toast.success('Download queue resumed.');
-        return;
-      }
-
-      toast.warning('Download queue paused.', {
-        timeout: 10000,
-      });
-    });
-
-    socket.value.on('presets_update', data => config.update('presets', JSON.parse(data)));
-  }
-
-  const on = (event, callback) => socket.value.on(event, callback);
-  const off = (event, callback) => socket.value.off(event, callback);
-  const emit = (event, data) => socket.value.emit(event, data);
-
-  if (false === isConnected.value) {
-    connect();
-  }
-
-  window.ws = socket.value;
-
-  return { connect, on, off, emit, socket, isConnected };
-});

ui/stores/SocketStore.ts

@@ -0,0 +1,157 @@
+import { io } from "socket.io-client";
+import { ag } from "~/utils/index"
+import type { Socket as IOSocket, SocketOptions } from "socket.io-client"
+import type { ManagerOptions } from "socket.io-client";
+
+export const useSocketStore = defineStore('socket', () => {
+  const runtimeConfig = useRuntimeConfig()
+  const config = useConfigStore()
+  const stateStore = useStateStore()
+  const toast = useNotification()
+
+  const socket = ref<IOSocket | null>(null)
+  const isConnected = ref<boolean>(false)
+
+  const connect = () => {
+    let opts = {
+      transports: ['websocket', 'polling'],
+      withCredentials: true,
+    } as Partial<ManagerOptions & SocketOptions>
+
+    let url = runtimeConfig.public.wss
+
+    if ('development' !== runtimeConfig.public?.APP_ENV) {
+      url = window.origin;
+      opts.path = `${runtimeConfig.app.baseURL.replace(/\/$/, '')}/socket.io`;
+    }
+
+    socket.value = io(url, opts)
+
+    socket.value.on('connect', () => isConnected.value = true);
+    socket.value.on('disconnect', () => isConnected.value = false);
+
+    socket.value.on('connected', stream => {
+      const json = JSON.parse(stream)
+
+      config.setAll({
+        app: json.data.config,
+        tasks: json.data.tasks,
+        folders: json.data.folders,
+        presets: json.data.presets,
+        paused: Boolean(json.data.paused)
+      })
+
+      stateStore.addAll('queue', json.data.queue || {})
+      stateStore.addAll('history', json.data.done || {})
+    })
+
+    socket.value.on('added', stream => {
+      const json = JSON.parse(stream);
+      stateStore.add('queue', json.data._id, json.data);
+      toast.success(`Item queued: ${ag(stateStore.get('queue', json.data._id, {}), 'title')}`);
+    });
+
+    ['log_info', 'log_success', 'log_warning', 'log_error'].forEach(event => socket.value?.on(event, stream => {
+      const json = JSON.parse(stream);
+      const message = json?.message || json?.data?.message;
+      const data = json.data?.data || json.data || {};
+      switch (event) {
+        case 'log_info':
+          toast.info(message, data);
+          break;
+        case 'log_success':
+          toast.success(message, data);
+          break;
+        case 'log_warning':
+          toast.warning(message, data);
+          break;
+        case 'log_error':
+          toast.error(message, data);
+          break;
+      }
+    }));
+
+    socket.value.on('completed', stream => {
+      const json = JSON.parse(stream);
+
+      if (true === stateStore.has('queue', json.data._id)) {
+        stateStore.remove('queue', json.data._id);
+      }
+
+      stateStore.add('history', json.data._id, json.data);
+    });
+
+    socket.value.on('cancelled', stream => {
+      const item = JSON.parse(stream);
+      const id = item.data._id
+
+      if (true !== stateStore.has('queue', id)) {
+        return
+      }
+
+      toast.warning(`Download cancelled: ${ag(stateStore.get('queue', id, {}), 'title')}`);
+
+      if (true === stateStore.has('queue', id)) {
+        stateStore.remove('queue', id);
+      }
+    });
+
+    socket.value.on('cleared', stream => {
+      const item = JSON.parse(stream);
+      const id = item.data._id
+
+      if (true !== stateStore.has('history', id)) {
+        return
+      }
+
+      stateStore.remove('history', id);
+    });
+
+    socket.value.on("updated", stream => {
+      const json = JSON.parse(stream);
+      const id = json.data._id;
+
+      if (true === stateStore.has('history', id)) {
+        stateStore.update('history', id, json.data);
+        return;
+      }
+
+      stateStore.update('queue', id, json.data);
+    });
+
+    socket.value.on("update", stream => {
+      const json = JSON.parse(stream);
+      if (true === stateStore.has('history', json.data._id)) {
+        stateStore.update('history', json.data._id, json.data);
+        return;
+      }
+    });
+
+    socket.value.on('paused', data => {
+      const json = JSON.parse(data);
+      const pausedState = Boolean(json.data.paused);
+      config.update('paused', pausedState);
+
+      if (false === pausedState) {
+        toast.success('Download queue resumed.');
+        return;
+      }
+
+      toast.warning('Download queue paused.', { timeout: 10000 });
+    });
+
+    socket.value.on('presets_update', (data: string) => config.update('presets', JSON.parse(data).data || []));
+  }
+
+  const emit = (event: string, data?: any): any => socket.value?.emit(event, data)
+  const on = (event: string, callback: (...args: any[]) => void): any => socket.value?.on(event, callback)
+  const off = (event: string, callback: (...args: any[]) => void): any => socket.value?.off(event, callback)
+
+  if (false === isConnected.value) {
+    connect();
+  }
+
+  window.ws = socket.value;
+
+  return { connect, on, off, emit, socket, isConnected };
+});

ui/stores/StateStore.js

@@ -1,43 +0,0 @@
-export const useStateStore = defineStore('state', () => {
-  const state = reactive({
-    queue: {},
-    history: {}
-  });
-
-  const actions = {
-    add(type, key, value) {
-      state[type][key] = value;
-    },
-    update(type, key, value) {
-      state[type][key] = value;
-    },
-    remove(type, key) {
-      if (state[type][key]) {
-        delete state[type][key];
-      }
-    },
-    get(type, key, defaultValue = null) {
-      return state[type][key] || defaultValue;
-    },
-    has(type, key) {
-      return !!state[type][key];
-    },
-    clearAll(type) {
-      state[type] = {};
-    },
-    addAll(type, data) {
-      state[type] = data;
-    },
-    move(fromType, toType, key) {
-      if (state[fromType][key]) {
-        state[toType][key] = state[fromType][key];
-        delete state[fromType][key];
-      }
-    },
-    count(type) {
-      return Object.keys(state[type]).length;
-    }
-  }
-
-  return { ...toRefs(state), ...actions };
-});

ui/stores/StateStore.ts

@@ -0,0 +1,59 @@
+import { defineStore } from 'pinia'
+import { reactive, toRefs } from 'vue'
+import type { StoreItem } from '~/types/store'
+
+type StateType = 'queue' | 'history'
+type KeyType = string
+type ValueType = StoreItem
+
+interface State {
+  queue: Record<KeyType, ValueType>
+  history: Record<KeyType, ValueType>
+}
+
+export const useStateStore = defineStore('state', () => {
+  const state = reactive<State>({ queue: {}, history: {} })
+
+  const add = (type: StateType, key: KeyType, value: ValueType): void => {
+    state[type][key] = value
+  }
+
+  const update = (type: StateType, key: KeyType, value: ValueType): void => {
+    state[type][key] = value
+  }
+
+  const remove = (type: StateType, key: KeyType): void => {
+    if (state[type][key]) {
+      delete state[type][key]
+    }
+  }
+
+  const get = (type: StateType, key: KeyType, defaultValue: ValueType | null | object = null): ValueType | null => {
+    return state[type][key] || defaultValue
+  }
+
+  const has = (type: StateType, key: KeyType): boolean => {
+    return !!state[type][key]
+  }
+
+  const clearAll = (type: StateType): void => {
+    state[type] = {}
+  }
+
+  const addAll = (type: StateType, data: Record<KeyType, ValueType>): void => {
+    state[type] = data
+  }
+
+  const move = (fromType: StateType, toType: StateType, key: KeyType): void => {
+    if (state[fromType][key]) {
+      state[toType][key] = state[fromType][key]
+      delete state[fromType][key]
+    }
+  }
+
+  const count = (type: StateType): number => {
+    return Object.keys(state[type]).length
+  }
+
+  return { ...toRefs(state), add, update, remove, get, has, clearAll, addAll, move, count }
+})

ui/tsconfig.json

@@ -2,6 +2,9 @@
   // https://nuxt.com/docs/guide/concepts/typescript
   "extends": "./.nuxt/tsconfig.json",
   "compilerOptions": {
-    "allowImportingTsExtensions": false
+    "allowImportingTsExtensions": false,
+    "types": [
+      "./types/globals"
+    ]
   }
 }

ui/types/globals.d.ts

@@ -0,0 +1,7 @@
+export { }
+
+declare global {
+  interface Window {
+    ws?: unknown
+  }
+}

ui/types/sockets.d.ts

@@ -0,0 +1,8 @@
+export type Event = {
+  id: str
+  created_at: str
+  event: str
+  title: str | null = null
+  message: str | null = null
+  data: Any = { }
+}

ui/types/store.d.ts

@@ -0,0 +1,76 @@
+export type ItemStatus = 'finished' | 'preparing' | 'error' | 'cancelled' | 'downloading' | 'postprocessing' | 'not_live' | null;
+export type StoreItem = {
+  /** Unique identifier for the item */
+  _id: string
+  /** Error message if available */
+  error: string | null
+  /** Item source id */
+  id: string
+  /** Title of the item */
+  title: string
+  /** URL of the item */
+  url: string
+  /** Preset used for the item */
+  preset: string
+  /** Folder where the item is saved */
+  folder: string
+  /** Download directory */
+  download_dir: string
+  /** Temporary directory for the item */
+  temp_dir: string
+  /** Status of the item */
+  status: ItemStatus
+  /** If the item has cookies */
+  cookies: string
+  /** If the item has custom output_template */
+  template: string
+  /** If the item has custom output_template for chapters */
+  template_chapter: string
+  /** When the item was created */
+  timestamp: number
+  /** If the item is a live stream */
+  is_live: boolean
+  /** ISO 8601 formatted start time of the item */
+  datetime: string
+  /** Live stream start time if available */
+  live_in: string | null
+  /** File size of the item if available */
+  file_size: number | null
+  /** Custom yt-dlp command line arguments */
+  cli: string
+  /** If the item is auto-started */
+  auto_start: boolean
+  /** Options for the item */
+  options: Record<string, unknown>
+  /** Extras for the item */
+  extras: {
+    /** Which channel the item belongs to */
+    channel?: string
+    /** The video duration if available */
+    duration?: number | null
+    /** The video release date if available */
+    release_in?: string
+    /** The video thumbnail URL if available */
+    thumbnail?: string
+    /** The uploader of the item if available */
+    uploader?: string
+  }
+  /** The item temporary filename */
+  tmpfilename?: string | null
+  /** The item filename */
+  filename?: string | null
+  /** Actual file size of the item if available */
+  total_bytes?: number | null
+  /** Estimated total bytes for the item if available */
+  total_bytes_estimate?: number | null
+  /** Downloaded bytes of the item if available */
+  downloaded_bytes?: number | null
+  /** Message attached with the item if available */
+  msg?: string | null
+  /** Progress percentage of the item if available */
+  percent?: number | null
+  /** Speed of the item download if available */
+  speed?: number | null
+  /** Time remaining for the item download if available */
+  eta?: number | null
+}

ui/utils/index.js

@@ -13,7 +13,7 @@ const getValue = (obj) => 'function' === typeof obj ? obj() : obj
 /**
  * Get value from object or function and return default value if it's undefined  or null
  *
- * @param {Object|Array} obj The object to get the value from.
+ * @param {Object|Array|any} obj The object to get the value from.
  * @param {string} path The path to the value.
  * @param {*} defaultValue The default value to return if the path is not found.
  *

ui/utils/utils.ts

@@ -1,4 +1,4 @@
-import type { convert_args_response } from "~/@types/responses";
+import type { convert_args_response } from "~/types/responses";
 
 const separators = [
   { name: 'Comma', value: ',', },