From dba70b3c585e0f2dce6cdcf157045992e617b236 Mon Sep 17 00:00:00 2001 From: dekzter Date: Tue, 27 Jan 2026 15:23:07 -0500 Subject: [PATCH] outstanding commits, may revisit later --- .../migrations/0004_user_stream_priority.py | 18 +++ apps/channels/models.py | 153 +++++++++++++++++- apps/channels/tasks.py | 5 +- apps/proxy/ts_proxy/client_manager.py | 16 +- apps/proxy/ts_proxy/server.py | 6 - apps/proxy/ts_proxy/utils.py | 4 +- apps/proxy/ts_proxy/views.py | 26 +-- docker/init/02-postgres.sh | 2 - docker/init/03-init-dispatcharr.sh | 2 +- docker/init/99-init-dev.sh | 50 +++--- docker/uwsgi.dev.ini | 4 +- frontend/prettier.config.js | 4 +- frontend/src/components/forms/User.jsx | 40 +++-- frontend/src/pages/Users.jsx | 35 ---- 14 files changed, 242 insertions(+), 123 deletions(-) create mode 100644 apps/accounts/migrations/0004_user_stream_priority.py diff --git a/apps/accounts/migrations/0004_user_stream_priority.py b/apps/accounts/migrations/0004_user_stream_priority.py new file mode 100644 index 00000000..35c8bbae --- /dev/null +++ b/apps/accounts/migrations/0004_user_stream_priority.py @@ -0,0 +1,18 @@ +# Generated by Django 5.2.4 on 2025-10-21 19:32 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('accounts', '0003_alter_user_custom_properties'), + ] + + operations = [ + migrations.AddField( + model_name='user', + name='stream_priority', + field=models.IntegerField(default=0, help_text='Priority level for streaming tasks. Lower values indicate higher priority.'), + ), + ] diff --git a/apps/channels/models.py b/apps/channels/models.py index 3dfb392b..7282d025 100644 --- a/apps/channels/models.py +++ b/apps/channels/models.py @@ -166,7 +166,7 @@ class Stream(models.Model): return stream_profile - 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. """ @@ -349,7 +349,145 @@ class Channel(models.Model): return stream_profile - def get_stream(self): + 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 get_stream(self, requester=None): """ Finds an available stream for the requested channel and returns the selected stream and profile. @@ -438,6 +576,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( diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index ffb2b97a..b63577d2 100755 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -1900,11 +1900,8 @@ 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) - def _d(bkey, cast=str): - v = _gv(bkey) + v = md.get(key) try: if v is None: return None diff --git a/apps/proxy/ts_proxy/client_manager.py b/apps/proxy/ts_proxy/client_manager.py index bffecdde..80be26f1 100644 --- a/apps/proxy/ts_proxy/client_manager.py +++ b/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 @@ -123,7 +121,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: @@ -224,7 +222,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") @@ -242,7 +240,8 @@ 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_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: @@ -303,8 +302,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) @@ -317,11 +314,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) diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index 9773e487..7d4a9f89 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -1121,12 +1121,6 @@ class ProxyServer: else: # Grace period expired with clients - mark channel as active logger.info(f"Grace period expired with {total_clients} clients - marking channel {channel_id} as active") -<<<<<<< HEAD - old_state = "unknown" - if metadata and 'state' in metadata: - old_state = metadata['state'] -======= ->>>>>>> origin/dev if self.update_channel_state(channel_id, ChannelState.ACTIVE, { "grace_period_ended_at": str(time.time()), "clients_at_activation": str(total_clients) diff --git a/apps/proxy/ts_proxy/utils.py b/apps/proxy/ts_proxy/utils.py index 20a6e140..ff9e1780 100644 --- a/apps/proxy/ts_proxy/utils.py +++ b/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) \ No newline at end of file + return logging.getLogger(logger_name) diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index 91f254a7..0f9c1228 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -45,7 +45,7 @@ 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) @@ -80,9 +80,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 [ @@ -120,7 +120,7 @@ def stream_ts(request, channel_id): else: owner_field = ChannelMetadataField.OWNER.encode("utf-8") 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 @@ -373,7 +373,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}" ) @@ -449,12 +449,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" ) @@ -490,7 +490,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 @@ -553,7 +553,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 @@ -681,7 +681,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) @@ -802,7 +802,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}" ) @@ -812,7 +812,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}" ) diff --git a/docker/init/02-postgres.sh b/docker/init/02-postgres.sh index e36dd744..233a5fb8 100644 --- a/docker/init/02-postgres.sh +++ b/docker/init/02-postgres.sh @@ -162,5 +162,3 @@ ensure_utf8_encoding() { echo "Database $POSTGRES_DB converted to UTF8 and permissions set." fi } - - diff --git a/docker/init/03-init-dispatcharr.sh b/docker/init/03-init-dispatcharr.sh index 5fbef23d..bfe081e2 100644 --- a/docker/init/03-init-dispatcharr.sh +++ b/docker/init/03-init-dispatcharr.sh @@ -64,4 +64,4 @@ if [ "$(id -u)" = "0" ]; then fi chmod +x /data -fi \ No newline at end of file +fi diff --git a/docker/init/99-init-dev.sh b/docker/init/99-init-dev.sh index 861c8307..fc8ef18a 100644 --- a/docker/init/99-init-dev.sh +++ b/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 pip dependencies -cd /app && pip install -r requirements.txt - -# Install debugpy for remote debugging -if [ "$DISPATCHARR_DEBUG" = "true" ]; then - echo "=== setting up debugpy ===" - pip install 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 pip dependencies + cd /app && pip install -r requirements.txt + + # Install debugpy for remote debugging + if [ "$DISPATCHARR_DEBUG" = "true" ]; then + echo "=== setting up debugpy ===" + pip install debugpy + fi + + if [[ "$DISPATCHARR_ENV" = "dev" ]]; then + touch /tmp/init + fi +else + echo "Development mode initialization already done. Skipping dev setup." fi diff --git a/docker/uwsgi.dev.ini b/docker/uwsgi.dev.ini index e476e216..e6d0806d 100644 --- a/docker/uwsgi.dev.ini +++ b/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 @@ -59,4 +59,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 \ No newline at end of file +log-buffering = 1024 # Add buffer size limit for logging diff --git a/frontend/prettier.config.js b/frontend/prettier.config.js index d2b86878..e143dfef 100644 --- a/frontend/prettier.config.js +++ b/frontend/prettier.config.js @@ -3,8 +3,8 @@ export default { semi: true, // Add semicolons at the end of statements singleQuote: true, // Use single quotes instead of double tabWidth: 2, // Set the indentation width - trailingComma: "es5", // Add trailing commas where valid in ES5 + trailingComma: 'es5', // Add trailing commas where valid in ES5 printWidth: 80, // Wrap lines at 80 characters bracketSpacing: true, // Add spaces inside object braces - arrowParens: "always", // Always include parentheses around arrow function parameters + arrowParens: 'always', // Always include parentheses around arrow function parameters }; diff --git a/frontend/src/components/forms/User.jsx b/frontend/src/components/forms/User.jsx index 619b156f..6ec8705e 100644 --- a/frontend/src/components/forms/User.jsx +++ b/frontend/src/components/forms/User.jsx @@ -12,8 +12,9 @@ import { Stack, MultiSelect, ActionIcon, + NumberInput, } from '@mantine/core'; -import { RotateCcwKey, X } from 'lucide-react'; +import { RotateCcwKey } from 'lucide-react'; import { useForm } from '@mantine/form'; import useChannelsStore from '../../store/channels'; import { USER_LEVELS, USER_LEVEL_LABELS } from '../../constants'; @@ -209,26 +210,23 @@ const User = ({ user = null, isOpen, onClose }) => { key={form.key('last_name')} /> - - - - - } - /> - + + + + } + /> {showPermissions && ( { const authUser = useAuthStore((s) => s.user); - const [selectedUser, setSelectedUser] = useState(null); - const [userModalOpen, setUserModalOpen] = useState(false); - const [confirmDeleteOpen, setConfirmDeleteOpen] = useState(false); - const [deleteTarget, setDeleteTarget] = useState(null); - const [userToDelete, setUserToDelete] = useState(null); - if (!authUser.id) { return <>; } - const closeUserModal = () => { - setSelectedUser(null); - setUserModalOpen(false); - }; - const editUser = (user) => { - setSelectedUser(user); - setUserModalOpen(true); - }; - - const deleteUser = (id) => { - // Get user details for the confirmation dialog - const user = users.find((u) => u.id === id); - setUserToDelete(user); - setDeleteTarget(id); - - // Skip warning if it's been suppressed - if (isWarningSuppressed('delete-user')) { - return executeDeleteUser(id); - } - - setConfirmDeleteOpen(true); - }; - - const executeDeleteUser = async (id) => { - await API.deleteUser(id); - setConfirmDeleteOpen(false); - }; - return (