Merge pull request #1242 from cedric-marcoux/feat/timeshift-native-v2

feat: built-in XC catch-up (timeshift) support

CHANGELOG.md

@@ -7,6 +7,30 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- **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. — Thanks [@cedric-marcoux](https://github.com/cedric-marcoux) (#1242)
+  - **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.
+  - **Provider pool accounting.** Catch-up reserves a provider profile slot (`connection_pool`) before connecting upstream — same contract as live and VOD — and releases it when the session ends. When the default profile is at capacity it walks the account's alternate profiles with their own resolved credentials, and answers `503` when every profile is full.
+  - **Per-client session pool.** The first request without `session_id` receives a `301` redirect to the same URL with a stable `?session_id=` query parameter; that ID becomes the client identity for stats, stop keys, and pool coordination. Each viewer gets their own Redis-backed pool entry even when watching the same programme; idle sessions can be reclaimed via fingerprint matching (same user and programme, IP + user-agent score). Real scrubs within a session stop the in-flight stream through the standard stop-key mechanism; parallel startup probes (full-file, open-ended, and EOF tail `Range` requests) are deferred with `503` instead of opening extra upstream connections. Stream-limit exemptions for `ignore_same_channel_connections` apply only to sibling requests from the same `session_id`.
+  - **`xmltv_prev_days_override` under `Settings → EPG`** (default `0` = auto-detect). Verbose timeshift logging follows the standard logger DEBUG level.
+  - **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.
+
+### Performance
+
+- **M3U/XC stream refresh is faster on large accounts.** Steady-state refreshes split `bulk_update` into a lightweight touch pass (`last_seen` / `is_stale` only) for unchanged streams and a full column update only when provider metadata or catch-up fields actually change.
+- **M3U stream filters compile once per refresh.** Account filters are regex-compiled before batch workers start; accounts with no filters skip per-stream filter checks entirely.
+- **M3U refresh releases parse catalogs sooner.** Standard accounts stream-parse the on-disk M3U instead of loading the full file into RAM; `extinf_data` is dropped immediately after batch DB work, before stale cleanup and auto-sync.
+- **XC live refresh avoids redundant work during catalog filtering.** `collect_xc_streams` skips disabled categories before building entries and uses a shared URL prefix instead of formatting each stream URL separately. Auto-sync releases per-group logo/EPG caches after each group iteration.
+- **Celery workers return RSS after memory-intensive tasks.** `cleanup_memory()` accepts an optional `trim_heap` flag (glibc `malloc_trim`); Celery `task_postrun` enables it after `close_old_connections()` for M3U account/group refresh, EPG, VOD, and channel-matching tasks so worker memory drops back toward baseline instead of ratcheting across successive large jobs.
+
+### Fixed
+
+- **M3U group processing retries on poisoned DB connections.** `_db_query_with_retry` now treats psycopg desync errors (`DatabaseError`, e.g. `lost synchronization with server`) as transient; `process_groups` relationship loading uses it so a stale Celery worker connection resets once instead of failing the whole refresh.
+
 ## [0.27.1] - 2026-06-25
 
 ### Security

apps/channels/migrations/0038_add_catchup_fields.py

@@ -0,0 +1,101 @@
+"""Add denormalized catch-up fields to Stream and Channel."""
+
+from django.db import migrations, models
+
+
+def backfill_stream_catchup(apps, schema_editor):
+    """Derive is_catchup/catchup_days from Stream.custom_properties JSON."""
+    with schema_editor.connection.cursor() as cursor:
+        cursor.execute("""
+            UPDATE dispatcharr_channels_stream
+            SET is_catchup = TRUE,
+                catchup_days = COALESCE(
+                    CASE WHEN (custom_properties->>'tv_archive_duration') ~ '^\\d+$'
+                         THEN (custom_properties->>'tv_archive_duration')::int
+                         ELSE NULL
+                    END, 7
+                )
+            WHERE custom_properties IS NOT NULL
+              AND custom_properties != 'null'::jsonb
+              AND (
+                  custom_properties->>'tv_archive' = '1'
+                  -- JSON booleans extract as lowercase 'true' via ->>; the
+                  -- 'True' spelling covers Python-str values stored by
+                  -- older import code.
+                  OR custom_properties->>'tv_archive' = 'true'
+                  OR custom_properties->>'tv_archive' = 'True'
+              )
+        """)
+
+
+def backfill_channel_catchup(apps, schema_editor):
+    """Roll up catch-up fields from streams to channels."""
+    with schema_editor.connection.cursor() as cursor:
+        cursor.execute("""
+            UPDATE dispatcharr_channels_channel c SET
+                is_catchup = EXISTS (
+                    SELECT 1 FROM dispatcharr_channels_channelstream cs
+                    JOIN dispatcharr_channels_stream s ON s.id = cs.stream_id
+                    WHERE cs.channel_id = c.id AND s.is_catchup = TRUE
+                ),
+                catchup_days = COALESCE((
+                    SELECT MAX(s.catchup_days) FROM dispatcharr_channels_channelstream cs
+                    JOIN dispatcharr_channels_stream s ON s.id = cs.stream_id
+                    WHERE cs.channel_id = c.id AND s.is_catchup = TRUE
+                ), 0)
+        """)
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("dispatcharr_channels", "0037_auto_sync_overhaul"),
+    ]
+
+    operations = [
+        # Stream fields
+        migrations.AddField(
+            model_name="stream",
+            name="is_catchup",
+            field=models.BooleanField(
+                default=False,
+                db_index=True,
+                help_text="Whether this stream supports catch-up/timeshift (tv_archive=1)",
+            ),
+        ),
+        migrations.AddField(
+            model_name="stream",
+            name="catchup_days",
+            field=models.PositiveIntegerField(
+                default=0,
+                help_text="Number of days of catch-up archive available (tv_archive_duration)",
+            ),
+        ),
+        # Channel fields
+        migrations.AddField(
+            model_name="channel",
+            name="is_catchup",
+            field=models.BooleanField(
+                default=False,
+                db_index=True,
+                help_text="Whether any stream on this channel supports catch-up (tv_archive=1)",
+            ),
+        ),
+        migrations.AddField(
+            model_name="channel",
+            name="catchup_days",
+            field=models.PositiveIntegerField(
+                default=0,
+                help_text="Max catch-up archive days across all streams on this channel",
+            ),
+        ),
+        # Backfill existing data
+        migrations.RunPython(
+            backfill_stream_catchup,
+            reverse_code=migrations.RunPython.noop,
+        ),
+        migrations.RunPython(
+            backfill_channel_catchup,
+            reverse_code=migrations.RunPython.noop,
+        ),
+    ]

apps/channels/models.py

@@ -135,6 +135,17 @@ class Stream(models.Model):
         db_index=True
     )
 
+    # Populated at import from tv_archive / tv_archive_duration.
+    is_catchup = models.BooleanField(
+        default=False,
+        db_index=True,
+        help_text="Whether this stream supports catch-up/timeshift (tv_archive=1)",
+    )
+    catchup_days = models.PositiveIntegerField(
+        default=0,
+        help_text="Number of days of catch-up archive available (tv_archive_duration)",
+    )
+
     class Meta:
         # If you use m3u_account, you might do unique_together = ('name','url','m3u_account')
         verbose_name = "Stream"
@@ -364,6 +375,17 @@ class Channel(models.Model):
         help_text="The M3U account that auto-created this channel"
     )
 
+    # Populated at import; rolled up via ChannelStream signal / m3u refresh.
+    is_catchup = models.BooleanField(
+        default=False,
+        db_index=True,
+        help_text="Whether any stream on this channel supports catch-up (tv_archive=1)",
+    )
+    catchup_days = models.PositiveIntegerField(
+        default=0,
+        help_text="Max catch-up archive days across all streams on this channel",
+    )
+
     # Hidden channels are excluded from HDHR, M3U, EPG, and XC output queries.
     # Auto-sync still recognizes them so they are not recreated when their
     # underlying provider stream persists; this is an output-layer concern, not

apps/channels/serializers.py

@@ -448,6 +448,7 @@ class ChannelSerializer(serializers.ModelSerializer):
             "logo_id",
             "user_level",
             "is_adult",
+            "is_catchup",
             "hidden_from_output",
             "auto_created",
             "auto_created_by",

apps/channels/signals.py

@@ -5,7 +5,7 @@ from django.dispatch import receiver
 from django.utils.timezone import now, is_aware, make_aware
 from celery.result import AsyncResult
 from django_celery_beat.models import ClockedSchedule, PeriodicTask
-from .models import Channel, Stream, ChannelProfile, ChannelProfileMembership, Recording
+from .models import Channel, Stream, ChannelStream, ChannelProfile, ChannelProfileMembership, Recording
 from apps.m3u.models import M3UAccount
 from apps.epg.tasks import parse_programs_for_tvg_id
 import json
@@ -369,3 +369,20 @@ def schedule_task_on_save(sender, instance, created, **kwargs):
 @receiver(post_delete, sender=Recording)
 def revoke_task_on_delete(sender, instance, **kwargs):
     revoke_task(instance.task_id)
+
+
+@receiver([post_save, post_delete], sender=ChannelStream)
+def update_channel_catchup_fields(sender, instance, **kwargs):
+    """Roll up catch-up flags from active streams (UI path; import uses SQL rollup)."""
+    from django.db.models import Max
+
+    channel = instance.channel
+    catchup_qs = channel.streams.filter(
+        is_catchup=True,
+        m3u_account__is_active=True,
+    )
+    max_days = catchup_qs.aggregate(max_days=Max("catchup_days"))["max_days"]
+    Channel.objects.filter(pk=channel.pk).update(
+        is_catchup=catchup_qs.exists(),
+        catchup_days=max_days or 0,
+    )

apps/channels/tests/test_catchup_utils.py

@@ -0,0 +1,100 @@
+from unittest.mock import patch
+
+from django.test import RequestFactory, SimpleTestCase, TestCase
+
+from apps.accounts.models import User
+from apps.channels.models import Channel, ChannelStream, Stream
+from apps.channels.utils import resolve_xc_epg_prev_days
+from apps.m3u.models import M3UAccount
+
+
+class ResolveXcEpgPrevDaysTests(SimpleTestCase):
+    def setUp(self):
+        self.factory = RequestFactory()
+        self.user = User(username="xc-prev", custom_properties={})
+
+    def test_url_prev_days_zero_is_explicit(self):
+        request = self.factory.get("/xmltv.php?prev_days=0")
+        self.assertEqual(resolve_xc_epg_prev_days(request, self.user), 0)
+
+    def test_user_epg_prev_days_used_when_url_omitted(self):
+        self.user.custom_properties = {"epg_prev_days": 5}
+        request = self.factory.get("/xmltv.php")
+        self.assertEqual(resolve_xc_epg_prev_days(request, self.user), 5)
+
+    @patch("apps.channels.utils.compute_provider_archive_days_capped", return_value=14)
+    def test_auto_detect_only_when_no_url_or_user_default(self, mock_compute):
+        request = self.factory.get("/xmltv.php")
+        self.assertEqual(resolve_xc_epg_prev_days(request, self.user), 14)
+        mock_compute.assert_called_once()
+
+    @patch("apps.channels.utils.compute_provider_archive_days_capped", return_value=14)
+    def test_per_channel_epg_skips_global_auto_detect(self, mock_compute):
+        request = self.factory.get("/player_api.php")
+        self.assertEqual(
+            resolve_xc_epg_prev_days(request, self.user, auto_detect_fallback=False),
+            0,
+        )
+        mock_compute.assert_not_called()
+
+    @patch("apps.channels.utils.compute_provider_archive_days_capped", return_value=14)
+    def test_user_default_prevents_auto_detect(self, mock_compute):
+        self.user.custom_properties = {"epg_prev_days": 3}
+        request = self.factory.get("/xmltv.php")
+        self.assertEqual(resolve_xc_epg_prev_days(request, self.user), 3)
+        mock_compute.assert_not_called()
+
+    @patch("apps.channels.utils.compute_provider_archive_days_capped", return_value=14)
+    @patch("core.models.CoreSettings.get_xmltv_prev_days_override", return_value=7)
+    def test_epg_settings_override_prevents_auto_detect(self, _override, mock_compute):
+        request = self.factory.get("/xmltv.php")
+        self.assertEqual(resolve_xc_epg_prev_days(request, self.user), 7)
+        mock_compute.assert_not_called()
+
+
+class CatchupRollupActiveAccountTests(TestCase):
+    """Denormalized catch-up flags ignore disabled M3U accounts."""
+
+    @classmethod
+    def setUpTestData(cls):
+        cls.inactive = M3UAccount.objects.create(
+            name="catchup-inactive",
+            server_url="http://example.test",
+            account_type="XC",
+            is_active=False,
+        )
+
+    def test_channelstream_signal_ignores_inactive_catchup_stream(self):
+        channel = Channel.objects.create(name="inactive-only")
+        stream = Stream.objects.create(
+            name="inactive-catchup",
+            url="http://example.test/inactive",
+            m3u_account=self.inactive,
+            is_catchup=True,
+            catchup_days=9,
+        )
+        ChannelStream.objects.create(channel=channel, stream=stream, order=0)
+
+        channel.refresh_from_db()
+        self.assertFalse(channel.is_catchup)
+        self.assertEqual(channel.catchup_days, 0)
+
+    def test_rollup_ignores_inactive_catchup_stream(self):
+        from apps.m3u.tasks import rollup_channel_catchup_fields
+
+        channel = Channel.objects.create(name="rollup-inactive-only")
+        stream = Stream.objects.create(
+            name="rollup-inactive-catchup",
+            url="http://example.test/rollup-inactive",
+            m3u_account=self.inactive,
+            is_catchup=True,
+            catchup_days=9,
+        )
+        ChannelStream.objects.create(channel=channel, stream=stream, order=0)
+        Channel.objects.filter(pk=channel.pk).update(is_catchup=True, catchup_days=9)
+
+        rollup_channel_catchup_fields(self.inactive.id)
+
+        channel.refresh_from_db()
+        self.assertFalse(channel.is_catchup)
+        self.assertEqual(channel.catchup_days, 0)

apps/channels/utils.py

@@ -1,8 +1,13 @@
 import logging
 import threading
 
+from django.core.cache import cache
+
 logger = logging.getLogger(__name__)
 
