feat(perf): implement direct-play bypass, HD artwork pipeline, multi-bot pooling, WAL checkpoints and read-ahead buffering

apex/apex_stream/stream.py

@@ -254,6 +254,14 @@ class StreamDriver:
                     for _ in range(to_skip):
                         await run.read_chunk()
                     while chunk_index <= run_end:
+                        # Predictive Read-Ahead: when reaching the end of the current run,
+                        # asynchronously prefetch the subsequent run into cache
+                        if chunk_index == run_end and run_end < layout.end_chunk:
+                            next_run_start = run_end + 1
+                            next_run_count = min(cfg.run_size, layout.end_chunk - next_run_start + 1)
+                            asyncio.create_task(
+                                self._prefetch_next_run(source_key, next_run_start, next_run_count, plan.dc_id)
+                            )
                         chunk = await run.read_chunk()
                         chunk_index += 1
                         if emitted_window == 0:
@@ -334,6 +342,22 @@ class StreamDriver:
 
         return _factory
 
+    async def _prefetch_next_run(self, source_key: tuple, next_run_start: int, next_run_count: int, dc_id: Optional[int]) -> None:
+        """Asynchronously pre-warm the next chunk run in RAM cache ahead of client playback."""
+        try:
+            client = await self.pool.select(dc_id=dc_id)
+            try:
+                factory = self._factory_for(client.client, source_key[0], source_key[1])
+                source_run = await self.registry.get_run(source_key, next_run_start, next_run_count, factory)
+                try:
+                    await asyncio.wait_for(source_run.read_chunk(), timeout=2.0)
+                finally:
+                    await source_run.close()
+            finally:
+                await self.pool.release(client.client_id)
+        except Exception:
+            pass
+
     # --- convenience ---------------------------------------------------
 
     def status_snapshot(self) -> dict:

apex/apex_stream/tmdb_meta.py

@@ -185,6 +185,14 @@ _MEDIA_EXT_RE = re.compile(
 )
 
 
+_URL_NOISE_RE = re.compile(
+    r"(?i)\b(?:https?://\S+|www\.\S+|\S+\.(?:com|org|net|vip|in|me|xyz|top|site|cc|to|is|pm|ms|link))\b"
+)
+_WATERMARK_PAREN_RE = re.compile(
+    r"(?i)\s*\([^)]*(?:www\.|t\.me|https?://|\.com|\.org|\.net|\.vip|\.in|\.xyz|tamil|bolly|cinema|esub|dual[- ]?audio|clean[- ]?audio|hq[- ]?line|rip)[^)]*\)"
+)
+
+
 def _strip_noise(text: str) -> str:
     """Strip group tags ([...], @Handle), the real media extension and collapse
     whitespace.
@@ -196,7 +204,9 @@ def _strip_noise(text: str) -> str:
     keeping them produces dotted folder names and breaks both Jellyfin
     matching and TMDB search queries.
     """
-    t = re.sub(r"\[[^\]]*\]", " ", text)
+    t = _URL_NOISE_RE.sub(" ", text)
+    t = _WATERMARK_PAREN_RE.sub(" ", t)
+    t = re.sub(r"\[[^\]]*\]", " ", t)
     t = re.sub(r"@[\w.\-]+", " ", t)
     t = _MEDIA_EXT_RE.sub(" ", t)
     # "+" and "&" are release/present separators too ("Dd+5.1 H" -> "Dd 5 1 H",

apex/combined-entrypoint.sh

@@ -42,6 +42,7 @@ restore_jellyfin_local() {
             \( -name metadata -o -name metadata.snapshot \) -prune -o \
             -exec cp -a -f {} "$LOCAL_ROOT/data/" \; 2>/dev/null || true
         echo "[PERSISTENCE] Jellyfin database snapshot found ($(stat -c '%s' /data/jellyfin/data/data/jellyfin.db 2>/dev/null || echo unknown) bytes)."
+        sqlite3 "$LOCAL_ROOT/data/data/jellyfin.db" "PRAGMA journal_mode=WAL; PRAGMA wal_checkpoint(TRUNCATE);" 2>/dev/null || true
     else
         echo "[PERSISTENCE] No Jellyfin database snapshot found; Jellyfin will run first-time initialization."
     fi
@@ -111,6 +112,7 @@ sync_jellyfin_back() {
     while :; do
         sleep 20
         if [ -f "$LOCAL_ROOT/data/data/jellyfin.db" ]; then
+            sqlite3 "$LOCAL_ROOT/data/data/jellyfin.db" "PRAGMA wal_checkpoint(TRUNCATE);" 2>/dev/null || true
             mkdir -p /data/jellyfin/data/data
             snapshot="/data/jellyfin/data/data/jellyfin.db.tmp"
             if sqlite3 "$LOCAL_ROOT/data/data/jellyfin.db" ".backup '$snapshot'" 2>/dev/null; then
@@ -228,6 +230,33 @@ for bcfg in "$LOCAL_ROOT/config/branding.xml" "/data/jellyfin/config/branding.xm
     fi
 done
 
+# Configure encoding.xml for direct play / stream remuxing (prevents CPU-heavy video transcode)
+ENCODING_XML="$LOCAL_ROOT/config/encoding.xml"
+python3 - "$ENCODING_XML" <<'PY'
+import sys
+from pathlib import Path
+
+path = Path(sys.argv[1])
+content = """<?xml version="1.0" encoding="utf-8"?>
+<EncodingOptions xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xsd="http://www.w3.org/2001/XMLSchema">
+  <EncodingThreadCount>0</EncodingThreadCount>
+  <TranscodingTempPath>/opt/jellyfin-local/cache/transcodes</TranscodingTempPath>
+  <DownMixAudioBoost>2</DownMixAudioBoost>
+  <EnableThrottling>true</EnableThrottling>
+  <ThrottleDelaySeconds>180</ThrottleDelaySeconds>
+  <HardwareAccelerationType>none</HardwareAccelerationType>
+  <EnableVideoEncoding>false</EnableVideoEncoding>
+  <EnableAudioEncoding>true</EnableAudioEncoding>
+  <AllowVideoStreamCopy>true</AllowVideoStreamCopy>
+  <AllowAudioStreamCopy>true</AllowAudioStreamCopy>
+</EncodingOptions>
+"""
+path.parent.mkdir(parents=True, exist_ok=True)
+path.write_text(content.strip(), encoding="utf-8")
+PY
+cp -f "$ENCODING_XML" "/data/jellyfin/config/encoding.xml" 2>/dev/null || true
+echo "[JELLYFIN] Configured zero-CPU video encoding & direct stream remux in $ENCODING_XML"
+
 echo "[HEALTH] Jellyfin starting in background (local datadir)..."
 launch_jellyfin
 

apex/gateway/main.py

@@ -66,6 +66,13 @@ JELLYFIN_PORT = int(os.environ.get("JELLYFIN_PORT", "8096"))
 TG_PORT = int(os.environ.get("TG_LISTEN_PORT", "8082"))
 FILETOLINK_PORT = int(os.environ.get("FILETOLINK_PORT", "8081"))
 
+@app.get("/ping")
+@app.get("/keepalive")
+@app.get("/heartbeat")
+async def keepalive_ping():
+    """Fast, zero-overhead heartbeat endpoint for UptimeRobot / cron pings to prevent Space sleep."""
+    return JSONResponse({"status": "ok", "service": "Apex-Jellyfin", "alive": True})
+
 @app.get("/health")
 @app.get("/debug/status")
 async def health_check():

apex/gateway/utils.py

@@ -298,6 +298,20 @@ async def proxy_http_request(
         except Exception:
             pass
 
+    # Direct Play Optimization: Force DirectPlay & DirectStream in PlaybackInfo responses
+    # for .strm items so modern browsers and devices stream natively without HLS CPU transcoding
+    if "application/json" in content_type and "playbackinfo" in request.url.path.lower() and content:
+        try:
+            import json
+            pb_data = json.loads(content.decode("utf-8"))
+            for m_src in pb_data.get("MediaSources") or []:
+                if isinstance(m_src, dict):
+                    m_src["SupportsDirectPlay"] = True
+                    m_src["SupportsDirectStream"] = True
+            content = json.dumps(pb_data).encode("utf-8")
+        except Exception:
+            pass
+
     normal_resp = Response(
         content=content,
         status_code=status_code,

apex/tests/test_gateway_plugin_routing.py

@@ -1,4 +1,5 @@
 import pytest
+import httpx
 from starlette.datastructures import Headers
 from unittest.mock import AsyncMock, patch
 
@@ -90,3 +91,58 @@ async def test_jellyfin_ui_assets_rewrite_to_web():
         target_url = mock_proxy.call_args[0][0]
         assert "8096" in target_url
         assert "/web/ui/abyss.css" in target_url
+
+
+@pytest.mark.asyncio
+async def test_keepalive_endpoints():
+    from gateway.main import keepalive_ping
+    resp = await keepalive_ping()
+    assert resp.status_code == 200
+    import json
+    data = json.loads(resp.body.decode("utf-8"))
+    assert data["status"] == "ok"
+    assert data["alive"] is True
+
+
+@pytest.mark.asyncio
+async def test_playbackinfo_directplay_override():
+    import json
+    from unittest.mock import MagicMock
+    from gateway.utils import proxy_http_request
+
+    # Mock response with MediaSources having SupportsDirectPlay=False
+    mock_resp = MagicMock()
+    mock_resp.status_code = 200
+    mock_resp.headers = httpx.Headers({"content-type": "application/json"})
+    payload = {
+        "MediaSources": [
+            {
+                "Id": "item1",
+                "Path": "/data/media/Movies/Spider-Man.strm",
+                "SupportsDirectPlay": False,
+                "SupportsDirectStream": False,
+            }
+        ]
+    }
+    mock_resp.aread = AsyncMock(return_value=json.dumps(payload).encode("utf-8"))
+    mock_resp.aclose = AsyncMock()
+
+    mock_client = MagicMock()
+    mock_client.build_request.return_value = MagicMock()
+    mock_client.send = AsyncMock(return_value=mock_resp)
+
+    with patch("gateway.utils.get_http_client", return_value=mock_client):
+        from starlette.datastructures import QueryParams
+        req = AsyncMock()
+        req.url.path = "/jellyfin/Items/123/PlaybackInfo"
+        req.headers = Headers({"accept": "application/json"})
+        req.method = "POST"
+        req.body = AsyncMock(return_value=b"{}")
+        req.query_params = QueryParams([])
+
+        result = await proxy_http_request("http://127.0.0.1:8096/Items/123/PlaybackInfo", req)
+        assert result.status_code == 200
+        out_data = json.loads(result.body.decode("utf-8"))
+        src = out_data["MediaSources"][0]
+        assert src["SupportsDirectPlay"] is True
+        assert src["SupportsDirectStream"] is True

apex/tg_streamer.py

@@ -759,10 +759,25 @@ apex_driver = StreamDriver(
 # they never consume updates so the webhook/primary bot owns ingestion.
 extra_clients = []
 if not APEX_CFG.extra_tokens:
-    APEX_CFG.extra_tokens = [
-        value for index in range(1, 100)
-        if (value := os.environ.get(f"JELLYFIN_MULTI_TOKEN{index}", "").strip())
-    ]
+    discovered_tokens = []
+    for index in range(1, 100):
+        if val := os.environ.get(f"JELLYFIN_MULTI_TOKEN{index}", "").strip():
+            if val not in discovered_tokens:
+                discovered_tokens.append(val)
+    for prefix in ("BOT_TOKEN_", "TG_BOT_TOKEN_", "MULTI_TOKEN_"):
+        for index in range(1, 25):
+            if val := os.environ.get(f"{prefix}{index}", "").strip():
+                if val not in discovered_tokens and val != BOT_TOKEN:
+                    discovered_tokens.append(val)
+    for env_name in ("EXTRA_BOT_TOKENS", "MULTI_BOT_TOKENS", "APEX_EXTRA_TOKENS", "APEX_TELEGRAM_EXTRA_TOKENS"):
+        if val := os.environ.get(env_name, "").strip():
+            for tok in val.split(","):
+                tok_clean = tok.strip()
+                if tok_clean and tok_clean not in discovered_tokens and tok_clean != BOT_TOKEN:
+                    discovered_tokens.append(tok_clean)
+    APEX_CFG.extra_tokens = discovered_tokens
+    if APEX_CFG.extra_tokens:
+        logger.info(f"[POOL] Discovered {len(APEX_CFG.extra_tokens)} extra streaming bot token(s).")
 for _ec_idx, _ec_token in enumerate(APEX_CFG.extra_tokens, start=2):
     if API_ID and API_HASH and _ec_token:
         try:
@@ -1444,10 +1459,24 @@ async def _enrich_media_metadata(entry):
                         f.write(build_tvshow_nfo(detail))
                 poster_url = _best_image("poster_path")
                 if poster_url and not os.path.exists(poster_path):
-                    await _download_image(session, _resize_tmdb_image(poster_url), poster_path)
+                    if await _download_image(session, _resize_tmdb_image(poster_url, "w780"), poster_path):
+                        folder_path = os.path.join(show_root, "folder.jpg")
+                        if not os.path.exists(folder_path):
+                            try:
+                                with open(poster_path, "rb") as src, open(folder_path, "wb") as dst:
+                                    dst.write(src.read())
+                            except OSError:
+                                pass
                 backdrop_url = _best_image("backdrop_path")
                 if backdrop_url and not os.path.exists(backdrop_path):
-                    await _download_image(session, _resize_tmdb_image(backdrop_url, "w1280"), backdrop_path)
+                    if await _download_image(session, _resize_tmdb_image(backdrop_url, "original"), backdrop_path):
+                        fanart_path = os.path.join(show_root, "fanart.jpg")
+                        if not os.path.exists(fanart_path):
+                            try:
+                                with open(backdrop_path, "rb") as src, open(fanart_path, "wb") as dst:
+                                    dst.write(src.read())
+                            except OSError:
+                                pass
 
                 season_detail = None
                 season_number = int(entry.get("season") or 1)
@@ -1476,12 +1505,35 @@ async def _enrich_media_metadata(entry):
                 )
                 with open(nfo_path, "w", encoding="utf-8") as f:
                     f.write(build_movie_nfo(detail))
+                movie_nfo = os.path.join(target_dir, "movie.nfo")
+                if not os.path.exists(movie_nfo):
+                    try:
+                        with open(nfo_path, "rb") as src, open(movie_nfo, "wb") as dst:
+                            dst.write(src.read())
+                    except OSError:
+                        pass
                 poster_url = _best_image("poster_path")
-                if poster_url and not os.path.exists(os.path.join(target_dir, "poster.jpg")):
-                    await _download_image(session, _resize_tmdb_image(poster_url), os.path.join(target_dir, "poster.jpg"))
+                movie_poster = os.path.join(target_dir, "poster.jpg")
+                if poster_url and not os.path.exists(movie_poster):
+                    if await _download_image(session, _resize_tmdb_image(poster_url, "w780"), movie_poster):
+                        folder_path = os.path.join(target_dir, "folder.jpg")
+                        if not os.path.exists(folder_path):
+                            try:
+                                with open(movie_poster, "rb") as src, open(folder_path, "wb") as dst:
+                                    dst.write(src.read())
+                            except OSError:
+                                pass
                 backdrop_url = _best_image("backdrop_path")
-                if backdrop_url and not os.path.exists(os.path.join(target_dir, "backdrop.jpg")):
-                    await _download_image(session, _resize_tmdb_image(backdrop_url, "w1280"), os.path.join(target_dir, "backdrop.jpg"))
+                movie_backdrop = os.path.join(target_dir, "backdrop.jpg")
+                if backdrop_url and not os.path.exists(movie_backdrop):
+                    if await _download_image(session, _resize_tmdb_image(backdrop_url, "original"), movie_backdrop):
+                        fanart_path = os.path.join(target_dir, "fanart.jpg")
+                        if not os.path.exists(fanart_path):
+                            try:
+                                with open(movie_backdrop, "rb") as src, open(fanart_path, "wb") as dst:
+                                    dst.write(src.read())
+                            except OSError:
+                                pass
 
         prev = FILE_ID_CACHE.get(key)
         if isinstance(prev, dict):
@@ -1542,9 +1594,11 @@ def _tmdb_year(detail):
     return None
 
 
-def _resize_tmdb_image(url, size="w500"):
-    """Rewrite a TMDB original image URL to a smaller requested size."""
-    if "image.tmdb.org/" not in url:
+def _resize_tmdb_image(url, size="w780"):
+    """Rewrite a TMDB image URL to requested size (or original)."""
+    if not url or "image.tmdb.org/" not in url:
+        return url
+    if size == "original":
         return url
     return url.replace("/original", f"/{size}", 1)