perf: extend WebSocket batching to item_moved events

Bursts of completions (e.g. small playlist entries) fire one item_moved
frame per item. Route item_moved through the same ItemBatcher mechanism
as item_updated: one frame per 500ms tick, payload always a list.

ItemBatcher gains an optional key callable since item_moved payloads are
{"to", "preset", "item"} dicts rather than ItemDTOs; moves coalesce by
item _id with last write wins, so an item that moves twice within a tick
delivers only its final destination. The notifications service consumes
ITEM_MOVED from the event bus directly and is unaffected.

BREAKING CHANGE: the item_moved WebSocket event payload (data) is now
always an array of {to, preset, item} objects instead of a single one.

app/library/HttpSocket.py

@@ -86,20 +86,29 @@ class HttpSocket:
         self.sio = sio or WebSocketHub(encoder=encoder)
         self.rootPath: Path = root_path
 
-        async def _emit_items(items: list) -> None:
-            ev = Event(event=Events.ITEM_UPDATED, data=items)
-            await self.sio.emit(event=Events.ITEM_UPDATED, data=json.loads(encoder.encode(ev)))
+        def _make_emit(event_name: str):
+            async def _emit_items(items: list) -> None:
+                ev = Event(event=event_name, data=items)
+                await self.sio.emit(event=event_name, data=json.loads(encoder.encode(ev)))
 
-        self._batcher = ItemBatcher(emit=_emit_items)
+            return _emit_items
+
+        self._batchers: dict[str, ItemBatcher] = {
+            Events.ITEM_UPDATED: ItemBatcher(emit=_make_emit(Events.ITEM_UPDATED)),
+            Events.ITEM_MOVED: ItemBatcher(
+                emit=_make_emit(Events.ITEM_MOVED),
+                key=lambda d: getattr(d.get("item"), "_id", None) if isinstance(d, dict) else None,
+            ),
+        }
 
         async def event_handler(e: Event, _, **kwargs):
-            if Events.ITEM_UPDATED == e.event and not kwargs:
-                await self._batcher.add(e.data)
-                return
-            if Events.ITEM_UPDATED == e.event and kwargs:
+            if batcher := self._batchers.get(e.event):
+                if not kwargs:
+                    await batcher.add(e.data)
+                    return
                 # Targeted delivery (e.g. to=sid): wrap data in a list so the
-                # wire shape is always an array for item_updated.
-                targeted = Event(event=Events.ITEM_UPDATED, data=[e.data])
+                # wire shape is always an array for batched events.
+                targeted = Event(event=e.event, data=[e.data])
                 payload = json.loads(encoder.encode(targeted))
                 await self.sio.emit(event=e.event, data=payload, **kwargs)
                 return
@@ -138,7 +147,8 @@ class HttpSocket:
 
     async def on_shutdown(self, _: web.Application):
         LOG.debug("Shutting down socket server.")
-        await self._batcher.flush()
+        for batcher in self._batchers.values():
+            await batcher.flush()
         await self.sio.disconnect_all()
         LOG.debug("Socket server shutdown complete.")
 

app/library/ItemBatcher.py

@@ -1,6 +1,6 @@
-"""Global coalescing batcher for ITEM_UPDATED WebSocket events.
+"""Global coalescing batcher for high-frequency WebSocket item events.
 
-Collects dirty ItemDTOs keyed by ``_id`` (last write wins) and flushes them
+Collects dirty items keyed per item (last write wins) and flushes them
 as a single list to an async emit callback at most once per *interval* seconds.
 
 Leading-edge semantics: the very first ``add()`` after an idle period fires
@@ -21,19 +21,27 @@ LOG = get_logger()
 
 
 class ItemBatcher:
-    """Coalescing batcher for WebSocket item-updated payloads.
+    """Coalescing batcher for WebSocket item event payloads.
 
     Args:
         emit: Async callable that receives a list of items and delivers them
               to all connected WebSocket clients.
         interval: Flush interval in seconds.  ``0`` or negative disables
                   batching – every ``add()`` emits immediately.