+PROVIDER_ARCHIVE_CACHE_TTL_SECONDS = 300
+MAX_AUTO_PREV_DAYS = 30
+
 # Bound memory/DB work per chunk for large libraries (20k+ channels).
 EPG_LOGO_APPLY_BATCH_SIZE = 500
 EPG_LOGO_APPLY_MAX_ERRORS = 100
@@ -23,6 +28,91 @@ def format_channel_number(value, empty=""):
         return int(value)
     return value
 
+
+def compute_provider_archive_days_capped():
+    """Max ``catchup_days`` across active XC catch-up streams (capped, cached).
+
+    Cached briefly so XC XMLTV exports without an explicit ``prev_days`` do not
+    repeat the aggregate query on every request.
+    """
+    def _scan():
+        from django.db.models import Max
+
+        from apps.channels.models import Stream
+
+        result = Stream.objects.filter(
+            m3u_account__account_type="XC",
+            m3u_account__is_active=True,
+            is_catchup=True,
+        ).aggregate(max_days=Max("catchup_days"))
+        return min(result["max_days"] or 0, MAX_AUTO_PREV_DAYS)
+
+    return cache.get_or_set(
+        "channels:provider_archive_days_capped",
+        _scan,
+        PROVIDER_ARCHIVE_CACHE_TTL_SECONDS,
+    )
+
+
+def resolve_xc_epg_prev_days(request, user, *, auto_detect_fallback=True):
+    """Resolve ``prev_days`` for XC XMLTV and player_api EPG.
+
+    Args:
+        request: HTTP request (reads ``?prev_days=``).
+        user: Authenticated user (reads ``custom_properties.epg_prev_days``).
+        auto_detect_fallback: When True (XC XMLTV), fall back to the largest
+            provider archive depth. When False (per-channel EPG), return 0 so
+            ``xc_get_epg`` can expand to each channel's ``catchup_days``.
+
+    Resolution order:
+        1. URL ``?prev_days=`` (explicit; 0 means no past programmes)
+        2. ``user.custom_properties.epg_prev_days``
+        3. ``CoreSettings.epg_settings.xmltv_prev_days_override`` when > 0
+        4. Auto-detect (only when *auto_detect_fallback* is True)
+    """
+    user_custom = (user.custom_properties or {}) if user else {}
+    url_prev = request.GET.get("prev_days")
+    user_prev = user_custom.get("epg_prev_days") if user_custom else None
+
+    if url_prev is not None:
+        try:
+            return max(0, min(int(url_prev), MAX_AUTO_PREV_DAYS))
+        except (ValueError, TypeError):
+            return 0
+    if user_prev not in (None, ""):
+        try:
+            return max(0, min(int(user_prev), MAX_AUTO_PREV_DAYS))
+        except (ValueError, TypeError):
+            return 0
+
+    from core.models import CoreSettings
+
+    try:
+        override = int(CoreSettings.get_xmltv_prev_days_override() or 0)
+    except (TypeError, ValueError):
+        override = 0
+    if override > 0:
+        return max(0, min(override, MAX_AUTO_PREV_DAYS))
+    if auto_detect_fallback:
+        return compute_provider_archive_days_capped()
+    return 0
+
+
+def get_channel_catchup_streams(channel):
+    """Active catch-up streams for a channel, in ``channelstream`` order.
+
+    Inactive M3U accounts are excluded, matching live dispatch.
+    """
+    if not getattr(channel, "is_catchup", False):
+        return []
+
+    return list(
+        channel.streams.filter(is_catchup=True, m3u_account__is_active=True)
+        .order_by("channelstream__order")
+        .select_related("m3u_account")
+    )
+
+
 def increment_stream_count(account):
     with lock:
         current_usage = active_streams_map.get(account.id, 0)

apps/m3u/tasks.py

