Merge pull request #1145 from Dispatcharr/user-priority-fixes

User stream limits

CHANGELOG.md

@@ -19,6 +19,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- User stream limits: administrators can now set a maximum number of concurrent streams per user account. When a user reaches their limit, the system can automatically terminate an existing stream to free a slot based on configurable rules. Limit enforcement applies to both live channels and VOD. (Closes #544)
+  - Each user account has a new **Stream Limit** field (0 = unlimited) configurable from the user edit form in Settings → Users.
+  - Global enforcement behaviour is configurable in Settings → User Limits:
+    - **Terminate on Limit Exceeded**: automatically stop an existing stream when the user's limit is reached (vs. rejecting the new connection).
+    - **Terminate Oldest**: prefer terminating the oldest stream when freeing a slot; disable to prefer the newest.
+    - **Prioritize Single-Client Channels**: prefer terminating streams on channels that only this user is watching.
+    - **Ignore Same-Channel Connections**: count multiple connections to the same live channel as one stream toward the limit. Same-channel reconnects are always allowed through. When this is enabled and a channel must be freed, all connections to the chosen channel are terminated together so that the unique-channel count actually decreases. VOD is explicitly excluded from this bypass since VOD connections are not shared upstream.
 - TLS and mutual TLS (mTLS) support for Redis and PostgreSQL connections in modular deployments. Supports encrypted connections, server certificate verification (Redis: on/off; PostgreSQL: verify-full, verify-ca, require), CA certificate configuration, and client certificate authentication. Configured via environment variables in the docker compose file. Includes startup validation for certificate paths and TLS/URL scheme conflicts, and a read-only Connection Security panel in System Settings. (Closes #950) — Thanks [@CodeBormen](https://github.com/CodeBormen)
 - Status filter for M3U group and VOD category filter modals: A new **All / Enabled / Disabled** segmented control is now shown alongside the text search input in the Live, VOD - Movies, and VOD - Series tabs of the M3U Group Filter modal. The status filter works in combination with the text search and also scopes the "Select Visible" / "Deselect Visible" buttons so they only act on the currently visible subset. (Closes #312)
 

apps/accounts/api_views.py

@@ -287,11 +287,10 @@ class UserViewSet(viewsets.ModelViewSet):
         if request.method == "PATCH":
             ALLOWED_FIELDS = {"custom_properties", "first_name", "last_name", "email", "password"}
             disallowed = set(request.data.keys()) - ALLOWED_FIELDS
-            if disallowed:
-                return Response(
-                    {"detail": f"Fields not allowed for self-update: {', '.join(disallowed)}"},
-                    status=400,
-                )
+
+            for key in disallowed:
+                request.data.pop(key, None)
+
             serializer = UserSerializer(user, data=request.data, partial=True)
             serializer.is_valid(raise_exception=True)
             serializer.save()

apps/accounts/migrations/0006_user_stream_limit.py

@@ -0,0 +1,18 @@
+# Generated by Django 5.2.11 on 2026-03-19 13:46
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('accounts', '0005_alter_user_managers'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='user',
+            name='stream_limit',
+            field=models.IntegerField(default=0),
+        ),
+    ]

apps/accounts/models.py

@@ -30,6 +30,7 @@ class User(AbstractUser):
     user_level = models.IntegerField(default=UserLevel.STREAMER)
     custom_properties = models.JSONField(default=dict, blank=True, null=True)
     api_key = models.CharField(max_length=200, blank=True, null=True, db_index=True)
+    stream_limit = models.IntegerField(default=0)
 
     def __str__(self):
         return self.username

apps/accounts/serializers.py

@@ -65,6 +65,7 @@ class UserSerializer(serializers.ModelSerializer):
             "channel_profiles",
             "custom_properties",
             "avatar_config",
+            "stream_limit",
             "is_staff",
             "is_superuser",
             "last_login",

apps/channels/api_views.py

@@ -2674,7 +2674,49 @@ class RecordingViewSet(viewsets.ModelViewSet):
         recording_id = instance.pk
         channel_name = instance.channel.name
 
-        # Capture state before the DB row is deleted
+        # Attempt to close the DVR client connection for this channel if active
+        try:
+            channel_uuid = str(instance.channel.uuid)
+            # Lazy imports to avoid module overhead if proxy isn't used
+            from core.utils import RedisClient
+            from apps.proxy.ts_proxy.redis_keys import RedisKeys
+            from apps.proxy.ts_proxy.services.channel_service import ChannelService
+
+            r = RedisClient.get_client()
+            if r:
+                client_set_key = RedisKeys.clients(channel_uuid)
+                client_ids = r.smembers(client_set_key) or []
+                stopped = 0
+                for cid in client_ids:
+                    try:
+                        meta_key = RedisKeys.client_metadata(channel_uuid, cid)
+                        ua = r.hget(meta_key, "user_agent")
+                        # Identify DVR recording client by its user agent
+                        if ua and "Dispatcharr-DVR" in ua:
+                            try:
+                                ChannelService.stop_client(channel_uuid, cid)
+                                stopped += 1
+                            except Exception as inner_e:
+                                logger.debug(f"Failed to stop DVR client {cid} for channel {channel_uuid}: {inner_e}")
+                    except Exception as inner:
+                        logger.debug(f"Error while checking client metadata: {inner}")
+                if stopped:
+                    logger.info(f"Stopped {stopped} DVR client(s) for channel {channel_uuid} due to recording cancellation")
+                # If no clients remain after stopping DVR clients, proactively stop the channel
+                try:
+                    remaining = r.scard(client_set_key) or 0
+                except Exception:
+                    remaining = 0
+                if remaining == 0:
+                    try:
+                        ChannelService.stop_channel(channel_uuid)
+                        logger.info(f"Stopped channel {channel_uuid} (no clients remain)")
+                    except Exception as sc_e:
+                        logger.debug(f"Unable to stop channel {channel_uuid}: {sc_e}")
+        except Exception as e:
+            logger.debug(f"Unable to stop DVR clients for cancelled recording: {e}")
+
+        # Capture paths before deletion
         cp = instance.custom_properties or {}
         rec_status = cp.get("status", "")
         file_path = cp.get("file_path")

apps/channels/models.py

@@ -204,7 +204,7 @@ class Stream(models.Model):
 
         return stream_profile
 
-    def get_stream(self):
+    def get_stream(self, requester=None):
         """
         Finds an available profile for this stream and reserves a connection slot.
 
@@ -400,6 +400,144 @@ class Channel(models.Model):
 
         return stream_profile
 
+    def _pick_channel_to_preempt(
+        self,
+        profile_id,
+        requester_level,
+        redis_client,
+        exclude_channel_ids=None,
+        cooldown_seconds=30,
+    ):
+        """
+        Pick the lowest-impact channel to terminate on the given profile.
+        Returns: Optional[int] channel_id to preempt
+        """
+        exclude_channel_ids = set(exclude_channel_ids or [])
+        candidates = []
+
+        # 1) Try to get active channel IDs for this profile from an index set if available
+        ch_set_key = f"ts_proxy:profile:{profile_id}:channels"
+        try:
+            ch_ids = { (int(x) if not isinstance(x, int) else x) for x in (redis_client.smembers(ch_set_key) or set()) }
+        except Exception:
+            ch_ids = set()
+
+        logger.debug("Candidate channels for preemption:")
+        logger.debug(ch_ids)
+
+        # 2) Fallback: scan metadata keys and filter by m3u_profile == profile_id
+        if not ch_ids:
+            cursor = 0
+            pattern = "ts_proxy:channel:*:metadata"
+            while True:
+                cursor, keys = redis_client.scan(cursor=cursor, match=pattern, count=500)
+                if keys:
+                    # Prefer HGET m3u_profile if metadata is a hash
+                    pipe = redis_client.pipeline()
+                    for k in keys:
+                        pipe.hget(k, "m3u_profile")
+                    prof_vals = pipe.execute()
+                    for k, prof_val in zip(keys, prof_vals):
+                        try:
+                            pid = int(prof_val) if prof_val is not None else None
+                        except Exception:
+                            pid = None
+
+                        if pid == profile_id:
+                            parts = k.split(":")  # ts_proxy:channel:{id}:metadata
+                            if len(parts) >= 4:
+                                try:
+                                    ch_ids.add(int(parts[2]))
+                                except Exception:
+                                    pass
+                if cursor == 0:
+                    break
+
+        logger.debug("Candidate channels for preemption:")
+        logger.debug(ch_ids)
+
+        if not ch_ids:
+            return None
+
+        # 3) Score candidates
+        for ch_id in ch_ids:
+            if ch_id in exclude_channel_ids:
+                continue
+
+            # Skip if recently preempted
+            last_preempt_key = f"ts_proxy:channel:{ch_id}:last_preempt"
+            try:
+                last_preempt = float(redis_client.get(last_preempt_key) or 0.0)
+            except Exception:
+                last_preempt = 0.0
+            if last_preempt and (time.time() - last_preempt) < cooldown_seconds:
+                continue
+
+            # Clients and their levels
+            clients_key = f"ts_proxy:channel:{ch_id}:clients"
+            member_ids = list(redis_client.smembers(clients_key) or [])
+            viewer_count = len(member_ids)
+            max_viewer_level = 0
+            if viewer_count:
+                pipe = redis_client.pipeline()
+                for cid in member_ids:
+                    pipe.hget(f"ts_proxy:channel:{ch_id}:clients:{cid}", "user_level")
+                levels_raw = pipe.execute()
+                levels = []
+                for lv in levels_raw:
+                    try:
+                        levels.append(int(lv or 0))
+                    except Exception:
+                        levels.append(0)
+                max_viewer_level = max(levels or [0])
+
+            # Only preempt if requester strictly outranks this channel's viewers
+            if requester_level <= max_viewer_level:
+                continue
+
+            # Metadata (protected/recording/started_at_ts)
+            meta_key = f"ts_proxy:channel:{ch_id}:metadata"
+            try:
+                protected, recording, started_at_ts = redis_client.hmget(
+                    meta_key, "protected", "recording", "started_at_ts"
+                )
+            except Exception:
+                protected = recording = started_at_ts = None
+
+            protected = str(protected or "0") in ("1", "true", "True")
+            recording = str(recording or "0") in ("1", "true", "True")
+            if protected or recording:
+                continue
+
+            try:
+                started_at_ts = float(started_at_ts) if started_at_ts is not None else None
+            except Exception:
+                started_at_ts = None
+            if started_at_ts is None:
+                started_at_ts = time.time()  # treat unknown as newest
+
+            # Score: lower is safer to terminate
+            has_viewers = 1 if viewer_count > 0 else 0
+            score = (has_viewers, max_viewer_level, viewer_count, started_at_ts)
+            candidates.append((score, ch_id))
+
+        logger.debug("Candidate channels after scoring:")
+        logger.debug(candidates)
+
+        if not candidates:
+            return None
+
+        candidates.sort(key=lambda x: x[0])
+        victim_id = candidates[0][1]
+
+        # Mark preempt timestamp to avoid thrashing
+        try:
+            redis_client.set(f"ts_proxy:channel:{victim_id}:last_preempt", str(time.time()), ex=3600)
+        except Exception:
+            pass
+
+        return victim_id
+
     def _check_and_reserve_profile_slot(self, profile, redis_client):
         """
         Atomically check and reserve a connection slot for the given profile.
@@ -432,7 +570,7 @@ class Channel(models.Model):
         redis_client.decr(profile_connections_key)
         return (False, new_count - 1)
 
-    def get_stream(self):
+    def get_stream(self, requester=None):
         """
         Finds an available stream for the requested channel and returns the selected stream and profile.
 
@@ -513,6 +651,17 @@ class Channel(models.Model):
                         None,
                     )  # Return newly assigned stream and matched profile
                 else:
+                    # At capacity: try to preempt a lower-impact channel on this profile
+                    victim_channel_id = self._pick_channel_to_preempt(
+                        profile_id=profile.id,
+                        requester_level=requester.user_level if requester else 100,
+                        redis_client=redis_client,
+                        exclude_channel_ids=None,
+                    )
+                    if victim_channel_id:
+                        logger.info(f"Preempting channel {victim_channel_id} for new stream on profile {profile.id}")
+                        # return self.id, profile.id, victim_channel_id
+
                     # This profile is at max connections
                     has_streams_but_maxed_out = True
                     logger.debug(

apps/channels/tasks.py

@@ -2493,15 +2493,12 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str):
                 metadata_key = RedisKeys.channel_metadata(str(channel.uuid))
                 md = r.hgetall(metadata_key)
                 if md:
-                    def _gv(bkey):
-                        return md.get(bkey.encode('utf-8'))
-
                     def _d(bkey, cast=str):
-                        v = _gv(bkey)
+                        v = md.get(bkey)
                         try:
                             if v is None:
                                 return None
-                            s = v.decode('utf-8')
+                            s = v
                             return cast(s) if cast is not str else s
                         except Exception:
                             return None

apps/proxy/tasks.py

@@ -31,7 +31,7 @@ def fetch_channel_stats():
         while True:
             cursor, keys = redis_client.scan(cursor, match=channel_pattern)
             for key in keys:
-                channel_id_match = re.search(r"ts_proxy:channel:(.*):metadata", key.decode('utf-8'))
+                channel_id_match = re.search(r"ts_proxy:channel:(.*):metadata", key)
                 if channel_id_match:
                     ch_id = channel_id_match.group(1)
                     channel_info = ChannelStatus.get_basic_channel_info(ch_id)

apps/proxy/ts_proxy/channel_status.py

@@ -39,19 +39,19 @@ class ChannelStatus:
 
         info = {
             'channel_id': channel_id,
-            'state': metadata.get(ChannelMetadataField.STATE.encode('utf-8'), b'unknown').decode('utf-8'),
-            'url': metadata.get(ChannelMetadataField.URL.encode('utf-8'), b'').decode('utf-8'),
-            'stream_profile': metadata.get(ChannelMetadataField.STREAM_PROFILE.encode('utf-8'), b'').decode('utf-8'),
-            'started_at': metadata.get(ChannelMetadataField.INIT_TIME.encode('utf-8'), b'0').decode('utf-8'),
-            'owner': metadata.get(ChannelMetadataField.OWNER.encode('utf-8'), b'unknown').decode('utf-8'),
-            'buffer_index': int(buffer_index_value.decode('utf-8')) if buffer_index_value else 0,
+            'state': metadata.get(ChannelMetadataField.STATE, 'unknown'),
+            'url': metadata.get(ChannelMetadataField.URL, ''),
+            'stream_profile': metadata.get(ChannelMetadataField.STREAM_PROFILE, ''),
+            'started_at': metadata.get(ChannelMetadataField.INIT_TIME, '0'),
+            'owner': metadata.get(ChannelMetadataField.OWNER, 'unknown'),
+            'buffer_index': int(buffer_index_value) if buffer_index_value else 0,
         }
 
         # Add stream ID and name information
-        stream_id_bytes = metadata.get(ChannelMetadataField.STREAM_ID.encode('utf-8'))
+        stream_id_bytes = metadata.get(ChannelMetadataField.STREAM_ID)
         if stream_id_bytes:
             try:
-                stream_id = int(stream_id_bytes.decode('utf-8'))
+                stream_id = int(stream_id_bytes)
                 info['stream_id'] = stream_id
 
                 # Look up stream name from database
@@ -66,10 +66,10 @@ class ChannelStatus:
                 logger.warning(f"Invalid stream_id format in Redis: {stream_id_bytes}")
 
         # Add M3U profile information
-        m3u_profile_id_bytes = metadata.get(ChannelMetadataField.M3U_PROFILE.encode('utf-8'))
+        m3u_profile_id_bytes = metadata.get(ChannelMetadataField.M3U_PROFILE)
         if m3u_profile_id_bytes:
             try:
-                m3u_profile_id = int(m3u_profile_id_bytes.decode('utf-8'))
+                m3u_profile_id = int(m3u_profile_id_bytes)
                 info['m3u_profile_id'] = m3u_profile_id
 
                 # Look up M3U profile name from database
@@ -84,22 +84,22 @@ class ChannelStatus:
                 logger.warning(f"Invalid m3u_profile_id format in Redis: {m3u_profile_id_bytes}")
 
         # Add timing information
-        state_changed_field = ChannelMetadataField.STATE_CHANGED_AT.encode('utf-8')
+        state_changed_field = ChannelMetadataField.STATE_CHANGED_AT
         if state_changed_field in metadata:
-            state_changed_at = float(metadata[state_changed_field].decode('utf-8'))
+            state_changed_at = float(metadata[state_changed_field])
             info['state_changed_at'] = state_changed_at
             info['state_duration'] = time.time() - state_changed_at
 
-        init_time_field = ChannelMetadataField.INIT_TIME.encode('utf-8')
+        init_time_field = ChannelMetadataField.INIT_TIME
         if init_time_field in metadata:
-            created_at = float(metadata[init_time_field].decode('utf-8'))
+            created_at = float(metadata[init_time_field])
             info['started_at'] = created_at
             info['uptime'] = time.time() - created_at
 
         # Add data throughput information
-        total_bytes_field = ChannelMetadataField.TOTAL_BYTES.encode('utf-8')
+        total_bytes_field = ChannelMetadataField.TOTAL_BYTES
         if total_bytes_field in metadata:
-            total_bytes = int(metadata[total_bytes_field].decode('utf-8'))
+            total_bytes = int(metadata[total_bytes_field])
             info['total_bytes'] = total_bytes
 
             # Format total bytes in human-readable form
@@ -130,7 +130,7 @@ class ChannelStatus:
 
         stale_client_ids = []
         for client_id in client_ids:
-            client_id_str = client_id.decode('utf-8')
+            client_id_str = client_id
             client_key = RedisKeys.client_metadata(channel_id, client_id_str)
             client_data = proxy_server.redis_client.hgetall(client_key)
 
@@ -141,33 +141,33 @@ class ChannelStatus:
 
             client_info = {
                 'client_id': client_id_str,
-                'user_agent': client_data.get(b'user_agent', b'unknown').decode('utf-8'),
-                'worker_id': client_data.get(b'worker_id', b'unknown').decode('utf-8'),
+                'user_agent': client_data.get('user_agent', 'unknown'),
+                'worker_id': client_data.get('worker_id', 'unknown'),
             }
 
-            if b'connected_at' in client_data:
-                connected_at = float(client_data[b'connected_at'].decode('utf-8'))
+            if 'connected_at' in client_data:
+                connected_at = float(client_data['connected_at'])
                 client_info['connected_at'] = connected_at
                 client_info['connection_duration'] = time.time() - connected_at
 
-            if b'last_active' in client_data:
-                last_active = float(client_data[b'last_active'].decode('utf-8'))
+            if 'last_active' in client_data:
+                last_active = float(client_data['last_active'])
                 client_info['last_active'] = last_active
                 client_info['last_active_ago'] = time.time() - last_active
 
             # Add transfer rate statistics
-            if b'bytes_sent' in client_data:
-                client_info['bytes_sent'] = int(client_data[b'bytes_sent'].decode('utf-8'))
+            if 'bytes_sent' in client_data:
+                client_info['bytes_sent'] = int(client_data['bytes_sent'])
 
             # Add average transfer rate
-            if b'avg_rate_KBps' in client_data:
-                client_info['avg_rate_KBps'] = float(client_data[b'avg_rate_KBps'].decode('utf-8'))
-            elif b'transfer_rate_KBps' in client_data:  # For backward compatibility
-                client_info['avg_rate_KBps'] = float(client_data[b'transfer_rate_KBps'].decode('utf-8'))
+            if 'avg_rate_KBps' in client_data:
+                client_info['avg_rate_KBps'] = float(client_data['avg_rate_KBps'])
+            elif 'transfer_rate_KBps' in client_data:  # For backward compatibility
+                client_info['avg_rate_KBps'] = float(client_data['transfer_rate_KBps'])
 
             # Add current transfer rate
-            if b'current_rate_KBps' in client_data:
-                client_info['current_rate_KBps'] = float(client_data[b'current_rate_KBps'].decode('utf-8'))
+            if 'current_rate_KBps' in client_data:
+                client_info['current_rate_KBps'] = float(client_data['current_rate_KBps'])
 
             clients.append(client_info)
 
@@ -249,7 +249,7 @@ class ChannelStatus:
                     while True:
                         cursor, keys = proxy_server.redis_client.scan(cursor, match=buffer_key_pattern, count=100)
                         if keys:
-                            all_buffer_keys.extend([k.decode('utf-8') for k in keys])
+                            all_buffer_keys.extend([k for k in keys])
                         if cursor == 0 or len(all_buffer_keys) >= 20:  # Limit to 20 keys
                             break
 
@@ -279,61 +279,64 @@ class ChannelStatus:
             }
 
         # Add FFmpeg stream information
-        video_codec = metadata.get(ChannelMetadataField.VIDEO_CODEC.encode('utf-8'))
+        video_codec = metadata.get(ChannelMetadataField.VIDEO_CODEC)
         if video_codec:
-            info['video_codec'] = video_codec.decode('utf-8')
+            info['video_codec'] = video_codec
 
-        resolution = metadata.get(ChannelMetadataField.RESOLUTION.encode('utf-8'))
+        resolution = metadata.get(ChannelMetadataField.RESOLUTION)
         if resolution:
-            info['resolution'] = resolution.decode('utf-8')
+            info['resolution'] = resolution
 
-        source_fps = metadata.get(ChannelMetadataField.SOURCE_FPS.encode('utf-8'))
+        source_fps = metadata.get(ChannelMetadataField.SOURCE_FPS)
         if source_fps:
-            info['source_fps'] = float(source_fps.decode('utf-8'))
+            info['source_fps'] = source_fps
 
-        pixel_format = metadata.get(ChannelMetadataField.PIXEL_FORMAT.encode('utf-8'))
+        pixel_format = metadata.get(ChannelMetadataField.PIXEL_FORMAT)
         if pixel_format:
-            info['pixel_format'] = pixel_format.decode('utf-8')
+            info['pixel_format'] = pixel_format
 
-        source_bitrate = metadata.get(ChannelMetadataField.SOURCE_BITRATE.encode('utf-8'))
+        source_bitrate = metadata.get(ChannelMetadataField.SOURCE_BITRATE)
         if source_bitrate:
-            info['source_bitrate'] = float(source_bitrate.decode('utf-8'))
+            info['source_bitrate'] = source_bitrate
 
-        audio_codec = metadata.get(ChannelMetadataField.AUDIO_CODEC.encode('utf-8'))
+        audio_codec = metadata.get(ChannelMetadataField.AUDIO_CODEC)
         if audio_codec:
-            info['audio_codec'] = audio_codec.decode('utf-8')
+            info['audio_codec'] = audio_codec
 
-        sample_rate = metadata.get(ChannelMetadataField.SAMPLE_RATE.encode('utf-8'))
+        sample_rate = metadata.get(ChannelMetadataField.SAMPLE_RATE)
         if sample_rate:
-            info['sample_rate'] = int(sample_rate.decode('utf-8'))
+            info['sample_rate'] = sample_rate
 
-        audio_channels = metadata.get(ChannelMetadataField.AUDIO_CHANNELS.encode('utf-8'))
+        audio_channels = metadata.get(ChannelMetadataField.AUDIO_CHANNELS)
         if audio_channels:
-            info['audio_channels'] = audio_channels.decode('utf-8')
+            info['audio_channels'] = audio_channels
 
-        audio_bitrate = metadata.get(ChannelMetadataField.AUDIO_BITRATE.encode('utf-8'))
+        audio_bitrate = metadata.get(ChannelMetadataField.AUDIO_BITRATE)
         if audio_bitrate:
-            info['audio_bitrate'] = float(audio_bitrate.decode('utf-8'))
+            info['audio_bitrate'] = audio_bitrate
+
 
         # Add FFmpeg performance stats
-        ffmpeg_speed = metadata.get(ChannelMetadataField.FFMPEG_SPEED.encode('utf-8'))
+        ffmpeg_speed = metadata.get(ChannelMetadataField.FFMPEG_SPEED)
         if ffmpeg_speed:
-            info['ffmpeg_speed'] = float(ffmpeg_speed.decode('utf-8'))
+            info['ffmpeg_speed'] = ffmpeg_speed
 
-        ffmpeg_fps = metadata.get(ChannelMetadataField.FFMPEG_FPS.encode('utf-8'))
+        ffmpeg_fps = metadata.get(ChannelMetadataField.FFMPEG_FPS)
         if ffmpeg_fps:
-            info['ffmpeg_fps'] = float(ffmpeg_fps.decode('utf-8'))
+            info['ffmpeg_fps'] = ffmpeg_fps
 
-        actual_fps = metadata.get(ChannelMetadataField.ACTUAL_FPS.encode('utf-8'))
+        actual_fps = metadata.get(ChannelMetadataField.ACTUAL_FPS)
         if actual_fps:
-            info['actual_fps'] = float(actual_fps.decode('utf-8'))
+            info['actual_fps'] = actual_fps
 
-        ffmpeg_bitrate = metadata.get(ChannelMetadataField.FFMPEG_BITRATE.encode('utf-8'))
+        ffmpeg_bitrate = metadata.get(ChannelMetadataField.FFMPEG_BITRATE)
         if ffmpeg_bitrate:
-            info['ffmpeg_bitrate'] = float(ffmpeg_bitrate.decode('utf-8'))
-        stream_type = metadata.get(ChannelMetadataField.STREAM_TYPE.encode('utf-8'))
+            info['ffmpeg_bitrate'] = ffmpeg_bitrate
+
+        stream_type = metadata.get(ChannelMetadataField.STREAM_TYPE)
         if stream_type:
-            info['stream_type'] = stream_type.decode('utf-8')
+            info['stream_type'] = stream_type
+
 
         return info
 
@@ -378,33 +381,27 @@ class ChannelStatus:
             client_count = proxy_server.redis_client.scard(client_set_key) or 0
 
             # Calculate uptime
-            init_time_bytes = metadata.get(ChannelMetadataField.INIT_TIME.encode('utf-8'), b'0')
-            created_at = float(init_time_bytes.decode('utf-8'))
+            init_time_bytes = metadata.get(ChannelMetadataField.INIT_TIME, '0')
+            created_at = float(init_time_bytes)
             uptime = time.time() - created_at if created_at > 0 else 0
 
-            # Safely decode bytes or use defaults
-            def safe_decode(bytes_value, default="unknown"):
-                if bytes_value is None:
-                    return default
-                return bytes_value.decode('utf-8')
-
             # Simplified info
             info = {
                 'channel_id': channel_id,
-                'state': safe_decode(metadata.get(ChannelMetadataField.STATE.encode('utf-8'))),
-                'url': safe_decode(metadata.get(ChannelMetadataField.URL.encode('utf-8')), ""),
-                'stream_profile': safe_decode(metadata.get(ChannelMetadataField.STREAM_PROFILE.encode('utf-8')), ""),
-                'owner': safe_decode(metadata.get(ChannelMetadataField.OWNER.encode('utf-8'))),
-                'buffer_index': int(buffer_index_value.decode('utf-8')) if buffer_index_value else 0,
+                'state': metadata.get(ChannelMetadataField.STATE),
+                'url': metadata.get(ChannelMetadataField.URL, ""),
+                'stream_profile': metadata.get(ChannelMetadataField.STREAM_PROFILE, ""),
+                'owner': metadata.get(ChannelMetadataField.OWNER),
+                'buffer_index': int(buffer_index_value) if buffer_index_value else 0,
                 'client_count': client_count,
                 'uptime': uptime
             }
 
             # Add stream ID and name information
-            stream_id_bytes = metadata.get(ChannelMetadataField.STREAM_ID.encode('utf-8'))
+            stream_id_bytes = metadata.get(ChannelMetadataField.STREAM_ID)
             if stream_id_bytes:
                 try:
-                    stream_id = int(stream_id_bytes.decode('utf-8'))
+                    stream_id = int(stream_id_bytes)
                     info['stream_id'] = stream_id
 
                     # Look up stream name from database
@@ -419,9 +416,9 @@ class ChannelStatus:
                     logger.warning(f"Invalid stream_id format in Redis: {stream_id_bytes}")
 
             # Add data throughput information to basic info
-            total_bytes_bytes = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.TOTAL_BYTES.encode('utf-8'))
+            total_bytes_bytes = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.TOTAL_BYTES)
             if total_bytes_bytes:
-                total_bytes = int(total_bytes_bytes.decode('utf-8'))
+                total_bytes = int(total_bytes_bytes)
                 info['total_bytes'] = total_bytes
 
                 # Calculate and add bitrate
@@ -458,23 +455,22 @@ class ChannelStatus:
                     if client_id in stale_client_ids:
                         continue
 
-                    client_id_str = client_id.decode('utf-8')
-                    client_key = RedisKeys.client_metadata(channel_id, client_id_str)
+                    client_key = RedisKeys.client_metadata(channel_id, client_id)
 
                     client_info = {
-                        'client_id': client_id_str,
+                        'client_id': client_id,
                     }
 
                     user_agent_bytes = proxy_server.redis_client.hget(client_key, 'user_agent')
-                    client_info['user_agent'] = safe_decode(user_agent_bytes)
+                    client_info['user_agent'] = user_agent_bytes
 
                     ip_address_bytes = proxy_server.redis_client.hget(client_key, 'ip_address')
                     if ip_address_bytes:
-                        client_info['ip_address'] = safe_decode(ip_address_bytes)
+                        client_info['ip_address'] = ip_address_bytes
 
                     connected_at_bytes = proxy_server.redis_client.hget(client_key, 'connected_at')
                     if connected_at_bytes:
-                        connected_at = float(connected_at_bytes.decode('utf-8'))
+                        connected_at = float(connected_at_bytes)
                         client_info['connected_since'] = time.time() - connected_at
 
                     clients.append(client_info)
@@ -484,10 +480,10 @@ class ChannelStatus:
             info['client_count'] = client_count
 
             # Add M3U profile information
-            m3u_profile_id_bytes = metadata.get(ChannelMetadataField.M3U_PROFILE.encode('utf-8'))
-            if m3u_profile_id_bytes:
+            m3u_profile_id = metadata.get(ChannelMetadataField.M3U_PROFILE)
+            if m3u_profile_id:
                 try:
-                    m3u_profile_id = int(m3u_profile_id_bytes.decode('utf-8'))
+                    m3u_profile_id = int(m3u_profile_id)
                     info['m3u_profile_id'] = m3u_profile_id
 
                     # Look up M3U profile name from database
@@ -499,32 +495,36 @@ class ChannelStatus:
                     except (ImportError, DatabaseError) as e:
                         logger.warning(f"Failed to get M3U profile name for ID {m3u_profile_id}: {e}")
                 except ValueError:
-                    logger.warning(f"Invalid m3u_profile_id format in Redis: {m3u_profile_id_bytes}")
+                    logger.warning(f"Invalid m3u_profile_id format in Redis: {m3u_profile_id}")
 
             # Add stream info to basic info as well
-            video_codec = metadata.get(ChannelMetadataField.VIDEO_CODEC.encode('utf-8'))
+            video_codec = metadata.get(ChannelMetadataField.VIDEO_CODEC)
             if video_codec:
-                info['video_codec'] = video_codec.decode('utf-8')
+                info['video_codec'] = video_codec
 
-            resolution = metadata.get(ChannelMetadataField.RESOLUTION.encode('utf-8'))
+            resolution = metadata.get(ChannelMetadataField.RESOLUTION)
             if resolution:
-                info['resolution'] = resolution.decode('utf-8')
+                info['resolution'] = resolution
 
-            source_fps = metadata.get(ChannelMetadataField.SOURCE_FPS.encode('utf-8'))
+            source_fps = metadata.get(ChannelMetadataField.SOURCE_FPS)
             if source_fps:
-                info['source_fps'] = float(source_fps.decode('utf-8'))
-            ffmpeg_speed = metadata.get(ChannelMetadataField.FFMPEG_SPEED.encode('utf-8'))
+                info['source_fps'] = float(source_fps)
+
+            ffmpeg_speed = metadata.get(ChannelMetadataField.FFMPEG_SPEED)
             if ffmpeg_speed:
-                info['ffmpeg_speed'] = float(ffmpeg_speed.decode('utf-8'))
-            audio_codec = metadata.get(ChannelMetadataField.AUDIO_CODEC.encode('utf-8'))
+                info['ffmpeg_speed'] = float(ffmpeg_speed)
+
+            audio_codec = metadata.get(ChannelMetadataField.AUDIO_CODEC)
             if audio_codec:
-                info['audio_codec'] = audio_codec.decode('utf-8')
-            audio_channels = metadata.get(ChannelMetadataField.AUDIO_CHANNELS.encode('utf-8'))
+                info['audio_codec'] = audio_codec
+
+            audio_channels = metadata.get(ChannelMetadataField.AUDIO_CHANNELS)
             if audio_channels:
-                info['audio_channels'] = audio_channels.decode('utf-8')
-            stream_type = metadata.get(ChannelMetadataField.STREAM_TYPE.encode('utf-8'))
+                info['audio_channels'] = audio_channels
+
+            stream_type = metadata.get(ChannelMetadataField.STREAM_TYPE)
             if stream_type:
-                info['stream_type'] = stream_type.decode('utf-8')
+                info['stream_type'] = stream_type
 
             return info
         except Exception as e:

apps/proxy/ts_proxy/client_manager.py

@@ -1,10 +1,8 @@
 """Client connection management for TS streams"""
 
 import threading
-import logging
 import time
 import json
-import gevent
 from typing import Set, Optional
 from apps.proxy.config import TSConfig as Config
 from redis.exceptions import ConnectionError, TimeoutError
@@ -130,7 +128,7 @@ class ClientManager:
                             # Check for stale activity using last_active field
                             last_active = self.redis_client.hget(client_key, "last_active")
                             if last_active:
-                                last_active_time = float(last_active.decode('utf-8'))
+                                last_active_time = float(last_active)
                                 ghost_timeout = self.heartbeat_interval * getattr(Config, 'GHOST_CLIENT_MULTIPLIER', 5.0)
 
                                 if current_time - last_active_time > ghost_timeout:
@@ -230,7 +228,7 @@ class ClientManager:
         except Exception as e:
             logger.error(f"Error notifying owner of client activity: {e}")
 
-    def add_client(self, client_id, client_ip, user_agent=None):
+    def add_client(self, client_id, client_ip, user_agent=None, user=None):
         """Add a client with duplicate prevention"""
         if client_id in self._registered_clients:
             logger.debug(f"Client {client_id} already registered, skipping")
@@ -248,7 +246,9 @@ class ClientManager:
             "ip_address": client_ip,
             "connected_at": current_time,
             "last_active": current_time,
-            "worker_id": self.worker_id or "unknown"
+            "worker_id": self.worker_id or "unknown",
+            "user_id": str(user.id) if user is not None else "0",
+            # "user_level": user.user_level if user is not None else 100, # default to a high value since no user means the non-user specific M3U/HDHR
         }
 
         try:
@@ -309,8 +309,6 @@ class ClientManager:
 
     def remove_client(self, client_id):
         """Remove a client from this channel and Redis"""
-        client_ip = None
-
         with self.lock:
             if client_id in self.clients:
                 self.clients.remove(client_id)
@@ -323,11 +321,6 @@ class ClientManager:
             if self.redis_client:
                 # Get client IP before removing the data
                 client_key = f"ts_proxy:channel:{self.channel_id}:clients:{client_id}"
-                client_data = self.redis_client.hgetall(client_key)
-                if client_data and b'ip_address' in client_data:
-                    client_ip = client_data[b'ip_address'].decode('utf-8')
-                elif client_data and 'ip_address' in client_data:
-                    client_ip = client_data['ip_address']
 
                 # Remove from channel's client set
                 self.redis_client.srem(self.client_set_key, client_id)
@@ -435,8 +428,7 @@ class ClientManager:
         client_id_list = list(client_ids)
         pipe = redis_client.pipeline()
         for cid in client_id_list:
-            cid_str = cid.decode('utf-8')
-            pipe.exists(RedisKeys.client_metadata(channel_id, cid_str))
+            pipe.exists(RedisKeys.client_metadata(channel_id, cid))
         results = pipe.execute()
 
         stale_ids = [

apps/proxy/ts_proxy/server.py

@@ -177,6 +177,7 @@ class ProxyServer:
                             socket_connect_timeout=10,
                             socket_keepalive=True,
                             health_check_interval=30,
+                            decode_responses=True,
                             **ssl_params
                         )
                         logger.info("Created fallback Redis PubSub client for event listener")
@@ -198,8 +199,8 @@ class ProxyServer:
                             continue
 
                         try:
-                            channel = message["channel"].decode("utf-8")
-                            data = json.loads(message["data"].decode("utf-8"))
+                            channel = message["channel"]
+                            data = json.loads(message["data"])
 
                             event_type = data.get("event")
                             channel_id = data.get("channel_id")
@@ -375,7 +376,7 @@ class ProxyServer:
             if result is None:
                 return None
             try:
-                return result.decode('utf-8')
+                return result
             except (AttributeError, UnicodeDecodeError) as e:
                 logger.error(f"Error decoding channel owner for {channel_id}: {e}, raw={result!r}")
                 return None
@@ -414,7 +415,7 @@ class ProxyServer:
             current_owner = self._execute_redis_command(
                 lambda: self.redis_client.get(lock_key)
             )
-            if current_owner and current_owner.decode('utf-8') == self.worker_id:
+            if current_owner and current_owner == self.worker_id:
                 # Refresh TTL
                 self._execute_redis_command(
                     lambda: self.redis_client.expire(lock_key, ttl)
@@ -439,7 +440,7 @@ class ProxyServer:
 
             # Only delete if we're the current owner to prevent race conditions
             current = self.redis_client.get(lock_key)
-            if current and current.decode('utf-8') == self.worker_id:
+            if current and current == self.worker_id:
                 self.redis_client.delete(lock_key)
                 logger.info(f"Released ownership of channel {channel_id}")
 
@@ -473,7 +474,7 @@ class ProxyServer:
                         return False
                 return False
 
-            if current.decode('utf-8') == self.worker_id:
+            if current == self.worker_id:
                 self.redis_client.expire(lock_key, ttl)
                 return True
 
@@ -490,15 +491,15 @@ class ProxyServer:
                 metadata_key = RedisKeys.channel_metadata(channel_id)
                 if self.redis_client.exists(metadata_key):
                     metadata = self.redis_client.hgetall(metadata_key)
-                    if b'state' in metadata:
-                        state = metadata[b'state'].decode('utf-8')
+                    if 'state' in metadata:
+                        state = metadata['state']
                         active_states = [ChannelState.INITIALIZING, ChannelState.CONNECTING,
                                         ChannelState.WAITING_FOR_CLIENTS, ChannelState.ACTIVE, ChannelState.BUFFERING]
                         if state in active_states:
                             logger.info(f"Channel {channel_id} already being initialized with state {state}")
                             # Create buffer and client manager only if we don't have them
                             if channel_id not in self.stream_buffers:
-                                self.stream_buffers[channel_id] = StreamBuffer(channel_id, redis_client=self.redis_client)
+                                self.stream_buffers[channel_id] = StreamBuffer(channel_id, redis_client=RedisClient.get_buffer())
                             if channel_id not in self.client_managers:
                                 self.client_managers[channel_id] = ClientManager(
                                     channel_id,
@@ -509,7 +510,7 @@ class ProxyServer:
 
             # Create buffer and client manager instances (or reuse if they exist)
             if channel_id not in self.stream_buffers:
-                buffer = StreamBuffer(channel_id, redis_client=self.redis_client)
+                buffer = StreamBuffer(channel_id, redis_client=RedisClient.get_buffer())
                 self.stream_buffers[channel_id] = buffer
 
             if channel_id not in self.client_managers:
@@ -548,18 +549,18 @@ class ProxyServer:
 
                 # If no url was passed, try to get from Redis
                 if not url and existing_metadata:
-                    url_bytes = existing_metadata.get(b'url')
+                    url_bytes = existing_metadata.get('url')
                     if url_bytes:
-                        channel_url = url_bytes.decode('utf-8')
+                        channel_url = url_bytes
 
-                    ua_bytes = existing_metadata.get(b'user_agent')
+                    ua_bytes = existing_metadata.get('user_agent')
                     if ua_bytes:
-                        channel_user_agent = ua_bytes.decode('utf-8')
+                        channel_user_agent = ua_bytes
 
                 # Get stream ID from metadata if not provided
-                if not channel_stream_id and b'stream_id' in existing_metadata:
+                if not channel_stream_id and 'stream_id' in existing_metadata:
                     try:
-                        channel_stream_id = int(existing_metadata[b'stream_id'].decode('utf-8'))
+                        channel_stream_id = int(existing_metadata['stream_id'])
                         logger.debug(f"Found stream_id {channel_stream_id} in metadata for channel {channel_id}")
                     except (ValueError, TypeError) as e:
                         logger.debug(f"Could not parse stream_id from metadata: {e}")
@@ -574,7 +575,7 @@ class ProxyServer:
 
                 # Create buffer but not stream manager (only if not already exists)
                 if channel_id not in self.stream_buffers:
-                    buffer = StreamBuffer(channel_id=channel_id, redis_client=self.redis_client)
+                    buffer = StreamBuffer(channel_id=channel_id, redis_client=RedisClient.get_buffer())
                     self.stream_buffers[channel_id] = buffer
 
                 # Create client manager with channel_id and redis_client (only if not already exists)
@@ -597,7 +598,7 @@ class ProxyServer:
 
                 # Create buffer but not stream manager (only if not already exists)
                 if channel_id not in self.stream_buffers:
-                    buffer = StreamBuffer(channel_id=channel_id, redis_client=self.redis_client)
+                    buffer = StreamBuffer(channel_id=channel_id, redis_client=RedisClient.get_buffer())
                     self.stream_buffers[channel_id] = buffer
 
                 # Create client manager with channel_id and redis_client (only if not already exists)
@@ -636,12 +637,12 @@ class ProxyServer:
                 # Verify the stream_id was set correctly in Redis
                 stream_id_value = self.redis_client.hget(metadata_key, "stream_id")
                 if stream_id_value:
-                    logger.info(f"Verified stream_id {stream_id_value.decode('utf-8')} is set in Redis for channel {channel_id}")
+                    logger.info(f"Verified stream_id {stream_id_value} is set in Redis for channel {channel_id}")
                 else:
                     logger.warning(f"Failed to set stream_id in Redis for channel {channel_id}")
 
             # Create stream buffer
-            buffer = StreamBuffer(channel_id=channel_id, redis_client=self.redis_client)
+            buffer = StreamBuffer(channel_id=channel_id, redis_client=RedisClient.get_buffer())
             logger.debug(f"Created StreamBuffer for channel {channel_id}")
             self.stream_buffers[channel_id] = buffer
 
@@ -735,8 +736,8 @@ class ProxyServer:
                 metadata = self.redis_client.hgetall(metadata_key)
 
                 # Get channel state and owner
-                state = metadata.get(b'state', b'unknown').decode('utf-8')
-                owner = metadata.get(b'owner', b'').decode('utf-8')
+                state = metadata.get('state', 'unknown')
+                owner = metadata.get('owner', '')
 
                 # States that indicate the channel is running properly or shutting down
                 valid_states = [ChannelState.ACTIVE, ChannelState.WAITING_FOR_CLIENTS,
@@ -774,8 +775,8 @@ class ProxyServer:
                     return False
                 else:
                     # Unknown or initializing state, check how long it's been in this state
-                    if b'state_changed_at' in metadata:
-                        state_changed_at = float(metadata[b'state_changed_at'].decode('utf-8'))
+                    if 'state_changed_at' in metadata:
+                        state_changed_at = float(metadata['state_changed_at'])
                         state_age = time.time() - state_changed_at
 
                         # If in initializing state for too long, consider it stale
@@ -813,8 +814,8 @@ class ProxyServer:
 
             # If we have metadata, log details for debugging
             if metadata:
-                state = metadata.get(b'state', b'unknown').decode('utf-8')
-                owner = metadata.get(b'owner', b'unknown').decode('utf-8')
+                state = metadata.get('state', 'unknown')
+                owner = metadata.get('owner', 'unknown')
                 logger.info(f"Zombie channel details - state: {state}, owner: {owner}")
 
             # Clean up Redis keys
@@ -939,16 +940,16 @@ class ProxyServer:
                         metadata = self.redis_client.hgetall(metadata_key)
                         if metadata:
                             # Calculate runtime from init_time
-                            if b'init_time' in metadata:
+                            if 'init_time' in metadata:
                                 try:
-                                    init_time = float(metadata[b'init_time'].decode('utf-8'))
+                                    init_time = float(metadata['init_time'])
                                     runtime = round(time.time() - init_time, 2)
                                 except Exception:
                                     pass
                             # Get total bytes transferred
-                            if b'total_bytes' in metadata:
+                            if 'total_bytes' in metadata:
                                 try:
-                                    total_bytes = int(metadata[b'total_bytes'].decode('utf-8'))
+                                    total_bytes = int(metadata['total_bytes'])
                                 except Exception:
                                     pass
 
@@ -1059,8 +1060,8 @@ class ProxyServer:
                             if self.redis_client:
                                 metadata_key = RedisKeys.channel_metadata(channel_id)
                                 metadata = self.redis_client.hgetall(metadata_key)
-                                if metadata and b'state' in metadata:
-                                    channel_state = metadata[b'state'].decode('utf-8')
+                                if metadata and 'state' in metadata:
+                                    channel_state = metadata['state']
 
                             # Check if channel has any clients left
                             total_clients = 0
@@ -1092,9 +1093,9 @@ class ProxyServer:
 
                                 # Get connection_ready_time from metadata (indicates if channel reached ready state)
                                 connection_ready_time = None
-                                if metadata and b'connection_ready_time' in metadata:
+                                if metadata and 'connection_ready_time' in metadata:
                                     try:
-                                        connection_ready_time = float(metadata[b'connection_ready_time'].decode('utf-8'))
+                                        connection_ready_time = float(metadata['connection_ready_time'])
                                     except (ValueError, TypeError):
                                         pass
 
@@ -1106,15 +1107,15 @@ class ProxyServer:
                                         attempt_value = self.redis_client.get(attempt_key)
                                         if attempt_value:
                                             try:
-                                                connection_attempt_time = float(attempt_value.decode('utf-8'))
+                                                connection_attempt_time = float(attempt_value)
                                             except (ValueError, TypeError):
                                                 pass
 
                                     # Also get init time as a fallback
                                     init_time = None
-                                    if metadata and b'init_time' in metadata:
+                                    if metadata and 'init_time' in metadata:
                                         try:
-                                            init_time = float(metadata[b'init_time'].decode('utf-8'))
+                                            init_time = float(metadata['init_time'])
                                         except (ValueError, TypeError):
                                             pass
 
@@ -1185,7 +1186,7 @@ class ProxyServer:
                                     disconnect_value = self.redis_client.get(disconnect_key)
                                     if disconnect_value:
                                         try:
-                                            disconnect_time = float(disconnect_value.decode('utf-8'))
+                                            disconnect_time = float(disconnect_value)
                                         except (ValueError, TypeError) as e:
                                             logger.error(f"Invalid disconnect time for channel {channel_id}: {e}")
 
@@ -1306,7 +1307,7 @@ class ProxyServer:
 
             for key in channel_keys:
                 try:
-                    channel_id = key.decode('utf-8').split(':')[2]
+                    channel_id = key.split(':')[2]
 
                     # Check if this channel has an owner
                     owner = self.get_channel_owner(channel_id)
@@ -1351,7 +1352,7 @@ class ProxyServer:
 
             for key in channel_keys:
                 try:
-                    channel_id = key.decode('utf-8').split(':')[2]
+                    channel_id = key.split(':')[2]
 
                     # Get metadata first
                     metadata = self.redis_client.hgetall(key)
@@ -1366,7 +1367,7 @@ class ProxyServer:
                         continue
 
                     # Get owner
-                    owner = metadata.get(b'owner', b'').decode('utf-8') if b'owner' in metadata else ''
+                    owner = metadata.get('owner', '') if 'owner' in metadata else ''
 
                     # Check if owner is still alive
                     owner_alive = False
@@ -1380,7 +1381,7 @@ class ProxyServer:
 
                     # If no owner and no clients, clean it up
                     if not owner_alive and client_count == 0:
-                        state = metadata.get(b'state', b'unknown').decode('utf-8') if b'state' in metadata else 'unknown'
+                        state = metadata.get('state', 'unknown')
                         logger.warning(f"Found orphaned metadata for channel {channel_id} (state: {state}, owner: {owner}, clients: {client_count}) - cleaning up")
 
                         # If we have it locally, stop it properly to clean up transcode/proxy processes
@@ -1399,7 +1400,7 @@ class ProxyServer:
                         real_count = max(0, client_count - len(stale_ids))
                         if real_count <= 0:
                             # No real clients remain — safe to clean up.
-                            state = metadata.get(b'state', b'unknown').decode('utf-8') if b'state' in metadata else 'unknown'
+                            state = metadata.get('state', 'unknown')
                             logger.warning(
                                 f"Orphaned channel {channel_id} (state: {state}, "
                                 f"owner: {owner}) had {client_count} ghost client(s) "
@@ -1494,8 +1495,8 @@ class ProxyServer:
             # Get current state for logging
             current_state = None
             metadata = self.redis_client.hgetall(metadata_key)
-            if metadata and b'state' in metadata:
-                current_state = metadata[b'state'].decode('utf-8')
+            if metadata and 'state' in metadata:
+                current_state = metadata['state']
 
             # Only update if state is actually changing
             if current_state == new_state:

apps/proxy/ts_proxy/services/channel_service.py

@@ -61,7 +61,7 @@ class ChannelService:
             # Verify the stream_id was set
             stream_id_value = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.STREAM_ID)
             if stream_id_value:
-                logger.debug(f"Verified stream_id {stream_id_value.decode('utf-8')} is now set in Redis")
+                logger.debug(f"Verified stream_id {stream_id_value} is now set in Redis")
             else:
                 logger.error(f"Failed to set stream_id {stream_id} in Redis before initialization")
 
@@ -131,7 +131,7 @@ class ChannelService:
             try:
                 # This is inefficient but used for diagnostics - in production would use more targeted checks
                 redis_keys = proxy_server.redis_client.keys(f"ts_proxy:*:{channel_id}*")
-                redis_keys = [k.decode('utf-8') for k in redis_keys] if redis_keys else []
+                redis_keys = [k for k in redis_keys] if redis_keys else []
             except Exception as e:
                 logger.error(f"Error checking Redis keys: {e}")
 
@@ -236,8 +236,8 @@ class ChannelService:
             metadata_key = RedisKeys.channel_metadata(channel_id)
             try:
                 metadata = proxy_server.redis_client.hgetall(metadata_key)
-                if metadata and b'state' in metadata:
-                    state = metadata[b'state'].decode('utf-8')
+                if metadata and 'state' in metadata:
+                    state = metadata['state']
                     channel_info = {"state": state}
 
                     # Immediately mark as stopping in metadata so clients detect it faster
@@ -382,8 +382,8 @@ class ChannelService:
             metadata = proxy_server.redis_client.hgetall(metadata_key)
 
             # Extract state and owner
-            state = metadata.get(ChannelMetadataField.STATE.encode(), b'unknown').decode('utf-8')
-            owner = metadata.get(ChannelMetadataField.OWNER.encode(), b'unknown').decode('utf-8')
+            state = metadata.get(ChannelMetadataField.STATE, 'unknown')
+            owner = metadata.get(ChannelMetadataField.OWNER, 'unknown')
 
             # Valid states indicate channel is running properly
             valid_states = [ChannelState.ACTIVE, ChannelState.WAITING_FOR_CLIENTS, ChannelState.CONNECTING]
@@ -409,7 +409,7 @@ class ChannelService:
             }
 
             if last_data:
-                last_data_time = float(last_data.decode('utf-8'))
+                last_data_time = float(last_data)
                 data_age = time.time() - last_data_time
                 details["last_data_age"] = data_age
 
@@ -432,13 +432,13 @@ class ChannelService:
         try:
             # Use factory to parse the line based on stream type
             parsed_data = LogParserFactory.parse(stream_type, stream_info_line)
-            
+
             if not parsed_data:
                 return
 
             # Update Redis and database with parsed data
             ChannelService._update_stream_info_in_redis(
-                channel_id, 
+                channel_id,
                 parsed_data.get('video_codec'),
                 parsed_data.get('resolution'),
                 parsed_data.get('width'),
@@ -579,7 +579,7 @@ class ChannelService:
         metadata_key = RedisKeys.channel_metadata(channel_id)
 
         # First check if the key exists and what type it is
-        key_type = proxy_server.redis_client.type(metadata_key).decode('utf-8')
+        key_type = proxy_server.redis_client.type(metadata_key)
         logger.debug(f"Redis key {metadata_key} is of type: {key_type}")
 
         # Build metadata update dict

apps/proxy/ts_proxy/stream_generator.py

@@ -141,13 +141,13 @@ class StreamGenerator:
                 metadata_key = RedisKeys.channel_metadata(self.channel_id)
                 metadata = proxy_server.redis_client.hgetall(metadata_key)
 
-                if metadata and b'state' in metadata:
-                    state = metadata[b'state'].decode('utf-8')
+                if metadata and 'state' in metadata:
+                    state = metadata['state']
                     if state in ['waiting_for_clients', 'active']:
                         logger.info(f"[{self.client_id}] Channel {self.channel_id} now ready (state={state})")
                         return True
                     elif state in ['error', 'stopped', 'stopping']:  # Added 'stopping' to error states
-                        error_message = metadata.get(b'error_message', b'Unknown error').decode('utf-8')
+                        error_message = metadata.get('error_message', 'Unknown error')
                         logger.error(f"[{self.client_id}] Channel {self.channel_id} in error state: {state}, message: {error_message}")
                         # Send error packet before giving up
                         yield create_ts_packet('error', f"Error: {error_message}")
@@ -155,9 +155,9 @@ class StreamGenerator:
                     else:
                         # Improved logging to track initialization progress
                         init_time = "unknown"
-                        if b'init_time' in metadata:
+                        if 'init_time' in metadata:
                             try:
-                                init_time_float = float(metadata[b'init_time'].decode('utf-8'))
+                                init_time_float = float(metadata['init_time'])
                                 init_duration = time.time() - init_time_float
                                 init_time = f"{init_duration:.1f}s ago"
                             except:
@@ -390,8 +390,8 @@ class StreamGenerator:
         # Channel state in metadata
         metadata_key = RedisKeys.channel_metadata(self.channel_id)
         metadata = proxy_server.redis_client.hgetall(metadata_key)
-        if metadata and b'state' in metadata:
-            state = metadata[b'state'].decode('utf-8')
+        if metadata and 'state' in metadata:
+            state = metadata['state']
             if state in ['error', 'stopped', 'stopping']:
                 logger.info(f"[{self.client_id}] Channel in {state} state, terminating stream")
                 return False
@@ -555,8 +555,6 @@ class StreamGenerator:
                 if metadata:
                     stream_id_bytes = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.STREAM_ID)
                     if stream_id_bytes:
-                        stream_id = int(stream_id_bytes.decode('utf-8'))
-
                         # Check if we're the last client
                         if self.channel_id in proxy_server.client_managers:
                             client_count = proxy_server.client_managers[self.channel_id].get_total_client_count()

apps/proxy/ts_proxy/stream_manager.py

@@ -90,7 +90,7 @@ class StreamManager:
                     # Try to get stream_id specifically
                     stream_id_bytes = buffer.redis_client.hget(metadata_key, "stream_id")
                     if stream_id_bytes:
-                        self.current_stream_id = int(stream_id_bytes.decode('utf-8'))
+                        self.current_stream_id = int(stream_id_bytes)
                         self.tried_stream_ids.add(self.current_stream_id)
                         logger.info(f"Loaded stream ID {self.current_stream_id} from Redis for channel {buffer.channel_id}")
                     else:
@@ -413,7 +413,7 @@ class StreamManager:
                     is_owner = (
                         current_owner
                         and self.worker_id
-                        and current_owner.decode('utf-8') == self.worker_id
+                        and current_owner == self.worker_id
                     )
                     no_owner = current_owner is None
 
@@ -423,7 +423,7 @@ class StreamManager:
                             metadata_key, ChannelMetadataField.STATE
                         )
                         current_state = (
-                            current_state_bytes.decode('utf-8')
+                            current_state_bytes
                             if current_state_bytes else None
                         )
                         should_update = current_state in ChannelState.PRE_ACTIVE
@@ -1503,9 +1503,9 @@ class StreamManager:
                     current_state = None
                     try:
                         metadata = redis_client.hgetall(metadata_key)
-                        state_field = ChannelMetadataField.STATE.encode('utf-8')
+                        state_field = ChannelMetadataField.STATE
                         if metadata and state_field in metadata:
-                            current_state = metadata[state_field].decode('utf-8')
+                            current_state = metadata[state_field]
                     except Exception as e:
                         logger.error(f"Error checking current state: {e}")
 
@@ -1710,4 +1710,4 @@ class StreamManager:
         """Safely reset the URL switching state if it gets stuck"""
         self.url_switching = False
         self.url_switch_start_time = 0
-        logger.info(f"Reset URL switching state for channel {self.channel_id}")
+        logger.info(f"Reset URL switching state for channel {self.channel_id}")

apps/proxy/ts_proxy/url_utils.py

@@ -211,9 +211,9 @@ def get_stream_info_for_switch(channel_id: str, target_stream_id: Optional[int]
                     existing_stream_id = redis_client.get(f"channel_stream:{channel.id}")
                     if existing_stream_id:
                         # Decode bytes to string/int for proper Redis key lookup
-                        existing_stream_id = existing_stream_id.decode('utf-8')
+                        existing_stream_id = existing_stream_id
                         existing_profile_id = redis_client.get(f"stream_profile:{existing_stream_id}")
-                        if existing_profile_id and int(existing_profile_id.decode('utf-8')) == profile.id:
+                        if existing_profile_id and int(existing_profile_id) == profile.id:
                             channel_using_profile = True
                             logger.debug(f"Channel {channel.id} already using profile {profile.id}")
 
@@ -349,9 +349,9 @@ def get_alternate_streams(channel_id: str, current_stream_id: Optional[int] = No
                         existing_stream_id = redis_client.get(f"channel_stream:{channel.id}")
                         if existing_stream_id:
                             # Decode bytes to string/int for proper Redis key lookup
-                            existing_stream_id = existing_stream_id.decode('utf-8')
+                            existing_stream_id = existing_stream_id
                             existing_profile_id = redis_client.get(f"stream_profile:{existing_stream_id}")
-                            if existing_profile_id and int(existing_profile_id.decode('utf-8')) == profile.id:
+                            if existing_profile_id and int(existing_profile_id) == profile.id:
                                 channel_using_profile = True
                                 logger.debug(f"Channel {channel.id} already using profile {profile.id}")
 

apps/proxy/ts_proxy/utils.py

@@ -83,7 +83,7 @@ def create_ts_packet(packet_type='null', message=None):
 
     # Add message to payload if provided
     if message:
-        msg_bytes = message.encode('utf-8')
+        msg_bytes = message
         packet[4:4+min(len(msg_bytes), 180)] = msg_bytes[:180]
 
     return bytes(packet)
@@ -113,4 +113,4 @@ def get_logger(component_name=None):
             # Default if detection fails
             logger_name = "ts_proxy"
 
-    return logging.getLogger(logger_name)
+    return logging.getLogger(logger_name)

apps/proxy/ts_proxy/views.py

@@ -40,12 +40,13 @@ from .utils import get_logger
 from uuid import UUID
 import gevent
 from dispatcharr.utils import network_access_allowed
+from apps.proxy.utils import check_user_stream_limits
 
 logger = get_logger()
 
 
 @api_view(["GET"])
-def stream_ts(request, channel_id):
+def stream_ts(request, channel_id, user=None):
     if not network_access_allowed(request, "STREAMS"):
         return JsonResponse({"error": "Forbidden"}, status=403)
 
@@ -71,6 +72,13 @@ def stream_ts(request, channel_id):
                 )
                 break
 
+        if user:
+            if not check_user_stream_limits(user, client_id, media_id=channel_id):
+                return JsonResponse(
+                    {"error": f"Stream limit exceeded ({user.stream_limit} concurrent streams allowed)"},
+                    status=429
+                )
+
         # Check if we need to reinitialize the channel
         needs_initialization = True
         channel_state = None
@@ -81,9 +89,9 @@ def stream_ts(request, channel_id):
             metadata_key = RedisKeys.channel_metadata(channel_id)
             if proxy_server.redis_client.exists(metadata_key):
                 metadata = proxy_server.redis_client.hgetall(metadata_key)
-                state_field = ChannelMetadataField.STATE.encode("utf-8")
+                state_field = ChannelMetadataField.STATE
                 if state_field in metadata:
-                    channel_state = metadata[state_field].decode("utf-8")
+                    channel_state = metadata[state_field]
 
                     # Active/running states - channel is operational, don't reinitialize
                     if channel_state in [
@@ -119,9 +127,9 @@ def stream_ts(request, channel_id):
                         )
                     # Unknown/empty state - check if owner is alive
                     else:
-                        owner_field = ChannelMetadataField.OWNER.encode("utf-8")
+                        owner_field = ChannelMetadataField.OWNER
                         if owner_field in metadata:
-                            owner = metadata[owner_field].decode("utf-8")
+                            owner = metadata[owner_field]
                             owner_heartbeat_key = f"ts_proxy:worker:{owner}:heartbeat"
                             if proxy_server.redis_client.exists(owner_heartbeat_key):
                                 # Owner is still active with unknown state - don't reinitialize
@@ -399,7 +407,7 @@ def stream_ts(request, channel_id):
                                         metadata_key, ChannelMetadataField.STATE
                                     )
                                     if state_bytes:
-                                        current_state = state_bytes.decode("utf-8")
+                                        current_state = state_bytes
                                         logger.debug(
                                             f"[{client_id}] Current state of channel {channel_id}: {current_state}"
                                         )
@@ -475,12 +483,12 @@ def stream_ts(request, channel_id):
                 )
 
                 if url_bytes:
-                    url = url_bytes.decode("utf-8")
+                    url = url_bytes
                 if ua_bytes:
-                    stream_user_agent = ua_bytes.decode("utf-8")
+                    stream_user_agent = ua_bytes
                 # Extract transcode setting from Redis
                 if profile_bytes:
-                    profile_str = profile_bytes.decode("utf-8")
+                    profile_str = profile_bytes
                     use_transcode = (
                         profile_str == PROXY_PROFILE_NAME or profile_str == "None"
                     )
@@ -516,7 +524,7 @@ def stream_ts(request, channel_id):
         # Register client
         buffer = proxy_server.stream_buffers[channel_id]
         client_manager = proxy_server.client_managers[channel_id]
-        client_manager.add_client(client_id, client_ip, client_user_agent)
+        client_manager.add_client(client_id, client_ip, client_user_agent, user)
         logger.info(f"[{client_id}] Client registered with channel {channel_id}")
 
         # Create a stream generator for this client
@@ -557,7 +565,6 @@ def stream_xc(request, username, password, channel_id):
     if custom_properties["xc_password"] != password:
         return Response({"error": "Invalid credentials"}, status=401)
 
-    print(f"Fetchin channel with ID: {channel_id}")
     if user.user_level < 10:
         user_profile_count = user.channel_profiles.count()
 
@@ -585,7 +592,7 @@ def stream_xc(request, username, password, channel_id):
         channel = get_object_or_404(Channel, id=channel_id)
 
     # @TODO: we've got the  file 'type' via extension, support this when we support multiple outputs
-    return stream_ts(request._request, str(channel.uuid))
+    return stream_ts(request._request, str(channel.uuid), user)
 
 
 @csrf_exempt
@@ -713,7 +720,7 @@ def channel_status(request, channel_id=None):
                 )
                 for key in keys:
                     channel_id_match = re.search(
-                        r"ts_proxy:channel:(.*):metadata", key.decode("utf-8")
+                        r"ts_proxy:channel:(.*):metadata", key
                     )
                     if channel_id_match:
                         ch_id = channel_id_match.group(1)
@@ -834,7 +841,7 @@ def next_stream(request, channel_id):
                     metadata_key, ChannelMetadataField.STREAM_ID
                 )
                 if stream_id_bytes:
-                    current_stream_id = int(stream_id_bytes.decode("utf-8"))
+                    current_stream_id = int(stream_id_bytes)
                     logger.info(
                         f"Found current stream ID {current_stream_id} in Redis for channel {channel_id}"
                     )
@@ -844,7 +851,7 @@ def next_stream(request, channel_id):
                         metadata_key, ChannelMetadataField.M3U_PROFILE
                     )
                     if profile_id_bytes:
-                        profile_id = int(profile_id_bytes.decode("utf-8"))
+                        profile_id = int(profile_id_bytes)
                         logger.info(
                             f"Found M3U profile ID {profile_id} in Redis for channel {channel_id}"
                         )

apps/proxy/urls.py

@@ -6,4 +6,4 @@ urlpatterns = [
     path('ts/', include('apps.proxy.ts_proxy.urls')),
     path('hls/', include('apps.proxy.hls_proxy.urls')),
     path('vod/', include('apps.proxy.vod_proxy.urls')),
-]
+]

apps/proxy/utils.py

@@ -0,0 +1,185 @@
+import logging
+from core.utils import RedisClient
+from apps.proxy.vod_proxy.multi_worker_connection_manager import MultiWorkerVODConnectionManager, get_vod_client_stop_key
+from core.models import CoreSettings
+from apps.proxy.ts_proxy.services.channel_service import ChannelService
+
+logger = logging.getLogger("proxy")
+
+
+def attempt_stream_termination(user_id, requesting_client_id, active_connections):
+    try:
+        logger.info("[stream limits]" f"[{requesting_client_id}] User {user_id} has {len(active_connections)} active connections, checking termination candidates")
+
+        user_limit_settings = CoreSettings.get_user_limits_settings()
+        terminate_oldest = user_limit_settings.get("terminate_oldest", True)
+        prioritize_single = user_limit_settings.get("prioritize_single_client_channels", True)
+        ignore_same_channel = user_limit_settings.get("ignore_same_channel_connections", False)
+
+        channel_counts = {}
+        for connection in active_connections:
+            media_id = connection['media_id']
+            channel_counts[media_id] = channel_counts.get(media_id, 0) + 1
+
+        def prioritize(connection):
+            is_multi = channel_counts[connection['media_id']] > 1
+
+            # if we're ignoring same-channel connections, put them at the end
+            same_ch_key = 1 if (ignore_same_channel and is_multi) else 0
+
+            # key for prioritizing single-client channels
+            single_key = 0 if (prioritize_single and not is_multi) else 1
+
+            # sort by age setting
+            time_key = connection['connected_at'] if terminate_oldest else -connection['connected_at']
+
+            return (same_ch_key, single_key, time_key)
+
+        termination_candidates = sorted(active_connections, key=prioritize)
+
+        if not termination_candidates:
+            logger.warning("[stream limits]" f"[{requesting_client_id}] No termination candidates found for user {user_id}")
+            return False
+
+        target = termination_candidates[0]
+        logger.info("[stream limits]"
+            f"[{requesting_client_id}] Terminating client {target['client_id']} "
+            f"on media {target['media_id']} (connected_at={target['connected_at']})"
+        )
+
+        # When counting by unique channel, freeing one connection from a multi-connection
+        # channel doesn't free a slot — terminate all connections to that channel so the
+        # unique-channel count actually drops by one.
+        targets = (
+            [c for c in active_connections if c['media_id'] == target['media_id']]
+            if ignore_same_channel
+            else [target]
+        )
+
+        for t in targets:
+            if t['type'] == 'live':
+                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']}")
+            else:
+                connection_manager = MultiWorkerVODConnectionManager.get_instance()
+                redis_client = connection_manager.redis_client
+
+                if not redis_client:
+                    return False
+
+                connection_key = f"vod_persistent_connection:{t['client_id']}"
+                connection_data = redis_client.hgetall(connection_key)
+                if not connection_data:
+                    logger.warning(f"VOD connection not found: {t['client_id']}")
+                    continue
+
+                stop_key = get_vod_client_stop_key(t['client_id'])
+                redis_client.setex(stop_key, 60, "true")  # 60 second TTL
+
+        return True
+    except Exception as e:
+        logger.error("[stream limits]" f"[{requesting_client_id}] Error during stream termination for user {user_id}: {e}")
+        return False
+
+def get_user_active_connections(user_id):
+    redis_client = RedisClient.get_client()
+    connections = []
+
+    try:
+        # Grab live streams
+        for key in redis_client.scan_iter(match="ts_proxy:channel:*:clients:*", count=1000):
+            parts = key.split(':')
+            if len(parts) >= 5:
+                channel_id = parts[2]
+                client_id = parts[4]
+
+                client_user_id = redis_client.hget(key, 'user_id')
+                connected_at = redis_client.hget(key, 'connected_at')
+
+                logger.debug(f"[stream limits] user_id = {user_id}")
+                logger.debug(f"[stream limits] channel_id = {channel_id}")
+                logger.debug(f"[stream limits] client_id = {client_id}")
+
+                if 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}")
+                        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',
+                        })
+                    except (ValueError, TypeError):
+                        pass
+
+        # Grab VOD
+        for key in redis_client.scan_iter(match="vod_persistent_connection:*", count=1000):
+            parts = key.split(':')
+            if len(parts) >= 2:
+                client_id = parts[1]
+
+                client_user_id = redis_client.hget(key, 'user_id')
+                connected_at = redis_client.hget(key, 'created_at')
+                content_uuid = redis_client.hget(key, 'content_uuid')
+
+                logger.debug(f"[stream limits] user_id = {user_id}")
+                logger.debug(f"[stream limits] client_id = {client_id}")
+
+                if client_user_id and int(client_user_id) == user_id:
+                    try:
+                        logger.debug(f"[stream limits] Found VOD connection for user {user_id} on content {content_uuid} with client ID {client_id}")
+                        connected_at = float(connected_at) if connected_at else 0
+                        connections.append({
+                            'media_id': content_uuid or client_id,
+                            'client_id': client_id,
+                            'connected_at': connected_at,
+                            'type': 'vod',
+                        })
+                    except (ValueError, TypeError):
+                        pass
+
+        return connections
+
+    except Exception as e:
+        logger.warning(f"Error getting active channel details for user {user_id}: {e}")
+        return []
+
+
+def check_user_stream_limits(user, client_id, media_id=None):
+    # Check user stream limits
+    if user and user.stream_limit > 0:
+        logger.debug("[stream limits]" f"[{client_id}] User {user.username} (ID: {user.id}) is requesting a stream (stream_limit: {user.stream_limit})")
+        user_limit_settings = CoreSettings.get_user_limits_settings()
+        ignore_same_channel = user_limit_settings.get("ignore_same_channel_connections", False)
+
+        active_connections = get_user_active_connections(user.id)
+        unique_channel_count = set([conn['media_id'] for conn in active_connections])
+        user_stream_count = len(unique_channel_count) if ignore_same_channel else len(active_connections)
+
+        logger.debug(f"[stream limits]" f"[{client_id}] User {user.username} currently has {len(active_connections)} active connections across {len(unique_channel_count)} unique channels (counting method: {'unique channels' if ignore_same_channel else 'total connections'})")
+
+        # If ignore_same_channel is enabled and this request is for a live channel the user
+        # is already watching, allow it through without counting against the limit.
+        # VOD is excluded: connections aren't shared so multiple VOD connections to the
+        # same content would mean multiple upstream connections.
+        live_channel_ids = {str(conn['media_id']) for conn in active_connections if conn['type'] == 'live'}
+        if ignore_same_channel and media_id and str(media_id) in live_channel_ids:
+            logger.debug(f"[stream limits][{client_id}] Same-channel reconnect 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
+
+            if user_stream_count >= user.stream_limit:
+                logger.warning("[stream limits]"
+                    f"[{client_id}] User {user.username} (ID: {user.id}) has reached stream limit "
+                    f"({user_stream_count}/{user.stream_limit} streams), attempting to free up slot"
+                )
+
+                if not attempt_stream_termination(user.id, client_id, active_connections):
+                    return False
+
+    return True

apps/proxy/vod_proxy/connection_manager.py

@@ -342,15 +342,15 @@ class VODConnectionManager:
                             continue
 
                         # Extract session info
-                        stored_content_type = session_data.get(b'content_type', b'').decode('utf-8')
-                        stored_content_uuid = session_data.get(b'content_uuid', b'').decode('utf-8')
+                        stored_content_type = session_data.get('content_type', '')
+                        stored_content_uuid = session_data.get('content_uuid', '')
 
                         # Check if content matches
                         if stored_content_type != content_type or stored_content_uuid != content_uuid:
                             continue
 
                         # Extract session ID from key
-                        session_id = key.decode('utf-8').replace('vod_session:', '')
+                        session_id = key.replace('vod_session:', '')
 
                         # Check if session has an active persistent connection
                         persistent_conn = self._persistent_connections.get(session_id)
@@ -364,13 +364,13 @@ class VODConnectionManager:
                             continue
 
                         # Get stored client info for comparison
-                        stored_client_ip = session_data.get(b'client_ip', b'').decode('utf-8')
-                        stored_user_agent = session_data.get(b'user_agent', b'').decode('utf-8')
+                        stored_client_ip = session_data.get('client_ip', '')
+                        stored_user_agent = session_data.get('user_agent', '')
 
                         # Check timeshift parameters match
-                        stored_utc_start = session_data.get(b'utc_start', b'').decode('utf-8')
-                        stored_utc_end = session_data.get(b'utc_end', b'').decode('utf-8')
-                        stored_offset = session_data.get(b'offset', b'').decode('utf-8')
+                        stored_utc_start = session_data.get('utc_start', '')
+                        stored_utc_end = session_data.get('utc_end', '')
+                        stored_offset = session_data.get('offset', '')
 
                         current_utc_start = utc_start or ""
                         current_utc_end = utc_end or ""
@@ -407,7 +407,7 @@ class VODConnectionManager:
                                 'session_id': session_id,
                                 'score': score,
                                 'reasons': match_reasons,
-                                'last_activity': float(session_data.get(b'last_activity', b'0').decode('utf-8'))
+                                'last_activity': float(session_data.get('last_activity', '0'))
                             })
 
                     except Exception as e:
@@ -588,7 +588,7 @@ class VODConnectionManager:
                 # Get current bytes and add to it
                 current_bytes = self.redis_client.hget(connection_key, "bytes_sent")
                 if current_bytes:
-                    total_bytes = int(current_bytes.decode('utf-8')) + bytes_sent
+                    total_bytes = int(current_bytes) + bytes_sent
                 else:
                     total_bytes = bytes_sent
                 update_data["bytes_sent"] = str(total_bytes)
@@ -623,7 +623,7 @@ class VODConnectionManager:
             profile_id = None
             if b"m3u_profile_id" in connection_data:
                 try:
-                    profile_id = int(connection_data[b"m3u_profile_id"].decode('utf-8'))
+                    profile_id = int(connection_data["m3u_profile_id"])
                 except ValueError:
                     pass
 
@@ -669,16 +669,14 @@ class VODConnectionManager:
             # Convert bytes to strings and parse numbers
             info = {}
             for key, value in connection_data.items():
-                key_str = key.decode('utf-8')
-                value_str = value.decode('utf-8')
 
                 # Parse numeric fields
-                if key_str in ['connected_at', 'last_activity']:
-                    info[key_str] = float(value_str)
-                elif key_str in ['bytes_sent', 'position_seconds', 'm3u_profile_id']:
-                    info[key_str] = int(value_str)
+                if key in ['connected_at', 'last_activity']:
+                    info[key] = float(value)
+                elif key in ['bytes_sent', 'position_seconds', 'm3u_profile_id']:
+                    info[key] = int(valuvaluee_str)
                 else:
-                    info[key_str] = value_str
+                    info[key] = value
 
             return info
 
@@ -728,14 +726,13 @@ class VODConnectionManager:
 
                 for key in keys:
                     try:
-                        key_str = key.decode('utf-8')
                         last_activity = self.redis_client.hget(key, "last_activity")
 
                         if last_activity:
-                            last_activity_time = float(last_activity.decode('utf-8'))
+                            last_activity_time = float(last_activity)
                             if current_time - last_activity_time > max_age_seconds:
                                 # Extract info for cleanup
-                                parts = key_str.split(':')
+                                parts = key.split(':')
                                 if len(parts) >= 5:
                                     content_type = parts[2]
                                     content_uuid = parts[3]
@@ -1009,7 +1006,7 @@ class VODConnectionManager:
             return HttpResponse(f"Streaming error: {str(e)}", status=500)
 
     def stream_content_with_session(self, session_id, content_obj, stream_url, m3u_profile, client_ip, user_agent, request,
-                                  utc_start=None, utc_end=None, offset=None, range_header=None):
+                                  utc_start=None, utc_end=None, offset=None, range_header=None, user=None):
         """
         Stream VOD content with persistent connection per session
 
@@ -1394,9 +1391,9 @@ class VODConnectionManager:
                     session_data = self.redis_client.hgetall(session_key)
                     if session_data:
                         # Get session details for connection cleanup
-                        content_type = session_data.get(b'content_type', b'').decode('utf-8')
-                        content_uuid = session_data.get(b'content_uuid', b'').decode('utf-8')
-                        profile_id = session_data.get(b'profile_id')
+                        content_type = session_data.get('content_type', '')
+                        content_uuid = session_data.get('content_uuid', '')
+                        profile_id = session_data.get('profile_id')
 
                         # Generate client_id from session_id (matches what's used during streaming)
                         client_id = session_id
@@ -1415,12 +1412,12 @@ class VODConnectionManager:
                         # Fallback DECR: only if the connection tracking key had already
                         # expired (remove_connection skips DECR in that case)
                         if not connection_key_exists:
-                            if session_data.get(b'connection_counted') == b'True' and profile_id:
-                                profile_key = self._get_profile_connections_key(int(profile_id.decode('utf-8')))
+                            if session_data.get('connection_counted') == 'True' and profile_id:
+                                profile_key = self._get_profile_connections_key(int(profile_id))
                                 current_count = int(self.redis_client.get(profile_key) or 0)
                                 if current_count > 0:
                                     self.redis_client.decr(profile_key)
-                                    logger.info(f"[{session_id}] Decremented profile {profile_id.decode('utf-8')} connections (fallback)")
+                                    logger.info(f"[{session_id}] Decremented profile {profile_id} connections (fallback)")
 
                     # Remove session tracking key
                     self.redis_client.delete(session_key)
@@ -1440,7 +1437,7 @@ class VODConnectionManager:
 
                         if session_related_keys:
                             # Filter out keys we already deleted
-                            remaining_keys = [k for k in session_related_keys if k.decode('utf-8') != session_key]
+                            remaining_keys = [k for k in session_related_keys if k != session_key]
                             if remaining_keys:
                                 self.redis_client.delete(*remaining_keys)
                                 logger.info(f"[{session_id}] Cleaned up {len(remaining_keys)} additional session-related keys")
@@ -1470,7 +1467,7 @@ class VODConnectionManager:
                 if self.redis_client:
                     session_data = self.redis_client.hgetall(session_key)
                     if session_data:
-                        created_at = float(session_data.get(b'created_at', b'0').decode('utf-8'))
+                        created_at = float(session_data.get('created_at', '0'))
                         if current_time - created_at > max_age_seconds:
                             logger.info(f"[{session_id}] Session older than {max_age_seconds}s")
                             stale_sessions.append(session_id)

apps/proxy/vod_proxy/multi_worker_connection_manager.py

@@ -93,7 +93,7 @@ class SerializableConnectionState:
                  content_name: str = None, client_ip: str = None,
                  client_user_agent: str = None, utc_start: str = None,
                  utc_end: str = None, offset: str = None,
-                 worker_id: str = None, connection_type: str = "redis_backed"):
+                 worker_id: str = None, connection_type: str = "redis_backed", user_id: str = "unknown"):
         self.session_id = session_id
         self.stream_url = stream_url
         self.headers = headers
@@ -104,6 +104,7 @@ class SerializableConnectionState:
         self.last_activity = time.time()
         self.request_count = 0
         self.active_streams = 0
+        self.user_id = user_id
 
         # Session metadata (consolidated from vod_session key)
         self.content_obj_type = content_obj_type
@@ -160,7 +161,8 @@ class SerializableConnectionState:
             'last_seek_byte': str(self.last_seek_byte),
             'last_seek_percentage': str(self.last_seek_percentage),
             'total_content_size': str(self.total_content_size),
-            'last_seek_timestamp': str(self.last_seek_timestamp)
+            'last_seek_timestamp': str(self.last_seek_timestamp),
+            'user_id': str(self.user_id),
         }
 
     @classmethod
@@ -184,7 +186,8 @@ class SerializableConnectionState:
             utc_end=data.get('utc_end') or '',
             offset=data.get('offset') or '',
             worker_id=data.get('worker_id') or None,
-            connection_type=data.get('connection_type', 'redis_backed')
+            connection_type=data.get('connection_type', 'redis_backed'),
+            user_id=data.get('user_id', 'unknown')
         )
         obj.last_activity = float(data.get('last_activity', time.time()))
         obj.request_count = int(data.get('request_count', 0))
@@ -224,7 +227,7 @@ class RedisBackedVODConnection:
 
             # Convert bytes keys/values to strings if needed
             if isinstance(list(data.keys())[0], bytes):
-                data = {k.decode('utf-8'): v.decode('utf-8') for k, v in data.items()}
+                data = {k: v for k, v in data.items()}
 
             return SerializableConnectionState.from_dict(data)
         except Exception as e:
@@ -281,7 +284,7 @@ class RedisBackedVODConnection:
                          content_name: str = None, client_ip: str = None,
                          client_user_agent: str = None, utc_start: str = None,
                          utc_end: str = None, offset: str = None,
-                         worker_id: str = None) -> bool:
+                         worker_id: str = None, user=None) -> bool:
         """Create a new connection state in Redis with consolidated session metadata"""
         if not self._acquire_lock():
             logger.warning(f"[{self.session_id}] Could not acquire lock for connection creation")
@@ -309,7 +312,8 @@ class RedisBackedVODConnection:
                 utc_start=utc_start,
                 utc_end=utc_end,
                 offset=offset,
-                worker_id=worker_id
+                worker_id=worker_id,
+                user_id=user.id if user else "unknown"
             )
             success = self._save_connection_state(state)
 
@@ -365,6 +369,24 @@ class RedisBackedVODConnection:
                 timeout=(10, 10),
                 allow_redirects=allow_redirects
             )
+
+            # If the cached final_url returned an error (e.g. an ephemeral dispatcharr session
+            # that has since expired), clear it and retry from the original stream_url.
+            if response.status_code >= 400 and state.final_url:
+                logger.warning(
+                    f"[{self.session_id}] Cached final_url returned {response.status_code}, "
+                    f"clearing and retrying from stream_url"
+                )
+                response.close()
+                state.final_url = None
+                response = self.local_session.get(
+                    state.stream_url,
+                    headers=headers,
+                    stream=True,
+                    timeout=(10, 10),
+                    allow_redirects=True
+                )
+
             response.raise_for_status()
 
             # Update state with response info on first request
@@ -761,7 +783,7 @@ class MultiWorkerVODConnectionManager:
 
     def stream_content_with_session(self, session_id, content_obj, stream_url, m3u_profile,
                                   client_ip, client_user_agent, request,
-                                  utc_start=None, utc_end=None, offset=None, range_header=None):
+                                  utc_start=None, utc_end=None, offset=None, range_header=None, user=None):
         """Stream content with Redis-backed persistent connection"""
 
         # Generate client ID
@@ -858,7 +880,8 @@ class MultiWorkerVODConnectionManager:
                     utc_start=utc_start,
                     utc_end=utc_end,
                     offset=str(offset) if offset else None,
-                    worker_id=self.worker_id
+                    worker_id=self.worker_id,
+                    user=user
                 ):
                     logger.error(f"[{client_id}] Worker {self.worker_id} - Failed to create Redis connection")
                     # Roll back the profile slot reservation since connection failed
@@ -1272,14 +1295,14 @@ class MultiWorkerVODConnectionManager:
 
                         # Convert bytes to strings if needed
                         if isinstance(list(data.keys())[0], bytes):
-                            data = {k.decode('utf-8'): v.decode('utf-8') for k, v in data.items()}
+                            data = {k: v for k, v in data.items()}
 
                         last_activity = float(data.get('last_activity', 0))
                         active_streams = int(data.get('active_streams', 0))
 
                         # Clean up if stale and no active streams
                         if (current_time - last_activity > max_age_seconds) and active_streams == 0:
-                            session_id = key.decode('utf-8').replace('vod_persistent_connection:', '')
+                            session_id = key.replace('vod_persistent_connection:', '')
                             logger.info(f"Cleaning up stale connection: {session_id}")
 
                             # Clean up connection and related keys
@@ -1376,7 +1399,7 @@ class MultiWorkerVODConnectionManager:
             if connection_data:
                 # Convert bytes to strings if needed
                 if isinstance(list(connection_data.keys())[0], bytes):
-                    connection_data = {k.decode('utf-8'): v.decode('utf-8') for k, v in connection_data.items()}
+                    connection_data = {k: v for k, v in connection_data.items()}
 
                 profile_id = connection_data.get('m3u_profile_id')
                 if profile_id:
@@ -1438,7 +1461,7 @@ class MultiWorkerVODConnectionManager:
 
                         # Convert bytes keys/values to strings if needed
                         if isinstance(list(connection_data.keys())[0], bytes):
-                            connection_data = {k.decode('utf-8'): v.decode('utf-8') for k, v in connection_data.items()}
+                            connection_data = {k: v for k, v in connection_data.items()}
 
                         # Check if content matches (using consolidated data)
                         stored_content_type = connection_data.get('content_obj_type', '')
@@ -1448,7 +1471,7 @@ class MultiWorkerVODConnectionManager:
                             continue
 
                         # Extract session ID
-                        session_id = key.decode('utf-8').replace('vod_persistent_connection:', '')
+                        session_id = key.replace('vod_persistent_connection:', '')
 
                         # Check if Redis-backed connection exists and has no active streams
                         redis_connection = RedisBackedVODConnection(session_id, self.redis_client)
@@ -1526,4 +1549,4 @@ class MultiWorkerVODConnectionManager:
             return redis_connection.get_session_metadata()
         except Exception as e:
             logger.error(f"Error getting session info for {session_id}: {e}")
-            return None
+            return None

apps/proxy/vod_proxy/urls.py

@@ -1,16 +1,17 @@
 from django.urls import path
 from . import views
+from .views import stream_vod
 
 app_name = 'vod_proxy'
 
 urlpatterns = [
     # Generic VOD streaming with session ID in path (for compatibility)
-    path('<str:content_type>/<uuid:content_id>/<str:session_id>', views.VODStreamView.as_view(), name='vod_stream_with_session'),
-    path('<str:content_type>/<uuid:content_id>/<str:session_id>/<int:profile_id>/', views.VODStreamView.as_view(), name='vod_stream_with_session_and_profile'),
+    path('<str:content_type>/<uuid:content_id>/<str:session_id>', stream_vod, name='vod_stream_with_session'),
+    path('<str:content_type>/<uuid:content_id>/<str:session_id>/<int:profile_id>/', stream_vod, name='vod_stream_with_session_and_profile'),
 
     # Generic VOD streaming (supports movies, episodes, series) - legacy patterns
-    path('<str:content_type>/<uuid:content_id>', views.VODStreamView.as_view(), name='vod_stream'),
-    path('<str:content_type>/<uuid:content_id>/<int:profile_id>/', views.VODStreamView.as_view(), name='vod_stream_with_profile'),
+    path('<str:content_type>/<uuid:content_id>', stream_vod, name='vod_stream'),
+    path('<str:content_type>/<uuid:content_id>/<int:profile_id>/', stream_vod, name='vod_stream_with_profile'),
 
     # VOD playlist generation
     path('playlist/', views.VODPlaylistView.as_view(), name='vod_playlist'),

apps/vod/api_views.py

@@ -578,13 +578,15 @@ class UnifiedContentViewSet(viewsets.ReadOnlyModelViewSet):
                 "series.id IN (SELECT DISTINCT series_id FROM vod_m3useriesrelation msr JOIN m3u_m3uaccount ma ON msr.m3u_account_id = ma.id WHERE ma.is_active = true)"
             ]
 
-            params = []
+            movie_params = []
+            series_params = []
 
             if search:
                 where_conditions[0] += " AND LOWER(movies.name) LIKE %s"
                 where_conditions[1] += " AND LOWER(series.name) LIKE %s"
                 search_param = f"%{search.lower()}%"
-                params.extend([search_param, search_param])
+                movie_params.append(search_param)
+                series_params.append(search_param)
 
             if category:
                 if '|' in category:
@@ -592,15 +594,20 @@ class UnifiedContentViewSet(viewsets.ReadOnlyModelViewSet):
                     if cat_type == 'movie':
                         where_conditions[0] += " AND movies.id IN (SELECT movie_id FROM vod_m3umovierelation mmr JOIN vod_vodcategory c ON mmr.category_id = c.id WHERE c.name = %s)"
                         where_conditions[1] = "1=0"  # Exclude series
-                        params.append(cat_name)
+                        movie_params.append(cat_name)
+                        series_params = []  # no params needed for "1=0"
                     elif cat_type == 'series':
                         where_conditions[1] += " AND series.id IN (SELECT series_id FROM vod_m3useriesrelation msr JOIN vod_vodcategory c ON msr.category_id = c.id WHERE c.name = %s)"
                         where_conditions[0] = "1=0"  # Exclude movies
-                        params.append(cat_name)
+                        series_params.append(cat_name)
+                        movie_params = []  # no params needed for "1=0"
                 else:
                     where_conditions[0] += " AND movies.id IN (SELECT movie_id FROM vod_m3umovierelation mmr JOIN vod_vodcategory c ON mmr.category_id = c.id WHERE c.name = %s)"
                     where_conditions[1] += " AND series.id IN (SELECT series_id FROM vod_m3useriesrelation msr JOIN vod_vodcategory c ON msr.category_id = c.id WHERE c.name = %s)"
-                    params.extend([category, category])
+                    movie_params.append(category)
+                    series_params.append(category)
+
+            params = movie_params + series_params
 
             # Use UNION ALL with ORDER BY and LIMIT/OFFSET for true unified pagination
             # This is much more efficient than Python sorting
@@ -896,4 +903,3 @@ class VODLogoViewSet(viewsets.ModelViewSet):
                 {"error": str(e)},
                 status=status.HTTP_500_INTERNAL_SERVER_ERROR
             )
-

core/migrations/022_default_user_limit_settings.py

@@ -0,0 +1,24 @@
+# Generated by Django 5.1.6 on 2025-03-01 14:01
+
+from django.db import migrations
+from django.utils.text import slugify
+
+
+def preload_user_limit_settings(apps, schema_editor):
+    CoreSettings = apps.get_model("core", "CoreSettings")
+    CoreSettings.objects.create(
+        key="user_limit_settings",
+        name="User Limit Settings",
+        value={},
+    )
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("core", "0021_systemnotification_notificationdismissal"),
+    ]
+
+    operations = [
+        migrations.RunPython(preload_user_limit_settings),
+    ]

core/models.py

@@ -156,6 +156,7 @@ PROXY_SETTINGS_KEY = "proxy_settings"
 NETWORK_ACCESS_KEY = "network_access"
 SYSTEM_SETTINGS_KEY = "system_settings"
 EPG_SETTINGS_KEY = "epg_settings"
+USER_LIMITS_SETTINGS_KEY = "user_limit_settings"
 
 
 class CoreSettings(models.Model):
@@ -362,6 +363,15 @@ class CoreSettings(models.Model):
         cls._update_group(SYSTEM_SETTINGS_KEY, "System Settings", {"time_zone": value})
         return value
 
+    @classmethod
+    def get_user_limits_settings(cls):
+        return cls._get_group(USER_LIMITS_SETTINGS_KEY, {
+            "terminate_on_limit_exceeded": True,
+            "prioritize_single_client_channels": True,
+            "ignore_same_channel_connections": False,
+            "terminate_oldest": True,
+        })
+
 
 class SystemEvent(models.Model):
     """

core/redis_pubsub.py

@@ -201,10 +201,6 @@ class RedisPubSubManager:
 
                     channel = message.get('channel')
                     if channel:
-                        # Decode binary channel name if needed
-                        if isinstance(channel, bytes):
-                            channel = channel.decode('utf-8')
-
                         # Find and call the appropriate handler
                         handler = self.message_handlers.get(channel)
                         if handler:

core/tasks.py

@@ -404,7 +404,7 @@ def fetch_channel_stats():
         while True:
             cursor, keys = redis_client.scan(cursor, match=channel_pattern)
             for key in keys:
-                channel_id_match = re.search(r"ts_proxy:channel:(.*):metadata", key.decode('utf-8'))
+                channel_id_match = re.search(r"ts_proxy:channel:(.*):metadata", key)
                 if channel_id_match:
                     ch_id = channel_id_match.group(1)
                     channel_info = ChannelStatus.get_basic_channel_info(ch_id)

core/utils.py

@@ -45,109 +45,116 @@ def natural_sort_key(text):
 
 class RedisClient:
     _client = None
+    _buffer = None
     _pubsub_client = None
 
     @classmethod
-    def get_client(cls, max_retries=5, retry_interval=1):
-        if cls._client is None:
-            retry_count = 0
-            while retry_count < max_retries:
+    def _init_client(cls, decode_responses=True, max_retries=5, retry_interval=1):
+        retry_count = 0
+        while retry_count < max_retries:
+            try:
+                # Get connection parameters from settings or environment
+                redis_host = os.environ.get("REDIS_HOST", getattr(settings, 'REDIS_HOST', 'localhost'))
+                redis_port = int(os.environ.get("REDIS_PORT", getattr(settings, 'REDIS_PORT', 6379)))
+                redis_db = int(os.environ.get("REDIS_DB", getattr(settings, 'REDIS_DB', 0)))
+                redis_password = os.environ.get("REDIS_PASSWORD", getattr(settings, 'REDIS_PASSWORD', ''))
+                redis_user = os.environ.get("REDIS_USER", getattr(settings, 'REDIS_USER', ''))
+
+                # Use standardized settings
+                socket_timeout = getattr(settings, 'REDIS_SOCKET_TIMEOUT', 5)
+                socket_connect_timeout = getattr(settings, 'REDIS_SOCKET_CONNECT_TIMEOUT', 5)
+                health_check_interval = getattr(settings, 'REDIS_HEALTH_CHECK_INTERVAL', 30)
+                socket_keepalive = getattr(settings, 'REDIS_SOCKET_KEEPALIVE', True)
+                retry_on_timeout = getattr(settings, 'REDIS_RETRY_ON_TIMEOUT', True)
+
+                # TLS params from settings (empty dict when TLS is disabled)
+                ssl_params = getattr(settings, 'REDIS_SSL_PARAMS', {})
+
+                # Create Redis client with better defaults
+                client = redis.Redis(
+                    host=redis_host,
+                    port=redis_port,
+                    db=redis_db,
+                    password=redis_password if redis_password else None,
+                    username=redis_user if redis_user else None,
+                    socket_timeout=socket_timeout,
+                    socket_connect_timeout=socket_connect_timeout,
+                    socket_keepalive=socket_keepalive,
+                    health_check_interval=health_check_interval,
+                    retry_on_timeout=retry_on_timeout,
+                    decode_responses=decode_responses,
+                    **ssl_params
+                )
+
+                # Validate connection with ping
+                client.ping()
+
+                # Disable persistence on first connection - improves performance
+                # Only try to disable if not in a read-only environment
                 try:
-                    # Get connection parameters from settings or environment
-                    redis_host = os.environ.get("REDIS_HOST", getattr(settings, 'REDIS_HOST', 'localhost'))
-                    redis_port = int(os.environ.get("REDIS_PORT", getattr(settings, 'REDIS_PORT', 6379)))
-                    redis_db = int(os.environ.get("REDIS_DB", getattr(settings, 'REDIS_DB', 0)))
-                    redis_password = os.environ.get("REDIS_PASSWORD", getattr(settings, 'REDIS_PASSWORD', ''))
-                    redis_user = os.environ.get("REDIS_USER", getattr(settings, 'REDIS_USER', ''))
+                    client.config_set('save', '')  # Disable RDB snapshots
+                    client.config_set('appendonly', 'no')  # Disable AOF logging
 
-                    # Use standardized settings
-                    socket_timeout = getattr(settings, 'REDIS_SOCKET_TIMEOUT', 5)
-                    socket_connect_timeout = getattr(settings, 'REDIS_SOCKET_CONNECT_TIMEOUT', 5)
-                    health_check_interval = getattr(settings, 'REDIS_HEALTH_CHECK_INTERVAL', 30)
-                    socket_keepalive = getattr(settings, 'REDIS_SOCKET_KEEPALIVE', True)
-                    retry_on_timeout = getattr(settings, 'REDIS_RETRY_ON_TIMEOUT', True)
+                    # Disable protected mode when in debug mode
+                    if os.environ.get('DISPATCHARR_DEBUG', '').lower() == 'true':
+                        client.config_set('protected-mode', 'no')  # Disable protected mode in debug
+                        logger.warning("Redis protected mode disabled for debug environment")
 
-                    # TLS params from settings (empty dict when TLS is disabled)
-                    ssl_params = getattr(settings, 'REDIS_SSL_PARAMS', {})
-
-                    # Create Redis client with better defaults
-                    client = redis.Redis(
-                        host=redis_host,
-                        port=redis_port,
-                        db=redis_db,
-                        password=redis_password if redis_password else None,
-                        username=redis_user if redis_user else None,
-                        socket_timeout=socket_timeout,
-                        socket_connect_timeout=socket_connect_timeout,
-                        socket_keepalive=socket_keepalive,
-                        health_check_interval=health_check_interval,
-                        retry_on_timeout=retry_on_timeout,
-                        **ssl_params
-                    )
-
-                    # Validate connection with ping
-                    client.ping()
-
-                    # Disable persistence on first connection - improves performance
-                    # Only try to disable if not in a read-only environment
-                    try:
-                        client.config_set('save', '')  # Disable RDB snapshots
-                        client.config_set('appendonly', 'no')  # Disable AOF logging
-
-                        # Set optimal memory settings with environment variable support
-                        # Get max memory from environment or use a larger default (512MB instead of 256MB)
-                        #max_memory = os.environ.get('REDIS_MAX_MEMORY', '512mb')
-                        #eviction_policy = os.environ.get('REDIS_EVICTION_POLICY', 'allkeys-lru')
-
-                        # Apply memory settings
-                        #client.config_set('maxmemory-policy', eviction_policy)
-                        #client.config_set('maxmemory', max_memory)
-
-                        #logger.info(f"Redis configured with maxmemory={max_memory}, policy={eviction_policy}")
-
-                        # Disable protected mode when in debug mode
-                        if os.environ.get('DISPATCHARR_DEBUG', '').lower() == 'true':
-                            client.config_set('protected-mode', 'no')  # Disable protected mode in debug
-                            logger.warning("Redis protected mode disabled for debug environment")
-
-                        logger.trace("Redis persistence disabled for better performance")
-                    except redis.exceptions.ResponseError as e:
-                        # Improve error handling for Redis configuration errors
-                        if "OOM" in str(e):
-                            logger.error(f"Redis OOM during configuration: {e}")
-                            # Try to increase maxmemory as an emergency measure
-                            try:
-                                client.config_set('maxmemory', '768mb')
-                                logger.warning("Applied emergency Redis memory increase to 768MB")
-                            except:
-                                pass
-                        else:
-                            logger.error(f"Redis configuration error: {e}")
-
-                    logger.info(f"Connected to Redis at {redis_host}:{redis_port}/{redis_db}")
-
-                    cls._client = client
-                    break
-
-                except (ConnectionError, TimeoutError) as e:
-                    retry_count += 1
-                    _tls_hint = _REDIS_TLS_HINT if ssl_params else ""
-                    if retry_count >= max_retries:
-                        logger.error(f"Failed to connect to Redis after {max_retries} attempts: {e}{_tls_hint}")
-                        return None
+                    logger.trace("Redis persistence disabled for better performance")
+                except redis.exceptions.ResponseError as e:
+                    # Improve error handling for Redis configuration errors
+                    if "OOM" in str(e):
+                        logger.error(f"Redis OOM during configuration: {e}")
+                        # Try to increase maxmemory as an emergency measure
+                        try:
+                            client.config_set('maxmemory', '768mb')
+                            logger.warning("Applied emergency Redis memory increase to 768MB")
+                        except:
+                            pass
                     else:
-                        # Use exponential backoff for retries
-                        wait_time = retry_interval * (2 ** (retry_count - 1))
-                        logger.warning(f"Redis connection failed. Retrying in {wait_time}s... ({retry_count}/{max_retries})")
-                        time.sleep(wait_time)
+                        logger.error(f"Redis configuration error: {e}")
 
-                except Exception as e:
-                    _tls_hint = _REDIS_TLS_HINT if ssl_params else ""
-                    logger.error(f"Unexpected error connecting to Redis: {e}{_tls_hint}")
+                logger.info(f"Connected to Redis at {redis_host}:{redis_port}/{redis_db}")
+
+                return client
+
+            except (ConnectionError, TimeoutError) as e:
+                retry_count += 1
+                _tls_hint = _REDIS_TLS_HINT if ssl_params else ""
+                if retry_count >= max_retries:
+                    logger.error(f"Failed to connect to Redis after {max_retries} attempts: {e}{_tls_hint}")
                     return None
+                else:
+                    # Use exponential backoff for retries
+                    wait_time = retry_interval * (2 ** (retry_count - 1))
+                    logger.warning(f"Redis connection failed. Retrying in {wait_time}s... ({retry_count}/{max_retries})")
+                    time.sleep(wait_time)
 
+            except Exception as e:
+                _tls_hint = ""
+                try:
+                    _tls_hint = _REDIS_TLS_HINT if ssl_params else ""
+                except NameError:
+                    pass
+                logger.error(f"Unexpected error connecting to Redis: {e}{_tls_hint}")
+                return None
+
+        return None
+
+    @classmethod
+    def get_client(cls, max_retries=5, retry_interval=1):
+        """Get Redis client optimized for non-binary data (decoded responses)"""
+        if cls._client is None:
+            cls._client = cls._init_client(decode_responses=True, max_retries=max_retries, retry_interval=retry_interval)
         return cls._client
 
+    @classmethod
+    def get_buffer(cls, max_retries=5, retry_interval=1):
+        """Get Redis client optimized for binary data (no decoding)"""
+        if cls._buffer is None:
+            cls._buffer = cls._init_client(decode_responses=False, max_retries=max_retries, retry_interval=retry_interval)
+        return cls._buffer
+
     @classmethod
     def get_pubsub_client(cls, max_retries=5, retry_interval=1):
         """Get Redis client optimized for PubSub operations"""
@@ -183,6 +190,7 @@ class RedisClient:
                         socket_keepalive=socket_keepalive,
                         health_check_interval=health_check_interval,
                         retry_on_timeout=retry_on_timeout,
+                        decode_responses=True,
                         **ssl_params
                     )
 

core/xtream_codes.py

@@ -62,7 +62,7 @@ class Client:
             url = f"{self.server_url}/{endpoint}"
             logger.debug(f"XC API Request: {url} with params: {params}")
 
-            response = self.session.get(url, params=params, timeout=30)
+            response = self.session.get(url, params=params, timeout=60)
             response.raise_for_status()
 
             # Check if response is empty

dispatcharr/persistent_lock.py

@@ -48,7 +48,7 @@ class PersistentLock:
         Returns True if the expiration was successfully extended.
         """
         current_value = self.redis_client.get(self.lock_key)
-        if current_value and current_value.decode("utf-8") == self.lock_token:
+        if current_value and current_value == self.lock_token:
             self.redis_client.expire(self.lock_key, self.lock_timeout)
             self.has_lock = False
             return True

dispatcharr/urls.py

@@ -6,7 +6,7 @@ from django.views.generic import TemplateView, RedirectView
 from .routing import websocket_urlpatterns
 from apps.output.views import xc_player_api, xc_panel_api, xc_get, xc_xmltv
 from apps.proxy.ts_proxy.views import stream_xc
-from apps.output.views import xc_movie_stream, xc_series_stream
+from apps.proxy.vod_proxy.views import stream_xc_movie, stream_xc_episode
 
 urlpatterns = [
     # API Routes
@@ -44,13 +44,13 @@ urlpatterns = [
     # XC VOD endpoints
     path(
         "movie/<str:username>/<str:password>/<str:stream_id>.<str:extension>",
-        xc_movie_stream,
-        name="xc_movie_stream",
+        stream_xc_movie,
+        name="stream_xc_movie",
     ),
     path(
         "series/<str:username>/<str:password>/<str:stream_id>.<str:extension>",
-        xc_series_stream,
-        name="xc_series_stream",
+        stream_xc_episode,
+        name="stream_xc_episode",
     ),
     # Admin
     path("admin", RedirectView.as_view(url="/admin/", permanent=True)),

docker/init/99-init-dev.sh

@@ -1,25 +1,33 @@
 #!/bin/bash
 
-echo "🚀 Development Mode - Setting up Frontend..."
+if [ ! -e "/tmp/init" ]; then
+    echo "🚀 Development Mode - Setting up Frontend..."
 
-# Install Node.js
-if ! command -v node 2>&1 >/dev/null
-then
-    echo "=== setting up nodejs ==="
-    curl -sL https://deb.nodesource.com/setup_23.x -o /tmp/nodesource_setup.sh
-    bash /tmp/nodesource_setup.sh
-    apt-get update
-    apt-get install -y --no-install-recommends \
-        nodejs
-fi
-
-# Install frontend dependencies
-cd /app/frontend && npm install
-# Install Python dependencies using UV
-cd /app && uv sync --python $UV_PROJECT_ENVIRONMENT/bin/python --no-install-project --no-dev
-
-# Install debugpy for remote debugging
-if [ "$DISPATCHARR_DEBUG" = "true" ]; then
-    echo "=== setting up debugpy ==="
-    uv pip install --python $UV_PROJECT_ENVIRONMENT/bin/python debugpy
+    # Install Node.js
+    if ! command -v node 2>&1 >/dev/null
+    then
+        echo "=== setting up nodejs ==="
+        curl -sL https://deb.nodesource.com/setup_23.x -o /tmp/nodesource_setup.sh
+        bash /tmp/nodesource_setup.sh
+        apt-get update
+        apt-get install -y --no-install-recommends \
+            nodejs
+    fi
+
+    # Install frontend dependencies
+    cd /app/frontend && npm install
+    # Install Python dependencies using UV
+    cd /app && uv sync --python $UV_PROJECT_ENVIRONMENT/bin/python --no-install-project --no-dev
+
+    # Install debugpy for remote debugging
+    if [ "$DISPATCHARR_DEBUG" = "true" ]; then
+        echo "=== setting up debugpy ==="
+        uv pip install --python $UV_PROJECT_ENVIRONMENT/bin/python debugpy
+    fi
+
+    if [[ "$DISPATCHARR_ENV" = "dev" ]]; then
+        touch /tmp/init
+    fi
+else
+    echo "Development mode initialization already done. Skipping dev setup."
 fi

docker/uwsgi.dev.ini

@@ -8,7 +8,7 @@
 exec-pre = python /app/scripts/wait_for_redis.py
 
 ; Start Redis first
-attach-daemon = redis-server
+attach-daemon = redis-server --protected-mode no
 ; Then start other services with configurable nice level (default: 5 for low priority)
 ; Users can override via CELERY_NICE_LEVEL environment variable in docker-compose
 attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr worker --autoscale=6,1
@@ -57,4 +57,4 @@ logformat-strftime = true
 log-date = %%Y-%%m-%%d %%H:%%M:%%S,000
 # Use formatted time with environment variable for log level
 log-format = %(ftime) $(DISPATCHARR_LOG_LEVEL) uwsgi.requests Worker ID: %(wid) %(method) %(status) %(uri) %(msecs)ms
-log-buffering = 1024  # Add buffer size limit for logging
+log-buffering = 1024  # Add buffer size limit for logging

frontend/src/api.js

@@ -2998,12 +2998,15 @@ export default class API {
     }
   }
 
-  static async updateUser(id, body) {
+  static async updateUser(id, body, self = false) {
     try {
-      const response = await request(`${host}/api/accounts/users/${id}/`, {
-        method: 'PATCH',
-        body,
-      });
+      const response = await request(
+        `${host}/api/accounts/users/${self ? 'me' : id}/`,
+        {
+          method: 'PATCH',
+          body,
+        }
+      );
 
       useUsersStore.getState().updateUser(response);
 

frontend/src/components/forms/User.jsx

@@ -17,6 +17,7 @@ import {
   Tooltip,
   Grid,
   SimpleGrid,
+  NumberInput,
   useMantineTheme,
 } from '@mantine/core';
 import { RotateCcwKey, RotateCw, X } from 'lucide-react';
@@ -48,6 +49,7 @@ const User = ({ user = null, isOpen, onClose }) => {
       last_name: '',
       email: '',
       user_level: '0',
+      stream_limit: 0,
       password: '',
       xc_password: '',
       channel_profiles: [],
@@ -117,7 +119,11 @@ const User = ({ user = null, isOpen, onClose }) => {
         delete values.password;
       }
 
-      const response = await API.updateUser(user.id, values);
+      const response = await API.updateUser(
+        user.id,
+        values,
+        isAdmin ? false : authUser.id === user.id
+      );
 
       if (user.id == authUser.id) {
         setUser(response);
@@ -139,6 +145,7 @@ const User = ({ user = null, isOpen, onClose }) => {
         last_name: user.last_name || '',
         email: user.email,
         user_level: `${user.user_level}`,
+        stream_limit: user.stream_limit || 0,
         channel_profiles:
           user.channel_profiles.length > 0
             ? user.channel_profiles.map((id) => `${id}`)
@@ -236,6 +243,7 @@ const User = ({ user = null, isOpen, onClose }) => {
               id="username"
               name="username"
               label="Username"
+              disabled={!isAdmin}
               {...form.getInputProps('username')}
               key={form.key('username')}
             />
@@ -257,19 +265,26 @@ const User = ({ user = null, isOpen, onClose }) => {
             />
 
             {showPermissions && (
-              <Select
-                label="User Level"
-                data={Object.entries(USER_LEVELS).map(([, value]) => {
-                  return {
-                    label: USER_LEVEL_LABELS[value],
-                    value: `${value}`,
-                  };
-                })}
-                {...form.getInputProps('user_level')}
-                key={form.key('user_level')}
-              />
-            )}
+              <>
+                <Select
+                  label="User Level"
+                  data={Object.entries(USER_LEVELS).map(([, value]) => {
+                    return {
+                      label: USER_LEVEL_LABELS[value],
+                      value: `${value}`,
+                    };
+                  })}
+                  {...form.getInputProps('user_level')}
+                  key={form.key('user_level')}
+                />
 
+                <NumberInput
+                  label="Stream Limit (0 = unlimited)"
+                  {...form.getInputProps('stream_limit')}
+                  key={form.key('stream_limit')}
+                />
+              </>
+            )}
           </Stack>
 
           <Stack gap="xs" style={{ flex: 1 }}>
@@ -289,26 +304,23 @@ const User = ({ user = null, isOpen, onClose }) => {
               key={form.key('last_name')}
             />
 
-            <Group align="flex-end">
-              <TextInput
-                label="XC Password"
-                description="Clear to disable XC API"
-                {...form.getInputProps('xc_password')}
-                key={form.key('xc_password')}
-                style={{ flex: 1 }}
-                rightSectionWidth={30}
-                rightSection={
-                  <ActionIcon
-                    variant="transparent"
-                    size="sm"
-                    color="white"
-                    onClick={generateXCPassword}
-                  >
-                    <RotateCcwKey />
-                  </ActionIcon>
-                }
-              />
-            </Group>
+            <TextInput
+              label="XC Password"
+              description="Clear to disable XC API"
+              {...form.getInputProps('xc_password')}
+              key={form.key('xc_password')}
+              rightSectionWidth={30}
+              rightSection={
+                <ActionIcon
+                  variant="transparent"
+                  size="sm"
+                  color="white"
+                  onClick={generateXCPassword}
+                >
+                  <RotateCcwKey />
+                </ActionIcon>
+              }
+            />
 
             {showPermissions && (
               <MultiSelect

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

@@ -0,0 +1,116 @@
+import useSettingsStore from '../../../store/settings.jsx';
+import React, { useEffect, useState } from 'react';
+import { useForm } from '@mantine/form';
+import { updateSetting } from '../../../utils/pages/SettingsUtils.js';
+import {
+  Alert,
+  Button,
+  Flex,
+  NumberInput,
+  Stack,
+  TextInput,
+  Checkbox,
+} from '@mantine/core';
+import { USER_LIMITS_OPTIONS } from '../../../constants.js';
+
+const USER_LIMIT_DEFAULTS = Object.keys(USER_LIMITS_OPTIONS).reduce(
+  (acc, key) => {
+    acc[key] = USER_LIMITS_OPTIONS[key].default;
+    return acc;
+  },
+  {}
+);
+
+const UserLimitsForm = React.memo(({ active }) => {
+  const settings = useSettingsStore((s) => s.settings);
+
+  const [saved, setSaved] = useState(false);
+
+  const userLimitSettingsForm = useForm({
+    mode: 'controlled',
+    initialValues: USER_LIMIT_DEFAULTS,
+  });
+
+  useEffect(() => {
+    if (!active) setSaved(false);
+  }, [active]);
+
+  useEffect(() => {
+    if (settings) {
+      if (settings['user_limit_settings']?.value) {
+        userLimitSettingsForm.setValues({
+          ...USER_LIMIT_DEFAULTS,
+          ...settings['user_limit_settings'].value,
+        });
+      }
+    }
+  }, [settings]);
+
+  const resetUserLimitsToDefaults = () => {
+    userLimitSettingsForm.setValues(USER_LIMIT_DEFAULTS);
+  };
+
+  const onUserLimitsSubmit = async () => {
+    setSaved(false);
+
+    try {
+      const result = await updateSetting({
+        ...settings['user_limit_settings'],
+        value: userLimitSettingsForm.getValues(),
+      });
+      if (result) {
+        setSaved(true);
+      }
+    } catch (error) {
+      console.error('Error saving user limit settings:', error);
+    }
+  };
+
+  return (
+    <form onSubmit={userLimitSettingsForm.onSubmit(onUserLimitsSubmit)}>
+      <Stack gap="sm">
+        {saved && (
+          <Alert
+            variant="light"
+            color="green"
+            title="Saved Successfully"
+          ></Alert>
+        )}
+
+        {Object.keys(USER_LIMITS_OPTIONS).reduce((acc, key) => {
+          const option = USER_LIMITS_OPTIONS[key];
+          acc.push(
+            <Checkbox
+              key={key}
+              label={option.label}
+              description={option.description}
+              {...userLimitSettingsForm.getInputProps(key, {
+                type: 'checkbox',
+              })}
+            />
+          );
+          return acc;
+        }, [])}
+
+        <Flex mih={50} gap="xs" justify="space-between" align="flex-end">
+          <Button
+            variant="subtle"
+            color="gray"
+            onClick={resetUserLimitsToDefaults}
+          >
+            Reset to Defaults
+          </Button>
+          <Button
+            type="submit"
+            disabled={userLimitSettingsForm.submitting}
+            variant="default"
+          >
+            Save
+          </Button>
+        </Flex>
+      </Stack>
+    </form>
+  );
+});
+
+export default UserLimitsForm;

frontend/src/constants.js

@@ -62,6 +62,33 @@ export const PROXY_SETTINGS_OPTIONS = {
   },
 };
 
+export const USER_LIMITS_OPTIONS = {
+  terminate_on_limit_exceeded: {
+    label: 'Terminate on Limit Exceeded',
+    description:
+      'Terminate a stream (based on below criteria) when the user exceeds the allowed limits',
+    default: true,
+  },
+  prioritize_single_client_channels: {
+    label: 'Prioritize Single Client Channels',
+    description:
+      'Prioritize terminating channels only a single client belonging to the user',
+    default: true,
+  },
+  ignore_same_channel_connections: {
+    label: 'Ignore Same-Channel Connections',
+    description:
+      'Multiple user connections to the same channel count as 1 connection toward user limits',
+    default: false,
+  },
+  terminate_oldest: {
+    label: 'Terminate Oldest',
+    description:
+      'Prioritize terminating the oldest stream when limits are exceeded. Setting to false prioritizes the newest stream.',
+    default: true,
+  },
+};
+
 export const M3U_FILTER_TYPES = [
   {
     label: 'Group',

frontend/src/pages/Settings.jsx

@@ -24,6 +24,9 @@ import useAuthStore from '../store/auth';
 import { USER_LEVELS } from '../constants';
 import UiSettingsForm from '../components/forms/settings/UiSettingsForm.jsx';
 import ErrorBoundary from '../components/ErrorBoundary.jsx';
+const UserLimitsForm = React.lazy(
+  () => import('../components/forms/settings/UserLimitsForm.jsx')
+);
 const NetworkAccessForm = React.lazy(
   () => import('../components/forms/settings/NetworkAccessForm.jsx')
 );
@@ -200,6 +203,19 @@ const SettingsPage = () => {
                   </ErrorBoundary>
                 </AccordionPanel>
               </AccordionItem>
+
+              <AccordionItem value="user-limits">
+                <AccordionControl>User Limits</AccordionControl>
+                <AccordionPanel>
+                  <ErrorBoundary>
+                    <Suspense fallback={<Loader />}>
+                      <UserLimitsForm
+                        active={accordianValue === 'user-limits'}
+                      />
+                    </Suspense>
+                  </ErrorBoundary>
+                </AccordionPanel>
+              </AccordionItem>
             </>
           )}
         </Accordion>

frontend/src/pages/__tests__/Settings.test.jsx

@@ -78,6 +78,13 @@ vi.mock('../../components/forms/settings/NavOrderForm', () => ({
     </div>
   ),
 }));
+vi.mock('../../components/forms/settings/UserLimitsForm', () => ({
+  default: ({ active }) => (
+    <div data-testid="user-limits-form">
+      UserLimitsForm {active ? 'active' : 'inactive'}
+    </div>
+  ),
+}));
 vi.mock('../../components/ErrorBoundary', () => ({
   default: ({ children }) => <div data-testid="error-boundary">{children}</div>,
 }));