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 (