@@ -52,9 +52,9 @@ def _db_query_with_retry(fn, *, label="DB query", max_retries=2):
     Poisoned Celery worker connections often surface as OperationalError or as
     ``IndexError: list index out of range`` inside Django's row converters.
     """
-    from django.db import InterfaceError, OperationalError
+    from django.db import DatabaseError, InterfaceError, OperationalError
 
-    transient_errors = (OperationalError, InterfaceError, IndexError)
+    transient_errors = (OperationalError, InterfaceError, IndexError, DatabaseError)
     for attempt in range(max_retries):
         try:
             return fn()
@@ -134,11 +134,23 @@ def _ensure_m3u_refresh_terminal_status(account_id):
 _EXTINF_ATTR_RE = re.compile(r'([^\s=]+)\s*=\s*(["\'])(.*?)\2')
 
 
+def _open_m3u_text_source(source_path):
+    """Open an on-disk M3U (or .m3u.gz) file for line-by-line parsing."""
+    if source_path.endswith(".gz"):
+        return gzip.open(source_path, "rt", encoding="utf-8")
+    return open(source_path, "r", encoding="utf-8")
+
+
 def fetch_m3u_lines(account, use_cache=False):
+    """Fetch M3U source for parsing.
+
+    On success returns ``(source, True)`` where *source* is either a filesystem
+    path (streamed during parse) or, for ZIP uploads only, an in-memory line
+    list. Failures return ``(None, False)``.
+    """
     os.makedirs(m3u_dir, exist_ok=True)
     file_path = os.path.join(m3u_dir, f"{account.id}.m3u")
 
-    """Fetch M3U file lines efficiently."""
     if account.server_url:
         if not use_cache or not os.path.exists(file_path):
             try:
@@ -213,7 +225,7 @@ def fetch_m3u_lines(account, use_cache=False):
                         status="error",
                         error=error_msg,
                     )
-                    return [], False
+                    return None, False
 
                 # Only call raise_for_status if we have a success code (this should not raise now)
                 response.raise_for_status()
@@ -288,7 +300,7 @@ def fetch_m3u_lines(account, use_cache=False):
                             status="error",
                             error=error_msg,
                         )
-                        return [], False
+                        return None, False
 
                     # Validate the file by reading only the first portion from
                     # disk — no need to load the entire file into memory just
@@ -360,7 +372,7 @@ def fetch_m3u_lines(account, use_cache=False):
                                 status="error",
                                 error=error_msg,
                             )
-                            return [], False
+                            return None, False
 
                     except UnicodeDecodeError:
                         with open(temp_path, "rb") as vf:
@@ -378,7 +390,7 @@ def fetch_m3u_lines(account, use_cache=False):
                             status="error",
                             error=error_msg,
                         )
-                        return [], False
+                        return None, False
 
                     # Validation passed — promote temp file to final path
                     os.replace(temp_path, file_path)
@@ -433,7 +445,7 @@ def fetch_m3u_lines(account, use_cache=False):
                     status="error",
                     error=error_msg,
                 )
-                return [], False
+                return None, False
             except requests.exceptions.RequestException as e:
                 # Handle other request errors (connection, timeout, etc.)
                 if "timeout" in str(e).lower():
@@ -454,7 +466,7 @@ def fetch_m3u_lines(account, use_cache=False):
                     status="error",
                     error=error_msg,
                 )
-                return [], False
+                return None, False
             except Exception as e:
                 # Handle any other unexpected errors
                 error_msg = f"Unexpected error while fetching M3U file from URL: {account.server_url} - {str(e)}"
@@ -469,7 +481,7 @@ def fetch_m3u_lines(account, use_cache=False):
                     status="error",
                     error=error_msg,
                 )
-                return [], False
+                return None, False
 
         # Check if the file exists and is not empty (fallback check - should not happen with new validation)
         if not os.path.exists(file_path) or os.path.getsize(file_path) == 0:
@@ -481,27 +493,14 @@ def fetch_m3u_lines(account, use_cache=False):
             send_m3u_update(
                 account.id, "downloading", 100, status="error", error=error_msg
             )
-            return [], False  # Return empty list and False for success
+            return None, False
 
-        try:
-            with open(file_path, "r", encoding="utf-8") as f:
-                return f.readlines(), True
-        except Exception as e:
-            error_msg = f"Error reading M3U file: {str(e)}"
-            logger.error(error_msg)
-            account.status = M3UAccount.Status.ERROR
-            account.last_message = error_msg
-            account.save(update_fields=["status", "last_message"])
-            send_m3u_update(
-                account.id, "downloading", 100, status="error", error=error_msg
-            )
-            return [], False
+        return file_path, True
 
     elif account.file_path:
         try:
             if account.file_path.endswith(".gz"):
-                with gzip.open(account.file_path, "rt", encoding="utf-8") as f:
-                    return f.readlines(), True
+                return account.file_path, True
 
             elif account.file_path.endswith(".zip"):
                 with zipfile.ZipFile(account.file_path, "r") as zip_file:
@@ -522,11 +521,10 @@ def fetch_m3u_lines(account, use_cache=False):
                     send_m3u_update(
                         account.id, "downloading", 100, status="error", error=error_msg
                     )
-                    return [], False
+                    return None, False
 
             else:
-                with open(account.file_path, "r", encoding="utf-8") as f:
-                    return f.readlines(), True
+                return account.file_path, True
 
         except (IOError, OSError, zipfile.BadZipFile, gzip.BadGzipFile) as e:
             error_msg = f"Error opening file {account.file_path}: {e}"
@@ -537,7 +535,7 @@ def fetch_m3u_lines(account, use_cache=False):
             send_m3u_update(
                 account.id, "downloading", 100, status="error", error=error_msg
             )
-            return [], False
+            return None, False
 
     # Neither server_url nor uploaded_file is available
     error_msg = "No M3U source available (missing URL and file)"
@@ -546,7 +544,7 @@ def fetch_m3u_lines(account, use_cache=False):
     account.last_message = error_msg
     account.save(update_fields=["status", "last_message"])
     send_m3u_update(account.id, "downloading", 100, status="error", error=error_msg)
-    return [], False
+    return None, False
 
 
 def get_case_insensitive_attr(attributes, key, default=""):
@@ -746,13 +744,16 @@ def process_groups(account, groups, scan_start_time=None):
     all_group_objs = existing_group_objs + newly_created_group_objs
 
     # Get existing relationships for this account
-    existing_relationships = {
-        rel.channel_group.name: rel
-        for rel in ChannelGroupM3UAccount.objects.filter(
-            m3u_account=account,
-            channel_group__name__in=groups.keys()
-        ).select_related('channel_group')
-    }
+    existing_relationships = _db_query_with_retry(
+        lambda: {
+            rel.channel_group.name: rel
+            for rel in ChannelGroupM3UAccount.objects.filter(
+                m3u_account=account,
+                channel_group__name__in=groups.keys(),
+            ).select_related("channel_group")
+        },
+        label=f"process_groups relationships for account {account.id}",
+    )
 
     relations_to_create = []
     relations_to_update = []
@@ -905,6 +906,11 @@ def collect_xc_streams(account_id, enabled_groups):
             account.get_user_agent(),
         ) as xc_client:
 
+            stream_url_prefix = (
+                f"{xc_client.server_url.rstrip('/')}/live/"
+                f"{xc_client.username}/{xc_client.password}/"
+            )
+
             # Fetch ALL live streams in a single API call (much more efficient)
             logger.info("Fetching ALL live streams from XC provider...")
             all_xc_streams = xc_client.get_all_live_streams()  # Get all streams without category filter
@@ -917,46 +923,49 @@ def collect_xc_streams(account_id, enabled_groups):
 
             # Filter streams based on enabled categories
             for stream in all_xc_streams:
-                # Fall back to a generated name if the provider returns null/empty
-                stream_name = stream.get("name") or f"{account.name} - {stream.get('stream_id', 'Unknown')}"
+                category_id = str(stream.get("category_id", ""))
+                if category_id not in enabled_category_ids:
+                    continue
+
+                group_info = enabled_category_ids[category_id]
+                stream_name = stream.get("name") or (
+                    f"{account.name} - {stream.get('stream_id', 'Unknown')}"
+                )
                 if not stream.get("name"):
                     logger.warning(
-                        f"XC stream has null/empty name; using generated name '{stream_name}' "
-                        f"(stream_id={stream.get('stream_id', 'unknown')})"
+                        "XC stream has null/empty name; using generated name '%s' "
+                        "(stream_id=%s)",
+                        stream_name, stream.get("stream_id", "unknown"),
                     )
 
-                # Get the category_id for this stream
-                category_id = str(stream.get("category_id", ""))
-
-                # Only include streams from enabled categories
-                if category_id in enabled_category_ids:
-                    group_info = enabled_category_ids[category_id]
-
-                    # Convert XC stream to our standard format with all properties preserved
-                    stream_data = {
-                        "name": stream_name,
-                        "url": xc_client.get_stream_url(stream["stream_id"]),
-                        "attributes": {
-                            "tvg-id": stream.get("epg_channel_id", ""),
-                            "tvg-logo": stream.get("stream_icon", ""),
-                            "group-title": group_info["name"],
-                            # Preserve all XC stream properties as custom attributes
-                            "stream_id": str(stream.get("stream_id", "")),
-                            "num": stream.get("num"),
-                            "category_id": category_id,
-                            "stream_type": stream.get("stream_type", ""),
-                            "added": stream.get("added", ""),
-                            "is_adult": str(stream.get("is_adult", "0")),
-                            "custom_sid": stream.get("custom_sid", ""),
-                            # Include any other properties that might be present
-                            **{k: str(v) for k, v in stream.items() if k not in [
-                                "name", "stream_id", "epg_channel_id", "stream_icon",
-                                "category_id", "stream_type", "added", "is_adult", "custom_sid", "num"
-                            ] and v is not None}
-                        }
-                    }
-                    all_streams.append(stream_data)
-                    filtered_count += 1
+                stream_id = stream.get("stream_id")
+                attributes = {
+                    "tvg-id": stream.get("epg_channel_id", ""),
+                    "tvg-logo": stream.get("stream_icon", ""),
+                    "group-title": group_info["name"],
+                    "stream_id": str(stream.get("stream_id", "")),
+                    "num": stream.get("num"),
+                    "category_id": category_id,
+                    "stream_type": stream.get("stream_type", ""),
+                    "added": stream.get("added", ""),
+                    "is_adult": str(stream.get("is_adult", "0")),
+                    "custom_sid": stream.get("custom_sid", ""),
+                    **{
+                        k: str(v)
+                        for k, v in stream.items()
+                        if k not in [
+                            "name", "stream_id", "epg_channel_id", "stream_icon",
+                            "category_id", "stream_type", "added", "is_adult",
+                            "custom_sid", "num",
+                        ] and v is not None
+                    },
+                }
+                all_streams.append({
+                    "name": stream_name,
+                    "url": f"{stream_url_prefix}{stream_id}.ts",
+                    "attributes": attributes,
+                })
+                filtered_count += 1
 
             # Drop the full provider catalog before returning; only filtered rows are needed.
             del all_xc_streams
@@ -971,6 +980,60 @@ def collect_xc_streams(account_id, enabled_groups):
     )
     return all_streams
 
+
+def _compile_m3u_stream_filters(filter_queryset):
+    """Compile account M3UFilter rows once per refresh for batch workers."""
+    compiled = []
+    for filter_obj in filter_queryset:
+        flags = (
+            re.IGNORECASE
+            if (filter_obj.custom_properties or {}).get("case_sensitive", True) is False
+            else 0
+        )
+        compiled.append((re.compile(filter_obj.regex_pattern, flags), filter_obj))
+    return compiled
+
+
+def _stream_passes_m3u_filters(name, url, group_title, compiled_filters):
+    """Return False when the first matching filter excludes the stream."""
+    for pattern, filter_obj in compiled_filters:
+        logger.trace("Checking filter pattern %s", pattern.pattern)
+        if filter_obj.filter_type == "url":
+            target = url
+        elif filter_obj.filter_type == "group":
+            target = group_title
+        else:
+            target = name
+
+        if pattern.search(target or ""):
+            logger.debug(
+                "Stream %s - %s matches filter pattern %s",
+                name, url, filter_obj.regex_pattern,
+            )
+            return not filter_obj.exclude
+    return True
+
+
+_STREAM_TOUCH_FIELDS = ("last_seen", "is_stale")
+_STREAM_CHANGED_FIELDS = (
+    "name", "url", "logo_url", "tvg_id", "custom_properties", "is_adult",
+    "last_seen", "updated_at", "is_stale", "stream_id", "stream_chno",
+    "channel_group_id", "is_catchup", "catchup_days",
+)
+
+
+def _bulk_update_stream_refresh_batches(changed_streams, touch_streams, *, batch_size):
+    """Unchanged streams only need last_seen/is_stale; changed rows get the full set."""
+    if touch_streams:
+        Stream.objects.bulk_update(
+            touch_streams, list(_STREAM_TOUCH_FIELDS), batch_size=batch_size,
+        )
+    if changed_streams:
+        Stream.objects.bulk_update(
+            changed_streams, list(_STREAM_CHANGED_FIELDS), batch_size=batch_size,
+        )
+
+
 def process_xc_category_direct(account_id, batch, groups, hash_keys):
     from django.db import connections
 
@@ -981,6 +1044,7 @@ def process_xc_category_direct(account_id, batch, groups, hash_keys):
 
     streams_to_create = []
     streams_to_update = []
+    streams_to_touch = []
     stream_hashes = {}
 
     try:
@@ -1053,6 +1117,13 @@ def process_xc_category_direct(account_id, batch, groups, hash_keys):
                             name, url, tvg_id, hash_keys, m3u_id=account_id, group=group_title,
                             account_type='XC', stream_id=provider_stream_id
                         )
+                        _tv_archive = str(stream.get("tv_archive", "0"))
+                        _is_catchup = _tv_archive in ("1", "True")
+                        try:
+                            _catchup_days = int(stream.get("tv_archive_duration", 0) or 0)
+                        except (TypeError, ValueError):
+                            _catchup_days = 0
+
                         stream_props = {
                             "name": name,
                             "url": url,
@@ -1066,6 +1137,8 @@ def process_xc_category_direct(account_id, batch, groups, hash_keys):
                             "is_stale": False,
                             "stream_id": provider_stream_id,
                             "stream_chno": stream_chno,
+                            "is_catchup": _is_catchup,
+                            "catchup_days": _catchup_days,
                         }
 
                         if stream_hash not in stream_hashes:
@@ -1080,7 +1153,7 @@ def process_xc_category_direct(account_id, batch, groups, hash_keys):
         existing_streams = {
             s.stream_hash: s
             for s in Stream.objects.filter(stream_hash__in=stream_hashes.keys()).select_related('m3u_account').only(
-                'id', 'stream_hash', 'name', 'url', 'logo_url', 'tvg_id', 'custom_properties', 'last_seen', 'updated_at', 'm3u_account', 'stream_id', 'stream_chno', 'channel_group_id'
+                'id', 'stream_hash', 'name', 'url', 'logo_url', 'tvg_id', 'custom_properties', 'last_seen', 'updated_at', 'm3u_account', 'stream_id', 'stream_chno', 'channel_group_id', 'is_catchup', 'catchup_days'
             )
         }
 
@@ -1097,7 +1170,9 @@ def process_xc_category_direct(account_id, batch, groups, hash_keys):
                     obj.is_adult != stream_props["is_adult"] or
                     obj.stream_id != stream_props["stream_id"] or
                     obj.stream_chno != stream_props["stream_chno"] or
-                    obj.channel_group_id != stream_props["channel_group_id"]
+                    obj.channel_group_id != stream_props["channel_group_id"] or
+                    obj.is_catchup != stream_props["is_catchup"] or
+                    obj.catchup_days != stream_props["catchup_days"]
                 )
 
                 if changed:
@@ -1108,11 +1183,9 @@ def process_xc_category_direct(account_id, batch, groups, hash_keys):
                     obj.is_stale = False
                     streams_to_update.append(obj)
                 else:
-                    # Always update last_seen, even if nothing else changed
                     obj.last_seen = timezone.now()
                     obj.is_stale = False
-                    # Don't update updated_at for unchanged streams
-                    streams_to_update.append(obj)
+                    streams_to_touch.append(obj)
 
                 # Remove from existing_streams since we've processed it
                 del existing_streams[stream_hash]
@@ -1129,13 +1202,9 @@ def process_xc_category_direct(account_id, batch, groups, hash_keys):
                 if streams_to_create:
                     Stream.objects.bulk_create(streams_to_create, ignore_conflicts=True)
 
-                if streams_to_update:
-                    # Simplified bulk update for better performance
-                    Stream.objects.bulk_update(
-                        streams_to_update,
-                        ['name', 'url', 'logo_url', 'tvg_id', 'custom_properties', 'is_adult', 'last_seen', 'updated_at', 'is_stale', 'stream_id', 'stream_chno', 'channel_group_id'],
-                        batch_size=150  # Smaller batch size for XC processing
-                    )
+                _bulk_update_stream_refresh_batches(
+                    streams_to_update, streams_to_touch, batch_size=150,
+                )
 
                 # Update last_seen for any remaining existing streams that weren't processed
                 if len(existing_streams.keys()) > 0:
@@ -1143,7 +1212,10 @@ def process_xc_category_direct(account_id, batch, groups, hash_keys):
         except Exception as e:
             logger.error(f"Bulk operation failed for XC streams: {str(e)}")
 
-        retval = f"Batch processed: {len(streams_to_create)} created, {len(streams_to_update)} updated."
+        retval = (
+            f"Batch processed: {len(streams_to_create)} created, "
+            f"{len(streams_to_update) + len(streams_to_touch)} updated."
+        )
 
     except Exception as e:
         logger.error(f"XC category processing error: {str(e)}")
@@ -1153,14 +1225,18 @@ def process_xc_category_direct(account_id, batch, groups, hash_keys):
         connections.close_all()
 
     # Aggressive garbage collection
-    del streams_to_create, streams_to_update, stream_hashes, existing_streams
+    del streams_to_create, streams_to_update, streams_to_touch, stream_hashes, existing_streams
     gc.collect()
 
     return retval
 
 
-def process_m3u_batch_direct(account_id, batch, groups, hash_keys):
-    """Processes a batch of M3U streams using bulk operations with thread-safe DB connections."""
+def process_m3u_batch_direct(account_id, batch, groups, hash_keys, compiled_filters=None):
+    """Processes a batch of M3U streams using bulk operations with thread-safe DB connections.
+
+    ``compiled_filters`` should be pre-built once per account refresh and shared
+    across batch workers. Pass an empty list when the account has no filters.
+    """
     from django.db import connections
 
     # Ensure clean database connections for threading
@@ -1168,33 +1244,22 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys):
 
     account = M3UAccount.objects.get(id=account_id)
 
-    compiled_filters = [
-        (
-            re.compile(
-                f.regex_pattern,
-                (
-                    re.IGNORECASE
-                    if (f.custom_properties or {}).get(
-                        "case_sensitive", True
-                    )
-                    == False
-                    else 0
-                ),
-            ),
-            f,
-        )
-        for f in account.filters.order_by("order")
-    ]
+    if compiled_filters is None:
+        compiled_filters = _compile_m3u_stream_filters(account.filters.order_by("order"))
 
     streams_to_create = []
     streams_to_update = []
+    streams_to_touch = []
     stream_hashes = {}
 
     name_max_length = Stream._meta.get_field('name').max_length
 
     logger.debug(f"Processing batch of {len(batch)} for M3U account {account_id}")
     if compiled_filters:
-        logger.debug(f"Using compiled filters: {[f[1].regex_pattern for f in compiled_filters]}")
+        logger.debug(
+            "Using compiled filters: %s",
+            [filter_obj.regex_pattern for _, filter_obj in compiled_filters],
+        )
     for stream_info in batch:
         try:
             name, url = stream_info["name"], stream_info["url"]
@@ -1215,25 +1280,12 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys):
             group_title = get_case_insensitive_attr(
                 stream_info["attributes"], "group-title", "Default Group"
             )
-            logger.debug(f"Processing stream: {name} - {url} in group {group_title}")
-            include = True
-            for pattern, filter in compiled_filters:
-                logger.trace(f"Checking filter pattern {pattern}")
-                target = name
-                if filter.filter_type == "url":
-                    target = url
-                elif filter.filter_type == "group":
-                    target = group_title
+            logger.trace("Processing stream: %s - %s in group %s", name, url, group_title)
 
-                if pattern.search(target or ""):
-                    logger.debug(
-                        f"Stream {name} - {url} matches filter pattern {filter.regex_pattern}"
-                    )
-                    include = not filter.exclude
-                    break
-
-            if not include:
-                logger.debug(f"Stream excluded by filter, skipping.")
+            if compiled_filters and not _stream_passes_m3u_filters(
+                name, url, group_title, compiled_filters,
+            ):
+                logger.debug("Stream excluded by filter, skipping.")
                 continue
 
             # Filter out disabled groups for this account
@@ -1279,6 +1331,14 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys):
                 account_type=account_type_for_hash, stream_id=provider_stream_id
             )
 
+            _attrs = stream_info["attributes"]
+            _tv_archive_m3u = str(_attrs.get("tv_archive", "0"))
+            _is_catchup_m3u = _tv_archive_m3u in ("1", "True")
+            try:
+                _catchup_days_m3u = int(_attrs.get("tv_archive_duration", 0) or 0)
+            except (TypeError, ValueError):
+                _catchup_days_m3u = 0
+
             stream_props = {
                 "name": name,
                 "url": url,
@@ -1287,11 +1347,13 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys):
                 "m3u_account": account,
                 "channel_group_id": int(groups.get(group_title)),
                 "stream_hash": stream_hash,
-                "custom_properties": {**stream_info["attributes"], "vlc_opts": stream_info["vlc_opts"]} if "vlc_opts" in stream_info else stream_info["attributes"],
-                "is_adult": parse_is_adult(stream_info["attributes"].get("is_adult", 0)),
+                "custom_properties": {**_attrs, "vlc_opts": stream_info["vlc_opts"]} if "vlc_opts" in stream_info else _attrs,
+                "is_adult": parse_is_adult(_attrs.get("is_adult", 0)),
                 "is_stale": False,
                 "stream_id": provider_stream_id,
                 "stream_chno": channel_num,
+                "is_catchup": _is_catchup_m3u,
+                "catchup_days": _catchup_days_m3u,
             }
 
             if stream_hash not in stream_hashes:
@@ -1303,7 +1365,7 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys):
     existing_streams = {
         s.stream_hash: s
         for s in Stream.objects.filter(stream_hash__in=stream_hashes.keys()).select_related('m3u_account').only(
-            'id', 'stream_hash', 'name', 'url', 'logo_url', 'tvg_id', 'custom_properties', 'last_seen', 'updated_at', 'm3u_account', 'stream_id', 'stream_chno', 'channel_group_id'
+            'id', 'stream_hash', 'name', 'url', 'logo_url', 'tvg_id', 'custom_properties', 'last_seen', 'updated_at', 'm3u_account', 'stream_id', 'stream_chno', 'channel_group_id', 'is_catchup', 'catchup_days'
         )
     }
 
@@ -1320,14 +1382,15 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys):
                 obj.is_adult != stream_props["is_adult"] or
                 obj.stream_id != stream_props["stream_id"] or
                 obj.stream_chno != stream_props["stream_chno"] or
-                obj.channel_group_id != stream_props["channel_group_id"]
+                obj.channel_group_id != stream_props["channel_group_id"] or
+                obj.is_catchup != stream_props["is_catchup"] or
+                obj.catchup_days != stream_props["catchup_days"]
             )
 
-            # Always update last_seen
             obj.last_seen = timezone.now()
+            obj.is_stale = False
 
             if changed:
-                # Only update fields that changed and set updated_at
                 obj.name = stream_props["name"]
                 obj.url = stream_props["url"]
                 obj.logo_url = stream_props["logo_url"]
@@ -1337,12 +1400,12 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys):
                 obj.stream_id = stream_props["stream_id"]
                 obj.stream_chno = stream_props["stream_chno"]
                 obj.channel_group_id = stream_props["channel_group_id"]
+                obj.is_catchup = stream_props["is_catchup"]
+                obj.catchup_days = stream_props["catchup_days"]
                 obj.updated_at = timezone.now()
-
-            # Always mark as not stale since we saw it in this refresh
-            obj.is_stale = False
-
-            streams_to_update.append(obj)
+                streams_to_update.append(obj)
+            else:
+                streams_to_touch.append(obj)
         else:
             # New stream
             stream_props["last_seen"] = timezone.now()
@@ -1355,23 +1418,22 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys):
             if streams_to_create:
                 Stream.objects.bulk_create(streams_to_create, ignore_conflicts=True)
 
-            if streams_to_update:
-                # Update all streams in a single bulk operation
-                Stream.objects.bulk_update(
-                    streams_to_update,
-                    ['name', 'url', 'logo_url', 'tvg_id', 'custom_properties', 'is_adult', 'last_seen', 'updated_at', 'is_stale', 'stream_id', 'stream_chno', 'channel_group_id'],
-                    batch_size=200
-                )
+            _bulk_update_stream_refresh_batches(
+                streams_to_update, streams_to_touch, batch_size=200,
+            )
     except Exception as e:
         logger.error(f"Bulk operation failed: {str(e)}")
 
-    retval = f"M3U account: {account_id}, Batch processed: {len(streams_to_create)} created, {len(streams_to_update)} updated."
+    retval = (
+        f"M3U account: {account_id}, Batch processed: {len(streams_to_create)} created, "
+        f"{len(streams_to_update) + len(streams_to_touch)} updated."
+    )
 
     # Clean up database connections for threading
     connections.close_all()
 
     # Free batch data structures (reference-counted deallocation)
-    del streams_to_create, streams_to_update, stream_hashes, existing_streams
+    del streams_to_create, streams_to_update, streams_to_touch, stream_hashes, existing_streams
     gc.collect()
 
     return retval
@@ -1658,28 +1720,49 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
             release_task_lock("refresh_m3u_account_groups", account_id)
             return error_msg, None
     else:
-        lines, success = fetch_m3u_lines(account, use_cache)
+        source, success = fetch_m3u_lines(account, use_cache)
         if not success:
             # If fetch failed, don't continue processing
             lock_renewer.stop()
             release_task_lock("refresh_m3u_account_groups", account_id)
             return f"Failed to fetch M3U data for account_id={account_id}.", None
 
-        # Log basic file structure for debugging
-        logger.debug(f"Processing {len(lines)} lines from M3U file")
-
         valid_stream_count = 0
 
-        for entry in iter_m3u_entries(lines):
-            valid_stream_count += 1
-            group_title_attr = get_case_insensitive_attr(entry["attributes"], "group-title", "")
-            if group_title_attr and group_title_attr not in groups:
-                logger.debug(f"Found new group for M3U account {account_id}: '{group_title_attr}'")
-                groups[group_title_attr] = {}
-            extinf_data.append(entry)
+        if isinstance(source, str):
+            logger.debug(f"Streaming M3U parse from {source}")
+            with _open_m3u_text_source(source) as m3u_file:
+                entry_iter = iter_m3u_entries(m3u_file)
+                for entry in entry_iter:
+                    valid_stream_count += 1
+                    group_title_attr = get_case_insensitive_attr(entry["attributes"], "group-title", "")
+                    if group_title_attr and group_title_attr not in groups:
+                        logger.debug(f"Found new group for M3U account {account_id}: '{group_title_attr}'")
+                        groups[group_title_attr] = {}
+                    extinf_data.append(entry)
 
-            if valid_stream_count % 1000 == 0:
-                logger.debug(f"Processed {valid_stream_count} valid streams so far for M3U account: {account_id}")
+                    if valid_stream_count % 1000 == 0:
+                        logger.debug(
+                            f"Processed {valid_stream_count} valid streams so far for M3U account: {account_id}"
+                        )
+        else:
+            logger.debug(f"Processing {len(source)} in-memory M3U lines (zip upload)")
+            try:
+                for entry in iter_m3u_entries(source):
+                    valid_stream_count += 1
+                    group_title_attr = get_case_insensitive_attr(entry["attributes"], "group-title", "")
+                    if group_title_attr and group_title_attr not in groups:
+                        logger.debug(f"Found new group for M3U account {account_id}: '{group_title_attr}'")
+                        groups[group_title_attr] = {}
+                    extinf_data.append(entry)
+
+                    if valid_stream_count % 1000 == 0:
+                        logger.debug(
+                            f"Processed {valid_stream_count} valid streams so far for M3U account: {account_id}"
+                        )
+            finally:
+                del source
+                gc.collect()
 
         logger.info(f"M3U parsing complete - Valid streams: {valid_stream_count}")
 
@@ -1865,6 +1948,60 @@ def _classify_sync_failure(exc):
     return "OTHER"
 
 
+def rollup_channel_catchup_fields(account_id):
+    """Roll up catch-up flags from streams to channels (active accounts only).
+
+    Both the aggregate update and the self-heal pass are limited to channels
+    that still have at least one stream from *account_id*.
+    """
+    from django.db import connection
+
+    account_channels = """
+        SELECT DISTINCT cs.channel_id
+        FROM dispatcharr_channels_channelstream cs
+        JOIN dispatcharr_channels_stream s ON s.id = cs.stream_id
+        WHERE s.m3u_account_id = %s
+    """
+
+    with connection.cursor() as cur:
+        cur.execute(f"""
+            WITH agg AS (
+                SELECT
+                    cs.channel_id,
+                    bool_or(s.is_catchup AND a.is_active)  AS any_catchup,
+                    MAX(s.catchup_days) FILTER (WHERE s.is_catchup AND a.is_active) AS max_days
+                FROM dispatcharr_channels_channelstream cs
+                JOIN dispatcharr_channels_stream s ON s.id = cs.stream_id
+                JOIN m3u_m3uaccount a ON a.id = s.m3u_account_id
+                WHERE cs.channel_id IN ({account_channels})
+                GROUP BY cs.channel_id
+            )
+            UPDATE dispatcharr_channels_channel c
+            SET
+                is_catchup   = COALESCE(agg.any_catchup, FALSE),
+                catchup_days = COALESCE(agg.max_days, 0)
+            FROM agg
+            WHERE c.id = agg.channel_id
+        """, [account_id])
+
+        # Self-heal stale is_catchup flags on account-linked channels only.
+        cur.execute(f"""
+            UPDATE dispatcharr_channels_channel c
+            SET is_catchup = FALSE, catchup_days = 0
+            WHERE c.is_catchup = TRUE
+              AND c.id IN ({account_channels})
+              AND NOT EXISTS (
+                  SELECT 1
+                  FROM dispatcharr_channels_channelstream cs
+                  JOIN dispatcharr_channels_stream s ON s.id = cs.stream_id
+                  JOIN m3u_m3uaccount a ON a.id = s.m3u_account_id
+                  WHERE cs.channel_id = c.id
+                    AND s.is_catchup = TRUE
+                    AND a.is_active = TRUE
+              )
+        """, [account_id])
+
+
 @shared_task
 def sync_auto_channels(account_id, scan_start_time=None):
     """
@@ -2798,6 +2935,17 @@ def sync_auto_channels(account_id, scan_start_time=None):
                     f"{pack_result['failed']} failed"
                 )
 
+            # Release per-group working sets before the next group iteration.
+            del (
+                current_streams,
+                logo_cache_by_url,
+                epg_cache_by_tvg_id,
+                existing_channel_map,
+                existing_channels_by_id,
+                existing_channel_streams,
+                processed_stream_ids,
+            )
+
         # Cleanup mode read from account.custom_properties.orphan_channel_cleanup:
         # "always" (default; key absent) removes every orphan auto channel;
         # "preserve_customized" keeps those with a ChannelOverride row;
@@ -2833,7 +2981,7 @@ def sync_auto_channels(account_id, scan_start_time=None):
             f"{channels_created} created, {channels_updated} updated, "
             f"{channels_deleted} deleted, {channels_failed} failed"
         )
-        return {
+        result = {
             "status": "ok",
             "channels_created": channels_created,
             "channels_updated": channels_updated,
@@ -2841,6 +2989,9 @@ def sync_auto_channels(account_id, scan_start_time=None):
             "channels_failed": channels_failed,
             "failed_stream_details": failed_stream_details,
         }
+        del failed_stream_details
+        gc.collect()
+        return result
 
     except Exception as e:
         logger.error(f"Error in auto channel sync for account {account_id}: {str(e)}")
@@ -3224,7 +3375,15 @@ def _refresh_single_m3u_account_impl(account_id):
         )
         account = _get_active_m3u_account(account_id)
 
-        filters = list(account.filters.all())
+        compiled_stream_filters = _compile_m3u_stream_filters(
+            account.filters.order_by("order")
+        )
+        if compiled_stream_filters:
+            logger.debug(
+                "Account %s has %s stream filter(s) for this refresh",
+                account_id,
+                len(compiled_stream_filters),
+            )
 
         # Check if VOD is enabled for this account
         vod_enabled = ensure_custom_properties_dict(account.custom_properties).get(
@@ -3401,7 +3560,14 @@ def _refresh_single_m3u_account_impl(account_id):
             with ThreadPoolExecutor(max_workers=max_workers) as executor:
                 # Submit batch processing tasks using direct functions (now thread-safe)
                 future_to_batch = {
-                    executor.submit(process_m3u_batch_direct, account_id, batch, existing_groups, hash_keys): i
+                    executor.submit(
+                        process_m3u_batch_direct,
+                        account_id,
+                        batch,
+                        existing_groups,
+                        hash_keys,
+                        compiled_stream_filters,
+                    ): i
                     for i, batch in enumerate(batches)
                 }
 
@@ -3456,6 +3622,10 @@ def _refresh_single_m3u_account_impl(account_id):
                         batches[batch_idx] = None
 
             logger.info(f"Thread-based processing completed for account {account_id}")
+
+            # Parsed catalog is no longer needed; drop before stale cleanup / auto-sync.
+            del extinf_data, batches
+            gc.collect()
         else:
             # For XC accounts, get the groups with their custom properties containing xc_id
             logger.debug(f"Processing XC account with groups: {existing_groups}")
@@ -3493,6 +3663,8 @@ def _refresh_single_m3u_account_impl(account_id):
             logger.info("Fetching all XC streams from provider and filtering by enabled categories...")
             all_xc_streams = collect_xc_streams(account_id, filtered_groups)
 
+            del channel_group_relationships, filtered_groups
+
             if not all_xc_streams:
                 logger.warning("No streams collected from XC groups")
             else:
@@ -3514,7 +3686,14 @@ def _refresh_single_m3u_account_impl(account_id):
                 with ThreadPoolExecutor(max_workers=max_workers) as executor:
                     # Submit stream batch processing tasks (reuse standard M3U processing)
                     future_to_batch = {
-                        executor.submit(process_m3u_batch_direct, account_id, batch, existing_groups, hash_keys): i
+                        executor.submit(
+                            process_m3u_batch_direct,
+                            account_id,
+                            batch,
+                            existing_groups,
+                            hash_keys,
+                            compiled_stream_filters,
+                        ): i
                         for i, batch in enumerate(batches)
                     }
 
@@ -3570,6 +3749,9 @@ def _refresh_single_m3u_account_impl(account_id):
 
                 logger.info(f"XC thread-based processing completed for account {account_id}")
 
+                del batches
+                gc.collect()
+
         # Ensure all database transactions are committed before cleanup
         logger.info(
             f"All thread processing completed, ensuring DB transactions are committed before cleanup"
@@ -3634,6 +3816,12 @@ def _refresh_single_m3u_account_impl(account_id):
                 f"Error running auto channel sync for account {account_id}: {str(e)}"
             )
 
+        try:
+            rollup_channel_catchup_fields(account_id)
+            logger.debug(f"Catch-up field rollup complete for account {account_id}")
+        except Exception as e:
+            logger.error(f"Error rolling up catch-up fields for account {account_id}: {str(e)}")
+
         # Calculate elapsed time
         elapsed_time = time.time() - start_time
 
@@ -3684,6 +3872,9 @@ def _refresh_single_m3u_account_impl(account_id):
             message=account.last_message,
         )
 
+        del auto_sync_result
+        gc.collect()
+
         # Trigger VOD refresh if enabled and account is XtreamCodes type
         if vod_enabled and account.account_type == M3UAccount.Types.XC:
             logger.info(f"VOD is enabled for account {account_id}, triggering VOD refresh")
@@ -3715,6 +3906,10 @@ def _refresh_single_m3u_account_impl(account_id):
             del filtered_groups
         if 'channel_group_relationships' in locals():
             del channel_group_relationships
+        if 'compiled_stream_filters' in locals():
+            del compiled_stream_filters
+
+        gc.collect()
 
         # Remove cache file after processing (success or failure)
         cache_path = os.path.join(m3u_dir, f"{account_id}.json")
@@ -3723,8 +3918,6 @@ def _refresh_single_m3u_account_impl(account_id):
         except OSError:
             pass
 
-        _release_task_db_connection()
-
     return f"Dispatched jobs complete."
 
 

apps/m3u/tests/test_memory_cleanup.py

@@ -30,7 +30,7 @@ class ProcessM3UBatchCleanupTests(SimpleTestCase):
         mock_stream_cls.generate_hash_key = MagicMock(return_value="hash123")
 
         with patch("django.db.connections") as mock_connections:
-            process_m3u_batch_direct(1, [], {}, ["name", "url"])
+            process_m3u_batch_direct(1, [], {}, ["name", "url"], compiled_filters=[])
             mock_connections.close_all.assert_called()
 
     @patch("apps.m3u.tasks.Stream")
@@ -48,9 +48,29 @@ class ProcessM3UBatchCleanupTests(SimpleTestCase):
         mock_stream_cls.generate_hash_key = MagicMock(return_value="hash123")
 
         with patch("gc.collect") as mock_gc, patch("django.db.connections"):
-            process_m3u_batch_direct(1, [], {}, ["name", "url"])
+            process_m3u_batch_direct(1, [], {}, ["name", "url"], compiled_filters=[])
             mock_gc.assert_called()
 
+    @patch("apps.m3u.tasks.Stream")
+    @patch("apps.m3u.tasks.M3UAccount")
+    def test_precompiled_empty_filters_skip_db_lookup(
+        self, mock_account_cls, mock_stream_cls,
+    ):
+        """When filters are precompiled as empty, batch workers must not query filters."""
+        from apps.m3u.tasks import process_m3u_batch_direct
+
+        mock_account = MagicMock()
+        mock_account_cls.objects.get.return_value = mock_account
+        mock_stream_cls.objects.filter.return_value.select_related.return_value.only.return_value = (
+            []
+        )
+        mock_stream_cls.generate_hash_key = MagicMock(return_value="hash123")
+
+        with patch("django.db.connections"):
+            process_m3u_batch_direct(1, [], {}, ["name", "url"], compiled_filters=[])
+
+        mock_account.filters.order_by.assert_not_called()
+
 
 class LockReleaseTests(SimpleTestCase):
     """Verify task lock is released on all exit paths."""

apps/m3u/tests/test_stream_filters.py

@@ -0,0 +1,156 @@
+"""Tests for M3U stream filter compilation and batch application."""
+from unittest.mock import MagicMock, patch
+
+from django.test import SimpleTestCase
+
+from apps.m3u.tasks import (
+    _compile_m3u_stream_filters,
+    _stream_passes_m3u_filters,
+    process_m3u_batch_direct,
+)
+
+
+class CompileM3UStreamFiltersTests(SimpleTestCase):
+    def test_compiles_case_insensitive_when_configured(self):
+        filter_obj = MagicMock()
+        filter_obj.regex_pattern = "news"
+        filter_obj.custom_properties = {"case_sensitive": False}
+
+        compiled = _compile_m3u_stream_filters([filter_obj])
+
+        self.assertEqual(len(compiled), 1)
+        pattern, _ = compiled[0]
+        self.assertTrue(pattern.search("NEWS"))
+
+    def test_compiles_case_sensitive_by_default(self):
+        filter_obj = MagicMock()
+        filter_obj.regex_pattern = "news"
+        filter_obj.custom_properties = {}
+
+        compiled = _compile_m3u_stream_filters([filter_obj])
+
+        pattern, _ = compiled[0]
+        self.assertIsNone(pattern.search("NEWS"))
+        self.assertTrue(pattern.search("news"))
+
+
+class StreamPassesM3UFiltersTests(SimpleTestCase):
+    def _compiled(self, *, filter_type="name", exclude=False, pattern="Adult"):
+        filter_obj = MagicMock()
+        filter_obj.filter_type = filter_type
+        filter_obj.exclude = exclude
+        filter_obj.regex_pattern = pattern
+        filter_obj.custom_properties = {}
+        return _compile_m3u_stream_filters([filter_obj])
+
+    def test_include_filter_passes_matching_stream(self):
+        compiled = self._compiled(exclude=False)
+        self.assertTrue(
+            _stream_passes_m3u_filters("Adult Channel", "http://x", "News", compiled)
+        )
+
+    def test_include_filter_passes_non_matching_stream(self):
+        """Non-matching streams still pass unless a matching exclude filter hits."""
+        compiled = self._compiled(exclude=False, pattern="news")
+        self.assertTrue(
+            _stream_passes_m3u_filters("Sports", "http://x", "Sports", compiled)
+        )
+
+    def test_exclude_filter_rejects_matching_stream(self):
+        compiled = self._compiled(exclude=True, pattern="Adult")
+        self.assertFalse(
+            _stream_passes_m3u_filters("Adult Channel", "http://x", "News", compiled)
+        )
+
+    def test_url_filter_type_targets_url(self):
+        compiled = self._compiled(filter_type="url", exclude=True, pattern="blocked")
+        self.assertFalse(
+            _stream_passes_m3u_filters("OK", "http://blocked.example/live", "News", compiled)
+        )
+        self.assertTrue(
+            _stream_passes_m3u_filters("blocked name", "http://ok.example/live", "News", compiled)
+        )
+
+    def test_group_filter_type_targets_group(self):
+        compiled = self._compiled(filter_type="group", exclude=True, pattern="Hidden")
+        self.assertFalse(
+            _stream_passes_m3u_filters("Channel", "http://x", "Hidden Group", compiled)
+        )
+
+
+class ProcessM3UBatchFilterTests(SimpleTestCase):
+    def _mock_stream_meta(self, mock_stream_cls, max_length=255):
+        mock_field = MagicMock()
+        mock_field.max_length = max_length
+        mock_stream_cls._meta.get_field.return_value = mock_field
+
+    @patch("apps.m3u.tasks._bulk_update_stream_refresh_batches")
+    @patch("apps.m3u.tasks.Stream")
+    @patch("apps.m3u.tasks.M3UAccount")
+    def test_exclude_filter_skips_stream_import(
+        self, mock_account_cls, mock_stream_cls, mock_bulk_update,
+    ):
+        self._mock_stream_meta(mock_stream_cls)
+        mock_account = MagicMock()
+        mock_account.account_type = "STD"
+        mock_account_cls.objects.get.return_value = mock_account
+        mock_stream_cls.objects.filter.return_value.select_related.return_value.only.return_value = (
+            []
+        )
+        mock_stream_cls.generate_hash_key = MagicMock(return_value="hash123")
+
+        filter_obj = MagicMock()
+        filter_obj.regex_pattern = "skip-me"
+        filter_obj.filter_type = "name"
+        filter_obj.exclude = True
+        filter_obj.custom_properties = {}
+        compiled = _compile_m3u_stream_filters([filter_obj])
+
+        batch = [{
+            "name": "skip-me channel",
+            "url": "http://example/live",
+            "attributes": {"group-title": "News"},
+            "vlc_opts": {},
+        }]
+
+        with patch("django.db.connections"):
+            result = process_m3u_batch_direct(
+                1, batch, {"News": 1}, ["name", "url"], compiled_filters=compiled,
+            )
+
+        self.assertIn("0 created", result)
+        mock_stream_cls.objects.bulk_create.assert_not_called()
+        mock_bulk_update.assert_called_once_with([], [], batch_size=200)
+
+    @patch("apps.m3u.tasks._bulk_update_stream_refresh_batches")
+    @patch("apps.m3u.tasks.Stream")
+    @patch("apps.m3u.tasks.M3UAccount")
+    def test_no_filters_imports_matching_stream(
+        self, mock_account_cls, mock_stream_cls, mock_bulk_update,
+    ):
+        self._mock_stream_meta(mock_stream_cls)
+        mock_account = MagicMock()
+        mock_account.account_type = "STD"
+        mock_account_cls.objects.get.return_value = mock_account
+        mock_stream_cls.objects.filter.return_value.select_related.return_value.only.return_value = (
+            []
+        )
+        mock_stream_cls.generate_hash_key = MagicMock(return_value="hash123")
+        mock_stream_cls.objects.bulk_create.return_value = []
+
+        batch = [{
+            "name": "News One",
+            "url": "http://example/live",
+            "attributes": {"group-title": "News"},
+            "vlc_opts": {},
+        }]
+
+        with patch("django.db.connections"), patch(
+            "apps.m3u.tasks.transaction.atomic",
+        ):
+            result = process_m3u_batch_direct(
+                1, batch, {"News": 1}, ["name", "url"], compiled_filters=[],
+            )
+
+        self.assertIn("1 created", result)
+        mock_stream_cls.objects.bulk_create.assert_called_once()

apps/output/epg.py

@@ -1088,7 +1088,7 @@ def generate_dummy_epg(
     return xml_lines
 
 
-def generate_epg(request, profile_name=None, user=None):
+def generate_epg(request, profile_name=None, user=None, *, xc_catchup_prev_days=False):
     """
     Dynamically generate an XMLTV (EPG) file using a streaming response.
     Since the EPG data is stored independently of Channels, we group programmes
@@ -1100,11 +1100,16 @@ def generate_epg(request, profile_name=None, user=None):
         num_days = max(0, min(num_days, 365))
     except (ValueError, TypeError):
         num_days = 0
-    try:
-        prev_days = int(request.GET.get('prev_days', user_custom.get('epg_prev_days', 0)))
-        prev_days = max(0, min(prev_days, 30))
-    except (ValueError, TypeError):
-        prev_days = 0
+    if xc_catchup_prev_days:
+        from apps.channels.utils import resolve_xc_epg_prev_days
+
+        prev_days = resolve_xc_epg_prev_days(request, user)
+    else:
+        try:
+            prev_days = int(request.GET.get('prev_days', user_custom.get('epg_prev_days', 0)))
+            prev_days = max(0, min(prev_days, 30))
+        except (ValueError, TypeError):
+            prev_days = 0
     use_cached_logos = request.GET.get('cachedlogos', 'true').lower() != 'false'
     tvg_id_source = request.GET.get('tvg_id_source', 'channel_number').lower()
     cache_params = (

apps/output/tests.py

@@ -896,3 +896,22 @@ class XcVodSeriesRegressionTests(TestCase):
         stream = xc_get_vod_streams(self.request, self.user)[0]
 
         self.assertEqual(stream["container_extension"], first.container_extension)
+
+
+class GenerateEpgPrevDaysTests(SimpleTestCase):
+    """Profile EPG keeps legacy prev_days=0 unless URL or user setting says otherwise."""
+
+    def setUp(self):
+        self.factory = RequestFactory()
+
+    @patch("apps.output.epg.stream_cached_response")
+    @patch("apps.output.epg.Channel.objects")
+    def test_non_xc_epg_defaults_prev_days_to_zero(self, _channels, mock_cache):
+        from apps.output.epg import generate_epg
+
+        mock_cache.side_effect = lambda cache_key, _source, **_kwargs: cache_key
+        request = self.factory.get("/epg/")
+
+        cache_key = generate_epg(request, profile_name="test", user=None)
+
+        self.assertIn(":p=0:", cache_key)

apps/output/views.py

@@ -11,10 +11,9 @@ from apps.accounts.models import User
 from dispatcharr.utils import network_access_allowed
 from django.utils import timezone as django_timezone
 from django.shortcuts import get_object_or_404
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone as dt_timezone
 import html
 import time
-from tzlocal import get_localzone
 from urllib.parse import urlencode
 import base64
 import logging
@@ -23,6 +22,7 @@ import os
 from apps.m3u.utils import calculate_tuner_count
 from apps.proxy.utils import get_user_active_connections
 import regex
+from core.models import CoreSettings
 from core.utils import log_system_event, build_absolute_uri_with_port
 import hashlib
 from apps.output.epg import generate_epg, generate_dummy_programs
@@ -368,6 +368,24 @@ def _xc_allowed_output_formats(user):
     return ['ts', 'mp4']
 
 
+def _build_xc_server_info(request, hostname, port):
+    """Build XC ``server_info``; keep timezone, ``time_now``, and EPG times in UTC.
+
+    XC clients use ``server_info.timezone`` to interpret EPG start/end strings.
+    Provider-local conversion happens in the timeshift proxy at request time.
+    """
+    # datetime.timezone.utc, not ZoneInfo("UTC"); avoids mis-set Docker /etc/timezone.
+    return {
+        "url": hostname,
+        "server_protocol": request.scheme,
+        "port": port,
+        "timezone": "UTC",
+        "timestamp_now": int(time.time()),
+        "time_now": datetime.now(dt_timezone.utc).strftime("%Y-%m-%d %H:%M:%S"),
+        "process": True,
+    }
+
+
 def xc_get_info(request, full=False):
     user = xc_get_user(request)
 
@@ -400,15 +418,7 @@ def xc_get_info(request, full=False):
             "max_connections": str(max_connections),
             "allowed_output_formats": _xc_allowed_output_formats(user),
         },
-        "server_info": {
-            "url": hostname,
-            "server_protocol": request.scheme,
-            "port": port,
-            "timezone": get_localzone().key,
-            "timestamp_now": int(time.time()),
-            "time_now": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
-            "process": True,
-        },
+        "server_info": _build_xc_server_info(request, hostname, port),
     }
 
     if full == True:
@@ -534,7 +544,7 @@ def xc_xmltv(request):
         )
         return JsonResponse({'error': 'Unauthorized'}, status=401)
 
-    return generate_epg(request, None, user)
+    return generate_epg(request, None, user, xc_catchup_prev_days=True)
 
 
 def xc_get_live_categories(user):
@@ -683,6 +693,14 @@ def _xc_channel_entry(channel, channel_num_map, _get_default_group_id, _logo_url
     effective_logo = channel.effective_logo_obj
     effective_group = channel.effective_channel_group_obj
     group_id = effective_group.id if effective_group else _get_default_group_id()
+
+    if channel.is_catchup:
+        tv_archive = 1
+        tv_archive_duration = channel.catchup_days
+    else:
+        tv_archive = 0
+        tv_archive_duration = 0
+
     return {
         "num": channel_num_int,
         "name": channel.effective_name,
@@ -697,10 +715,10 @@ def _xc_channel_entry(channel, channel_num_map, _get_default_group_id, _logo_url
         "is_adult": int(channel.is_adult),
         "category_id": str(group_id),
         "category_ids": [group_id],
-        "custom_sid": None,
-        "tv_archive": 0,
+        "custom_sid": "",
+        "tv_archive": tv_archive,
         "direct_source": "",
-        "tv_archive_duration": 0,
+        "tv_archive_duration": tv_archive_duration,
     }
 
 
@@ -733,10 +751,12 @@ def xc_get_epg(request, user, short=False):
     if not channel_id:
         raise Http404()
 
+    try:
+        resolved_channel_id = int(channel_id)
+    except (TypeError, ValueError):
+        raise Http404()
+
     channel = None
-    # Apply effective-value annotation + hidden-exclusion at every channel
-    # resolution path so a single channel lookup honors the same visibility
-    # rules as xc_get_live_streams.
     def _annotate(qs):
         return with_effective_values(qs, select_related_fks=True).exclude(hidden_from_output=True)
 
@@ -747,7 +767,7 @@ def xc_get_epg(request, user, short=False):
         if user_profile_count == 0:
             # No profile filtering - user sees all channels based on user_level
             filters = {
-                "id": channel_id,
+                "id": resolved_channel_id,
                 "user_level__lte": user.user_level
             }
             # Hide adult content if user preference is set
@@ -757,7 +777,7 @@ def xc_get_epg(request, user, short=False):
         else:
             # User has specific limited profiles assigned
             filters = {
-                "id": channel_id,
+                "id": resolved_channel_id,
                 "channelprofilemembership__enabled": True,
                 "user_level__lte": user.user_level,
                 "channelprofilemembership__channel_profile__in": user.channel_profiles.all()
@@ -770,7 +790,7 @@ def xc_get_epg(request, user, short=False):
         if not channel:
             raise Http404()
     else:
-        channel = _annotate(Channel.objects.filter(id=channel_id).select_related('epg_data__epg_source')).first()
+        channel = _annotate(Channel.objects.filter(id=resolved_channel_id).select_related('epg_data__epg_source')).first()
         if not channel:
             raise Http404()
 
@@ -818,6 +838,8 @@ def xc_get_epg(request, user, short=False):
         int(channel.effective_channel_number) if channel.effective_channel_number is not None else 0,
     )
 
+    from apps.channels.utils import resolve_xc_epg_prev_days
+
     limit = int(request.GET.get('limit', 4))
     user_custom = user.custom_properties or {}
     try:
@@ -825,12 +847,15 @@ def xc_get_epg(request, user, short=False):
         num_days = max(0, min(num_days, 365))
     except (ValueError, TypeError):
         num_days = 0
-    try:
-        prev_days = int(request.GET.get('prev_days', user_custom.get('epg_prev_days', 0)))
-        prev_days = max(0, min(prev_days, 30))
-    except (ValueError, TypeError):
-        prev_days = 0
+    prev_days = resolve_xc_epg_prev_days(request, user, auto_detect_fallback=False)
     now = django_timezone.now()
+
+    # XC catch-up clients expect past programmes when prev_days was not set.
+    _channel_is_catchup = getattr(channel, "is_catchup", False)
+    _channel_catchup_days = min(getattr(channel, "catchup_days", 0) or 0, 30)
+    if _channel_is_catchup and prev_days == 0:
+        prev_days = _channel_catchup_days
+
     lookback_cutoff = now - timedelta(days=prev_days)
     forward_cutoff = now + timedelta(days=num_days) if num_days > 0 else None
     effective_epg_data = channel.effective_epg_data_obj
@@ -876,6 +901,13 @@ def xc_get_epg(request, user, short=False):
 
     output = {"epg_listings": []}
 
+    if _channel_is_catchup:
+        archive_window = timedelta(days=_channel_catchup_days)
+    else:
+        archive_window = None
+
+    _epg_utc = dt_timezone.utc
+
     for program in programs:
         title = program['title'] if isinstance(program, dict) else program.title
         description = program['description'] if isinstance(program, dict) else program.description
@@ -900,8 +932,8 @@ def xc_get_epg(request, user, short=False):
             "epg_id": epg_id,
             "title": base64.b64encode((title or "").encode()).decode(),
             "lang": "",
-            "start": start.strftime("%Y-%m-%d %H:%M:%S"),
-            "end": end.strftime("%Y-%m-%d %H:%M:%S"),
+            "start": start.astimezone(_epg_utc).strftime("%Y-%m-%d %H:%M:%S"),
+            "end": end.astimezone(_epg_utc).strftime("%Y-%m-%d %H:%M:%S"),
             "description": base64.b64encode((description or "").encode()).decode(),
             "channel_id": str(channel_num_int),
             "start_timestamp": str(int(start.timestamp())),
@@ -909,10 +941,14 @@ def xc_get_epg(request, user, short=False):
             "stream_id": f"{channel_id}",
         }
 
-        if short == False:
-            program_output["now_playing"] = 1 if start <= django_timezone.now() <= end else 0
+        if archive_window is not None and end < now and end > now - archive_window:
+            program_output["has_archive"] = 1
+        else:
             program_output["has_archive"] = 0
 
+        if short == False:
+            program_output["now_playing"] = 1 if start <= now <= end else 0
+
         output['epg_listings'].append(program_output)
 
     return output

apps/proxy/live_proxy/channel_status.py

@@ -407,6 +407,20 @@ 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),
+            ):
+                raw = metadata.get(field)
+                if not raw:
+                    continue
+                try:
+                    info[key] = int(raw)
+                except (TypeError, ValueError):
+                    pass
+
             stream_id_bytes = metadata.get(ChannelMetadataField.STREAM_ID)
             if stream_id_bytes:
                 try:

apps/proxy/live_proxy/constants.py

@@ -54,6 +54,8 @@ class ChannelMetadataField:
     STREAM_ID = "stream_id"
     CHANNEL_NAME = "channel_name"
     STREAM_NAME = "stream_name"
+    IS_TIMESHIFT = "is_timeshift"
+    LOGO_ID = "logo_id"
 
     # Profile fields
     STREAM_PROFILE = "stream_profile"

apps/proxy/live_proxy/server.py

@@ -2079,6 +2079,9 @@ 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:

apps/proxy/utils.py

@@ -1,6 +1,7 @@
 import logging
 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 core.models import CoreSettings
 from apps.proxy.live_proxy.services.channel_service import ChannelService
 
@@ -61,6 +62,15 @@ def attempt_stream_termination(user_id, requesting_client_id, active_connections
                 result = ChannelService.stop_client(t['media_id'], t['client_id'])
                 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']}")
             else:
                 connection_manager = MultiWorkerVODConnectionManager.get_instance()
                 redis_client = connection_manager.redis_client
@@ -106,13 +116,14 @@ def get_user_active_connections(user_id):
 
                 if user_id is None or (client_user_id and int(client_user_id) == user_id):
                     try:
-                        logger.debug(f"[stream limits] Found LIVE connection for user {user_id} on channel {channel_id} with client ID {client_id}")
+                        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': 'live',
+                            'type': conn_type,
                         })
                     except (ValueError, TypeError):
                         pass
@@ -172,6 +183,22 @@ def check_user_stream_limits(user, client_id, media_id=None):
             logger.debug(f"[stream limits][{client_id}] Same-channel reconnect for {media_id} allowed (ignore_same_channel=True)")
             return True
 
+        # Timeshift sibling range/probe requests share one provider slot per
+        # session_id. Each distinct client/session still consumes its own slot.
+        if ignore_same_channel and media_id:
+            media_id_str = str(media_id)
+            for conn in active_connections:
+                if conn.get('type') != 'timeshift':
+                    continue
+                if conn.get('client_id') != client_id:
+                    continue
+                conn_media_id = str(conn.get('media_id') or '')
+                if conn_media_id == media_id_str or conn_media_id.startswith(f"{media_id_str}_"):
+                    logger.debug(
+                        f"[stream limits][{client_id}] Same timeshift session probe for {media_id} allowed (ignore_same_channel=True)"
+                    )
+                    return True
+
         if user_stream_count >= user.stream_limit:
             if user_limit_settings.get("terminate_on_limit_exceeded", True) == False:
                 return False
@@ -186,3 +213,28 @@ def check_user_stream_limits(user, client_id, media_id=None):
                     return False
 
     return True
+
+
+_TS_PACKET_SIZE = 188
+_TS_SYNC_BYTE = 0x47
+
+
+def find_ts_sync(buf):
+    """Return byte offset of the first valid MPEG-TS sync chain in *buf*, or -1.
+
+    Args:
+        buf: Raw bytes from an upstream HTTP response (typically the first 1 KB).
+
+    Returns:
+        Offset of the first 0x47 byte that starts three consecutive 188-byte
+        packets, or -1. Used to strip PHP/HTML preamble before streaming.
+    """
+    end = len(buf) - 2 * _TS_PACKET_SIZE
+    for i in range(0, end):
+        if (
+            buf[i] == _TS_SYNC_BYTE
+            and buf[i + _TS_PACKET_SIZE] == _TS_SYNC_BYTE
+            and buf[i + 2 * _TS_PACKET_SIZE] == _TS_SYNC_BYTE
+        ):
+            return i
+    return -1

apps/timeshift/apps.py

@@ -0,0 +1,7 @@
+from django.apps import AppConfig
+
+
+class TimeshiftConfig(AppConfig):
+    default_auto_field = "django.db.models.BigAutoField"
+    name = "apps.timeshift"
+    verbose_name = "Timeshift"

apps/timeshift/helpers.py

@@ -0,0 +1,252 @@
+"""URL builders and timestamp helpers for XC catch-up."""
+
+import logging
+import re
+from collections import namedtuple
+from datetime import datetime, timezone
+from urllib.parse import quote
+from zoneinfo import ZoneInfo
+
+logger = logging.getLogger(__name__)
+
+# Credentials for the profile whose pool slot was reserved (not raw account fields).
+TimeshiftCredentials = namedtuple(
+    "TimeshiftCredentials", ("server_url", "username", "password")
+)
+
+DEFAULT_DURATION_MINUTES = 120
+DURATION_BUFFER_MINUTES = 5
+MAX_DURATION_MINUTES = 480
+
+# Wall-clock shapes seen from XC / iPlayTV / TiviMate clients. Compiled once.
+_CATCHUP_WALL_CLOCK_RE = re.compile(
+    r"^"
+    r"(?P<date>\d{4}-\d{2}-\d{2})"
+    r"(?P<dtsep>[:_]| )"
+    r"(?P<hour>\d{2})"
+    r"(?P<hmsep>[-:])"
+    r"(?P<minute>\d{2})"
+    r"(?:"
+    r":"
+    r"(?P<second>\d{2})"
+    r")?"
+    r"$"
+)
+
+
+def normalize_catchup_timestamp_input(timestamp_str):
+    """Map a client catch-up timestamp to an ISO-8601 string for ``fromisoformat``.
+
+    Supported inputs:
+        - ``YYYY-MM-DD:HH-MM`` (iPlayTV/TiviMate colon-dash)
+        - ``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)
+        - Unix epoch seconds (10 digits) or milliseconds (13 digits)
+
+    Returns:
+        An ISO-8601 date-time string (``YYYY-MM-DDTHH:MM:SS``), or None if
+        the value does not match a known catch-up shape.
+    """
+    if timestamp_str is None:
+        return None
+    if not isinstance(timestamp_str, str):
+        timestamp_str = str(timestamp_str)
+    value = timestamp_str.strip()
+    if not value:
+        return None
+
+    if value.isdigit():
+        length = len(value)
+        if length == 10:
+            dt = datetime.fromtimestamp(int(value), tz=timezone.utc)
+            return dt.replace(tzinfo=None).isoformat(timespec="seconds")
+        if length == 13:
+            dt = datetime.fromtimestamp(int(value) / 1000, tz=timezone.utc)
+            return dt.replace(tzinfo=None).isoformat(timespec="seconds")
+        return None
+
+    match = _CATCHUP_WALL_CLOCK_RE.match(value)
+    if not match:
+        return None
+
+    parts = match.groupdict()
+    second = parts["second"] or "00"
+    return f"{parts['date']}T{parts['hour']}:{parts['minute']}:{second}"
+
+
+def parse_catchup_timestamp(timestamp_str):
+    """Parse a catch-up timestamp string into a naive UTC wall-clock datetime.
+
+    See ``normalize_catchup_timestamp_input`` for supported input shapes.
+
+    Returns:
+        A naive datetime on success, or None.
+    """
+    iso_value = normalize_catchup_timestamp_input(timestamp_str)
+    if iso_value is None:
+        if timestamp_str is not None and str(timestamp_str).strip():
+            logger.debug(
+                "Timeshift: unrecognised catch-up timestamp: %r", timestamp_str
+            )
+        return None
+    try:
+        return datetime.fromisoformat(iso_value)
+    except ValueError:
+        logger.debug(
+            "Timeshift: invalid catch-up timestamp after normalize: %r -> %r",
+            timestamp_str,
+            iso_value,
+        )
+        return None
+
+
+def _reshape_timestamp(timestamp, strftime_fmt, label):
+    dt = parse_catchup_timestamp(timestamp)
+    if dt is None:
+        logger.error(
+            "Timeshift %s reshape failed for %r: unrecognised format", label, timestamp
+        )
+        return timestamp
+    return dt.strftime(strftime_fmt)
+
+
+def convert_timestamp_to_provider_tz(timestamp_str, provider_tz_name):
+    """Convert a UTC catch-up timestamp to the provider's local zone.
+
+    Args:
+        timestamp_str: UTC wall-clock in ``YYYY-MM-DD:HH-MM`` or underscore form.
+        provider_tz_name: IANA zone from the provider's ``server_info.timezone``
+            (e.g. ``Europe/Brussels``). Falsy, ``UTC``, or unknown: no conversion.
+
+    Returns:
+        ``YYYY-MM-DD:HH-MM`` in the provider zone, or the input unchanged on skip/failure.
+    """
+    if not provider_tz_name or provider_tz_name == "UTC":
+        return timestamp_str
+    dt = parse_catchup_timestamp(timestamp_str)
+    if dt is None:
+        return timestamp_str
+    try:
+        target = ZoneInfo(provider_tz_name)
+    except Exception:
+        logger.warning(
+            "Timeshift: unknown provider timezone %r, no conversion applied",
+            provider_tz_name,
+        )
+        return timestamp_str
+    # timezone.utc, not ZoneInfo("UTC"): avoids mis-set Docker /etc/timezone.
+    local_dt = dt.replace(tzinfo=timezone.utc).astimezone(target)
+    return local_dt.strftime("%Y-%m-%d:%H-%M")
+
+
+def get_programme_duration(channel, timestamp_str):
+    """Look up catch-up duration in minutes from EPG.
+
+    Args:
+        channel: Channel with optional ``epg_data`` relation loaded.
+        timestamp_str: Programme start in UTC (same shape as the client URL).
+
+    Returns:
+        Programme length plus a small buffer, capped at ``MAX_DURATION_MINUTES``,
+        or ``DEFAULT_DURATION_MINUTES`` when EPG lookup fails.
+    """
+    try:
+        dt = parse_catchup_timestamp(timestamp_str)
+        if dt is None:
+            return DEFAULT_DURATION_MINUTES
+        # EPG times are timezone-aware; parsed value must be too.
+        dt = dt.replace(tzinfo=timezone.utc)
+        if not channel.epg_data:
+            return DEFAULT_DURATION_MINUTES
+
+        programme = channel.epg_data.programs.filter(
+            start_time__lte=dt, end_time__gt=dt
+        ).first()
+        if not programme:
+            return DEFAULT_DURATION_MINUTES
+
+        duration_seconds = (programme.end_time - programme.start_time).total_seconds()
+        duration_minutes = int(duration_seconds / 60) + DURATION_BUFFER_MINUTES
+        return min(duration_minutes, MAX_DURATION_MINUTES)
+    except Exception:
+        return DEFAULT_DURATION_MINUTES
+
+
+def build_timeshift_url_format_a(creds, stream_id, timestamp, duration_minutes):
+    """QUERY layout: ``/streaming/timeshift.php?username=...&start=...``"""
+    return (
+        f"{creds.server_url.rstrip('/')}/streaming/timeshift.php"
+        f"?username={quote(str(creds.username), safe='')}"
+        f"&password={quote(str(creds.password), safe='')}"
+        f"&stream={stream_id}"
+        f"&start={timestamp}"
+        f"&duration={duration_minutes}"
+    )
+
+
+def build_timeshift_url_format_b(creds, stream_id, timestamp, duration_minutes):
+    """PATH layout: ``/timeshift/{user}/{pass}/{dur}/{start}/{id}.ts``"""
+    return (
+        f"{creds.server_url.rstrip('/')}/timeshift"
+        f"/{quote(str(creds.username), safe='')}"
+        f"/{quote(str(creds.password), safe='')}"
+        f"/{duration_minutes}"
+        f"/{timestamp}"
+        f"/{stream_id}.ts"
+    )
+
+
+def build_timeshift_candidate_urls(creds, stream_id, timestamp, duration_minutes):
+    """Build ordered upstream URL candidates (PATH forms first, QUERY last).
+
+    Args:
+        creds: ``TimeshiftCredentials`` for the reserved profile.
+        stream_id: Provider stream id from the catch-up stream's custom properties.
+        timestamp: Already converted to the serving provider's local zone.
+        duration_minutes: Archive window length passed to the provider.
+
+    Returns:
+        List of URL strings to try in order. QUERY forms are last because some
+        providers return live TV even when ``start`` is set.
+    """
+    dt = parse_catchup_timestamp(timestamp)
+    if dt is None:
+        colon_dash_ts = timestamp
+        underscore_ts = timestamp
+        colon_seconds_ts = timestamp
+        sql_ts = timestamp
+    else:
+        colon_dash_ts = dt.strftime("%Y-%m-%d:%H-%M")
+        underscore_ts = dt.strftime("%Y-%m-%d_%H-%M")
+        colon_seconds_ts = dt.strftime("%Y-%m-%d:%H:%M:%S")
+        sql_ts = dt.strftime("%Y-%m-%d %H:%M:%S")
+    return [
+        build_timeshift_url_format_b(creds, stream_id, colon_dash_ts, duration_minutes),
+        build_timeshift_url_format_b(creds, stream_id, underscore_ts, duration_minutes),
+        build_timeshift_url_format_b(creds, stream_id, colon_seconds_ts, duration_minutes),
+        build_timeshift_url_format_a(creds, stream_id, underscore_ts, duration_minutes),
+        build_timeshift_url_format_a(creds, stream_id, sql_ts, duration_minutes),
+        build_timeshift_url_format_a(creds, stream_id, colon_dash_ts, duration_minutes),
+        build_timeshift_url_format_a(creds, stream_id, colon_seconds_ts, duration_minutes),
+    ]
+
+
+def format_timestamp_as_colon_dash(timestamp):
+    """Reshape to ``YYYY-MM-DD:HH-MM`` without timezone conversion."""
+    return _reshape_timestamp(timestamp, "%Y-%m-%d:%H-%M", "colon-dash")
+
+
+def format_timestamp_as_colon_seconds(timestamp):
+    """Reshape to ``YYYY-MM-DD:HH:MM:SS`` without timezone conversion."""
+    return _reshape_timestamp(timestamp, "%Y-%m-%d:%H:%M:%S", "colon-seconds")
+
+
+def format_timestamp_as_underscore(timestamp):
+    """Reshape to ``YYYY-MM-DD_HH-MM`` without timezone conversion."""
+    return _reshape_timestamp(timestamp, "%Y-%m-%d_%H-%M", "underscore")
+
+
+def format_timestamp_as_sql_datetime(timestamp):
+    """Reshape to ``YYYY-MM-DD HH:MM:SS`` without timezone conversion."""
+    return _reshape_timestamp(timestamp, "%Y-%m-%d %H:%M:%S", "SQL")

apps/timeshift/tests/test_helpers.py

@@ -0,0 +1,325 @@
+"""Tests for `apps.timeshift.helpers`: timestamp shape conversion and URL build."""
+
+from datetime import datetime, timezone
+
+from django.test import TestCase
+
+from apps.timeshift.helpers import (
+    TimeshiftCredentials,
+    build_timeshift_candidate_urls,
+    build_timeshift_url_format_a,
+    build_timeshift_url_format_b,
+    convert_timestamp_to_provider_tz,
+    format_timestamp_as_colon_dash,
+    format_timestamp_as_colon_seconds,
+    format_timestamp_as_sql_datetime,
+    format_timestamp_as_underscore,
+    normalize_catchup_timestamp_input,
+    parse_catchup_timestamp,
+)
+
+
+def _make_creds():
+    # The builders consume resolved per-profile credentials, never an account
+    # object — get_transformed_credentials() produces these in the view.
+    return TimeshiftCredentials("http://example.test", "user", "pass")
+
+
+class TimestampFormatTests(TestCase):
+    """Timestamp reshape functions change format only; no timezone conversion."""
+
+    def test_normalize_colon_dash_shape(self):
+        self.assertEqual(
+            normalize_catchup_timestamp_input("2026-05-21:12-55"),
+            "2026-05-21T12:55:00",
+        )
+
+    def test_normalize_colon_seconds_xc_format(self):
+        self.assertEqual(
+            normalize_catchup_timestamp_input("2026-06-23:04:00:00"),
+            "2026-06-23T04:00:00",
+        )
+
+    def test_normalize_epg_sql_format(self):
+        self.assertEqual(
+            normalize_catchup_timestamp_input("2026-06-23 04:00:00"),
+            "2026-06-23T04:00:00",
+        )
+
+    def test_normalize_unix_epoch_seconds(self):
+        epoch = str(int(datetime(2026, 6, 23, 4, 0, 0, tzinfo=timezone.utc).timestamp()))
+        self.assertEqual(
+            normalize_catchup_timestamp_input(epoch),
+            "2026-06-23T04:00:00",
+        )
+
+    def test_normalize_unix_epoch_milliseconds(self):
+        epoch_ms = str(
+            int(datetime(2026, 6, 23, 4, 0, 0, tzinfo=timezone.utc).timestamp() * 1000)
+        )
+        self.assertEqual(
+            normalize_catchup_timestamp_input(epoch_ms),
+            "2026-06-23T04:00:00",
+        )
+
+    def test_normalize_rejects_garbage(self):
+        self.assertIsNone(normalize_catchup_timestamp_input("garbage"))
+        self.assertIsNone(normalize_catchup_timestamp_input(""))
+        self.assertIsNone(normalize_catchup_timestamp_input("12345"))
+
+    def test_parse_rejects_invalid_calendar_date(self):
+        self.assertIsNone(parse_catchup_timestamp("2026-13-45:04-00"))
+
+    def test_parse_colon_dash_format(self):
+        dt = parse_catchup_timestamp("2026-05-21:12-55")
+        self.assertEqual(dt, datetime(2026, 5, 21, 12, 55, 0))
+
+    def test_parse_underscore_format(self):
+        dt = parse_catchup_timestamp("2026-05-21_12-55")
+        self.assertEqual(dt, datetime(2026, 5, 21, 12, 55, 0))
+
+    def test_parse_colon_minutes_without_seconds(self):
+        dt = parse_catchup_timestamp("2026-06-23:04:00")
+        self.assertEqual(dt, datetime(2026, 6, 23, 4, 0, 0))
+
+    def test_parse_colon_seconds_xc_format(self):
+        dt = parse_catchup_timestamp("2026-06-23:04:00:00")
+        self.assertEqual(dt, datetime(2026, 6, 23, 4, 0, 0))
+
+    def test_parse_epg_sql_format(self):
+        dt = parse_catchup_timestamp("2026-06-23 04:00:00")
+        self.assertEqual(dt, datetime(2026, 6, 23, 4, 0, 0))
+
+    def test_format_colon_dash_from_colon_seconds(self):
+        self.assertEqual(
+            format_timestamp_as_colon_dash("2026-06-23:04:00:00"),
+            "2026-06-23:04-00",
+        )
+
+    def test_format_colon_seconds_from_colon_dash(self):
+        self.assertEqual(
+            format_timestamp_as_colon_seconds("2026-06-23:04-00"),
+            "2026-06-23:04:00:00",
+        )
+
+    def test_format_colon_seconds_from_unix_epoch(self):
+        epoch = str(int(datetime(2026, 6, 23, 4, 0, 0, tzinfo=timezone.utc).timestamp()))
+        self.assertEqual(
+            format_timestamp_as_colon_dash(epoch),
+            "2026-06-23:04-00",
+        )
+
+    def test_format_sql_reshapes_without_tz_conversion(self):
+        self.assertEqual(
+            format_timestamp_as_sql_datetime("2026-05-12:17-00"),
+            "2026-05-12 17:00:00",
+        )
+
+    def test_format_sql_accepts_underscore_input(self):
+        self.assertEqual(
+            format_timestamp_as_sql_datetime("2026-05-12_17-00"),
+            "2026-05-12 17:00:00",
+        )
+
+    def test_format_sql_invalid_falls_back(self):
+        self.assertEqual(format_timestamp_as_sql_datetime("garbage"), "garbage")
+
+    def test_format_underscore_from_colon_dash(self):
+        self.assertEqual(
+            format_timestamp_as_underscore("2026-05-21:12-55"),
+            "2026-05-21_12-55",
+        )
+
+    def test_format_underscore_idempotent(self):
+        # Underscore input → underscore output (no change)
+        self.assertEqual(
+            format_timestamp_as_underscore("2026-05-21_12-55"),
+            "2026-05-21_12-55",
+        )
+
+    def test_format_underscore_invalid_falls_back(self):
+        self.assertEqual(format_timestamp_as_underscore("garbage"), "garbage")
+
+
+class BuildTimeshiftUrlTests(TestCase):
+    def setUp(self):
+        self.creds = _make_creds()
+
+    def test_format_a_passes_dash_shape_unchanged(self):
+        url = build_timeshift_url_format_a(
+            self.creds, "22372", "2026-05-12:19-00", 40
+        )
+        self.assertIn("start=2026-05-12:19-00", url)
+        self.assertIn("stream=22372", url)
+        self.assertIn("duration=40", url)
+
+    def test_format_a_passes_sql_shape_unchanged(self):
+        url = build_timeshift_url_format_a(
+            self.creds, "22372", "2026-05-12 19:00:00", 40
+        )
+        self.assertIn("start=2026-05-12 19:00:00", url)
+
+    def test_format_b_path_with_dash_shape(self):
+        url = build_timeshift_url_format_b(
+            self.creds, "22372", "2026-05-12:19-00", 40
+        )
+        self.assertIn("/40/2026-05-12:19-00/22372.ts", url)
+
+
+class CandidateOrderingTests(TestCase):
+    """`build_timeshift_candidate_urls` must try the PATH form (which seeks the
+    archive) before the QUERY form (which returns LIVE on some providers,
+    silently ignoring the requested timestamp). Regression guard for the
+    "catch-up plays the live stream instead of the requested programme" bug."""
+
+    def setUp(self):
+        self.creds = _make_creds()
+
+    def _is_path_form(self, url):
+        return "/timeshift/" in url and url.endswith(".ts") and "timeshift.php" not in url
+
+    def _is_query_form(self, url):
+        return "timeshift.php?" in url
+
+    def test_every_path_candidate_precedes_every_query_candidate(self):
+        urls = build_timeshift_candidate_urls(self.creds, "22372", "2026-05-12:19-00", 40)
+        path_indices = [i for i, u in enumerate(urls) if self._is_path_form(u)]
+        query_indices = [i for i, u in enumerate(urls) if self._is_query_form(u)]
+        # Each URL is classified as exactly one form.
+        self.assertEqual(len(path_indices) + len(query_indices), len(urls))
+        self.assertTrue(path_indices and query_indices)
+        # The last PATH candidate still comes before the first QUERY candidate.
+        self.assertLess(max(path_indices), min(query_indices))
+
+    def test_first_candidate_is_path_form_with_canonical_dash_timestamp(self):
+        urls = build_timeshift_candidate_urls(self.creds, "22372", "2026-05-12:19-00", 40)
+        self.assertTrue(self._is_path_form(urls[0]))
+        # Canonical colon-dash timestamp, passed through unchanged.
+        self.assertIn("/40/2026-05-12:19-00/22372.ts", urls[0])
+
+    def test_accepts_colon_seconds_input_timestamp(self):
+        urls = build_timeshift_candidate_urls(
+            self.creds, "22372", "2026-06-23:04:00:00", 40
+        )
+        self.assertTrue(self._is_path_form(urls[0]))
+        self.assertIn("/40/2026-06-23:04-00/22372.ts", urls[0])
+        self.assertIn("/40/2026-06-23:04:00:00/22372.ts", urls[2])
+
+    def test_accepts_underscore_input_timestamp(self):
+        # Client may send the underscore shape; PATH form still leads.
+        urls = build_timeshift_candidate_urls(self.creds, "22372", "2026-05-12_19-00", 40)
+        self.assertTrue(self._is_path_form(urls[0]))
+
+
+class ConvertTimestampToProviderTzTests(TestCase):
+    """`convert_timestamp_to_provider_tz` shifts a UTC catch-up timestamp into the
+    serving provider's local zone (XC providers index archives in their own zone),
+    DST-correct, and is a no-op when the zone is UTC/unknown/missing."""
+
+    def test_utc_to_brussels_summer_is_plus_two(self):
+        # June → CEST (+02:00): 17:00 UTC == 19:00 Brussels (the 19h JT case).
+        self.assertEqual(
+            convert_timestamp_to_provider_tz("2026-06-08:17-00", "Europe/Brussels"),
+            "2026-06-08:19-00",
+        )
+
+    def test_utc_to_brussels_winter_is_plus_one(self):
+        # January → CET (+01:00): 17:00 UTC == 18:00 Brussels (DST handled).
+        self.assertEqual(
+            convert_timestamp_to_provider_tz("2026-01-08:17-00", "Europe/Brussels"),
+            "2026-01-08:18-00",
+        )
+
+    def test_day_rollover(self):
+        # 23:30 UTC + 2h (CEST) crosses midnight into the next day.
+        self.assertEqual(
+            convert_timestamp_to_provider_tz("2026-06-08:23-30", "Europe/Brussels"),
+            "2026-06-09:01-30",
+        )
+
+    def test_underscore_input_returns_colon_dash(self):
+        self.assertEqual(
+            convert_timestamp_to_provider_tz("2026-06-08_17-00", "Europe/Brussels"),
+            "2026-06-08:19-00",
+        )
+
+    def test_utc_zone_is_noop(self):
+        self.assertEqual(
+            convert_timestamp_to_provider_tz("2026-06-08:17-00", "UTC"),
+            "2026-06-08:17-00",
+        )
+
+    def test_none_zone_is_noop(self):
+        self.assertEqual(
+            convert_timestamp_to_provider_tz("2026-06-08:17-00", None),
+            "2026-06-08:17-00",
+        )
+
+    def test_unknown_zone_is_noop(self):
+        self.assertEqual(
+            convert_timestamp_to_provider_tz("2026-06-08:17-00", "Mars/Phobos"),
+            "2026-06-08:17-00",
+        )
+
+    def test_utc_to_brussels_from_unix_epoch(self):
+        epoch = str(int(datetime(2026, 6, 8, 17, 0, 0, tzinfo=timezone.utc).timestamp()))
+        self.assertEqual(
+            convert_timestamp_to_provider_tz(epoch, "Europe/Brussels"),
+            "2026-06-08:19-00",
+        )
+
+    def test_garbage_timestamp_passthrough(self):
+        self.assertEqual(
+            convert_timestamp_to_provider_tz("garbage", "Europe/Brussels"),
+            "garbage",
+        )
+
+
+class GetProgrammeDurationTests(TestCase):
+    """Duration window resolution: programme length + buffer, capped, with a
+    safe default whenever the EPG lookup cannot resolve."""
+
+    def _channel_with_programme(self, minutes):
+        from datetime import datetime, timedelta, timezone as dt_timezone
+        from unittest.mock import MagicMock
+
+        start = datetime(2026, 6, 8, 17, 0, tzinfo=dt_timezone.utc)
+        programme = MagicMock(
+            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_duration_is_programme_length_plus_buffer(self):
+        from apps.timeshift.helpers import get_programme_duration
+        # 40-minute programme + 5-minute buffer.
+        self.assertEqual(
+            get_programme_duration(self._channel_with_programme(40), "2026-06-08:17-00"),
+            45,
+        )
+
+    def test_duration_capped_at_max(self):
+        from apps.timeshift.helpers import get_programme_duration
+        self.assertEqual(
+            get_programme_duration(self._channel_with_programme(1000), "2026-06-08:17-00"),
+            480,
+        )
+
+    def test_no_epg_data_falls_back_to_default(self):
+        from unittest.mock import MagicMock
+        from apps.timeshift.helpers import get_programme_duration
+        channel = MagicMock(epg_data=None)
+        self.assertEqual(get_programme_duration(channel, "2026-06-08:17-00"), 120)
+
+    def test_no_matching_programme_falls_back_to_default(self):
+        from unittest.mock import MagicMock
+        from apps.timeshift.helpers import get_programme_duration
+        channel = MagicMock()
+        channel.epg_data.programs.filter.return_value.first.return_value = None
+        self.assertEqual(get_programme_duration(channel, "2026-06-08:17-00"), 120)
+
+    def test_garbage_timestamp_falls_back_to_default(self):
+        from unittest.mock import MagicMock
+        from apps.timeshift.helpers import get_programme_duration
+        self.assertEqual(get_programme_duration(MagicMock(), "garbage"), 120)

core/models.py

@@ -280,8 +280,18 @@ class CoreSettings(models.Model):
             "epg_match_ignore_prefixes": [],
             "epg_match_ignore_suffixes": [],
             "epg_match_ignore_custom": [],
+            # XC catch-up: forced XMLTV lookback (0 = auto-detect, capped at 30).
+            "xmltv_prev_days_override": 0,
         })
 
+    @classmethod
+    def get_xmltv_prev_days_override(cls):
+        """Global XC XMLTV prev_days default (0 = auto-detect from provider archives)."""
+        try:
+            return int(cls.get_epg_settings().get("xmltv_prev_days_override", 0) or 0)
+        except (TypeError, ValueError):
+            return 0
+
     @classmethod
     def _safe_string_list(cls, value):
         """Return a list of strings, filtering out non-list or non-string values."""

core/utils.py

@@ -565,13 +565,15 @@ def trim_c_allocator_heap():
         return False
 
 
-def cleanup_memory(log_usage=False, force_collection=True):
+def cleanup_memory(log_usage=False, force_collection=True, trim_heap=False):
     """
     Comprehensive memory cleanup function to reduce memory footprint
 
     Args:
         log_usage: Whether to log memory usage before and after cleanup
         force_collection: Whether to force garbage collection
+        trim_heap: Return freed C heap pages to the OS. Only use after DB
+            connections are closed (e.g. Celery task_postrun).
     """
     logger.trace("Starting memory cleanup django memory cleanup")
     # Skip logging if log level is not set to debug or more verbose (like trace)
@@ -606,6 +608,8 @@ def cleanup_memory(log_usage=False, force_collection=True):
             logger.debug(f"Memory after cleanup: {after_mem:.2f} MB (change: {after_mem-before_mem:.2f} MB)")
         except (ImportError, Exception):
             pass
+    if trim_heap:
+        trim_c_allocator_heap()
     logger.trace("Memory cleanup complete for django")
 
 
@@ -618,8 +622,7 @@ def spawn_memory_trim(close_connections=False):
     so the pooled DB connection is released first.
     """
     def _run():
-        cleanup_memory(force_collection=True)
-        trim_c_allocator_heap()
+        cleanup_memory(force_collection=True, trim_heap=True)
 
     if close_connections:
         from django.db import close_old_connections

dispatcharr/celery.py

@@ -99,6 +99,7 @@ def cleanup_task_memory(**kwargs):
     memory_intensive_tasks = [
         'apps.m3u.tasks.refresh_single_m3u_account',
         'apps.m3u.tasks.refresh_m3u_accounts',
+        'apps.m3u.tasks.refresh_m3u_groups',
         'apps.m3u.tasks.process_m3u_batch',
         'apps.m3u.tasks.process_xc_category',
         'apps.m3u.tasks.sync_auto_channels',
@@ -121,7 +122,7 @@ def cleanup_task_memory(**kwargs):
         from core.utils import cleanup_memory
 
         # Use the comprehensive cleanup function
-        cleanup_memory(log_usage=True, force_collection=True)
+        cleanup_memory(log_usage=True, force_collection=True, trim_heap=True)
 
         # Log memory usage if psutil is installed
         try:

dispatcharr/settings.py

@@ -99,6 +99,7 @@ INSTALLED_APPS = [
     "django_filters",
     "django_celery_beat",
     "apps.plugins",
+    "apps.timeshift.apps.TimeshiftConfig",
 ]
 
 # EPG Processing optimization settings

dispatcharr/urls.py

@@ -7,6 +7,7 @@ from .routing import websocket_urlpatterns
 from apps.output.views import xc_player_api, xc_panel_api, xc_get, xc_xmltv
 from apps.proxy.live_proxy.views import stream_xc
 from apps.proxy.vod_proxy.views import stream_xc_movie, stream_xc_episode
+from apps.timeshift.views import timeshift_proxy
 
 urlpatterns = [
     # API Routes
@@ -41,6 +42,11 @@ urlpatterns = [
         stream_xc,
         name="xc_stream_endpoint",
     ),
+    path(
+        "timeshift/<str:username>/<str:password>/<str:stream_id>/<str:timestamp>/<str:duration>",
+        timeshift_proxy,
+        name="timeshift_proxy",
+    ),
     # XC VOD endpoints
     path(
         "movie/<str:username>/<str:password>/<str:stream_id>.<str:extension>",

frontend/src/components/cards/StreamConnectionCard.jsx

@@ -22,6 +22,7 @@ import {
   Gauge,
   HardDriveDownload,
   HardDriveUpload,
+  Rewind,
   SquareX,
   Timer,
   Users,
@@ -486,7 +487,10 @@ const StreamConnectionCard = ({
   }, [channel.name, channel.stream_id]);
 
   const channelName =
-    channel.name || previewedStream?.name || 'Unnamed Channel';
+    channel.name ||
+    channel.channel_name ||
+    previewedStream?.name ||
+    'Unnamed Channel';
   const uptime = channel.uptime || 0;
   const bitrates = channel.bitrates || [];
   const totalBytes = channel.total_bytes || 0;
@@ -661,6 +665,18 @@ 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">

frontend/src/components/forms/settings/EpgSettingsForm.jsx

@@ -0,0 +1,81 @@
+import useSettingsStore from '../../../store/settings.jsx';
+import React, { useEffect, useState } from 'react';
+import { useForm } from '@mantine/form';
+import {
+  Alert,
+  Button,
+  Flex,
+  NumberInput,
+  Stack,
+  Text,
+} from '@mantine/core';
+import { EPG_SETTINGS_OPTIONS } from '../../../constants.js';
+import {
+  getChangedSettings,
+  parseSettings,
+  saveChangedSettings,
+} from '../../../utils/pages/SettingsUtils.js';
+import { getEpgSettingsFormInitialValues } from '../../../utils/forms/settings/EpgSettingsFormUtils.js';
+
+const EpgSettingsForm = React.memo(({ active }) => {
+  const settings = useSettingsStore((s) => s.settings);
+  const [saved, setSaved] = useState(false);
+
+  const form = useForm({
+    mode: 'controlled',
+    initialValues: getEpgSettingsFormInitialValues(),
+  });
+
+  useEffect(() => {
+    if (!active) setSaved(false);
+  }, [active]);
+
+  useEffect(() => {
+    if (settings) {
+      const parsed = parseSettings(settings);
+      form.setFieldValue(
+        'xmltv_prev_days_override',
+        parsed.xmltv_prev_days_override ?? 0,
+      );
+    }
+  }, [settings]);
+
+  const onSubmit = async () => {
+    setSaved(false);
+    const changedSettings = getChangedSettings(form.getValues(), settings);
+    try {
+      await saveChangedSettings(settings, changedSettings);
+      setSaved(true);
+    } catch (error) {
+      console.error('Error saving EPG settings:', error);
+    }
+  };
+
+  const prevDaysConfig = EPG_SETTINGS_OPTIONS.xmltv_prev_days_override;
+
+  return (
+    <form onSubmit={form.onSubmit(onSubmit)}>
+      <Stack gap="md">
+        {saved && (
+          <Alert variant="light" color="green" title="Saved Successfully" />
+        )}
+        <NumberInput
+          label={prevDaysConfig.label}
+          description={prevDaysConfig.description}
+          min={0}
+          max={30}
+          {...form.getInputProps('xmltv_prev_days_override')}
+        />
+        <Text size="xs" c="dimmed">
+          Per-user defaults and URL parameters still override this global value.
+          EPG channel matching options are configured from the Channels page.
+        </Text>
+        <Flex justify="flex-end">
+          <Button type="submit">Save</Button>
+        </Flex>
+      </Stack>
+    </form>
+  );
+});
+
+export default EpgSettingsForm;

frontend/src/constants.js

@@ -63,6 +63,14 @@ export const PROXY_SETTINGS_OPTIONS = {
   },
 };
 
+export const EPG_SETTINGS_OPTIONS = {
+  xmltv_prev_days_override: {
+    label: 'XMLTV prev_days Override (catch-up)',
+    description:
+      'Days of past programmes in the XC EPG output. 0 = auto-detect from the providers’ tv_archive_duration (capped at 30).',
+  },
+};
+
 export const USER_LIMITS_OPTIONS = {
   terminate_on_limit_exceeded: {
     label: 'Terminate on Limit Exceeded',

frontend/src/pages/Settings.jsx

@@ -45,6 +45,9 @@ const DvrSettingsForm = React.lazy(
 const SystemSettingsForm = React.lazy(
   () => import('../components/forms/settings/SystemSettingsForm.jsx')
 );
+const EpgSettingsForm = React.lazy(
+  () => import('../components/forms/settings/EpgSettingsForm.jsx')
+);
 const NavOrderForm = React.lazy(
   () => import('../components/forms/settings/NavOrderForm.jsx')
 );
@@ -122,6 +125,19 @@ const SettingsPage = () => {
                 </AccordionPanel>
               </AccordionItem>
 
+              <AccordionItem value="epg-settings">
+                <AccordionControl>EPG</AccordionControl>
+                <AccordionPanel>
+                  <ErrorBoundary>
+                    <Suspense fallback={<Loader />}>
+                      <EpgSettingsForm
+                        active={accordianValue === 'epg-settings'}
+                      />
+                    </Suspense>
+                  </ErrorBoundary>
+                </AccordionPanel>
+              </AccordionItem>
+
               <AccordionItem value="system-settings">
                 <AccordionControl>System Settings</AccordionControl>
                 <AccordionPanel>

frontend/src/utils/forms/settings/EpgSettingsFormUtils.js

@@ -0,0 +1,3 @@
+export const getEpgSettingsFormInitialValues = () => ({
+  xmltv_prev_days_override: 0,
+});

frontend/src/utils/pages/SettingsUtils.js

@@ -39,6 +39,7 @@ export const saveChangedSettings = async (settings, changedSettings) => {
     'epg_match_ignore_prefixes',
     'epg_match_ignore_suffixes',
     'epg_match_ignore_custom',
+    'xmltv_prev_days_override',
   ];
   const dvrFields = [
     'tv_template',
@@ -125,6 +126,7 @@ export const saveChangedSettings = async (settings, changedSettings) => {
       'retention_count',
       'schedule_day_of_week',
       'max_system_events',
+      'xmltv_prev_days_override',
     ];
     if (numericFields.includes(formKey) && value != null) {
       value = typeof value === 'number' ? value : parseInt(value, 10);
@@ -222,6 +224,20 @@ export const getChangedSettings = (values, settings) => {
       continue;
     }
 
+    if (settingKey === 'xmltv_prev_days_override') {
+      const baseline = Number(
+        settings['epg_settings']?.value?.xmltv_prev_days_override ?? 0,
+      );
+      const nextVal =
+        typeof actualValue === 'number'
+          ? actualValue
+          : parseInt(actualValue, 10) || 0;
+      if (nextVal !== baseline) {
+        changedSettings[settingKey] = nextVal;
+      }
+      continue;
+    }
+
     // Convert array values (like m3u_hash_key) to comma-separated strings for comparison
     if (Array.isArray(actualValue)) {
       actualValue = actualValue.join(',');
@@ -296,6 +312,12 @@ export const parseSettings = (settings) => {
     epgSettings && Array.isArray(epgSettings.epg_match_ignore_custom)
       ? epgSettings.epg_match_ignore_custom
       : [];
+  parsed.xmltv_prev_days_override =
+    epgSettings && epgSettings.xmltv_prev_days_override != null
+      ? typeof epgSettings.xmltv_prev_days_override === 'number'
+        ? epgSettings.xmltv_prev_days_override
+        : parseInt(epgSettings.xmltv_prev_days_override, 10) || 0
+      : 0;
 
   // DVR settings - direct mapping with underscore keys
   const dvrSettings = settings['dvr_settings']?.value;