stream: let listeners remove a track from the queue

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

ingest/radieo/api.py

@@ -12,6 +12,9 @@ Endpoints:
   POST /enqueue?url= -> resolve a yt-dlp URL (single track or whole playlist/
                         album) and queue it as priority requests; returns
                         {queued: N}. Proxied by the stream.
+  POST /dequeue?id=  -> drop one upcoming track (the opaque id from /queue)
+                        from the queue; returns {removed: bool}. Proxied by
+                        the stream.
   GET /healthz       -> "ok"
 """
 
@@ -87,6 +90,8 @@ class _Handler(BaseHTTPRequestHandler):
             self._serve_share(parse_qs(parsed.query))
         elif parsed.path == "/enqueue":
             self._serve_enqueue(parse_qs(parsed.query))
+        elif parsed.path == "/dequeue":
+            self._serve_dequeue(parse_qs(parsed.query))
         else:
             self._text(404, "not found\n")
 
@@ -110,6 +115,24 @@ class _Handler(BaseHTTPRequestHandler):
             200, json.dumps({"queued": count}) + "\n", "application/json; charset=utf-8"
         )
 
+    def _serve_dequeue(self, query: dict[str, list[str]]):
+        # Remove one upcoming track by the opaque id handed out by /queue.
+        # Proxied here by the stream. Unknown id (already played, or gone) → 404.
+        raw = (query.get("id") or [""])[0].strip()
+        try:
+            entry_id = int(raw)
+        except ValueError:
+            self._text(400, "missing or invalid id\n")
+            return
+        removed = self.server.queue.remove(entry_id)
+        if not removed:
+            self._text(404, "not in queue\n")
+            return
+        self._text(
+            200, json.dumps({"removed": True}) + "\n",
+            "application/json; charset=utf-8",
+        )
+
     def _serve_share(self, query: dict[str, list[str]]):
         # Mint a public Subsonic share for one song id, on demand. Called by the
         # stream when a listener clicks a subsonic track's source link, so no

ingest/radieo/db.py

@@ -141,6 +141,16 @@ class Database:
                 (path, track_key),
             )
 
+    def forget_download(self, path: str) -> None:
+        """Drop a cache-file row without touching history.
+
+        Used when a queued-but-unplayed track is removed from the queue: its
+        row (played_at NULL) would otherwise linger forever, since retention
+        only ever considers already-played files.
+        """
+        with self._lock:
+            self._conn.execute("DELETE FROM cache_files WHERE path = ?", (path,))
+
     def played_files(self, limit: int) -> list[str]:
         """Files already aired, newest first (the stream's fallback pool).
 

ingest/radieo/queue.py

@@ -156,7 +156,10 @@ class TrackQueue:
         references), then the automatic radio buffer — the same order
         ``pop_next`` serves them. Mirrors the fields exposed for the current
         track (see ``annotate_uri``): a source ``url`` only for http(s)
-        locators.
+        locators. Each entry also carries an opaque ``id`` (the track object's
+        handle) that ``remove`` accepts to drop it from the queue — stable
+        while the track sits in a buffer, and immune to reordering between
+        polls (unlike a positional index).
         """
         with self._lock:
             upcoming = (
@@ -167,6 +170,7 @@ class TrackQueue:
         items = []
         for track in upcoming:
             entry = {
+                "id": str(id(track)),
                 "title": track.title,
                 "artist": track.artist,
                 "origin": track.origin,
@@ -176,6 +180,40 @@ class TrackQueue:
             items.append(entry)
         return items
 
+    def remove(self, entry_id: int) -> bool:
+        """Drop an upcoming track from the queue by its ``snapshot`` id.
+
+        ``entry_id`` is the ``id()`` handle exposed by :meth:`snapshot`, stable
+        while the track sits in a buffer. A still-pending request simply
+        vanishes; an already-fetched entry also has its cache file removed,
+        since once out of the queue it would neither play nor be LRU-evicted
+        (eviction only touches files that have aired). Returns whether anything
+        matched.
+        """
+        removed_path = None
+        with self._lock:
+            for track in self._requests:
+                if id(track) == entry_id:
+                    self._requests.remove(track)
+                    return True
+            for buf in (self._ready_req, self._ready):
+                for item in buf:
+                    if id(item[1]) == entry_id:
+                        buf.remove(item)
+                        removed_path = item[0]
+                        break
+                if removed_path is not None:
+                    break
+        if removed_path is None:
+            return False
+        self._db.forget_download(str(removed_path))
+        try:
+            removed_path.unlink(missing_ok=True)
+            log.info("removed queued %s", removed_path.name)
+        except OSError:
+            log.exception("could not remove queued file %s", removed_path)
+        return True
+
     # --- serving ----------------------------------------------------------
 
     def pop_next(self) -> tuple[Path, Track] | None: