feat(timeshift): add catch-up playback, stats, and admin APIs

Add end-to-end catch-up support for XC clients and native apps: provider
proxy with failover and per-viewer session pooling, REST session minting
for tokenless playback URLs, catch-up admin stats, combined connection
stats, and Stats UI with dedicated cards plus websocket updates.

Includes Redis namespace consolidation under timeshift:* (dropping legacy
timeshift_ id prefixes), dedicated catch-up stop by session_id, and
cleaner channel/client metadata split for stats keys.

Closes #133
This commit is contained in:
SergeantPanda 2026-07-11 17:14:31 +00:00
parent 3c2c42ce66
commit 223dff33ed
43 changed files with 7114 additions and 537 deletions

View file

@ -10,7 +10,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Added
- **Native XZ decompression for EPG sources and uploaded M3U playlists.** `.xz`-compressed XMLTV and M3U files are now decompressed using Python's stdlib `lzma` module. EPG ingestion detects XZ via magic bytes, file extension (including compound names like `.xml.xz`), or MIME type; uploaded M3U `.xz` playlists stream line-by-line like `.gz`. The `/data/epgs` auto-import scanner now includes `.xz` files alongside `.xml`, `.gz`, and `.zip`. (Closes #1414) — Thanks [@MotWakorb](https://github.com/MotWakorb)
- **XC catch-up (timeshift) support**. Adds a native `/timeshift/{user}/{pass}/{stream_id}/{timestamp}/{duration}.ts` endpoint that proxies replay sessions from an Xtream Codes provider to clients such as iPlayTV (Apple TV) and TiviMate. Auth reuses the same `xc_password` custom-property the live `/live/.../{id}.ts` endpoint already uses. Catch-up sessions appear on `/stats` with a violet `TIMESHIFT` badge alongside live sessions and respect per-channel access rules. (Closes #133) — Thanks [@cedric-marcoux](https://github.com/cedric-marcoux)
- **XC catch-up (timeshift) support**. Adds a native `/timeshift/{user}/{pass}/{stream_id}/{timestamp}/{duration}.ts` endpoint that proxies replay sessions from an Xtream Codes provider to clients such as iPlayTV (Apple TV) and TiviMate. Auth reuses the same `xc_password` custom-property the live `/live/.../{id}.ts` endpoint already uses. Catch-up sessions appear on the Stats page in dedicated connection cards (separate from live and VOD) and respect per-channel access rules. (Closes #133) — Thanks [@cedric-marcoux](https://github.com/cedric-marcoux)
- **REST catch-up API for native apps.** `POST /api/catchup/sessions/` (JWT or API key) mints a server-side playback session and returns a tokenless `playback_url` at `/proxy/catchup/<channel_uuid>?session_id=...` for headerless video players. `DELETE /api/catchup/sessions/<session_id>/` ends the session. OpenAPI docs cover handshake and idle TTL behaviour.
- **Catch-up admin APIs.** `GET /proxy/catchup/stats/` lists active catch-up viewers; `POST /proxy/catchup/programs/` returns batch EPG metadata for those sessions; `POST /proxy/catchup/stop_client/` stops a viewer from the admin UI.
- **Catch-up Stats UI.** Dedicated catch-up connection cards show channel logo, programme preview (watched/remaining timeline), playback position, bitrate, and a stop button. Stats refresh in real time via WebSocket (`timeshift_stats`) with desktop notifications when catch-up sessions start or end.
- **Catch-up flags** (`tv_archive`, `tv_archive_duration`) denormalized at import time for zero-cost output queries; end-of-refresh SQL rollup updates channels linked to the refreshed account and self-heals stale flags on those channels only (e.g. after catch-up streams are removed or an account is deactivated).
- **Strictly-UTC API surface, automatic per-provider timezone.** `server_info.timezone` is always `UTC` and the XC EPG `start`/`end` strings are emitted in UTC; the proxy converts the requested instant to the serving provider's own reported timezone (`server_info.timezone` captured on account refresh) at request time. No timezone to configure.
- **Multi-provider failover.** Catch-up walks the channel's catch-up streams in order (like live playback): if one provider cannot serve the archive the next is tried, each attempt with its own account context (credentials, reported timezone, user-agent). Accounts that fail with an auth/ban-class status are not retried via their other streams.
@ -20,6 +23,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- **EPG XMLTV `prev_days` auto-detection** from the provider's largest `tv_archive_duration` (capped at 30 days), overridable via setting or `?prev_days=` URL parameter.
- **Byte path streams directly via `iter_content` + generator yield** (same pattern as the VOD proxy), with an upfront MPEG-TS sync peek that strips any PHP-warning preamble before bytes reach strict demuxers. Throughput stays comfortably above the typical 5 Mbps FHD bitrate so clients don't buffer.
- **Catch-up indicators in Channels and Streams tables.** Channels and streams with catch-up enabled show a grey history icon beside the name (tooltip includes archive days when known). Expanded channel stream rows show a catch-up badge. Channel and stream API serializers expose `is_catchup` and `catchup_days` for the UI. The Channels and Streams table filter menus include **Only Catch-up** to narrow each list to catch-up entries.
- **Combined connection stats API.** `GET /proxy/stats/` (admin) returns live, VOD, and catch-up connection stats in one JSON response (`live`, `vod`, `catchup`, `timestamp`). The Stats page polls this endpoint instead of separate live and VOD stats requests.
- **Frontend unit tests extended to table components.** `ChannelsTable`, `StreamsTable`, `M3UsTable`, `EPGsTable`, `LogosTable`, `CustomTable`, and related header/editable subcomponents now have Vitest + Testing Library suites. Table business logic was extracted into `frontend/src/utils/tables/*` (with shared `M3uTableUtils` for M3U/EPG sort headers) and covered by dedicated util tests. `dateTimeUtils.formatDuration` gained a `human` precision mode for readable content lengths. — Thanks [@nick4810](https://github.com/nick4810)
### Changed

View file

@ -12,6 +12,7 @@ urlpatterns = [
path('core/', include(('core.api_urls', 'core'), namespace='core')),
path('plugins/', include(('apps.plugins.api_urls', 'plugins'), namespace='plugins')),
path('vod/', include(('apps.vod.api_urls', 'vod'), namespace='vod')),
path('catchup/', include(('apps.timeshift.api_urls', 'catchup'), namespace='catchup')),
path('backups/', include(('apps.backups.api_urls', 'backups'), namespace='backups')),
path('connect/', include(('apps.connect.api_urls', 'connect'), namespace='connect')),
# path('output/', include(('apps.output.api_urls', 'output'), namespace='output')),

View file

@ -214,11 +214,12 @@ class DoStatsUpdateTests(TestCase):
"""_do_stats_update must call send_websocket_update with channel_stats."""
cm = self._make_client_manager()
mock_redis = MagicMock()
mock_redis.scan.return_value = (0, [])
with patch("apps.proxy.live_proxy.client_manager.send_websocket_update") as mock_ws, \
patch("core.utils.RedisClient.get_client", return_value=mock_redis):
patch(
"apps.proxy.live_proxy.channel_status.build_live_channel_stats_data",
return_value={"channels": [], "count": 0},
), \
patch("core.utils.RedisClient.get_client", return_value=MagicMock()):
cm._do_stats_update()
mock_ws.assert_called_once()
@ -237,19 +238,20 @@ class DoStatsUpdateTests(TestCase):
except Exception as e:
self.fail(f"_do_stats_update raised an exception: {e}")
def test_do_stats_update_scans_channel_metadata_keys(self):
"""Must scan for live:channel:*:metadata pattern."""
def test_do_stats_update_uses_build_live_channel_stats_data(self):
"""Must build live stats via the shared builder."""
cm = self._make_client_manager()
mock_redis = MagicMock()
mock_redis.scan.return_value = (0, [])
with patch("apps.proxy.live_proxy.client_manager.send_websocket_update"), \
patch(
"apps.proxy.live_proxy.channel_status.build_live_channel_stats_data",
return_value={"channels": [{"channel_id": "ch-1"}], "count": 1},
) as mock_build, \
patch("core.utils.RedisClient.get_client", return_value=mock_redis):
cm._do_stats_update()
scan_call = mock_redis.scan.call_args
self.assertIn("live:channel:*:metadata", str(scan_call))
mock_build.assert_called_once_with(mock_redis)
# ---------------------------------------------------------------------------

View file

@ -1,3 +1,4 @@
import re
import time
from .server import ProxyServer
from .redis_keys import RedisKeys
@ -407,8 +408,6 @@ class ChannelStatus:
if channel_name:
info['channel_name'] = channel_name
info['is_timeshift'] = bool(metadata.get(ChannelMetadataField.IS_TIMESHIFT))
for key, field in (
('logo_id', ChannelMetadataField.LOGO_ID),
('m3u_profile_id', ChannelMetadataField.M3U_PROFILE),
@ -433,7 +432,7 @@ class ChannelStatus:
info['stream_name'] = stream_name
# Add data throughput information to basic info
# TOTAL_BYTES is already present in the hgetall result — avoid a redundant round-trip
# TOTAL_BYTES is already in the hgetall result; skip a redundant round-trip.
total_bytes_bytes = metadata.get(ChannelMetadataField.TOTAL_BYTES)
if total_bytes_bytes:
total_bytes = int(total_bytes_bytes)
@ -552,3 +551,34 @@ class ChannelStatus:
except Exception as e:
logger.error(f"Error getting channel info: {e}", exc_info=True)
return None
def build_live_channel_stats_data(redis_client):
"""Scan Redis for live channel metadata and build the stats payload."""
empty = {"channels": [], "count": 0}
if not redis_client:
return empty
try:
all_channels = []
channel_pattern = "live:channel:*:metadata"
cursor = 0
while True:
cursor, keys = redis_client.scan(cursor, match=channel_pattern)
for key in keys:
key_str = key.decode() if isinstance(key, bytes) else key
channel_id_match = re.search(r"live:channel:(.*):metadata", key_str)
if not channel_id_match:
continue
ch_id = channel_id_match.group(1)
channel_info = ChannelStatus.get_basic_channel_info(ch_id)
if channel_info:
all_channels.append(channel_info)
if cursor == 0:
break
return {"channels": all_channels, "count": len(all_channels)}
except Exception as e:
logger.error(f"Error building live channel stats: {e}", exc_info=True)
return empty

View file

@ -49,27 +49,14 @@ class ClientManager:
def _do_stats_update(self):
"""Perform the stats update in the background."""
try:
from apps.proxy.live_proxy.channel_status import ChannelStatus
from apps.proxy.live_proxy.channel_status import build_live_channel_stats_data
from core.utils import RedisClient
redis_client = RedisClient.get_client()
if not redis_client:
return
all_channels = []
cursor = 0
while True:
cursor, keys = redis_client.scan(cursor, match="live:channel:*:metadata", count=100)
for key in keys:
parts = key.split(':')
if len(parts) >= 4:
ch_id = parts[2]
channel_info = ChannelStatus.get_basic_channel_info(ch_id)
if channel_info:
all_channels.append(channel_info)
if cursor == 0:
break
live_stats = build_live_channel_stats_data(redis_client)
send_websocket_update(
"updates",
@ -77,7 +64,7 @@ class ClientManager:
{
"success": True,
"type": "channel_stats",
"stats": json.dumps({'channels': all_channels, 'count': len(all_channels)})
"stats": json.dumps(live_stats),
}
)
except Exception as e:

View file

@ -54,7 +54,8 @@ class ChannelMetadataField:
STREAM_ID = "stream_id"
CHANNEL_NAME = "channel_name"
STREAM_NAME = "stream_name"
IS_TIMESHIFT = "is_timeshift"
CHANNEL_ID = "channel_id"
CHANNEL_UUID = "channel_uuid"
LOGO_ID = "logo_id"
# Profile fields

View file

@ -2095,9 +2095,6 @@ class ProxyServer:
if not channel_id:
continue
if channel_id.startswith("timeshift_"):
continue
# Get metadata first
metadata = self.redis_client.hgetall(key)
if not metadata:

View file

@ -8,7 +8,7 @@ from django.http import StreamingHttpResponse, JsonResponse, HttpResponseRedirec
from django.views.decorators.csrf import csrf_exempt
from django.shortcuts import get_object_or_404
from .server import ProxyServer
from .channel_status import ChannelStatus
from .channel_status import ChannelStatus, build_live_channel_stats_data
from .output.ts.generator import create_stream_generator
from .output.fmp4.generator import create_fmp4_stream_generator
from .utils import get_client_ip
@ -872,28 +872,7 @@ def channel_status(request, channel_id=None):
{"error": f"Channel {channel_id} not found"}, status=404
)
else:
# Basic info for all channels
channel_pattern = "live:channel:*:metadata"
all_channels = []
# Extract channel IDs from keys
cursor = 0
while True:
cursor, keys = proxy_server.redis_client.scan(
cursor, match=channel_pattern
)
for key in keys:
channel_id_match = re.search(
r"live:channel:(.*):metadata", key
)
if channel_id_match:
ch_id = channel_id_match.group(1)
channel_info = ChannelStatus.get_basic_channel_info(ch_id)
if channel_info:
all_channels.append(channel_info)
if cursor == 0:
break
live_stats = build_live_channel_stats_data(proxy_server.redis_client)
# Send WebSocket update with the stats
# Format it the same way the original Celery task did
@ -903,11 +882,11 @@ def channel_status(request, channel_id=None):
{
"success": True,
"type": "channel_stats",
"stats": json.dumps({'channels': all_channels, 'count': len(all_channels)})
"stats": json.dumps(live_stats),
}
)
return JsonResponse({"channels": all_channels, "count": len(all_channels)})
return JsonResponse(live_stats)
except Exception as e:
logger.error(f"Error in channel_status: {e}", exc_info=True)

31
apps/proxy/stats_views.py Normal file
View file

@ -0,0 +1,31 @@
"""Combined connection stats for live, VOD, and catch-up."""
import logging
import time
from django.http import JsonResponse
from rest_framework.decorators import api_view, permission_classes
from apps.accounts.permissions import IsAdmin
from apps.proxy.live_proxy.channel_status import build_live_channel_stats_data
from apps.proxy.vod_proxy.views import build_vod_stats_data
from apps.timeshift.stats import build_timeshift_stats_data
from core.utils import RedisClient
logger = logging.getLogger(__name__)
@api_view(["GET"])
@permission_classes([IsAdmin])
def combined_stats(request):
"""Return live, VOD, and catch-up stats in one response."""
redis_client = RedisClient.get_client()
if not redis_client:
return JsonResponse({"error": "Redis not available"}, status=500)
return JsonResponse({
"live": build_live_channel_stats_data(redis_client),
"vod": build_vod_stats_data(redis_client),
"catchup": build_timeshift_stats_data(redis_client),
"timestamp": time.time(),
})

View file

@ -1,10 +1,9 @@
from celery import shared_task
import json
import logging
import re
import gc
from core.utils import RedisClient
from apps.proxy.live_proxy.channel_status import ChannelStatus
from apps.proxy.live_proxy.channel_status import build_live_channel_stats_data
from core.utils import send_websocket_update
logger = logging.getLogger(__name__)
@ -17,29 +16,10 @@ def fetch_channel_stats():
redis_client = RedisClient.get_client()
try:
# Basic info for all channels
channel_pattern = "live:channel:*:metadata"
all_channels = []
# Extract channel IDs from keys
cursor = 0
while True:
cursor, keys = redis_client.scan(cursor, match=channel_pattern)
for key in keys:
channel_id_match = re.search(r"live:channel:(.*):metadata", key)
if channel_id_match:
ch_id = channel_id_match.group(1)
channel_info = ChannelStatus.get_basic_channel_info(ch_id)
if channel_info:
all_channels.append(channel_info)
if cursor == 0:
break
live_stats = build_live_channel_stats_data(redis_client)
except Exception as e:
logger.error(f"Error in channel_status: {e}", exc_info=True)
return
# return JsonResponse({'error': str(e)}, status=500)
send_websocket_update(
"updates",
@ -47,13 +27,11 @@ def fetch_channel_stats():
{
"success": True,
"type": "channel_stats",
"stats": json.dumps({'channels': all_channels, 'count': len(all_channels)})
"stats": json.dumps(live_stats),
},
collect_garbage=True
)
# Explicitly clean up large data structures
all_channels = None
gc.collect()

View file

@ -0,0 +1 @@

View file

@ -0,0 +1,110 @@
"""Tests for combined connection stats API and live stats builder."""
import json
from unittest.mock import MagicMock, patch
from django.test import TestCase
from rest_framework.test import APIRequestFactory, force_authenticate
from apps.accounts.models import User
from apps.proxy import stats_views
from apps.proxy.live_proxy.channel_status import build_live_channel_stats_data
class BuildLiveChannelStatsDataTests(TestCase):
@patch("apps.proxy.live_proxy.channel_status.ChannelStatus.get_basic_channel_info")
def test_builds_channel_list_from_metadata_scan(self, mock_get_info):
mock_get_info.side_effect = lambda ch_id: {"channel_id": ch_id}
redis = MagicMock()
redis.scan.return_value = (
0,
[
"live:channel:abc-uuid:metadata",
"live:channel:def-uuid:metadata",
],
)
result = build_live_channel_stats_data(redis)
self.assertEqual(result["count"], 2)
self.assertEqual(
[ch["channel_id"] for ch in result["channels"]],
["abc-uuid", "def-uuid"],
)
def test_returns_empty_when_redis_unavailable(self):
result = build_live_channel_stats_data(None)
self.assertEqual(result, {"channels": [], "count": 0})
@patch("apps.proxy.live_proxy.channel_status.ChannelStatus.get_basic_channel_info")
def test_returns_empty_on_error(self, mock_get_info):
mock_get_info.side_effect = RuntimeError("redis blew up")
redis = MagicMock()
redis.scan.return_value = (0, ["live:channel:abc-uuid:metadata"])
result = build_live_channel_stats_data(redis)
self.assertEqual(result, {"channels": [], "count": 0})
class CombinedStatsApiTests(TestCase):
@classmethod
def setUpTestData(cls):
cls.admin = User.objects.create(
username="combined-stats-admin",
user_level=User.UserLevel.ADMIN,
)
def setUp(self):
self.factory = APIRequestFactory()
@patch("apps.proxy.stats_views.build_timeshift_stats_data")
@patch("apps.proxy.stats_views.build_vod_stats_data")
@patch("apps.proxy.stats_views.build_live_channel_stats_data")
@patch("apps.proxy.stats_views.RedisClient.get_client")
def test_combined_stats_returns_all_sections(
self,
redis_mock,
live_mock,
vod_mock,
catchup_mock,
):
redis_mock.return_value = MagicMock()
live_mock.return_value = {"channels": [{"channel_id": "ch-1"}], "count": 1}
vod_mock.return_value = {
"vod_connections": [],
"total_connections": 0,
"timestamp": 100.0,
}
catchup_mock.return_value = {
"timeshift_sessions": [],
"total_connections": 0,
"timestamp": 100.0,
}
request = self.factory.get("/proxy/stats/")
force_authenticate(request, user=self.admin)
response = stats_views.combined_stats(request)
self.assertEqual(response.status_code, 200)
payload = json.loads(response.content)
self.assertEqual(payload["live"]["count"], 1)
self.assertIn("vod_connections", payload["vod"])
self.assertIn("timeshift_sessions", payload["catchup"])
self.assertIn("timestamp", payload)
live_mock.assert_called_once()
vod_mock.assert_called_once()
catchup_mock.assert_called_once()
@patch("apps.proxy.stats_views.RedisClient.get_client")
def test_combined_stats_redis_unavailable(self, redis_mock):
redis_mock.return_value = None
request = self.factory.get("/proxy/stats/")
force_authenticate(request, user=self.admin)
response = stats_views.combined_stats(request)
self.assertEqual(response.status_code, 500)
self.assertIn("error", json.loads(response.content))

View file

@ -0,0 +1,285 @@
"""Tests for per-user stream limit enforcement in apps.proxy.utils."""
from unittest.mock import MagicMock, patch
from django.test import TestCase
from apps.timeshift.redis_keys import TimeshiftRedisKeys as RedisKeys
from apps.proxy.utils import (
_STOP_REASON_ADMIN,
_timeshift_stop_channel_id,
attempt_stream_termination,
check_user_stream_limits,
stop_timeshift_client,
)
class _FakeRedis:
def __init__(self):
self.store = {}
def setex(self, key, ttl, value):
self.store[key] = str(value)
def hget(self, key, field):
hash_value = self.store.get(key)
return hash_value.get(field) if isinstance(hash_value, dict) else None
def hset(self, key, field=None, value=None, mapping=None, **kwargs):
hash_value = self.store.get(key)
if not isinstance(hash_value, dict):
hash_value = {}
self.store[key] = hash_value
if field is not None and value is not None:
hash_value[str(field)] = str(value)
for f, v in (mapping or {}).items():
hash_value[str(f)] = str(v)
def delete(self, *keys):
count = 0
for key in keys:
if self.store.pop(key, None) is not None:
count += 1
return count
def exists(self, key):
return key in self.store
def sadd(self, key, *members):
existing = self.store.get(key)
if not isinstance(existing, set):
existing = set()
self.store[key] = existing
before = len(existing)
existing.update(str(m) for m in members)
return len(existing) - before
def smembers(self, key):
value = self.store.get(key)
return set(value) if isinstance(value, set) else set()
def scan_iter(self, match=None, count=None): # noqa: ARG002
import fnmatch
for key in list(self.store):
if match is None or fnmatch.fnmatch(str(key), match):
yield key
def eval(self, script, numkeys, *keys_and_args):
if numkeys != 1 or len(keys_and_args) != 2:
raise NotImplementedError("FakeRedis eval only supports claim script")
key, token = keys_and_args
current = self.store.get(key)
if current is not None and str(current) == str(token):
self.store.pop(key, None)
return 1
return 0
class TimeshiftStopChannelIdTests(TestCase):
def setUp(self):
self.redis = _FakeRedis()
self.stats_channel_id = "8_victim"
self.client_id = "victim"
self.programme_vid = "8_2026-06-08-17-00_111"
def test_resolves_programme_vid_from_stats_metadata(self):
client_key = RedisKeys.client_metadata(
self.stats_channel_id, self.client_id,
)
self.redis.hset(client_key, "programme_vid", self.programme_vid)
resolved = _timeshift_stop_channel_id(
self.redis, self.stats_channel_id, self.client_id,
)
self.assertEqual(resolved, self.programme_vid)
def test_falls_back_to_stats_channel_id_for_legacy_entries(self):
resolved = _timeshift_stop_channel_id(
self.redis, self.stats_channel_id, self.client_id,
)
self.assertEqual(resolved, self.stats_channel_id)
class AttemptStreamTerminationTests(TestCase):
def setUp(self):
self.redis = _FakeRedis()
self.user_id = 5
self.requesting_client_id = "newsession"
self.stats_channel_id = "8_victim"
self.programme_vid = "8_2026-06-08-17-00_111"
self.victim_client_id = "victim"
client_key = RedisKeys.client_metadata(
self.stats_channel_id, self.victim_client_id,
)
self.redis.hset(client_key, "programme_vid", self.programme_vid)
def _connections(self):
return [{
"media_id": self.stats_channel_id,
"client_id": self.victim_client_id,
"connected_at": 1000.0,
"type": "timeshift",
}]
def _limits_settings(self):
return {
"terminate_oldest": True,
"prioritize_single_client_channels": True,
"ignore_same_channel_connections": False,
}
def test_timeshift_stop_key_targets_programme_vid_not_stats_channel(self):
with patch("apps.proxy.utils.RedisClient.get_client",
return_value=self.redis), \
patch("apps.proxy.utils.CoreSettings.get_user_limits_settings",
return_value=self._limits_settings()):
ok = attempt_stream_termination(
self.user_id, self.requesting_client_id, self._connections(),
)
self.assertTrue(ok)
programme_stop = RedisKeys.client_stop(
self.programme_vid, self.victim_client_id,
)
stats_stop = RedisKeys.client_stop(
self.stats_channel_id, self.victim_client_id,
)
self.assertIn(programme_stop, self.redis.store)
self.assertNotIn(stats_stop, self.redis.store)
def test_timeshift_limit_stop_uses_generation_key_when_metadata_missing(self):
client_key = RedisKeys.client_metadata(
self.stats_channel_id, self.victim_client_id,
)
self.redis.store.pop(client_key, None)
generation_key = (
f"timeshift:stream_gen:{self.programme_vid}:{self.victim_client_id}"
)
self.redis.store[generation_key] = "4"
with patch("apps.proxy.utils.RedisClient.get_client",
return_value=self.redis), \
patch("apps.proxy.utils.CoreSettings.get_user_limits_settings",
return_value=self._limits_settings()):
ok = attempt_stream_termination(
self.user_id, self.requesting_client_id, self._connections(),
)
self.assertTrue(ok)
programme_stop = RedisKeys.client_stop(
self.programme_vid, self.victim_client_id,
)
self.assertIn(programme_stop, self.redis.store)
def test_limit_termination_allows_new_timeshift_when_victim_stopped(self):
user = MagicMock(id=5, username="viewer", stream_limit=1)
new_session = "newsession"
connections = self._connections()
settings = {
**self._limits_settings(),
"terminate_on_limit_exceeded": True,
}
with patch("apps.proxy.utils.get_user_active_connections",
return_value=connections), \
patch("apps.proxy.utils.RedisClient.get_client",
return_value=self.redis), \
patch("apps.proxy.utils.CoreSettings.get_user_limits_settings",
return_value=settings):
allowed = check_user_stream_limits(
user, new_session, media_id="8_2026-06-08-17-30",
)
self.assertTrue(allowed)
programme_stop = RedisKeys.client_stop(
self.programme_vid, self.victim_client_id,
)
self.assertIn(programme_stop, self.redis.store)
def test_live_termination_still_uses_channel_service(self):
connections = [{
"media_id": "42",
"client_id": "live_client_1",
"connected_at": 1000.0,
"type": "live",
}]
with patch("apps.proxy.utils.CoreSettings.get_user_limits_settings",
return_value=self._limits_settings()), \
patch("apps.proxy.utils.ChannelService.stop_client",
return_value={"status": "ok"}) as stop_mock:
ok = attempt_stream_termination(
self.user_id, self.requesting_client_id, connections,
)
self.assertTrue(ok)
stop_mock.assert_called_once_with("42", "live_client_1")
class TimeshiftAdminStopTests(TestCase):
def setUp(self):
self.redis = _FakeRedis()
self.stats_channel_id = "8_victim"
self.programme_vid = "8_2026-06-08-17-00_111"
self.client_id = "victim"
client_key = RedisKeys.client_metadata(
self.stats_channel_id, self.client_id,
)
self.redis.hset(client_key, "programme_vid", self.programme_vid)
self.redis.sadd(RedisKeys.clients(self.stats_channel_id), self.client_id)
self.redis.hset(
RedisKeys.channel_metadata(self.stats_channel_id),
"state", "active",
)
def test_admin_stop_client_signals_programme_and_unregisters_stats(self):
session_key = RedisKeys.api_session(self.client_id)
self.redis.hset(session_key, "user_id", "1")
with patch("apps.timeshift.views._unregister_stats_client") as unregister_mock, \
patch("apps.timeshift.views._trigger_timeshift_stats_update") as trigger_mock:
result = stop_timeshift_client(
self.redis, self.stats_channel_id, self.client_id,
)
self.assertEqual(result["status"], "success")
programme_stop = RedisKeys.client_stop(self.programme_vid, self.client_id)
stats_stop = RedisKeys.client_stop(self.stats_channel_id, self.client_id)
self.assertEqual(self.redis.store.get(programme_stop), _STOP_REASON_ADMIN)
self.assertNotIn(stats_stop, self.redis.store)
self.assertNotIn(session_key, self.redis.store)
unregister_mock.assert_called_once_with(
self.redis, self.stats_channel_id, self.client_id,
)
trigger_mock.assert_called_once_with(self.redis)
def test_admin_stop_closes_local_upstream_when_registered(self):
from apps.timeshift.views import _close_active_upstream, _register_active_upstream
upstream = MagicMock()
_register_active_upstream(self.programme_vid, self.client_id, upstream)
with patch("apps.timeshift.views._unregister_stats_client"):
result = stop_timeshift_client(
self.redis, self.stats_channel_id, self.client_id,
)
self.assertEqual(result["status"], "success")
upstream.close.assert_called_once()
_close_active_upstream(self.programme_vid, self.client_id)
upstream.close.assert_called_once()
def test_admin_stop_uses_generation_key_when_metadata_missing(self):
client_key = RedisKeys.client_metadata(self.stats_channel_id, self.client_id)
self.redis.store.pop(client_key, None)
generation_key = f"timeshift:stream_gen:{self.programme_vid}:{self.client_id}"
self.redis.store[generation_key] = "4"
with patch("apps.timeshift.views._unregister_stats_client"):
result = stop_timeshift_client(
self.redis, self.stats_channel_id, self.client_id,
)
self.assertEqual(result["status"], "success")
programme_stop = RedisKeys.client_stop(self.programme_vid, self.client_id)
self.assertEqual(self.redis.store.get(programme_stop), _STOP_REASON_ADMIN)
self.assertIn(self.programme_vid, result["stop_channel_ids"])

View file

@ -1,9 +1,13 @@
from django.urls import path, include
from apps.proxy import stats_views
app_name = 'proxy'
urlpatterns = [
path('stats/', stats_views.combined_stats, name='combined_stats'),
path('ts/', include('apps.proxy.live_proxy.urls')),
path('catchup/', include('apps.timeshift.urls')),
path('hls/', include('apps.proxy.hls_proxy.urls')),
path('vod/', include('apps.proxy.vod_proxy.urls')),
]

View file

@ -1,12 +1,145 @@
import logging
import re
from core.utils import RedisClient
from apps.proxy.vod_proxy.multi_worker_connection_manager import MultiWorkerVODConnectionManager, get_vod_client_stop_key
from apps.proxy.live_proxy.redis_keys import RedisKeys
from apps.timeshift.redis_keys import (
TimeshiftRedisKeys,
parse_stats_channel_id,
)
from core.models import CoreSettings
from apps.proxy.live_proxy.services.channel_service import ChannelService
logger = logging.getLogger("proxy")
_STOP_REASON_LIMIT = "limit"
_STOP_REASON_ADMIN = "admin"
def _timeshift_stop_channel_id(redis_client, stats_channel_id, client_id, fallback=None):
"""Return the programme virtual_channel_id used for generator stop keys."""
client_key = TimeshiftRedisKeys.client_metadata(stats_channel_id, client_id)
programme_vid = redis_client.hget(client_key, "programme_vid")
if programme_vid:
if isinstance(programme_vid, bytes):
programme_vid = programme_vid.decode()
return programme_vid
return fallback if fallback is not None else stats_channel_id
def _timeshift_generation_channel_ids(redis_client, client_id):
"""Return virtual channel ids with active generation counters for a client."""
if redis_client is None or not client_id:
return []
prefix = "timeshift:stream_gen:"
suffix = f":{client_id}"
try:
keys = redis_client.scan_iter(
match=f"{prefix}*{suffix}", count=100,
)
channel_ids = []
for key in keys:
if isinstance(key, bytes):
key = key.decode()
key = str(key)
if not key.startswith(prefix) or not key.endswith(suffix):
continue
channel_id = key[len(prefix):-len(suffix)]
if channel_id:
channel_ids.append(channel_id)
return channel_ids
except Exception as exc:
logger.debug(
"Timeshift generation stop-key scan failed for %s: %s",
client_id, exc,
)
return []
def _timeshift_stop_channel_ids(redis_client, stats_channel_id, client_id, fallback=None):
"""Return all Redis stop-key channel ids that may be observed by a stream."""
candidates = [
_timeshift_stop_channel_id(
redis_client, stats_channel_id, client_id, fallback=fallback,
)
]
candidates.extend(_timeshift_generation_channel_ids(redis_client, client_id))
seen = set()
ordered = []
for channel_id in candidates:
if not channel_id or channel_id in seen:
continue
seen.add(channel_id)
ordered.append(channel_id)
return ordered
def _set_timeshift_stop_keys(redis_client, stats_channel_id, client_id, reason, fallback=None):
"""Signal a timeshift stream to stop across workers."""
stop_channel_ids = _timeshift_stop_channel_ids(
redis_client, stats_channel_id, client_id, fallback=fallback,
)
for stop_channel_id in stop_channel_ids:
stop_key = TimeshiftRedisKeys.client_stop(stop_channel_id, client_id)
redis_client.setex(stop_key, 60, reason)
return stop_channel_ids
def stop_timeshift_client(redis_client, stats_channel_id, client_id):
"""Stop one catch-up viewer (admin Stats UI and ``POST /proxy/catchup/stop_client/``)."""
if redis_client is None:
return {"status": "error", "message": "Redis unavailable"}
if not stats_channel_id or not client_id:
return {"status": "error", "message": "Missing channel or client id"}
from apps.timeshift.views import (
_cancel_stats_disconnect_grace,
_cleanup_all_stream_generations,
_close_active_upstream,
_finalize_playback_session_auth,
_trigger_timeshift_stats_update,
_unregister_stats_client,
)
try:
stop_channel_ids = _set_timeshift_stop_keys(
redis_client, stats_channel_id, client_id, _STOP_REASON_ADMIN,
)
for stop_channel_id in stop_channel_ids:
_close_active_upstream(stop_channel_id, client_id)
_cancel_stats_disconnect_grace(redis_client, stats_channel_id, client_id)
_unregister_stats_client(redis_client, stats_channel_id, client_id)
_cleanup_all_stream_generations(redis_client, client_id)
_finalize_playback_session_auth(redis_client, client_id)
try:
from apps.timeshift.views import _superseded_pool_key
redis_client.delete(_superseded_pool_key(client_id))
except Exception:
pass
except Exception as exc:
logger.error(
"Timeshift admin stop failed for %s on %s: %s",
client_id, stats_channel_id, exc,
)
return {"status": "error", "message": str(exc)}
_trigger_timeshift_stats_update(redis_client)
logger.info(
"Timeshift admin stop: client=%s stats_channel=%s stop_channels=%s",
client_id, stats_channel_id, stop_channel_ids,
)
return {
"status": "success",
"message": "Timeshift client stop signal sent",
"channel_id": stats_channel_id,
"client_id": client_id,
"stop_channel_id": stop_channel_ids[0] if stop_channel_ids else stats_channel_id,
"stop_channel_ids": stop_channel_ids,
"stop_key_set": True,
"locally_processed": False,
}
def attempt_stream_termination(user_id, requesting_client_id, active_connections):
try:
@ -48,9 +181,7 @@ def attempt_stream_termination(user_id, requesting_client_id, active_connections
f"on media {target['media_id']} (connected_at={target['connected_at']})"
)
# When counting by unique channel, freeing one connection from a multi-connection
# channel doesn't free a slot — terminate all connections to that channel so the
# unique-channel count actually drops by one.
# When counting by unique channel, stop all connections on that channel.
targets = (
[c for c in active_connections if c['media_id'] == target['media_id']]
if ignore_same_channel
@ -63,14 +194,20 @@ def attempt_stream_termination(user_id, requesting_client_id, active_connections
if result.get("status") == "error":
logger.warning(f"[stream limits][{requesting_client_id}] Failed to stop client {t['client_id']} on channel {t['media_id']}")
elif t['type'] == 'timeshift':
# Same Redis stop key as live; timeshift generator polls it every 5s.
redis_client = RedisClient.get_client()
if not redis_client:
# Deny the new stream if we cannot stop the old one.
return False
stop_key = RedisKeys.client_stop(t['media_id'], t['client_id'])
redis_client.setex(stop_key, 60, "true")
logger.info(f"[stream limits][{requesting_client_id}] Set stop key for timeshift client {t['client_id']}")
stop_channel_ids = _set_timeshift_stop_keys(
redis_client, t['media_id'], t['client_id'], _STOP_REASON_LIMIT,
)
from apps.timeshift.views import _close_active_upstream
for stop_channel_id in stop_channel_ids:
_close_active_upstream(stop_channel_id, t['client_id'])
logger.info(
f"[stream limits][{requesting_client_id}] Set stop key for "
f"timeshift client {t['client_id']} on {stop_channel_ids}",
)
else:
connection_manager = MultiWorkerVODConnectionManager.get_instance()
redis_client = connection_manager.redis_client
@ -101,32 +238,35 @@ def get_user_active_connections(user_id):
connections = []
try:
# Grab live streams
for key in redis_client.scan_iter(match="live:channel:*:clients:*", count=1000):
parts = key.split(':')
if len(parts) >= 5:
channel_id = parts[2]
client_id = parts[4]
# Grab live and timeshift streams (same key layout, separate namespaces)
for pattern, conn_type in (
("live:channel:*:clients:*", "live"),
("timeshift:channel:*:clients:*", "timeshift"),
):
for key in redis_client.scan_iter(match=pattern, count=1000):
parts = key.split(':')
if len(parts) >= 5:
channel_id = parts[2]
client_id = parts[4]
client_user_id, connected_at = redis_client.hmget(key, 'user_id', 'connected_at')
client_user_id, connected_at = redis_client.hmget(key, 'user_id', 'connected_at')
logger.debug(f"[stream limits] user_id = {user_id}")
logger.debug(f"[stream limits] channel_id = {channel_id}")
logger.debug(f"[stream limits] client_id = {client_id}")
logger.debug(f"[stream limits] user_id = {user_id}")
logger.debug(f"[stream limits] channel_id = {channel_id}")
logger.debug(f"[stream limits] client_id = {client_id}")
if user_id is None or (client_user_id and int(client_user_id) == user_id):
try:
conn_type = 'timeshift' if channel_id.startswith('timeshift_') else 'live'
logger.debug(f"[stream limits] Found {conn_type.upper()} connection for user {user_id} on channel {channel_id} with client ID {client_id}")
connected_at = float(connected_at) if connected_at else 0
connections.append({
'media_id': channel_id,
'client_id': client_id,
'connected_at': connected_at,
'type': conn_type,
})
except (ValueError, TypeError):
pass
if user_id is None or (client_user_id and int(client_user_id) == user_id):
try:
logger.debug(f"[stream limits] Found {conn_type.upper()} connection for user {user_id} on channel {channel_id} with client ID {client_id}")
connected_at = float(connected_at) if connected_at else 0
connections.append({
'media_id': channel_id,
'client_id': client_id,
'connected_at': connected_at,
'type': conn_type,
})
except (ValueError, TypeError):
pass
# Grab VOD
for key in redis_client.scan_iter(match="vod_persistent_connection:*", count=1000):
@ -187,6 +327,10 @@ def check_user_stream_limits(user, client_id, media_id=None):
# session_id. Each distinct client/session still consumes its own slot.
if media_id and client_id:
media_id_str = str(media_id)
req_channel = None
channel_match = re.match(r"^(\d+)_", media_id_str)
if channel_match:
req_channel = channel_match.group(1)
for conn in active_connections:
if conn.get('type') != 'timeshift':
continue
@ -198,6 +342,17 @@ def check_user_stream_limits(user, client_id, media_id=None):
f"[stream limits][{client_id}] Same timeshift session probe for {media_id} allowed"
)
return True
if req_channel:
parsed_conn = parse_stats_channel_id(conn_media_id)
if parsed_conn:
conn_channel = str(parsed_conn["channel_id"])
else:
conn_channel = conn_media_id.split("_", 1)[0]
if conn_channel == req_channel:
logger.debug(
f"[stream limits][{client_id}] Same timeshift channel {req_channel} allowed"
)
return True
if user_stream_count >= user.stream_limit:
if user_limit_settings.get("terminate_on_limit_exceeded", True) == False:

View file

@ -0,0 +1,18 @@
from django.urls import path
from . import api_views
app_name = "catchup"
urlpatterns = [
path(
"sessions/",
api_views.CatchupSessionCreateAPIView.as_view(),
name="catchup-session-create",
),
path(
"sessions/<str:session_id>/",
api_views.CatchupSessionDestroyAPIView.as_view(),
name="catchup-session-destroy",
),
]

168
apps/timeshift/api_views.py Normal file
View file

@ -0,0 +1,168 @@
"""REST API for native catch-up playback session setup."""
from django.http import Http404
from drf_spectacular.utils import extend_schema, inline_serializer
from rest_framework import serializers, status
from rest_framework.response import Response
from rest_framework.views import APIView
from apps.accounts.permissions import IsStandardUser
from apps.channels.models import Channel
from apps.channels.utils import get_channel_catchup_streams
from dispatcharr.utils import network_access_allowed
from .helpers import parse_catchup_timestamp
from .sessions import (
HANDSHAKE_TTL_SECONDS,
SESSION_IDLE_TTL_SECONDS,
create_catchup_session,
delete_catchup_session,
user_owns_catchup_session,
)
from .views import _user_can_access_channel
class CatchupSessionCreateSerializer(serializers.Serializer):
channel_uuid = serializers.UUIDField(
help_text="Dispatcharr channel UUID to play catch-up from.",
)
start = serializers.CharField(
help_text=(
"UTC programme start time (which archived show to play). "
"ISO-8601 (2026-07-09T14:00:00Z), Unix epoch, or XC wall-clock shapes."
),
)
class CatchupSessionResponseSerializer(serializers.Serializer):
session_id = serializers.CharField()
playback_url = serializers.CharField()
expires_at = serializers.IntegerField(
help_text="Unix timestamp. Handshake deadline: first playback GET within this window.",
)
channel_uuid = serializers.UUIDField()
start = serializers.CharField()
class CatchupSessionCreateAPIView(APIView):
"""Mint a server-side playback session for headerless video players."""
permission_classes = [IsStandardUser]
@extend_schema(
description=(
"Create a catch-up playback session for a single archived programme.\n\n"
"Call this once per programme with a JWT or API key. The response "
"includes a ``playback_url`` for the video player **without** an "
"embedded access token.\n\n"
"**``start``** is the programme's broadcast start time in UTC "
"(from EPG ``start_time``). It selects *which* archived show to "
"fetch, not when the viewer pressed play.\n\n"
f"The player should open ``playback_url`` within "
f"**{HANDSHAKE_TTL_SECONDS} seconds** (see ``expires_at``). After the "
"first byte request, the session stays valid with a "
f"**{SESSION_IDLE_TTL_SECONDS // 60}-minute sliding idle window** "
"(refreshed on each range/seek request).\n\n"
"Start a **new** session for each different programme."
),
request=CatchupSessionCreateSerializer,
responses={
201: CatchupSessionResponseSerializer,
400: inline_serializer(
name="CatchupSessionCreateError",
fields={"error": serializers.CharField()},
),
403: inline_serializer(
name="CatchupSessionForbidden",
fields={"error": serializers.CharField()},
),
404: inline_serializer(
name="CatchupSessionNotFound",
fields={"error": serializers.CharField()},
),
},
tags=["catchup"],
)
def post(self, request):
user = request.user
if not network_access_allowed(request, "STREAMS", user):
return Response({"error": "Forbidden"}, status=status.HTTP_403_FORBIDDEN)
body = CatchupSessionCreateSerializer(data=request.data)
body.is_valid(raise_exception=True)
start = body.validated_data["start"].strip()
if parse_catchup_timestamp(start) is None:
return Response(
{"error": "Invalid start timestamp"},
status=status.HTTP_400_BAD_REQUEST,
)
channel_uuid = body.validated_data["channel_uuid"]
try:
channel = Channel.objects.get(uuid=channel_uuid)
except Channel.DoesNotExist:
raise Http404("Channel not found") from None
if not _user_can_access_channel(user, channel):
return Response({"error": "Access denied"}, status=status.HTTP_403_FORBIDDEN)
if not channel.is_catchup:
return Response(
{"error": "Catch-up not supported for this channel"},
status=status.HTTP_400_BAD_REQUEST,
)
if not get_channel_catchup_streams(channel):
return Response(
{"error": "Catch-up not supported for this channel"},
status=status.HTTP_400_BAD_REQUEST,
)
try:
payload = create_catchup_session(user=user, channel=channel, start=start)
except RuntimeError:
return Response(
{"error": "Session service unavailable"},
status=status.HTTP_503_SERVICE_UNAVAILABLE,
)
return Response(
CatchupSessionResponseSerializer(payload).data,
status=status.HTTP_201_CREATED,
)
class CatchupSessionDestroyAPIView(APIView):
"""Revoke a catch-up playback session early."""
permission_classes = [IsStandardUser]
@extend_schema(
description=(
"Delete a catch-up session before it expires. Only the user who "
"created the session may revoke it. Returns 404 when the session "
"is missing or owned by another user."
),
responses={
204: None,
403: inline_serializer(
name="CatchupSessionDestroyForbidden",
fields={"error": serializers.CharField()},
),
404: inline_serializer(
name="CatchupSessionDestroyNotFound",
fields={"error": serializers.CharField()},
),
},
tags=["catchup"],
)
def delete(self, request, session_id):
if not network_access_allowed(request, "STREAMS", request.user):
return Response({"error": "Forbidden"}, status=status.HTTP_403_FORBIDDEN)
if not user_owns_catchup_session(session_id, request.user.id):
return Response({"error": "Session not found"}, status=status.HTTP_404_NOT_FOUND)
delete_catchup_session(session_id)
return Response(status=status.HTTP_204_NO_CONTENT)

View file

@ -42,6 +42,7 @@ def normalize_catchup_timestamp_input(timestamp_str):
- ``YYYY-MM-DD_HH-MM`` (XC underscore)
- ``YYYY-MM-DD:HH:MM[:SS]`` (XC colon time in catch-up URLs)
- ``YYYY-MM-DD HH:MM[:SS]`` (EPG / SQL datetime)
- ISO-8601 UTC (``2026-07-09T14:00:00Z`` or with offset)
- Unix epoch seconds (10 digits) or milliseconds (13 digits)
Returns:
@ -66,6 +67,15 @@ def normalize_catchup_timestamp_input(timestamp_str):
return dt.replace(tzinfo=None).isoformat(timespec="seconds")
return None
if "T" in value:
try:
dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
if dt.tzinfo is not None:
dt = dt.astimezone(timezone.utc).replace(tzinfo=None)
return dt.isoformat(timespec="seconds")
except ValueError:
return None
match = _CATCHUP_WALL_CLOCK_RE.match(value)
if not match:
return None
@ -173,6 +183,85 @@ def get_programme_duration(channel, timestamp_str):
return DEFAULT_DURATION_MINUTES
def get_programme_info(channel, timestamp_str):
"""Return EPG metadata for the programme airing at *timestamp_str*."""
try:
dt = parse_catchup_timestamp(timestamp_str)
if dt is None:
return None
dt = dt.replace(tzinfo=timezone.utc)
if not channel or not getattr(channel, "epg_data", None):
return None
programme = channel.epg_data.programs.filter(
start_time__lte=dt, end_time__gt=dt
).first()
if not programme:
return None
duration_seconds = (programme.end_time - programme.start_time).total_seconds()
return {
"title": programme.title,
"sub_title": programme.sub_title or "",
"description": programme.description or "",
"start_time": programme.start_time.isoformat(),
"end_time": programme.end_time.isoformat(),
"duration_secs": int(duration_seconds),
}
except Exception:
return None
def get_catchup_programmes_for_sessions(sessions):
"""Resolve EPG metadata for catch-up stats cards (batch, on demand).
Each session dict needs ``session_id``, ``channel_uuid``, and
``programme_start``. Called by the frontend when active sessions or seek
position change, not on every stats poll.
"""
from apps.channels.models import Channel
if not sessions:
return []
valid = [
s for s in sessions
if s.get("channel_uuid") and s.get("programme_start") and s.get("session_id")
]
if not valid:
return []
valid = valid[:50]
uuids = list({str(s["channel_uuid"]) for s in valid})
channels_by_uuid = {
str(ch.uuid): ch
for ch in Channel.objects.filter(uuid__in=uuids).select_related("epg_data")
}
results = []
for session in valid:
channel_uuid = str(session["channel_uuid"])
programme_start = session["programme_start"]
channel = channels_by_uuid.get(channel_uuid)
info = get_programme_info(channel, programme_start) if channel else None
entry = {
"session_id": session["session_id"],
"channel_uuid": channel_uuid,
"programme_start": programme_start,
}
if info:
entry.update({
"title": info["title"],
"sub_title": info.get("sub_title", ""),
"description": info.get("description", ""),
"start_time": info["start_time"],
"end_time": info["end_time"],
"duration_secs": info["duration_secs"],
})
results.append(entry)
return results
def build_timeshift_url_format_a(creds, stream_id, timestamp, duration_minutes):
"""QUERY layout: ``/streaming/timeshift.php?username=...&start=...``"""
return (

View file

@ -0,0 +1,96 @@
"""Redis key patterns and logical ids for catch-up (timeshift).
All catch-up state lives under the ``timeshift:`` prefix so live-proxy
janitors and live stats scans never see it.
"""
import re
import secrets
# Logical id: stats channel key segment ``timeshift:channel:{id}:...``
_STATS_CHANNEL_RE = re.compile(r"^(\d+)_(.+)$")
class TimeshiftRedisKeys:
@staticmethod
def channel_metadata(channel_id):
return f"timeshift:channel:{channel_id}:metadata"
@staticmethod
def clients(channel_id):
return f"timeshift:channel:{channel_id}:clients"
@staticmethod
def client_metadata(channel_id, client_id):
return f"timeshift:channel:{channel_id}:clients:{client_id}"
@staticmethod
def client_stop(channel_id, client_id):
return f"timeshift:channel:{channel_id}:client:{client_id}:stop"
@staticmethod
def pool(session_id):
return f"timeshift:pool:{session_id}"
@staticmethod
def pool_lock(session_id):
return f"timeshift:pool:{session_id}:lock"
@staticmethod
def pool_superseded(session_id):
return f"timeshift:pool:{session_id}:superseded"
@staticmethod
def pool_scan_pattern():
return "timeshift:pool:*"
@staticmethod
def api_session(session_id):
return f"timeshift:session:{session_id}"
@staticmethod
def stats_grace(stats_channel_id, client_id):
return f"timeshift:grace:{stats_channel_id}:{client_id}"
@staticmethod
def stream_generation(virtual_channel_id, client_id):
return f"timeshift:stream_gen:{virtual_channel_id}:{client_id}"
@staticmethod
def stream_generation_scan_pattern(client_id):
return f"timeshift:stream_gen:*:{client_id}"
@staticmethod
def format_cache(account_id):
return f"timeshift:format_idx:{account_id}"
def mint_session_id():
"""Opaque per-viewer session id (URL query param and pool key suffix)."""
return secrets.token_urlsafe(16)
def stats_channel_id(channel_id, session_id):
"""Stable stats/redis channel id for one viewer on a channel."""
return f"{channel_id}_{session_id}"
def programme_media_id(channel_id, safe_ts):
"""Catch-up position identity (channel + programme timestamp)."""
return f"{channel_id}_{safe_ts}"
def virtual_channel_id(channel_id, safe_ts, stream_id_value):
"""Programme virtual id for stop keys and stream generation."""
return f"{channel_id}_{safe_ts}_{stream_id_value}"
def parse_stats_channel_id(stats_channel_id):
"""Split a stats channel id into numeric channel id and session id."""
match = _STATS_CHANNEL_RE.match(str(stats_channel_id or ""))
if not match:
return None
return {
"channel_id": int(match.group(1)),
"session_id": match.group(2),
}

197
apps/timeshift/sessions.py Normal file
View file

@ -0,0 +1,197 @@
"""Redis-backed catch-up playback sessions for native API clients.
A session is minted by ``POST /api/catchup/sessions/`` (JWT/API key). The
returned ``session_id`` lets a headerless video player call
``GET /proxy/catchup/{uuid}?session_id=...`` without embedding a JWT in the URL.
Lifecycle:
* **Handshake**: unused sessions expire ``HANDSHAKE_TTL_SECONDS`` after POST.
* **Playback**: the first GET extends TTL to ``SESSION_IDLE_TTL_SECONDS``;
each subsequent GET (or active-stream heartbeat) refreshes that sliding
window. Pausing longer than this without a new request requires minting
a new session.
* **End of viewing**: when the client disconnects for real (not a seek within
the same session), the playback layer deletes the record after a short grace
window so stale ``session_id`` values cannot be replayed until TTL expiry.
* **User resolution**: prefer ``timeshift:pool:{session_id}.user_id`` while
the provider pool entry exists; fall back to the API session record when the
pool is idle/expired (pause gaps between HTTP range requests).
"""
import logging
import time
from apps.accounts.models import User
from apps.channels.models import Channel
from apps.timeshift.redis_keys import TimeshiftRedisKeys, mint_session_id
from core.utils import RedisClient
logger = logging.getLogger(__name__)
HANDSHAKE_TTL_SECONDS = 60
# Max idle pause between range/seek requests (refreshed on each playback GET).
SESSION_IDLE_TTL_SECONDS = 10 * 60
def mint_catchup_session_id():
"""Backward-compatible alias for :func:`mint_session_id`."""
return mint_session_id()
def create_catchup_session(*, user, channel, start):
"""Persist a new playback session and return metadata for the API response."""
redis_client = RedisClient.get_client()
if redis_client is None:
raise RuntimeError("Redis unavailable")
session_id = mint_session_id()
now = int(time.time())
key = TimeshiftRedisKeys.api_session(session_id)
redis_client.hset(
key,
mapping={
"user_id": str(user.id),
"channel_uuid": str(channel.uuid),
"channel_id": str(channel.id),
"start": str(start),
"created_at": str(now),
},
)
redis_client.expire(key, HANDSHAKE_TTL_SECONDS)
handshake_expires_at = now + HANDSHAKE_TTL_SECONDS
playback_url = f"/proxy/catchup/{channel.uuid}?session_id={session_id}"
return {
"session_id": session_id,
"playback_url": playback_url,
"expires_at": handshake_expires_at,
"channel_uuid": str(channel.uuid),
"start": str(start),
}
def get_catchup_session(session_id):
"""Return session fields as a dict, or None if missing."""
redis_client = RedisClient.get_client()
if redis_client is None or not session_id:
return None
try:
data = redis_client.hgetall(TimeshiftRedisKeys.api_session(session_id))
except Exception as exc:
logger.warning("Catchup session read failed for %s: %s", session_id, exc)
return None
if not data:
return None
return data
def touch_catchup_session(session_id, *, redis_client=None):
"""Extend sliding idle TTL after a playback request uses the session."""
if redis_client is None:
redis_client = RedisClient.get_client()
if redis_client is None or not session_id:
return False
key = TimeshiftRedisKeys.api_session(session_id)
try:
if not redis_client.exists(key):
return False
redis_client.expire(key, SESSION_IDLE_TTL_SECONDS)
return True
except Exception as exc:
logger.warning("Catchup session touch failed for %s: %s", session_id, exc)
return False
def delete_catchup_session(session_id, *, redis_client=None):
if not session_id:
return False
if redis_client is None:
redis_client = RedisClient.get_client()
if redis_client is None:
return False
try:
deleted = bool(redis_client.delete(TimeshiftRedisKeys.api_session(session_id)))
if deleted:
logger.debug("Catchup session deleted: %s", session_id)
return deleted
except Exception as exc:
logger.warning("Catchup session delete failed for %s: %s", session_id, exc)
return False
def catchup_session_exists(session_id, *, redis_client=None):
"""Return True when *session_id* has an API session record."""
if not session_id:
return False
if redis_client is None:
redis_client = RedisClient.get_client()
if redis_client is None:
return False
try:
return bool(redis_client.exists(TimeshiftRedisKeys.api_session(session_id)))
except Exception:
return False
def _user_id_from_pool(session_id):
redis_client = RedisClient.get_client()
if redis_client is None or not session_id:
return None
try:
data = redis_client.hgetall(TimeshiftRedisKeys.pool(session_id))
except Exception:
return None
if not data:
return None
uid = data.get("user_id")
if not uid:
return None
try:
return int(uid)
except (TypeError, ValueError):
return None
def resolve_catchup_playback(session_id, channel_uuid):
"""Resolve user and programme start for a tokenless playback request.
Returns:
``(user, start)`` on success, or ``None`` if the session is invalid,
expired, or bound to a different channel.
"""
record = get_catchup_session(session_id)
if not record:
return None
if str(record.get("channel_uuid") or "") != str(channel_uuid):
return None
touch_catchup_session(session_id)
user_id = _user_id_from_pool(session_id)
if user_id is None:
try:
user_id = int(record.get("user_id") or "")
except (TypeError, ValueError):
return None
user = User.objects.filter(id=user_id, is_active=True).first()
if user is None:
return None
start = record.get("start")
if not start:
return None
return user, str(start)
def user_owns_catchup_session(session_id, user_id):
record = get_catchup_session(session_id)
if not record:
return False
try:
return int(record.get("user_id") or "") == int(user_id)
except (TypeError, ValueError):
return False

480
apps/timeshift/stats.py Normal file
View file

@ -0,0 +1,480 @@
"""Build catch-up (timeshift) connection stats for the admin UI."""
import logging
import re
import time
from django.db import close_old_connections
from datetime import timezone as dt_timezone
from apps.channels.models import Channel
from apps.m3u.models import M3UAccountProfile
from apps.proxy.live_proxy.constants import ChannelMetadataField
from apps.timeshift.redis_keys import TimeshiftRedisKeys, parse_stats_channel_id
from apps.timeshift.helpers import parse_catchup_timestamp
logger = logging.getLogger(__name__)
_STREAM_STATS_TO_METADATA = {
"video_codec": ChannelMetadataField.VIDEO_CODEC,
"resolution": ChannelMetadataField.RESOLUTION,
"source_fps": ChannelMetadataField.SOURCE_FPS,
"pixel_format": ChannelMetadataField.PIXEL_FORMAT,
"video_bitrate": ChannelMetadataField.VIDEO_BITRATE,
"audio_codec": ChannelMetadataField.AUDIO_CODEC,
"sample_rate": ChannelMetadataField.SAMPLE_RATE,
"audio_channels": ChannelMetadataField.AUDIO_CHANNELS,
"audio_bitrate": ChannelMetadataField.AUDIO_BITRATE,
"stream_type": ChannelMetadataField.STREAM_TYPE,
}
# Redis metadata → stats API / UI display keys (matches live stats cards).
_METADATA_TO_DISPLAY = {
ChannelMetadataField.RESOLUTION: "resolution",
ChannelMetadataField.SOURCE_FPS: "source_fps",
ChannelMetadataField.VIDEO_CODEC: "video_codec",
ChannelMetadataField.AUDIO_CODEC: "audio_codec",
ChannelMetadataField.AUDIO_CHANNELS: "audio_channels",
ChannelMetadataField.STREAM_TYPE: "stream_type",
}
def stream_stats_to_metadata_fields(stream_stats):
"""Map a ``Stream.stream_stats`` dict to Redis channel metadata fields."""
if not stream_stats:
return {}
out = {}
for src_key, field_name in _STREAM_STATS_TO_METADATA.items():
value = stream_stats.get(src_key)
if value is not None and value != "":
out[field_name] = str(value)
if out:
out[ChannelMetadataField.STREAM_INFO_UPDATED] = str(time.time())
return out
def stream_info_from_metadata(metadata):
"""Extract display-oriented stream stats from Redis channel metadata."""
if not metadata:
return {}
out = {}
for field_name, display_key in _METADATA_TO_DISPLAY.items():
value = metadata.get(field_name)
if value is not None and value != "":
out[display_key] = _decode_value(value)
return out
def seed_stream_stats_metadata(
redis_client,
metadata_key,
metadata_payload,
*,
stats_stream_id,
stream_stats=None,
):
"""Copy stream stats into catch-up Redis metadata when the upstream stream changes.
``stream_stats`` should come from the already-loaded ``catchup_stream`` row
when available. A DB lookup is only attempted when the pool reuses a session
whose metadata was evicted and no in-memory stats were passed.
"""
if stats_stream_id is None or redis_client is None:
return
sid = str(stats_stream_id)
try:
existing = redis_client.hget(metadata_key, ChannelMetadataField.STREAM_ID)
if isinstance(existing, bytes):
existing = existing.decode()
except Exception:
existing = None
if existing == sid:
return
stats = stream_stats
if not stats:
try:
from apps.channels.models import Stream
row = Stream.objects.filter(id=int(sid)).values("stream_stats").first()
stats = row.get("stream_stats") if row else None
except Exception as exc:
logger.debug("Timeshift stream stats lookup failed for %s: %s", sid, exc)
stats = None
fields = stream_stats_to_metadata_fields(stats)
if not fields:
return
metadata_payload.update(fields)
metadata_payload[ChannelMetadataField.STREAM_ID] = sid
def _decode_value(value):
if isinstance(value, bytes):
return value.decode()
return value
def _decode_hash(data):
if not data:
return {}
return {_decode_value(k): _decode_value(v) for k, v in data.items()}
def compute_playback_base_from_byte_range(range_start, content_length, duration_secs):
"""Map a byte offset into an approximate programme position in seconds."""
if range_start is None or range_start <= 0:
return None
try:
range_start = int(range_start)
content_length = int(content_length)
duration_secs = float(duration_secs)
except (TypeError, ValueError):
return None
if content_length <= 0 or duration_secs <= 0:
return None
ratio = min(1.0, max(0.0, range_start / content_length))
return ratio * duration_secs
def resolve_stats_playback_fields(
*,
timestamp_utc,
existing_programme_start,
existing_position_anchor,
existing_playback_base,
range_start,
representation_length,
programme_duration_secs,
now,
):
"""Return ``(playback_base_secs, position_anchor_at)`` for stats registration.
``playback_base_secs`` is set for byte-range seeks (VLC); ``None`` means
derive position from the URL programme timestamp instead.
"""
if isinstance(existing_programme_start, bytes):
existing_programme_start = existing_programme_start.decode()
if isinstance(existing_position_anchor, bytes):
existing_position_anchor = existing_position_anchor.decode()
if isinstance(existing_playback_base, bytes):
existing_playback_base = existing_playback_base.decode()
programme_changed = (
existing_programme_start is not None
and existing_programme_start != timestamp_utc
)
byte_base = compute_playback_base_from_byte_range(
range_start, representation_length, programme_duration_secs,
)
if programme_changed:
return None, now
if byte_base is not None:
return byte_base, now
if range_start == 0:
return None, now
if existing_programme_start == timestamp_utc and existing_position_anchor:
existing_base = None
if existing_playback_base is not None:
try:
existing_base = float(existing_playback_base)
except (TypeError, ValueError):
existing_base = None
return existing_base, existing_position_anchor
return None, now
def compute_playback_position_secs(
programme_start_url,
epg_start_iso,
position_anchor_at,
current_time,
duration_secs=None,
playback_base_secs=None,
):
"""Best-effort catch-up play position in seconds within the programme.
IPTV clients seek by opening a new catch-up URL at the target time, so the
URL timestamp (``programme_start_url``) is the requested position. Offset it
from the programme's EPG start, then advance by wall-clock since the current
stream opened (``position_anchor_at``).
Native players (VLC) often keep the programme URL fixed and seek via
``Range: bytes=``; in that case ``playback_base_secs`` carries the mapped
position at the anchor instead of the URL timestamp offset.
"""
elapsed_since_anchor = 0.0
if position_anchor_at:
try:
elapsed_since_anchor = max(0.0, current_time - float(position_anchor_at))
except (TypeError, ValueError):
elapsed_since_anchor = 0.0
if playback_base_secs is not None:
try:
position = float(playback_base_secs) + elapsed_since_anchor
except (TypeError, ValueError):
position = elapsed_since_anchor
else:
if not programme_start_url or not epg_start_iso:
return None
url_dt = parse_catchup_timestamp(programme_start_url)
if url_dt is None:
return None
try:
from datetime import datetime
epg_dt = datetime.fromisoformat(epg_start_iso)
except (TypeError, ValueError):
return None
if epg_dt.tzinfo is not None:
epg_dt = epg_dt.astimezone(dt_timezone.utc).replace(tzinfo=None)
url_offset = (url_dt - epg_dt).total_seconds()
position = url_offset + elapsed_since_anchor
if position < 0:
position = 0.0
if duration_secs:
position = min(position, float(duration_secs))
return position
def find_stats_channel_for_session(redis_client, session_id):
"""Locate the Redis stats channel id for a catch-up session."""
if not redis_client or not session_id:
return None
pattern = f"timeshift:channel:*:clients:{session_id}"
cursor = 0
while True:
cursor, keys = redis_client.scan(cursor, match=pattern, count=100)
for key in keys:
key_str = _decode_value(key)
match = re.match(r"timeshift:channel:(.+):clients:.+", key_str)
if match:
return match.group(1)
if cursor == 0:
break
return None
def build_timeshift_stats_data(redis_client):
"""Build catch-up stats payload from Redis session metadata."""
empty = {
"timeshift_sessions": [],
"total_connections": 0,
"timestamp": time.time(),
}
if redis_client is None:
return empty
try:
current_time = time.time()
connections = []
cursor = 0
metadata_pattern = "timeshift:channel:*:metadata"
while True:
cursor, keys = redis_client.scan(
cursor, match=metadata_pattern, count=100,
)
for key in keys:
key_str = _decode_value(key)
match = re.search(r"timeshift:channel:(.+):metadata", key_str)
if not match:
continue
stats_channel_id = match.group(1)
parsed_stats = parse_stats_channel_id(stats_channel_id)
if not parsed_stats:
continue
metadata = _decode_hash(redis_client.hgetall(key))
client_set_key = TimeshiftRedisKeys.clients(stats_channel_id)
client_ids = redis_client.smembers(client_set_key) or []
if not client_ids:
continue
init_time = float(metadata.get(ChannelMetadataField.INIT_TIME, 0) or 0)
uptime = current_time - init_time if init_time > 0 else 0
total_bytes = int(metadata.get(ChannelMetadataField.TOTAL_BYTES, 0) or 0)
avg_bitrate_kbps = 0
if uptime > 0 and total_bytes > 0:
avg_bitrate_kbps = (total_bytes * 8) / uptime / 1000
logo_id = metadata.get(ChannelMetadataField.LOGO_ID)
m3u_profile_id = metadata.get(ChannelMetadataField.M3U_PROFILE)
channel_name = metadata.get(
ChannelMetadataField.CHANNEL_NAME, "Catch-up",
)
channel_id_raw = metadata.get(ChannelMetadataField.CHANNEL_ID)
if not channel_id_raw:
channel_id_raw = str(parsed_stats["channel_id"])
channel_uuid = metadata.get(ChannelMetadataField.CHANNEL_UUID, "")
stream_info = stream_info_from_metadata(metadata)
for raw_client_id in client_ids:
client_id = _decode_value(raw_client_id)
client_key = TimeshiftRedisKeys.client_metadata(
stats_channel_id, client_id,
)
client_data = _decode_hash(redis_client.hgetall(client_key))
if not client_data:
continue
programme_start = client_data.get("programme_start")
if not programme_start:
continue
try:
channel_id = int(channel_id_raw)
except (TypeError, ValueError):
continue
playback_base_raw = client_data.get("playback_base_secs")
playback_base_secs = None
if playback_base_raw is not None and playback_base_raw != "":
try:
playback_base_secs = float(playback_base_raw)
except (TypeError, ValueError):
playback_base_secs = None
connected_at = client_data.get("connected_at")
duration = 0
if connected_at:
try:
duration = int(current_time - float(connected_at))
except (TypeError, ValueError):
duration = 0
connections.append({
"stats_channel_id": stats_channel_id,
"session_id": parsed_stats["session_id"],
"client_id": client_id,
"channel_id": channel_id,
"channel_uuid": channel_uuid,
"channel_name": channel_name,
"logo_id": int(logo_id) if logo_id else None,
"programme_start": programme_start,
"position_anchor_at": client_data.get("position_anchor_at"),
"playback_base_secs": playback_base_secs,
"m3u_profile_id": int(m3u_profile_id) if m3u_profile_id else None,
"ip_address": client_data.get("ip_address", "Unknown"),
"user_agent": client_data.get("user_agent", "unknown"),
"user_id": client_data.get("user_id", "0"),
"username": client_data.get("username", "unknown"),
"connected_at": float(connected_at) if connected_at else None,
"duration": duration,
"bytes_streamed": total_bytes,
"avg_bitrate_kbps": round(avg_bitrate_kbps, 2),
"uptime": uptime,
**stream_info,
})
if cursor == 0:
break
channel_ids = {conn["channel_id"] for conn in connections if conn.get("channel_id")}
channels_by_id = {}
if channel_ids:
for channel in Channel.objects.select_related("logo", "epg_data").filter(
id__in=channel_ids,
):
channels_by_id[channel.id] = channel
profile_ids = {
conn["m3u_profile_id"]
for conn in connections
if conn.get("m3u_profile_id")
}
profiles_by_id = {}
if profile_ids:
for profile in M3UAccountProfile.objects.select_related("m3u_account").filter(
id__in=profile_ids,
):
profiles_by_id[profile.id] = profile
session_stats = {}
for conn in connections:
session_key = conn["session_id"]
channel = channels_by_id.get(conn["channel_id"])
if channel:
conn["channel_name"] = channel.name
if channel.logo_id:
conn["logo_id"] = channel.logo_id
if channel.logo:
conn["logo_url"] = channel.logo.url
profile = profiles_by_id.get(conn.get("m3u_profile_id"))
if profile:
conn["m3u_profile"] = {
"profile_name": profile.name,
"account_name": profile.m3u_account.name,
"account_id": profile.m3u_account.id,
"m3u_profile_id": profile.id,
}
position_anchor_at = conn.get("position_anchor_at")
try:
position_anchor_at = (
float(position_anchor_at) if position_anchor_at else None
)
except (TypeError, ValueError):
position_anchor_at = None
playback_base_secs = conn.get("playback_base_secs")
if session_key not in session_stats:
session_stats[session_key] = {
"session_id": session_key,
"stats_channel_id": conn["stats_channel_id"],
"channel_id": conn["channel_id"],
"channel_uuid": conn.get("channel_uuid", ""),
"channel_name": conn["channel_name"],
"logo_id": conn.get("logo_id"),
"logo_url": conn.get("logo_url"),
"programme_start": conn["programme_start"],
"position_anchor_at": position_anchor_at,
"playback_base_secs": playback_base_secs,
"resolution": conn.get("resolution"),
"source_fps": conn.get("source_fps"),
"video_codec": conn.get("video_codec"),
"audio_codec": conn.get("audio_codec"),
"audio_channels": conn.get("audio_channels"),
"stream_type": conn.get("stream_type"),
"connection_count": 0,
"connections": [],
}
session_stats[session_key]["connection_count"] += 1
session_stats[session_key]["connections"].append({
"client_id": conn["client_id"],
"session_id": conn["session_id"],
"ip_address": conn["ip_address"],
"user_agent": conn["user_agent"],
"user_id": conn["user_id"],
"username": conn["username"],
"connected_at": conn["connected_at"],
"duration": conn["duration"],
"bytes_streamed": conn["bytes_streamed"],
"avg_bitrate_kbps": conn["avg_bitrate_kbps"],
"m3u_profile": conn.get("m3u_profile", {}),
"m3u_profile_id": conn.get("m3u_profile_id"),
})
return {
"timeshift_sessions": list(session_stats.values()),
"total_connections": len(connections),
"timestamp": current_time,
}
except Exception as exc:
logger.error("Error building timeshift stats: %s", exc, exc_info=True)
return empty
finally:
close_old_connections()

View file

@ -0,0 +1,78 @@
"""Admin API views for catch-up connection statistics."""
import json
import logging
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from rest_framework.decorators import api_view, permission_classes
from apps.accounts.permissions import IsAdmin
from apps.proxy.utils import stop_timeshift_client
from apps.timeshift.helpers import get_catchup_programmes_for_sessions
from apps.timeshift.stats import (
build_timeshift_stats_data,
find_stats_channel_for_session,
)
from core.utils import RedisClient
logger = logging.getLogger(__name__)
@api_view(["GET"])
@permission_classes([IsAdmin])
def timeshift_stats(request):
"""Return active catch-up viewer sessions for the stats page."""
redis_client = RedisClient.get_client()
if not redis_client:
return JsonResponse({"error": "Redis not available"}, status=500)
return JsonResponse(build_timeshift_stats_data(redis_client))
@api_view(["POST"])
@permission_classes([IsAdmin])
def catchup_programmes(request):
"""Return EPG metadata for active catch-up sessions (batch)."""
sessions = request.data.get("sessions")
if sessions is None:
return JsonResponse({"error": "sessions is required"}, status=400)
if not isinstance(sessions, list):
return JsonResponse({"error": "sessions must be an array"}, status=400)
return JsonResponse({
"sessions": get_catchup_programmes_for_sessions(sessions),
})
@csrf_exempt
@api_view(["POST"])
@permission_classes([IsAdmin])
def stop_timeshift_session(request):
"""Stop a catch-up viewer by session id (one session = one stats channel)."""
try:
data = json.loads(request.body)
except json.JSONDecodeError:
return JsonResponse({"error": "Invalid JSON"}, status=400)
session_id = data.get("session_id")
if not session_id:
return JsonResponse({"error": "No session_id provided"}, status=400)
redis_client = RedisClient.get_client()
if not redis_client:
return JsonResponse({"error": "Redis not available"}, status=500)
stats_channel_id = find_stats_channel_for_session(redis_client, session_id)
if not stats_channel_id:
return JsonResponse({"error": "Connection not found"}, status=404)
result = stop_timeshift_client(redis_client, stats_channel_id, session_id)
if result.get("status") != "success":
return JsonResponse(
{"error": result.get("message", "Stop failed")},
status=500,
)
return JsonResponse({
"message": "Timeshift session stop signal sent",
"session_id": session_id,
})

View file

@ -62,6 +62,18 @@ class TimestampFormatTests(TestCase):
"2026-06-23T04:00:00",
)
def test_normalize_iso8601_utc(self):
self.assertEqual(
normalize_catchup_timestamp_input("2026-06-23T04:00:00Z"),
"2026-06-23T04:00:00",
)
def test_normalize_iso8601_with_offset(self):
self.assertEqual(
normalize_catchup_timestamp_input("2026-06-23T06:00:00+02:00"),
"2026-06-23T04:00:00",
)
def test_normalize_rejects_garbage(self):
self.assertIsNone(normalize_catchup_timestamp_input("garbage"))
self.assertIsNone(normalize_catchup_timestamp_input(""))

View file

@ -0,0 +1,318 @@
"""Tests for catch-up playback session API and Redis helpers."""
import time
import uuid
from unittest.mock import MagicMock, patch
from django.http import HttpResponse
from django.test import TestCase, override_settings
from rest_framework.test import APIClient
from apps.accounts.models import User
from apps.channels.models import Channel, ChannelStream, Stream
from apps.m3u.models import M3UAccount
from apps.timeshift import sessions, views
from apps.timeshift.redis_keys import TimeshiftRedisKeys
from apps.timeshift.tests.test_views import _proxy_url
from rest_framework.test import APIRequestFactory, force_authenticate
class FakeRedisSessionStore:
"""Minimal Redis stand-in for session module tests."""
def __init__(self):
self.store = {}
self.ttl = {}
def hset(self, key, mapping=None, **kwargs):
if mapping is None:
mapping = kwargs
else:
mapping = {**mapping, **kwargs}
bucket = self.store.setdefault(key, {})
bucket.update({k: str(v) for k, v in mapping.items()})
def hgetall(self, key):
return dict(self.store.get(key, {}))
def expire(self, key, seconds):
self.ttl[key] = seconds
def exists(self, key):
return key in self.store and bool(self.store[key])
def delete(self, key):
self.store.pop(key, None)
self.ttl.pop(key, None)
return 1
@override_settings(
CACHES={
"default": {
"BACKEND": "django.core.cache.backends.locmem.LocMemCache",
"LOCATION": "catchup-session-tests",
}
},
REST_FRAMEWORK={
"DEFAULT_AUTHENTICATION_CLASSES": [
"rest_framework_simplejwt.authentication.JWTAuthentication",
"apps.accounts.authentication.ApiKeyAuthentication",
],
"DEFAULT_PERMISSION_CLASSES": [
"apps.accounts.permissions.IsAdmin",
],
},
)
class CatchupSessionApiTests(TestCase):
@classmethod
def setUpTestData(cls):
cls.user = User.objects.create(
username="catchup-session-user",
user_level=User.UserLevel.STANDARD,
)
cls.other = User.objects.create(
username="catchup-session-other",
user_level=User.UserLevel.STANDARD,
)
cls.account = M3UAccount.objects.create(
name="catchup-session-acct",
server_url="http://example.test",
account_type="XC",
is_active=True,
)
cls.channel = Channel.objects.create(
name="Catchup Session Channel",
is_catchup=True,
catchup_days=7,
)
cls.stream = Stream.objects.create(
name="catchup-session-stream",
url="http://example.test/live",
m3u_account=cls.account,
is_catchup=True,
catchup_days=7,
custom_properties={"stream_id": "111"},
)
ChannelStream.objects.create(
channel=cls.channel, stream=cls.stream, order=0,
)
def setUp(self):
self.client = APIClient()
self.client.force_authenticate(user=self.user)
self.redis = FakeRedisSessionStore()
def _create_url(self):
return "/api/catchup/sessions/"
@patch.object(sessions.RedisClient, "get_client")
@patch("apps.timeshift.api_views.network_access_allowed", return_value=True)
def test_post_creates_session_without_start_in_playback_url(self, _net, redis_mock):
redis_mock.return_value = self.redis
response = self.client.post(
self._create_url(),
{
"channel_uuid": str(self.channel.uuid),
"start": "2026-06-08T17:00:00Z",
},
format="json",
)
self.assertEqual(response.status_code, 201)
data = response.json()
self.assertGreater(len(data["session_id"]), 8)
self.assertIn(f"session_id={data['session_id']}", data["playback_url"])
self.assertNotIn("start=", data["playback_url"])
self.assertEqual(data["channel_uuid"], str(self.channel.uuid))
self.assertEqual(data["start"], "2026-06-08T17:00:00Z")
self.assertGreater(data["expires_at"], int(time.time()))
@patch.object(sessions.RedisClient, "get_client")
@patch("apps.timeshift.api_views.network_access_allowed", return_value=True)
def test_post_rejects_non_catchup_channel(self, _net, redis_mock):
redis_mock.return_value = self.redis
plain = Channel.objects.create(name="no-catchup")
response = self.client.post(
self._create_url(),
{"channel_uuid": str(plain.uuid), "start": "2026-06-08T17:00:00Z"},
format="json",
)
self.assertEqual(response.status_code, 400)
@patch.object(sessions.RedisClient, "get_client")
@patch("apps.timeshift.api_views.network_access_allowed", return_value=True)
def test_delete_revokes_own_session(self, _net, redis_mock):
redis_mock.return_value = self.redis
created = self.client.post(
self._create_url(),
{
"channel_uuid": str(self.channel.uuid),
"start": "2026-06-08T17:00:00Z",
},
format="json",
)
session_id = created.json()["session_id"]
deleted = self.client.delete(f"/api/catchup/sessions/{session_id}/")
self.assertEqual(deleted.status_code, 204)
self.assertFalse(sessions.get_catchup_session(session_id))
@patch.object(sessions.RedisClient, "get_client")
@patch("apps.timeshift.api_views.network_access_allowed", return_value=True)
def test_delete_rejects_other_users_session(self, _net, redis_mock):
redis_mock.return_value = self.redis
created = self.client.post(
self._create_url(),
{
"channel_uuid": str(self.channel.uuid),
"start": "2026-06-08T17:00:00Z",
},
format="json",
)
session_id = created.json()["session_id"]
self.client.force_authenticate(user=self.other)
deleted = self.client.delete(f"/api/catchup/sessions/{session_id}/")
self.assertEqual(deleted.status_code, 404)
class CatchupSessionResolveTests(TestCase):
@classmethod
def setUpTestData(cls):
cls.user = User.objects.create(
username="catchup-resolve-user",
user_level=User.UserLevel.STANDARD,
)
cls.channel = Channel.objects.create(
name="resolve-channel",
is_catchup=True,
)
def setUp(self):
self.redis = FakeRedisSessionStore()
@patch.object(sessions.RedisClient, "get_client")
def test_resolve_prefers_pool_user_id(self, redis_mock):
redis_mock.return_value = self.redis
session_id = sessions.mint_catchup_session_id()
self.redis.hset(
TimeshiftRedisKeys.api_session(session_id),
mapping={
"user_id": "999",
"channel_uuid": str(self.channel.uuid),
"channel_id": str(self.channel.id),
"start": "2026-06-08T17:00:00Z",
"created_at": "1",
},
)
self.redis.hset(
TimeshiftRedisKeys.pool(session_id),
mapping={"user_id": str(self.user.id)},
)
resolved = sessions.resolve_catchup_playback(session_id, self.channel.uuid)
self.assertIsNotNone(resolved)
self.assertEqual(resolved[0].id, self.user.id)
self.assertEqual(resolved[1], "2026-06-08T17:00:00Z")
self.assertEqual(
self.redis.ttl[TimeshiftRedisKeys.api_session(session_id)],
sessions.SESSION_IDLE_TTL_SECONDS,
)
@patch.object(sessions.RedisClient, "get_client")
def test_resolve_rejects_wrong_channel(self, redis_mock):
redis_mock.return_value = self.redis
session_id = sessions.mint_catchup_session_id()
self.redis.hset(
TimeshiftRedisKeys.api_session(session_id),
mapping={
"user_id": str(self.user.id),
"channel_uuid": str(self.channel.uuid),
"channel_id": str(self.channel.id),
"start": "2026-06-08T17:00:00Z",
"created_at": "1",
},
)
other_uuid = uuid.uuid4()
self.assertIsNone(
sessions.resolve_catchup_playback(session_id, other_uuid),
)
class CatchupProxySessionAuthTests(TestCase):
"""Playback via API session without JWT."""
def setUp(self):
self.factory = APIRequestFactory()
self.channel_uuid = uuid.uuid4()
@patch.object(views, "resolve_catchup_playback")
@patch.object(views, "network_access_allowed", return_value=True)
@patch.object(views, "_serve_catchup", return_value=HttpResponse("ok"))
@patch.object(views, "_user_can_access_channel", return_value=True)
@patch.object(views, "Channel")
def test_session_auth_without_jwt(
self, channel_cls, _access, serve, _net, resolve_mock,
):
user = MagicMock(id=42, is_authenticated=False)
resolve_mock.return_value = (user, "2026-06-08T17:00:00Z")
channel_cls.objects.get.return_value = MagicMock(
id=8, uuid=self.channel_uuid,
)
request = self.factory.get(
f"/proxy/catchup/{self.channel_uuid}?session_id=test",
)
response = views.catchup_proxy(request, self.channel_uuid)
self.assertEqual(response.status_code, 200)
serve.assert_called_once()
_args, kwargs = serve.call_args
self.assertEqual(_args[3], "2026-06-08T17:00:00Z")
@patch.object(views, "resolve_catchup_playback", return_value=None)
@patch.object(views, "network_access_allowed", return_value=True)
def test_expired_session_without_jwt_returns_401(self, _net, _resolve):
request = self.factory.get(
f"/proxy/catchup/{self.channel_uuid}?session_id=gone",
)
response = views.catchup_proxy(request, self.channel_uuid)
self.assertEqual(response.status_code, 401)
@patch.object(views, "resolve_catchup_playback")
@patch.object(views, "network_access_allowed", return_value=True)
def test_mismatched_jwt_and_session_returns_403(self, _net, resolve_mock):
resolve_mock.return_value = (MagicMock(id=1), "2026-06-08T17:00:00Z")
request = self.factory.get(
f"/proxy/catchup/{self.channel_uuid}?session_id=test",
)
other = MagicMock(id=2, is_authenticated=True)
force_authenticate(request, user=other)
response = views.catchup_proxy(request, self.channel_uuid)
self.assertEqual(response.status_code, 403)
@patch.object(views, "network_access_allowed", return_value=True)
@patch.object(views, "_serve_catchup", return_value=HttpResponse("ok"))
@patch.object(views, "_user_can_access_channel", return_value=True)
@patch.object(views, "Channel")
def test_legacy_jwt_start_still_works(self, channel_cls, _access, serve, _net):
user = MagicMock(id=1, is_authenticated=True)
channel_cls.objects.get.return_value = MagicMock(
id=8, uuid=self.channel_uuid,
)
request = self.factory.get(
f"/proxy/catchup/{self.channel_uuid}?start=2026-06-08T17:00:00Z",
)
force_authenticate(request, user=user)
response = views.catchup_proxy(request, self.channel_uuid)
self.assertEqual(response.status_code, 200)
serve.assert_called_once()
def test_xc_path_unchanged(self):
request = self.factory.get(_proxy_url())
with patch.object(views, "_authenticate_user", return_value=MagicMock(id=1)), \
patch.object(views, "network_access_allowed", return_value=True), \
patch.object(views, "Channel") as channel_cls, \
patch.object(views, "_user_can_access_channel", return_value=True), \
patch.object(views, "_serve_catchup", return_value=HttpResponse("ok")) as serve:
channel_cls.objects.get.return_value = MagicMock(id=8)
response = views.timeshift_proxy(
request, "u", "p", "8", "2026-06-08:17-00", "8.ts",
)
self.assertEqual(response.status_code, 200)
serve.assert_called_once()

View file

@ -0,0 +1,347 @@
"""Tests for catch-up stats API and builders."""
import json
import time
from datetime import datetime, timedelta, timezone as dt_timezone
from unittest.mock import MagicMock, patch
from django.test import TestCase
from rest_framework.test import APIRequestFactory, force_authenticate
from apps.accounts.models import User
from apps.proxy.live_proxy.constants import ChannelMetadataField, ChannelState
from apps.timeshift.redis_keys import TimeshiftRedisKeys as RedisKeys, parse_stats_channel_id
from apps.timeshift import stats_views
from apps.timeshift.helpers import get_programme_info
from apps.timeshift.stats import (
build_timeshift_stats_data,
compute_playback_base_from_byte_range,
compute_playback_position_secs,
find_stats_channel_for_session,
resolve_stats_playback_fields,
seed_stream_stats_metadata,
stream_stats_to_metadata_fields,
)
from apps.timeshift.tests.test_views import _FakeRedis
TEST_SESSION_ID = "testsession1"
STATS_CHANNEL_ID = f"8_{TEST_SESSION_ID}"
class TimeshiftHelperParsingTests(TestCase):
def test_parse_stats_channel_id(self):
parsed = parse_stats_channel_id(STATS_CHANNEL_ID)
self.assertEqual(parsed["channel_id"], 8)
self.assertEqual(parsed["session_id"], TEST_SESSION_ID)
class GetProgrammeInfoTests(TestCase):
def _channel_with_programme(self, title="Evening News", sub_title="Local Edition", minutes=45):
start = datetime(2026, 6, 8, 17, 0, tzinfo=dt_timezone.utc)
programme = MagicMock(
title=title,
sub_title=sub_title,
description="Details",
start_time=start,
end_time=start + timedelta(minutes=minutes),
)
channel = MagicMock()
channel.epg_data.programs.filter.return_value.first.return_value = programme
return channel
def test_get_programme_info_resolves_title(self):
info = get_programme_info(self._channel_with_programme(), "2026-06-08:17-00")
self.assertEqual(info["title"], "Evening News")
self.assertEqual(info["sub_title"], "Local Edition")
self.assertEqual(info["duration_secs"], 45 * 60)
class ComputePlaybackPositionTests(TestCase):
EPG_START = "2026-07-10T14:00:00+00:00"
def test_byte_range_base_plus_elapsed(self):
pos = compute_playback_position_secs(
"2026-07-10:14-00",
self.EPG_START,
position_anchor_at=1000.0,
current_time=1030.0,
duration_secs=3600,
playback_base_secs=1800.0,
)
self.assertAlmostEqual(pos, 1830.0)
def test_url_offset_plus_elapsed(self):
# Seeked to 14:19 (19 min into a 14:00 programme), stream opened 30s ago.
pos = compute_playback_position_secs(
"2026-07-10:14-19",
self.EPG_START,
position_anchor_at=1000.0,
current_time=1030.0,
duration_secs=3600,
)
self.assertAlmostEqual(pos, 19 * 60 + 30)
def test_capped_at_duration(self):
pos = compute_playback_position_secs(
"2026-07-10:14-19",
self.EPG_START,
position_anchor_at=1000.0,
current_time=100000.0,
duration_secs=1800,
)
self.assertEqual(pos, 1800)
def test_missing_epg_returns_none(self):
self.assertIsNone(
compute_playback_position_secs(
"2026-07-10:14-19", None, 1000.0, 1030.0,
)
)
def test_no_anchor_uses_url_offset_only(self):
pos = compute_playback_position_secs(
"2026-07-10:14-05",
self.EPG_START,
position_anchor_at=None,
current_time=1030.0,
duration_secs=3600,
)
self.assertAlmostEqual(pos, 5 * 60)
class ByteRangePlaybackTests(TestCase):
def test_compute_playback_base_from_byte_range(self):
base = compute_playback_base_from_byte_range(
506786520, 833563944, 3900,
)
self.assertAlmostEqual(base, 2371.0, delta=2.0)
def test_resolve_byte_seek_reanchors(self):
base, anchor = resolve_stats_playback_fields(
timestamp_utc="2026-07-09:20:00:00",
existing_programme_start="2026-07-09:20:00:00",
existing_position_anchor="1000.0",
existing_playback_base=None,
range_start=506786520,
representation_length=833563944,
programme_duration_secs=3900,
now="2000.0",
)
self.assertAlmostEqual(base, 2371.0, delta=2.0)
self.assertEqual(anchor, "2000.0")
def test_resolve_same_programme_preserves_without_range(self):
base, anchor = resolve_stats_playback_fields(
timestamp_utc="2026-06-08:17-00",
existing_programme_start="2026-06-08:17-00",
existing_position_anchor="1000.0",
existing_playback_base="900.0",
range_start=None,
representation_length=833563944,
programme_duration_secs=3900,
now="2000.0",
)
self.assertAlmostEqual(base, 900.0)
self.assertEqual(anchor, "1000.0")
class TimeshiftStreamStatsTests(TestCase):
def test_stream_stats_to_metadata_fields(self):
fields = stream_stats_to_metadata_fields({
"resolution": "1920x1080",
"source_fps": 30,
"video_codec": "h264",
})
self.assertEqual(fields[ChannelMetadataField.RESOLUTION], "1920x1080")
self.assertEqual(fields[ChannelMetadataField.SOURCE_FPS], "30")
self.assertEqual(fields[ChannelMetadataField.VIDEO_CODEC], "h264")
self.assertIn(ChannelMetadataField.STREAM_INFO_UPDATED, fields)
def test_seed_stream_stats_metadata_skips_when_stream_unchanged(self):
redis = _FakeRedis()
metadata_key = RedisKeys.channel_metadata(STATS_CHANNEL_ID)
redis.hset(metadata_key, ChannelMetadataField.STREAM_ID, "7")
payload = {}
seed_stream_stats_metadata(
redis, metadata_key, payload,
stats_stream_id=7,
stream_stats={"resolution": "1920x1080"},
)
self.assertEqual(payload, {})
def test_seed_stream_stats_metadata_applies_new_stream(self):
redis = _FakeRedis()
metadata_key = RedisKeys.channel_metadata(STATS_CHANNEL_ID)
payload = {}
seed_stream_stats_metadata(
redis, metadata_key, payload,
stats_stream_id=7,
stream_stats={"resolution": "1280x720", "video_codec": "h264"},
)
self.assertEqual(payload[ChannelMetadataField.STREAM_ID], "7")
self.assertEqual(payload[ChannelMetadataField.RESOLUTION], "1280x720")
class BuildTimeshiftStatsDataTests(TestCase):
def setUp(self):
self.channel_id = 42
self.session_id = TEST_SESSION_ID
self.stats_channel_id = f"{self.channel_id}_{self.session_id}"
self.redis = _FakeRedis()
now = time.time()
metadata_key = RedisKeys.channel_metadata(self.stats_channel_id)
client_set_key = RedisKeys.clients(self.stats_channel_id)
client_key = RedisKeys.client_metadata(self.stats_channel_id, self.session_id)
self.redis.hset(
metadata_key,
mapping={
ChannelMetadataField.STATE: ChannelState.ACTIVE,
ChannelMetadataField.CHANNEL_ID: str(self.channel_id),
ChannelMetadataField.CHANNEL_UUID: "00000000-0000-0000-0000-000000000042",
ChannelMetadataField.CHANNEL_NAME: "Catch-up Stats Channel",
ChannelMetadataField.INIT_TIME: str(now - 120),
ChannelMetadataField.TOTAL_BYTES: "1048576",
ChannelMetadataField.LOGO_ID: "0",
ChannelMetadataField.RESOLUTION: "1920x1080",
ChannelMetadataField.SOURCE_FPS: "29.97",
ChannelMetadataField.VIDEO_CODEC: "h264",
},
)
self.redis.sadd(client_set_key, self.session_id)
self.anchor_at = now - 30
self.redis.hset(
client_key,
mapping={
"user_agent": "VLC/3.0.0",
"ip_address": "10.0.0.5",
"connected_at": str(now - 60),
"user_id": "1",
"username": "viewer",
"programme_vid": f"8_{self.channel_id}_2026-06-08-17-00_111",
"programme_start": "2026-06-08:17-15",
"position_anchor_at": str(self.anchor_at),
},
)
@patch("apps.timeshift.stats.Channel")
def test_build_timeshift_stats_data_redis_only_programme_fields(
self, mock_channel_model,
):
channel = MagicMock()
channel.id = self.channel_id
channel.name = "Catch-up Stats Channel"
channel.uuid = "00000000-0000-0000-0000-000000000042"
channel.logo_id = None
channel.logo = None
mock_channel_model.objects.select_related.return_value.filter.return_value = [
channel,
]
payload = build_timeshift_stats_data(self.redis)
self.assertEqual(payload["total_connections"], 1)
self.assertEqual(len(payload["timeshift_sessions"]), 1)
session = payload["timeshift_sessions"][0]
self.assertEqual(session["session_id"], self.session_id)
self.assertEqual(session["programme_start"], "2026-06-08:17-15")
self.assertAlmostEqual(session["position_anchor_at"], self.anchor_at, places=3)
self.assertNotIn("programme_title", session)
self.assertNotIn("playback_position_secs", session)
self.assertEqual(session["channel_name"], "Catch-up Stats Channel")
self.assertEqual(session["resolution"], "1920x1080")
self.assertEqual(session["connections"][0]["ip_address"], "10.0.0.5")
def test_find_stats_channel_for_session(self):
found = find_stats_channel_for_session(self.redis, self.session_id)
self.assertEqual(found, self.stats_channel_id)
@patch("apps.timeshift.stats.Channel")
def test_skips_clients_missing_required_metadata(self, mock_channel_model):
mock_channel_model.objects.select_related.return_value.filter.return_value = []
orphan_key = RedisKeys.client_metadata(self.stats_channel_id, "orphan")
self.redis.hset(orphan_key, mapping={"user_agent": "VLC"})
self.redis.sadd(RedisKeys.clients(self.stats_channel_id), "orphan")
payload = build_timeshift_stats_data(self.redis)
self.assertEqual(payload["total_connections"], 1)
class TimeshiftStatsApiTests(TestCase):
@classmethod
def setUpTestData(cls):
cls.admin = User.objects.create(
username="catchup-stats-admin",
user_level=User.UserLevel.ADMIN,
)
def setUp(self):
self.factory = APIRequestFactory()
self.redis = _FakeRedis()
self.session_id = TEST_SESSION_ID
self.stats_channel_id = f"8_{self.session_id}"
@patch("apps.timeshift.stats_views.RedisClient.get_client")
@patch("apps.timeshift.stats_views.build_timeshift_stats_data")
def test_timeshift_stats_requires_admin(self, mock_build, redis_mock):
mock_build.return_value = {
"timeshift_sessions": [],
"total_connections": 0,
"timestamp": 1,
}
redis_mock.return_value = self.redis
request = self.factory.get("/proxy/catchup/stats/")
force_authenticate(request, user=self.admin)
response = stats_views.timeshift_stats(request)
self.assertEqual(response.status_code, 200)
self.assertIn("timeshift_sessions", json.loads(response.content))
@patch("apps.timeshift.stats_views.stop_timeshift_client")
@patch("apps.timeshift.stats_views.RedisClient.get_client")
def test_stop_timeshift_session(self, redis_mock, stop_mock):
redis_mock.return_value = self.redis
stop_mock.return_value = {"status": "success"}
metadata_key = RedisKeys.channel_metadata(self.stats_channel_id)
client_key = RedisKeys.client_metadata(self.stats_channel_id, self.session_id)
self.redis.hset(metadata_key, mapping={ChannelMetadataField.STATE: ChannelState.ACTIVE})
self.redis.sadd(RedisKeys.clients(self.stats_channel_id), self.session_id)
self.redis.hset(
client_key,
mapping={
"programme_start": "2026-06-08:17-00",
},
)
request = self.factory.post(
"/proxy/catchup/stop_client/",
{"session_id": self.session_id},
format="json",
)
force_authenticate(request, user=self.admin)
response = stats_views.stop_timeshift_session(request)
self.assertEqual(response.status_code, 200)
stop_mock.assert_called_once_with(
self.redis, self.stats_channel_id, self.session_id,
)
@patch("apps.timeshift.stats_views.get_catchup_programmes_for_sessions")
def test_catchup_programmes_batch(self, programmes_mock):
programmes_mock.return_value = [{
"session_id": self.session_id,
"title": "Evening News",
"start_time": "2026-06-08T17:00:00+00:00",
}]
request = self.factory.post(
"/proxy/catchup/programs/",
{"sessions": [{
"session_id": self.session_id,
"channel_uuid": "00000000-0000-0000-0000-000000000099",
"programme_start": "2026-06-08:17-00",
}]},
format="json",
)
force_authenticate(request, user=self.admin)
response = stats_views.catchup_programmes(request)
self.assertEqual(response.status_code, 200)
body = json.loads(response.content)
self.assertEqual(body["sessions"][0]["title"], "Evening News")
programmes_mock.assert_called_once()

File diff suppressed because it is too large Load diff

12
apps/timeshift/urls.py Normal file
View file

@ -0,0 +1,12 @@
from django.urls import path
from . import stats_views, views
app_name = "catchup"
urlpatterns = [
path("stats/", stats_views.timeshift_stats, name="catchup_stats"),
path("programs/", stats_views.catchup_programmes, name="catchup_programmes"),
path("stop_client/", stats_views.stop_timeshift_session, name="catchup_stop_client"),
path("<uuid:channel_id>", views.catchup_proxy, name="catchup"),
]

File diff suppressed because it is too large Load diff

View file

@ -7,7 +7,7 @@ import re
import time
import os
from core.utils import RedisClient, send_websocket_update, acquire_task_lock, release_task_lock
from apps.proxy.live_proxy.channel_status import ChannelStatus
from apps.proxy.live_proxy.channel_status import build_live_channel_stats_data
from apps.m3u.models import M3UAccount
from apps.epg.models import EPGSource
from apps.m3u.tasks import refresh_single_m3u_account
@ -425,43 +425,22 @@ def fetch_channel_stats():
redis_client = RedisClient.get_client()
try:
# Basic info for all channels
channel_pattern = "live:channel:*:metadata"
all_channels = []
# Extract channel IDs from keys
cursor = 0
while True:
cursor, keys = redis_client.scan(cursor, match=channel_pattern)
for key in keys:
channel_id_match = re.search(r"live:channel:(.*):metadata", key)
if channel_id_match:
ch_id = channel_id_match.group(1)
channel_info = ChannelStatus.get_basic_channel_info(ch_id)
if channel_info:
all_channels.append(channel_info)
if cursor == 0:
break
send_websocket_update(
"updates",
"update",
{
"success": True,
"type": "channel_stats",
"stats": json.dumps({'channels': all_channels, 'count': len(all_channels)})
},
collect_garbage=True
)
# Explicitly clean up large data structures
all_channels = None
live_stats = build_live_channel_stats_data(redis_client)
except Exception as e:
logger.error(f"Error in channel_status: {e}", exc_info=True)
return
send_websocket_update(
"updates",
"update",
{
"success": True,
"type": "channel_stats",
"stats": json.dumps(live_stats),
},
collect_garbage=True
)
@shared_task
def rehash_streams(keys):
"""

View file

@ -320,6 +320,10 @@ export const WebsocketProvider = ({ children }) => {
setVodStats(JSON.parse(parsedEvent.data.stats));
break;
case 'timeshift_stats':
setTimeshiftStats(JSON.parse(parsedEvent.data.stats));
break;
case 'vod_started':
case 'vod_stopped': {
const { content_name, client_ip, user_id } = parsedEvent.data;
@ -1055,6 +1059,7 @@ export const WebsocketProvider = ({ children }) => {
const setChannelStats = useChannelsStore((s) => s.setChannelStats);
const setVodStats = useChannelsStore((s) => s.setVodStats);
const setTimeshiftStats = useChannelsStore((s) => s.setTimeshiftStats);
const fetchPlaylists = usePlaylistsStore((s) => s.fetchPlaylists);
const setRefreshProgress = usePlaylistsStore((s) => s.setRefreshProgress);
const setProfilePreview = usePlaylistsStore((s) => s.setProfilePreview);

View file

@ -2483,6 +2483,56 @@ export default class API {
}
}
static async getTimeshiftStats() {
try {
const response = await request(`${host}/proxy/catchup/stats/`);
return response;
} catch (e) {
errorNotification('Failed to retrieve catch-up stats', e);
}
}
static async getAllConnectionStats() {
try {
const response = await request(`${host}/proxy/stats/`);
return response;
} catch (e) {
errorNotification('Failed to retrieve connection stats', e);
throw e;
}
}
static async getCatchupPrograms(sessions) {
try {
const response = await request(`${host}/proxy/catchup/programs/`, {
method: 'POST',
body: { sessions },
});
return response;
} catch (e) {
console.error('Failed to retrieve catch-up programmes', e);
return { sessions: [] };
}
}
static async stopTimeshiftSession(sessionId) {
try {
const response = await request(`${host}/proxy/catchup/stop_client/`, {
method: 'POST',
body: {
session_id: sessionId,
},
});
return response;
} catch (e) {
errorNotification('Failed to stop catch-up session', e);
}
}
static async stopVODClient(clientId) {
try {
const response = await request(`${host}/proxy/vod/stop_client/`, {

View file

@ -21,7 +21,70 @@ const formatProgramTime = (seconds) => {
return `${minutes}:${secs.toString().padStart(2, '0')}`;
};
const ProgramPreview = ({ program, loading, fetched, label = 'Now Playing:' }) => {
const formatAirWindow = (startTime, endTime) => {
if (!startTime || !endTime) {
return null;
}
const start = new Date(startTime);
const end = new Date(endTime);
const dateOpts = { month: 'short', day: 'numeric', hour: '2-digit', minute: '2-digit' };
const endOpts = { hour: '2-digit', minute: '2-digit' };
return `${start.toLocaleString(undefined, dateOpts)} to ${end.toLocaleTimeString(undefined, endOpts)}`;
};
const getTimelineState = ({
timelineMode,
program,
playbackElapsedSeconds,
now,
}) => {
const startTime = program.start_time ? new Date(program.start_time) : null;
const endTime = program.end_time ? new Date(program.end_time) : null;
if (!startTime || !endTime) {
return { hasValidTime: false };
}
const totalDuration = (endTime - startTime) / 1000;
if (totalDuration <= 0) {
return { hasValidTime: false };
}
if (timelineMode === 'catchup') {
const watched = Math.max(0, Math.floor(playbackElapsedSeconds ?? 0));
const cappedWatched = Math.min(watched, totalDuration);
const remaining = Math.max(0, totalDuration - cappedWatched);
return {
hasValidTime: true,
elapsed: cappedWatched,
remaining,
percentage: Math.min(100, (cappedWatched / totalDuration) * 100),
elapsedLabel: 'watched',
remainingLabel: 'remaining',
airWindow: formatAirWindow(program.start_time, program.end_time),
};
}
const elapsed = (now - startTime) / 1000;
const remaining = (endTime - now) / 1000;
return {
hasValidTime: true,
elapsed,
remaining,
percentage: Math.min(100, Math.max(0, (elapsed / totalDuration) * 100)),
elapsedLabel: 'elapsed',
remainingLabel: 'remaining',
airWindow: null,
};
};
const ProgramPreview = ({
program,
loading,
fetched,
label = 'Now Playing:',
timelineMode = 'live',
playbackElapsedSeconds = 0,
}) => {
const [isExpanded, setIsExpanded] = useState(false);
if (loading) {
@ -46,21 +109,12 @@ const ProgramPreview = ({ program, loading, fetched, label = 'Now Playing:' }) =
return null;
}
const now = new Date();
const startTime = program.start_time ? new Date(program.start_time) : null;
const endTime = program.end_time ? new Date(program.end_time) : null;
let elapsed = 0, remaining = 0, percentage = 0;
let hasValidTime = false;
if (startTime && endTime) {
const totalDuration = (endTime - startTime) / 1000;
if (totalDuration > 0) {
hasValidTime = true;
elapsed = (now - startTime) / 1000;
remaining = (endTime - now) / 1000;
percentage = Math.min(100, Math.max(0, (elapsed / totalDuration) * 100));
}
}
const timeline = getTimelineState({
timelineMode,
program,
playbackElapsedSeconds,
now: new Date(),
});
return (
<>
@ -92,18 +146,23 @@ const ProgramPreview = ({ program, loading, fetched, label = 'Now Playing:' }) =
</Box>
)}
{isExpanded && hasValidTime && (
{isExpanded && timeline.hasValidTime && (
<Stack gap="xs" mt={4} ml={24}>
{timeline.airWindow && (
<Text size="xs" c="dimmed">
Aired {timeline.airWindow}
</Text>
)}
<Group justify="space-between" align="center">
<Text size="xs" c="dimmed">
{formatProgramTime(elapsed)} elapsed
{formatProgramTime(timeline.elapsed)} {timeline.elapsedLabel}
</Text>
<Text size="xs" c="dimmed">
{formatProgramTime(remaining)} remaining
{formatProgramTime(timeline.remaining)} {timeline.remainingLabel}
</Text>
</Group>
<Progress
value={percentage}
value={timeline.percentage}
size="sm"
color="#3BA882"
style={{

View file

@ -165,6 +165,35 @@ describe('ProgramPreview', () => {
expect(screen.getByText(/elapsed/)).toBeTruthy();
expect(screen.queryByText('null')).toBeNull();
});
it('uses session-based progress for catch-up programmes', () => {
const program = {
title: 'Archived Show',
description: 'Catch-up description',
start_time: '2026-06-08T17:00:00+00:00',
end_time: '2026-06-08T18:00:00+00:00',
};
render(
<ProgramPreview
loading={false}
fetched={true}
program={program}
timelineMode="catchup"
label="Watching:"
playbackElapsedSeconds={900}
/>
);
expect(screen.getByText('Watching:')).toBeTruthy();
const expandButton = screen.getByTestId('chevron-right').closest('button');
fireEvent.click(expandButton);
expect(screen.getByText(/Aired /)).toBeTruthy();
expect(screen.getByText('15:00 watched')).toBeTruthy();
expect(screen.getByText('45:00 remaining')).toBeTruthy();
expect(screen.getByTestId('progress').getAttribute('data-value')).toBe('25');
});
});
describe('formatProgramTime', () => {

View file

@ -22,7 +22,6 @@ import {
Gauge,
HardDriveDownload,
HardDriveUpload,
Rewind,
SquareX,
Timer,
Users,
@ -665,18 +664,6 @@ const StreamConnectionCard = ({
{/* Add stream information badges */}
<Group gap="xs" mt="5">
{channel.is_timeshift && (
<Tooltip label="Catch-up (timeshift)">
<Badge
size="sm"
variant="light"
color="violet"
leftSection={<Rewind size={12} />}
>
TIMESHIFT
</Badge>
</Tooltip>
)}
{channel.resolution && (
<Tooltip label="Video resolution">
<Badge size="sm" variant="light" color="red">

View file

@ -0,0 +1,405 @@
import React, { useCallback, useEffect, useMemo, useRef, useState } from 'react';
import logo from '../../images/logo.png';
import {
ActionIcon,
Badge,
Box,
Card,
Center,
Group,
Stack,
Text,
Tooltip,
} from '@mantine/core';
import {
ChevronDown,
HardDriveUpload,
History,
SquareX,
Timer,
} from 'lucide-react';
import ProgramPreview from '../ProgramPreview.jsx';
import {
calculateConnectionDuration,
calculateConnectionStartTime,
computeCatchupPlaybackSeconds,
getConnectionDurationSeconds,
} from '../../utils/cards/TimeshiftConnectionCardUtils.js';
import { useDateTimeFormat } from '../../utils/dateTimeUtils.js';
import { getLogoUrl } from '../../utils/cards/StreamConnectionCardUtils.js';
import useUsersStore from '../../store/users.jsx';
const ClientDetails = ({ connection, connectionStartTime }) => (
<Stack
gap="xs"
style={{ backgroundColor: 'rgba(255, 255, 255, 0.02)' }}
p={12}
bdrs={6}
bd={'1px solid rgba(255, 255, 255, 0.08)'}
>
{connection.user_agent && connection.user_agent !== 'Unknown' && (
<Group gap={8} align="flex-start">
<Text size="xs" fw={500} c="dimmed" miw={80}>
User Agent:
</Text>
<Text size="xs" ff={'monospace'} flex={1}>
{connection.user_agent.length > 100
? `${connection.user_agent.substring(0, 100)}...`
: connection.user_agent}
</Text>
</Group>
)}
<Group gap={8}>
<Text size="xs" fw={500} c="dimmed" miw={80}>
Session ID:
</Text>
<Text size="xs" ff={'monospace'}>
{connection.session_id || connection.client_id || 'Unknown'}
</Text>
</Group>
{connection.connected_at && (
<Group gap={8}>
<Text size="xs" fw={500} c="dimmed" miw={80}>
Connected:
</Text>
<Text size="xs">{connectionStartTime}</Text>
</Group>
)}
{connection.duration > 0 && (
<Group gap={8}>
<Text size="xs" fw={500} c="dimmed" miw={80}>
Watch Duration:
</Text>
<Text size="xs">
{calculateConnectionDuration(connection)}
</Text>
</Group>
)}
{connection.bytes_streamed > 0 && (
<Group gap={8}>
<Text size="xs" fw={500} c="dimmed" miw={80}>
Data Sent:
</Text>
<Text size="xs">
{(connection.bytes_streamed / (1024 * 1024)).toFixed(1)} MB
</Text>
</Group>
)}
{connection.avg_bitrate_kbps > 0 && (
<Group gap={8}>
<Text size="xs" fw={500} c="dimmed" miw={80}>
Avg Bitrate:
</Text>
<Text size="xs">
{connection.avg_bitrate_kbps > 1000
? `${(connection.avg_bitrate_kbps / 1000).toFixed(2)} Mbps`
: `${connection.avg_bitrate_kbps.toFixed(0)} Kbps`}
</Text>
</Group>
)}
</Stack>
);
const TimeshiftConnectionCard = ({
timeshiftSession,
currentProgram,
stopTimeshiftSession,
logos,
}) => {
const { fullDateTimeFormat } = useDateTimeFormat();
const [isClientExpanded, setIsClientExpanded] = useState(false);
const users = useUsersStore((s) => s.users);
const [, setUpdateTrigger] = useState(0);
const usersMap = useMemo(() => {
const map = {};
users.forEach((u) => {
map[String(u.id)] = u.username;
});
return map;
}, [users]);
useEffect(() => {
const interval = setInterval(() => {
setUpdateTrigger((prev) => prev + 1);
}, 1000);
return () => clearInterval(interval);
}, []);
const connection =
timeshiftSession.individual_connection ||
(timeshiftSession.connections && timeshiftSession.connections[0]);
const logoUrl =
timeshiftSession.logo_url ||
getLogoUrl(timeshiftSession.logo_id, logos) ||
logo;
const programmePreview = useMemo(() => {
if (currentProgram?.title) {
return {
title: currentProgram.title,
description: currentProgram.description,
start_time: currentProgram.start_time,
end_time: currentProgram.end_time,
duration_secs: currentProgram.duration_secs,
};
}
if (timeshiftSession.programme_start) {
return {
title: `Catch-up @ ${timeshiftSession.programme_start}`,
};
}
return null;
}, [currentProgram, timeshiftSession.programme_start]);
const m3uProfileName =
connection?.m3u_profile?.profile_name ||
connection?.m3u_profile?.account_name ||
null;
// Play position from URL timestamp + EPG window + stream-open anchor.
const playbackBaseRef = useRef({ base: null, receivedAtMs: 0 });
useEffect(() => {
const computed = computeCatchupPlaybackSeconds({
programmeStart: timeshiftSession.programme_start,
programStartTime: programmePreview?.start_time,
programDurationSecs: programmePreview?.duration_secs,
positionAnchorAt: timeshiftSession.position_anchor_at,
playbackBaseSecs: timeshiftSession.playback_base_secs,
nowMs: Date.now(),
});
if (computed != null) {
playbackBaseRef.current = {
base: computed,
receivedAtMs: Date.now(),
};
}
}, [
timeshiftSession.programme_start,
timeshiftSession.position_anchor_at,
timeshiftSession.playback_base_secs,
programmePreview?.start_time,
programmePreview?.duration_secs,
]);
const { base: playbackBase, receivedAtMs: playbackReceivedAtMs } =
playbackBaseRef.current;
const playbackElapsedSeconds =
playbackBase != null
? playbackBase + (Date.now() - playbackReceivedAtMs) / 1000
: getConnectionDurationSeconds(connection);
const getConnectionStartTime = useCallback(
(conn) => calculateConnectionStartTime(conn, fullDateTimeFormat),
[fullDateTimeFormat]
);
return (
<Card
shadow="sm"
padding="md"
radius="md"
withBorder
style={{ backgroundColor: '#27272A' }}
color="#FFF"
maw={700}
w={'100%'}
>
<Stack pos="relative">
<Group justify="space-between" align="flex-start">
<Box
style={{ alignItems: 'center', justifyContent: 'center' }}
w={140}
h={70}
display="flex"
>
<img
src={logoUrl}
style={{
maxWidth: '100%',
maxHeight: '100%',
objectFit: 'contain',
}}
alt="channel logo"
/>
</Box>
<Group mt={10}>
{connection && (
<Tooltip
label={`Connected at ${getConnectionStartTime(connection)}`}
>
<Center>
<Timer pr={5} />
{calculateConnectionDuration(connection)}
</Center>
</Tooltip>
)}
{connection && stopTimeshiftSession && (
<Center>
<Tooltip label="Stop Catch-up Session">
<ActionIcon
variant="transparent"
color="red.9"
onClick={() =>
stopTimeshiftSession(
connection.session_id || connection.client_id,
)
}
>
<SquareX size="24" />
</ActionIcon>
</Tooltip>
</Center>
)}
</Group>
</Group>
{m3uProfileName && (
<Box pos="absolute" top={95} right={16} style={{ zIndex: 1 }}>
<Group gap={5}>
<HardDriveUpload size="18" />
<Tooltip label="Current M3U Profile">
<Text size="xs">{m3uProfileName}</Text>
</Tooltip>
</Group>
</Box>
)}
<Group gap={6} mt={4} align="center" wrap="nowrap">
<Text fw={500}>{timeshiftSession.channel_name || 'Catch-up'}</Text>
<Tooltip label="Catch-up session">
<Box
component="span"
role="img"
aria-label="Catch-up session"
style={{ display: 'inline-flex' }}
>
<History size={16} color="#9ca3af" aria-hidden="true" />
</Box>
</Tooltip>
</Group>
{programmePreview && (
<Box mt={-9}>
<ProgramPreview
program={programmePreview}
timelineMode="catchup"
label="Watching:"
playbackElapsedSeconds={playbackElapsedSeconds}
/>
</Box>
)}
<Group gap="xs" mt="5">
{timeshiftSession.resolution && (
<Tooltip label="Video resolution">
<Badge size="sm" variant="light" color="red">
{timeshiftSession.resolution}
</Badge>
</Tooltip>
)}
{timeshiftSession.source_fps && (
<Tooltip label="Source frames per second">
<Badge size="sm" variant="light" color="orange">
{timeshiftSession.source_fps} FPS
</Badge>
</Tooltip>
)}
{timeshiftSession.video_codec && (
<Tooltip label="Video codec">
<Badge size="sm" variant="light" color="blue">
{timeshiftSession.video_codec.toUpperCase()}
</Badge>
</Tooltip>
)}
{timeshiftSession.audio_codec && (
<Tooltip label="Audio codec">
<Badge size="sm" variant="light" color="pink">
{timeshiftSession.audio_codec.toUpperCase()}
</Badge>
</Tooltip>
)}
{timeshiftSession.audio_channels && (
<Tooltip label="Audio channel configuration">
<Badge size="sm" variant="light" color="pink">
{timeshiftSession.audio_channels}
</Badge>
</Tooltip>
)}
{timeshiftSession.stream_type && (
<Tooltip label="Stream type">
<Badge size="sm" variant="light" color="cyan">
{timeshiftSession.stream_type.toUpperCase()}
</Badge>
</Tooltip>
)}
</Group>
{connection && (
<Stack gap="xs" mt="xs">
<Group
justify="space-between"
align="center"
style={{
cursor: 'pointer',
backgroundColor: 'rgba(255, 255, 255, 0.05)',
}}
p={'8px 12px'}
bdrs={6}
bd={'1px solid rgba(255, 255, 255, 0.1)'}
onClick={() => setIsClientExpanded(!isClientExpanded)}
>
<Group gap={8}>
<Text size="sm" fw={500} color="dimmed">
Client IP:
</Text>
<Text size="sm" ff={'monospace'}>
{connection.ip_address || 'Unknown IP'}
</Text>
{usersMap[String(connection.user_id)] && (
<>
<Text size="sm" c="dimmed">
User:
</Text>
<Text size="sm">
{usersMap[String(connection.user_id)]}
</Text>
</>
)}
</Group>
<Group gap={8}>
<Text size="xs" color="dimmed">
{isClientExpanded ? 'Hide Details' : 'Show Details'}
</Text>
<ChevronDown
size={16}
style={{
transform: isClientExpanded ? 'rotate(0deg)' : 'rotate(180deg)',
transition: 'transform 0.2s',
}}
/>
</Group>
</Group>
{isClientExpanded && (
<ClientDetails
connection={connection}
connectionStartTime={getConnectionStartTime(connection)}
/>
)}
</Stack>
)}
</Stack>
</Card>
);
};
export default TimeshiftConnectionCard;

View file

@ -23,19 +23,23 @@ import useLocalStorage from '../hooks/useLocalStorage';
import SystemEvents from '../components/SystemEvents';
import ErrorBoundary from '../components/ErrorBoundary.jsx';
import {
fetchActiveChannelStats,
fetchAllConnectionStats,
getClientStats,
getCatchupPrograms,
getCombinedConnections,
getCurrentPrograms,
getStatsByChannelId,
getVODStats,
stopChannel,
stopClient,
stopTimeshiftSession,
stopVODClient,
} from '../utils/pages/StatsUtils.js';
const VodConnectionCard = React.lazy(
() => import('../components/cards/VodConnectionCard.jsx')
);
const TimeshiftConnectionCard = React.lazy(
() => import('../components/cards/TimeshiftConnectionCard.jsx')
);
const StreamConnectionCard = React.lazy(
() => import('../components/cards/StreamConnectionCard.jsx')
);
@ -46,7 +50,9 @@ const Connections = ({
channelsByUUID,
channels,
handleStopVODClient,
handleStopTimeshiftSession,
currentPrograms,
catchupPrograms,
}) => {
const logos = useLogosStore((s) => s.logos);
@ -88,6 +94,16 @@ const Connections = ({
stopVODClient={handleStopVODClient}
/>
);
} else if (connection.type === 'timeshift') {
return (
<TimeshiftConnectionCard
key={connection.id}
timeshiftSession={connection.data}
currentProgram={catchupPrograms[connection.data.session_id]}
stopTimeshiftSession={handleStopTimeshiftSession}
logos={logos}
/>
);
}
return null;
})}
@ -101,17 +117,20 @@ const StatsPage = () => {
const setChannelStats = useChannelsStore((s) => s.setChannelStats);
const vodConnections = useChannelsStore((s) => s.activeVodConnections);
const setVodStats = useChannelsStore((s) => s.setVodStats);
const timeshiftSessions = useChannelsStore((s) => s.activeTimeshiftSessions);
const setTimeshiftStats = useChannelsStore((s) => s.setTimeshiftStats);
const streamProfiles = useStreamProfilesStore((s) => s.profiles);
const [clients, setClients] = useState([]);
const [channelHistory, setChannelHistory] = useState({});
const [isPollingActive, setIsPollingActive] = useState(false);
const [currentPrograms, setCurrentPrograms] = useState({});
const [catchupPrograms, setCatchupPrograms] = useState({});
const [channels, setChannels] = useState({}); // id -> channel
const [channelsByUUID, setChannelsByUUID] = useState({}); // uuid -> id
// Compute needed channel UUIDs from the current active channels.
// Stream previews use a non-UUID hash as channel_id filter those out.
// Stream previews use a non-UUID hash as channel_id; filter those out.
const UUID_REGEX =
/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
const neededUUIDs = useMemo(
@ -168,55 +187,51 @@ const StatsPage = () => {
(total, vodContent) => total + (vodContent.connections?.length || 0),
0
);
const timeshiftConnectionsCount = timeshiftSessions.reduce(
(total, session) => total + (session.connections?.length || 0),
0
);
const handleStopVODClient = async (clientId) => {
await stopVODClient(clientId);
// Refresh VOD stats after stopping to update the UI
fetchVODStats();
fetchAllStats();
};
// Function to fetch channel stats from API
const fetchChannelStats = useCallback(async () => {
try {
const response = await fetchActiveChannelStats();
if (response) {
setChannelStats(response);
} else {
console.log('API response was empty or null');
}
} catch (error) {
console.error('Error fetching channel stats:', error);
console.error('Error details:', {
message: error.message,
status: error.status,
body: error.body,
});
}
}, [setChannelStats]);
const handleStopTimeshiftSession = async (sessionId) => {
await stopTimeshiftSession(sessionId);
fetchAllStats();
};
const fetchVODStats = useCallback(async () => {
const fetchAllStats = useCallback(async () => {
try {
const response = await getVODStats();
const response = await fetchAllConnectionStats();
if (response) {
setVodStats(response);
if (response.live) {
setChannelStats(response.live);
}
if (response.vod) {
setVodStats(response.vod);
}
if (response.catchup) {
setTimeshiftStats(response.catchup);
}
} else {
console.log('VOD API response was empty or null');
console.log('Combined stats API response was empty or null');
}
} catch (error) {
console.error('Error fetching VOD stats:', error);
console.error('Error fetching connection stats:', error);
console.error('Error details:', {
message: error.message,
status: error.status,
body: error.body,
});
}
}, [setVodStats]);
}, [setChannelStats, setVodStats, setTimeshiftStats]);
// Always fetch once on mount, regardless of polling interval setting
useEffect(() => {
fetchChannelStats();
fetchVODStats();
}, [fetchChannelStats, fetchVODStats]);
fetchAllStats();
}, [fetchAllStats]);
// Set up polling for stats when on stats page
useEffect(() => {
@ -226,8 +241,7 @@ const StatsPage = () => {
setIsPollingActive(true);
const interval = setInterval(() => {
fetchChannelStats();
fetchVODStats();
fetchAllStats();
}, refreshInterval);
return () => {
@ -237,7 +251,7 @@ const StatsPage = () => {
} else {
setIsPollingActive(false);
}
}, [refreshInterval, fetchChannelStats, fetchVODStats]);
}, [refreshInterval, fetchAllStats]);
useEffect(() => {
console.log('Processing channel stats:', channelStats);
@ -328,10 +342,76 @@ const StatsPage = () => {
};
}, [activeChannelIds]); // Only re-run when active channel set changes
// Combine active streams and VOD connections into a single mixed list
// Programme metadata for catch-up cards: fetch when sessions or seek
// position (programme_start) change, then reschedule at programme end.
const timeshiftProgrammeKey = useMemo(() => {
return timeshiftSessions
.map((session) => `${session.session_id}:${session.programme_start || ''}`)
.sort()
.join(',');
}, [timeshiftSessions]);
const timeshiftSessionsRef = useRef(timeshiftSessions);
useEffect(() => {
timeshiftSessionsRef.current = timeshiftSessions;
}, [timeshiftSessions]);
useEffect(() => {
if (!timeshiftProgrammeKey) {
setCatchupPrograms({});
return;
}
let timer = null;
const fetchPrograms = async () => {
const sessions = timeshiftSessionsRef.current.map((session) => ({
session_id: session.session_id,
channel_uuid: session.channel_uuid,
programme_start: session.programme_start,
}));
const programs = await getCatchupPrograms(sessions);
setCatchupPrograms(programs);
if (programs && Object.keys(programs).length > 0) {
const now = new Date();
let nearestEndTime = null;
Object.values(programs).forEach((program) => {
if (program && program.end_time) {
const endTime = new Date(program.end_time);
if (
endTime > now &&
(!nearestEndTime || endTime < nearestEndTime)
) {
nearestEndTime = endTime;
}
}
});
if (nearestEndTime) {
const timeUntilChange = nearestEndTime.getTime() - now.getTime();
const fetchDelay = Math.max(timeUntilChange + 5000, 0);
timer = setTimeout(fetchPrograms, fetchDelay);
}
}
};
fetchPrograms();
return () => {
if (timer) clearTimeout(timer);
};
}, [timeshiftProgrammeKey]);
// Combine active streams, VOD, and catch-up into a single mixed list
const combinedConnections = useMemo(() => {
return getCombinedConnections(channelHistory, vodConnections);
}, [channelHistory, vodConnections]);
return getCombinedConnections(
channelHistory,
vodConnections,
timeshiftSessions
);
}, [channelHistory, vodConnections, timeshiftSessions]);
return (
<>
@ -348,6 +428,11 @@ const StatsPage = () => {
{vodConnectionsCount !== 1
? 'VOD connections'
: 'VOD connection'}
{' • '}
{timeshiftConnectionsCount}{' '}
{timeshiftConnectionsCount !== 1
? 'catch-up sessions'
: 'catch-up session'}
</Text>
<Group align="center" gap="xs">
<Text size="sm">Refresh Interval (seconds):</Text>
@ -374,10 +459,7 @@ const StatsPage = () => {
<Button
size="xs"
variant="subtle"
onClick={() => {
fetchChannelStats();
fetchVODStats();
}}
onClick={fetchAllStats}
loading={false}
>
Refresh Now
@ -402,7 +484,9 @@ const StatsPage = () => {
channelsByUUID={channelsByUUID}
channels={channels}
handleStopVODClient={handleStopVODClient}
handleStopTimeshiftSession={handleStopTimeshiftSession}
currentPrograms={currentPrograms}
catchupPrograms={catchupPrograms}
/>
</Box>
</Box>

View file

@ -13,12 +13,11 @@ import useLocalStorage from '../../hooks/useLocalStorage';
import useChannelsStore from '../../store/channels';
import useLogosStore from '../../store/logos';
import {
fetchActiveChannelStats,
fetchAllConnectionStats,
getCurrentPrograms,
getClientStats,
getCombinedConnections,
getStatsByChannelId,
getVODStats,
stopChannel,
stopClient,
stopVODClient,
@ -91,8 +90,7 @@ vi.mock('@mantine/core', () => ({
//mock stats utils
vi.mock('../../utils/pages/StatsUtils', () => {
return {
fetchActiveChannelStats: vi.fn(),
getVODStats: vi.fn(),
fetchAllConnectionStats: vi.fn(),
getCurrentPrograms: vi.fn(),
getClientStats: vi.fn(),
getCombinedConnections: vi.fn(),
@ -136,6 +134,12 @@ describe('StatsPage', () => {
],
};
const mockCombinedStats = {
live: mockChannelStats,
vod: mockVODStats,
catchup: { timeshift_sessions: [], total_connections: 0 },
};
const mockProcessedChannelHistory = {
1: { id: 1, uuid: 'channel-1', connections: 2 },
2: { id: 2, uuid: 'channel-2', connections: 1 },
@ -160,6 +164,7 @@ describe('StatsPage', () => {
let mockSetChannelStats;
let mockSetRefreshInterval;
let mockSetVodStats;
let mockSetTimeshiftStats;
beforeEach(() => {
vi.clearAllMocks();
@ -167,6 +172,7 @@ describe('StatsPage', () => {
mockSetChannelStats = vi.fn();
mockSetRefreshInterval = vi.fn();
mockSetVodStats = vi.fn();
mockSetTimeshiftStats = vi.fn();
// Setup store mocks
useChannelsStore.mockImplementation((selector) => {
@ -177,6 +183,8 @@ describe('StatsPage', () => {
setChannelStats: mockSetChannelStats,
activeVodConnections: mockVODStats.vod_connections,
setVodStats: mockSetVodStats,
activeTimeshiftSessions: [],
setTimeshiftStats: mockSetTimeshiftStats,
};
return selector ? selector(state) : state;
});
@ -198,8 +206,7 @@ describe('StatsPage', () => {
useLocalStorage.mockReturnValue([5, mockSetRefreshInterval]);
// Setup API mocks
fetchActiveChannelStats.mockResolvedValue(mockChannelStats);
getVODStats.mockResolvedValue(mockVODStats);
fetchAllConnectionStats.mockResolvedValue(mockCombinedStats);
getCurrentPrograms.mockResolvedValue({});
getStatsByChannelId.mockReturnValue(mockProcessedChannelHistory);
getClientStats.mockReturnValue(mockClients);
@ -220,8 +227,9 @@ describe('StatsPage', () => {
render(<StatsPage />);
await waitFor(() => {
expect(fetchActiveChannelStats).toHaveBeenCalledTimes(1);
expect(getVODStats).toHaveBeenCalledTimes(1);
expect(fetchAllConnectionStats).toHaveBeenCalledTimes(1);
expect(mockSetChannelStats).toHaveBeenCalledWith(mockChannelStats);
expect(mockSetVodStats).toHaveBeenCalledWith(mockVODStats);
});
});
@ -283,16 +291,14 @@ describe('StatsPage', () => {
render(<StatsPage />);
expect(fetchActiveChannelStats).toHaveBeenCalledTimes(1);
expect(getVODStats).toHaveBeenCalledTimes(1);
expect(fetchAllConnectionStats).toHaveBeenCalledTimes(1);
// Advance timers by 5 seconds
await act(async () => {
vi.advanceTimersByTime(5000);
});
expect(fetchActiveChannelStats).toHaveBeenCalledTimes(2);
expect(getVODStats).toHaveBeenCalledTimes(2);
expect(fetchAllConnectionStats).toHaveBeenCalledTimes(2);
vi.useRealTimers();
});
@ -304,16 +310,14 @@ describe('StatsPage', () => {
render(<StatsPage />);
// Should still fetch once on mount even with interval = 0
expect(fetchActiveChannelStats).toHaveBeenCalledTimes(1);
expect(getVODStats).toHaveBeenCalledTimes(1);
expect(fetchAllConnectionStats).toHaveBeenCalledTimes(1);
await act(async () => {
vi.advanceTimersByTime(10000);
});
// Should not have polled count stays at 1
expect(fetchActiveChannelStats).toHaveBeenCalledTimes(1);
expect(getVODStats).toHaveBeenCalledTimes(1);
expect(fetchAllConnectionStats).toHaveBeenCalledTimes(1);
vi.useRealTimers();
});
@ -323,7 +327,7 @@ describe('StatsPage', () => {
const { unmount } = render(<StatsPage />);
expect(fetchActiveChannelStats).toHaveBeenCalledTimes(1);
expect(fetchAllConnectionStats).toHaveBeenCalledTimes(1);
unmount();
@ -332,7 +336,7 @@ describe('StatsPage', () => {
});
// Should not fetch again after unmount
expect(fetchActiveChannelStats).toHaveBeenCalledTimes(1);
expect(fetchAllConnectionStats).toHaveBeenCalledTimes(1);
vi.useRealTimers();
});
@ -342,14 +346,13 @@ describe('StatsPage', () => {
it('refreshes stats when Refresh Now button is clicked', async () => {
render(<StatsPage />);
expect(fetchActiveChannelStats).toHaveBeenCalledTimes(1);
expect(fetchAllConnectionStats).toHaveBeenCalledTimes(1);
const refreshButton = screen.getByText('Refresh Now');
fireEvent.click(refreshButton);
await waitFor(() => {
expect(fetchActiveChannelStats).toHaveBeenCalledTimes(2);
expect(getVODStats).toHaveBeenCalledTimes(2);
expect(fetchAllConnectionStats).toHaveBeenCalledTimes(2);
});
});
});
@ -410,14 +413,14 @@ describe('StatsPage', () => {
render(<StatsPage />);
await waitFor(() => {
expect(getVODStats).toHaveBeenCalledTimes(1);
expect(fetchAllConnectionStats).toHaveBeenCalledTimes(1);
});
const stopButton = await screen.findByTestId('stop-vod-client-client-1');
fireEvent.click(stopButton);
await waitFor(() => {
expect(getVODStats).toHaveBeenCalledTimes(2);
expect(fetchAllConnectionStats).toHaveBeenCalledTimes(2);
});
});
});
@ -453,35 +456,17 @@ describe('StatsPage', () => {
});
describe('Error Handling', () => {
it('handles fetchActiveChannelStats error gracefully', async () => {
it('handles fetchAllConnectionStats error gracefully', async () => {
const consoleError = vi
.spyOn(console, 'error')
.mockImplementation(() => {});
fetchActiveChannelStats.mockRejectedValue(new Error('API Error'));
fetchAllConnectionStats.mockRejectedValue(new Error('API Error'));
render(<StatsPage />);
await waitFor(() => {
expect(consoleError).toHaveBeenCalledWith(
'Error fetching channel stats:',
expect.any(Error)
);
});
consoleError.mockRestore();
});
it('handles getVODStats error gracefully', async () => {
const consoleError = vi
.spyOn(console, 'error')
.mockImplementation(() => {});
getVODStats.mockRejectedValue(new Error('VOD API Error'));
render(<StatsPage />);
await waitFor(() => {
expect(consoleError).toHaveBeenCalledWith(
'Error fetching VOD stats:',
'Error fetching connection stats:',
expect.any(Error)
);
});
@ -511,7 +496,10 @@ describe('StatsPage', () => {
{ content_uuid: 'vod-2', connections: [{ client_id: 'c2' }] },
],
};
getVODStats.mockResolvedValue(multiVODStats);
fetchAllConnectionStats.mockResolvedValue({
...mockCombinedStats,
vod: multiVODStats,
});
useChannelsStore.mockImplementation((selector) => {
const state = {
@ -521,6 +509,8 @@ describe('StatsPage', () => {
setChannelStats: mockSetChannelStats,
activeVodConnections: multiVODStats.vod_connections,
setVodStats: mockSetVodStats,
activeTimeshiftSessions: [],
setTimeshiftStats: mockSetTimeshiftStats,
};
return selector ? selector(state) : state;
});

View file

@ -22,6 +22,7 @@ describe('useChannelsStore', () => {
stats: {},
activeChannels: {},
activeClients: {},
activeTimeshiftSessions: [],
recordings: [],
recurringRules: [],
isLoading: false,
@ -324,6 +325,116 @@ describe('useChannelsStore', () => {
});
});
describe('setTimeshiftStats', () => {
const now = Date.now() / 1000;
it('should show a notification when a new catch-up session starts', () => {
const { result } = renderHook(() => useChannelsStore());
act(() => {
result.current.setTimeshiftStats({
timeshift_sessions: [
{
session_id: 'session-1',
channel_name: 'News HD',
connections: [
{
client_id: 'session-1',
ip_address: '1.2.3.4',
connected_at: now,
},
],
},
],
});
});
expect(result.current.activeTimeshiftSessions).toHaveLength(1);
expect(showNotification).toHaveBeenCalledWith(
expect.objectContaining({
title: 'Catch-up started',
color: 'blue.5',
})
);
});
it('should show a notification when a catch-up session ends', () => {
const { result } = renderHook(() => useChannelsStore());
act(() => {
useChannelsStore.setState({
activeTimeshiftSessions: [
{
session_id: 'session-1',
channel_name: 'News HD',
connections: [
{
client_id: 'session-1',
ip_address: '1.2.3.4',
connected_at: now,
},
],
},
],
});
});
act(() => {
result.current.setTimeshiftStats({ timeshift_sessions: [] });
});
expect(result.current.activeTimeshiftSessions).toHaveLength(0);
expect(showNotification).toHaveBeenCalledWith(
expect.objectContaining({
title: 'Catch-up ended',
color: 'blue.5',
})
);
});
it('should not show start notifications for sessions already active on first poll', () => {
const { result } = renderHook(() => useChannelsStore());
act(() => {
useChannelsStore.setState({
activeTimeshiftSessions: [
{
session_id: 'session-1',
channel_name: 'News HD',
connections: [
{
client_id: 'session-1',
ip_address: '1.2.3.4',
connected_at: now - 3600,
},
],
},
],
});
});
act(() => {
result.current.setTimeshiftStats({
timeshift_sessions: [
{
session_id: 'session-1',
channel_name: 'News HD',
connections: [
{
client_id: 'session-1',
ip_address: '1.2.3.4',
connected_at: now - 3600,
},
],
},
],
});
});
expect(showNotification).not.toHaveBeenCalled();
});
});
describe('recordings operations', () => {
it('should fetch recordings', async () => {
const mockRecordings = [{ id: 1, title: 'Recording 1' }];

View file

@ -119,6 +119,7 @@ const useChannelsStore = create((set, get) => ({
activeChannels: {},
activeClients: {},
activeVodConnections: [],
activeTimeshiftSessions: [],
recordings: [],
recurringRules: [],
isLoading: false,
@ -507,6 +508,71 @@ const useChannelsStore = create((set, get) => ({
set({ activeVodConnections: stats.vod_connections || [] });
},
setTimeshiftStats: (stats) => {
return set((state) => {
const sessions = stats.timeshift_sessions || [];
const oldSessions = state.activeTimeshiftSessions.reduce((acc, session) => {
acc[session.session_id] = session;
return acc;
}, {});
const newSessions = sessions.reduce((acc, session) => {
acc[session.session_id] = session;
return acc;
}, {});
sessions.forEach((session) => {
const isNewSession = oldSessions[session.session_id] === undefined;
const primaryClient = session.connections?.[0];
if (!primaryClient) {
return;
}
if (isNewSession && isClientNewSincePageLoad(primaryClient)) {
showNotification({
title: 'Catch-up started',
message: clientMessage(session.channel_name, primaryClient),
color: 'blue.5',
});
return;
}
const oldConnIds = new Set(
(oldSessions[session.session_id]?.connections || []).map(
(client) => client.client_id
)
);
session.connections.forEach((client) => {
if (
!oldConnIds.has(client.client_id) &&
isClientNewSincePageLoad(client)
) {
showNotification({
title: 'Catch-up started',
message: clientMessage(session.channel_name, client),
color: 'blue.5',
});
}
});
});
for (const sessionId in oldSessions) {
if (newSessions[sessionId] === undefined) {
const session = oldSessions[sessionId];
const client = session.connections?.[0];
showNotification({
title: 'Catch-up ended',
message: client
? clientMessage(session.channel_name, client)
: session.channel_name,
color: 'blue.5',
});
}
}
return { activeTimeshiftSessions: sessions };
});
},
fetchRecordings: async () => {
try {
set({ recordings: await api.getRecordings() });

View file

@ -0,0 +1,86 @@
import { format, getNowMs, toFriendlyDuration } from '../dateTimeUtils.js';
/** Parse a catch-up URL timestamp (UTC wall clock) into epoch ms. */
export const parseCatchupTimestampMs = (timestampStr) => {
if (!timestampStr) {
return null;
}
const match = String(timestampStr).match(
/^(\d{4}-\d{2}-\d{2})[:_ ](\d{2})[-:](\d{2})/
);
if (!match) {
return null;
}
const [, datePart, hour, minute] = match;
const parsed = Date.parse(`${datePart}T${hour}:${minute}:00Z`);
return Number.isNaN(parsed) ? null : parsed;
};
/** Best-effort play position within the programme (seconds). */
export const computeCatchupPlaybackSeconds = ({
programmeStart,
programStartTime,
programDurationSecs,
positionAnchorAt,
playbackBaseSecs,
nowMs = getNowMs(),
}) => {
let elapsedSinceAnchor = 0;
if (positionAnchorAt != null) {
const anchor = Number(positionAnchorAt);
if (!Number.isNaN(anchor)) {
elapsedSinceAnchor = Math.max(0, nowMs / 1000 - anchor);
}
}
if (playbackBaseSecs != null && !Number.isNaN(Number(playbackBaseSecs))) {
let position = Number(playbackBaseSecs) + elapsedSinceAnchor;
if (position < 0) {
position = 0;
}
if (programDurationSecs != null) {
position = Math.min(position, Number(programDurationSecs));
}
return position;
}
const urlMs = parseCatchupTimestampMs(programmeStart);
const epgStartMs = programStartTime ? Date.parse(programStartTime) : null;
if (urlMs == null || epgStartMs == null || Number.isNaN(epgStartMs)) {
return null;
}
const urlOffsetSec = (urlMs - epgStartMs) / 1000;
let position = urlOffsetSec + elapsedSinceAnchor;
if (position < 0) {
position = 0;
}
if (programDurationSecs != null) {
position = Math.min(position, Number(programDurationSecs));
}
return position;
};
export const calculateConnectionDuration = (connection) => {
const seconds = getConnectionDurationSeconds(connection);
return toFriendlyDuration(seconds, 'seconds');
};
export const getConnectionDurationSeconds = (connection) => {
if (!connection) {
return 0;
}
if (connection.duration && connection.duration > 0) {
return connection.duration;
}
if (connection.connected_at) {
return Math.max(0, Math.floor(getNowMs() / 1000 - connection.connected_at));
}
return 0;
};
export const calculateConnectionStartTime = (connection, fullDateTimeFormat) => {
if (!connection?.connected_at) {
return 'Unknown';
}
return format(new Date(connection.connected_at * 1000), fullDateTimeFormat);
};

View file

@ -0,0 +1,38 @@
import { describe, it, expect } from 'vitest';
import {
computeCatchupPlaybackSeconds,
parseCatchupTimestampMs,
} from '../TimeshiftConnectionCardUtils.js';
describe('parseCatchupTimestampMs', () => {
it('parses colon-dash catch-up timestamps as UTC', () => {
const ms = parseCatchupTimestampMs('2026-07-10:14-19');
expect(new Date(ms).toISOString()).toBe('2026-07-10T14:19:00.000Z');
});
});
describe('computeCatchupPlaybackSeconds', () => {
it('combines URL offset with elapsed time since anchor', () => {
const epgStart = '2026-07-10T14:00:00+00:00';
const position = computeCatchupPlaybackSeconds({
programmeStart: '2026-07-10:14-19',
programStartTime: epgStart,
programDurationSecs: 3600,
positionAnchorAt: 1000,
nowMs: 1030 * 1000,
});
expect(position).toBeCloseTo(19 * 60 + 30, 5);
});
it('uses playback base for byte-range seeks', () => {
const position = computeCatchupPlaybackSeconds({
programmeStart: '2026-07-10:14-00',
programStartTime: '2026-07-10T14:00:00+00:00',
programDurationSecs: 3600,
positionAnchorAt: 1000,
playbackBaseSecs: 1800,
nowMs: 1030 * 1000,
});
expect(position).toBeCloseTo(1830, 5);
});
});

View file

@ -12,14 +12,47 @@ export const stopVODClient = async (clientId) => {
await API.stopVODClient(clientId);
};
export const stopTimeshiftSession = async (sessionId) => {
await API.stopTimeshiftSession(sessionId);
};
export const fetchActiveChannelStats = async () => {
return await API.fetchActiveChannelStats();
};
export const fetchAllConnectionStats = async () => {
return await API.getAllConnectionStats();
};
export const getVODStats = async () => {
return await API.getVODStats();
};
export const getTimeshiftStats = async () => {
return await API.getTimeshiftStats();
};
export const getCatchupPrograms = async (sessions) => {
try {
if (!sessions || sessions.length === 0) {
return {};
}
const response = await API.getCatchupPrograms(sessions);
const programmes = response?.sessions || [];
const programmesMap = {};
programmes.forEach((program) => {
if (program.session_id) {
programmesMap[program.session_id] = program;
}
});
return programmesMap;
} catch (error) {
console.error('Error fetching catch-up programmes:', error);
return {};
}
};
export const getCurrentPrograms = async (activeChannelUUIDs) => {
try {
if (!activeChannelUUIDs || activeChannelUUIDs.length === 0) {
@ -45,7 +78,11 @@ export const getCurrentPrograms = async (activeChannelUUIDs) => {
}
};
export const getCombinedConnections = (channelHistory, vodConnections) => {
export const getCombinedConnections = (
channelHistory,
vodConnections,
timeshiftSessions = []
) => {
const activeStreams = Object.values(channelHistory).map((channel) => ({
type: 'stream',
data: channel,
@ -70,8 +107,24 @@ export const getCombinedConnections = (channelHistory, vodConnections) => {
}));
});
const timeshiftItems = timeshiftSessions.flatMap((session) => {
return (session.connections || []).map((connection, index) => ({
type: 'timeshift',
data: {
...session,
connections: [connection],
connection_count: 1,
individual_connection: connection,
},
id: `timeshift-${session.session_id}-${connection.client_id}-${index}`,
sortKey: connection.connected_at || Date.now() / 1000,
}));
});
// Combine and sort by newest connections first (higher sortKey = more recent)
return [...activeStreams, ...vodItems].sort((a, b) => b.sortKey - a.sortKey);
return [...activeStreams, ...vodItems, ...timeshiftItems].sort(
(a, b) => b.sortKey - a.sortKey
);
};
const getChannelWithMetadata = (

View file

@ -8,6 +8,8 @@ vi.mock('../../../api.js', () => ({
stopClient: vi.fn(),
stopVODClient: vi.fn(),
fetchActiveChannelStats: vi.fn(),
fetchAllConnectionStats: vi.fn(),
getAllConnectionStats: vi.fn(),
getVODStats: vi.fn(),
},
}));
@ -110,6 +112,34 @@ describe('StatsUtils', () => {
});
});
describe('fetchAllConnectionStats', () => {
it('should call API getAllConnectionStats', async () => {
const mockStats = {
live: { channels: [], count: 0 },
vod: { vod_connections: [] },
catchup: { timeshift_sessions: [] },
};
API.getAllConnectionStats.mockResolvedValue(mockStats);
const result = await StatsUtils.fetchAllConnectionStats();
expect(API.getAllConnectionStats).toHaveBeenCalledWith();
expect(API.getAllConnectionStats).toHaveBeenCalledTimes(1);
expect(result).toEqual(mockStats);
});
it('should propagate API errors', async () => {
const error = new Error('Failed to fetch combined stats');
API.getAllConnectionStats.mockRejectedValue(error);
await expect(StatsUtils.fetchAllConnectionStats()).rejects.toThrow(
'Failed to fetch combined stats'
);
});
});
describe('fetchActiveChannelStats', () => {
it('should call API fetchActiveChannelStats', async () => {
const mockStats = { channels: [] };