+        key: Optional callable that extracts the coalescing key from an item.
+             Defaults to the item's ``_id`` attribute.
 
     """
 
-    def __init__(self, emit: Callable[[list], Awaitable[None]], interval: float = 0.5) -> None:
+    def __init__(
+        self,
+        emit: Callable[[list], Awaitable[None]],
+        interval: float = 0.5,
+        key: Callable[[Any], str | None] | None = None,
+    ) -> None:
         self._emit = emit
         self._interval = interval
+        self._key = key
         self._pending: dict[str, Any] = {}
         self._handle: asyncio.TimerHandle | None = None
         self._last_flush: float = 0.0  # monotonic
@@ -50,11 +58,11 @@ class ItemBatcher:
         flush.
 
         Args:
-            item: The ItemDTO (or any object with a ``_id`` attribute) to
-                  batch.
+            item: The payload to batch. Without a ``key`` callable, this is
+                  expected to expose a ``_id`` attribute.
 
         """
-        item_id: str = getattr(item, "_id", None) or str(id(item))
+        item_id: str = (self._key(item) if self._key else getattr(item, "_id", None)) or str(id(item))
 
         # interval <= 0  →  always emit immediately, no batching
         if self._interval <= 0:

app/tests/test_item_batcher.py

@@ -189,6 +189,39 @@ class TestItemBatcher:
 
         assert emitted == []
 
+    @pytest.mark.asyncio
+    async def test_custom_key_coalesces_dict_payloads(self) -> None:
+        """A key callable extracts the coalescing key from non-ItemDTO payloads."""
+        emitted: list[list] = []
+
+        async def emit(items: list) -> None:
+            emitted.append(items)
+
+        batcher = ItemBatcher(
+            emit=emit,
+            interval=0.1,
+            key=lambda d: getattr(d.get("item"), "_id", None) if isinstance(d, dict) else None,
+        )
+
+        # Consume the leading-edge slot
+        await batcher.add({"to": "history", "item": make_item("id1")})
+        assert len(emitted) == 1
+
+        # Two moves for the same item within the interval → last wins
+        await batcher.add({"to": "history", "item": make_item("id2", "first")})
+        await batcher.add({"to": "queue", "item": make_item("id2", "second")})
+        await batcher.add({"to": "history", "item": make_item("id3")})
+
+        await asyncio.sleep(0.15)
+
+        assert len(emitted) == 2
+        flush_items = emitted[1]
+        assert len(flush_items) == 2
+        by_id = {moved["item"]._id: moved for moved in flush_items}
+        assert set(by_id) == {"id2", "id3"}
+        assert by_id["id2"]["to"] == "queue"
+        assert by_id["id2"]["item"].title == "second"
+
     @pytest.mark.asyncio
     async def test_interval_zero_emits_every_add_immediately(self) -> None:
         """interval=0 → every add emits immediately as a single-item list."""

ui/app/composables/useAppSocket.ts

@@ -345,16 +345,18 @@ on('item_updated', (data: WSEP['item_updated']) => {
 
 on('item_moved', (data: WSEP['item_moved']) => {
   const queueState = getQueueState();
-  const to = data.data.to;
-  const id = data.data.item._id;
 
-  if ('queue' === to) {
-    queueState.add(id, data.data.item);
-  }
+  for (const moved of data.data) {
+    const id = moved.item._id;
 
-  if ('history' === to) {
-    if (true === queueState.has(id)) {
-      queueState.remove(id);
+    if ('queue' === moved.to) {
+      queueState.add(id, moved.item);
+    }
+
+    if ('history' === moved.to) {
+      if (true === queueState.has(id)) {
+        queueState.remove(id);
+      }
     }
   }
 });

ui/app/composables/useHistoryState.ts

@@ -231,11 +231,15 @@ const moveHandler = (
   shouldHandle: () => boolean = () => isLoaded.value,
 ): ((payload: WSEP['item_moved']) => void) => {
   return (payload: WSEP['item_moved']): void => {
-    if ('history' !== payload.data.to || !shouldHandle()) {
+    if (!shouldHandle()) {
       return;
     }
 
-    upsert(payload.data.item);
+    for (const moved of payload.data) {
+      if ('history' === moved.to) {
+        upsert(moved.item);
+      }
+    }
   };
 };
 

ui/app/types/sockets.d.ts

@@ -26,7 +26,7 @@ export type WSEP = {
   item_cancelled: EventPayload<StoreItem>;
   item_deleted: EventPayload<StoreItem>;
   item_bulk_deleted: EventPayload<{ count: number; status?: string; ids?: string[] }>;
-  item_moved: EventPayload<{ to: 'queue' | 'history'; item: StoreItem }>;
+  item_moved: EventPayload<Array<{ to: 'queue' | 'history'; item: StoreItem }>>;
   item_status: EventPayload<{ status?: string; msg?: string; preset?: string }>;
   paused: EventPayload<{ paused?: boolean }>;
   resumed: EventPayload<{ paused?: boolean }>;