Mirrors/Dispatcharr
1
0
Fork 1
mirror of https://github.com/Dispatcharr/Dispatcharr.git synced 2026-07-18 00:55:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Merge pull request #1254 from Goldenfreddy0703/feature/1137-credential-pools

Browse source 
Enforce shared credential connection pools across M3U accounts (#1137)
This commit is contained in:
SergeantPanda 2026-06-09 18:48:14 -05:00 • committed by GitHub
commit d3ecefb7d6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
21 changed files with 2263 additions and 367 deletions
Download patch file Download diff file Expand all files Collapse all files

14
CHANGELOG.md
View file

@ -7,8 +7,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
### Added
- **Manual Server Groups for shared M3U connection limits.** The existing `ServerGroup` model and account FK are now wired into live and VOD playback. Accounts assigned to the same group share a credential-scoped Redis counter when their provider logins match (hashed fingerprint); unrelated logins in the same group keep separate counters. Enforcement uses each profile's `max_streams` - not a group-wide cap - and profiles with `max_streams=0` skip credential pooling for that profile while still rotating on their own per-profile counter. New `apps/m3u/connection_pool.py` centralizes reserve/release, profile rotation, and credential moves on profile switch. (Closes #1137) — Thanks [@Goldenfreddy0703](https://github.com/Goldenfreddy0703)
- **Server Groups manager UI** on the M3U Accounts page: create, rename, and delete groups with account counts and delete confirmation. Groups load on login via a new Zustand store and REST helpers (`/api/m3u/server-groups/`).
- **M3U account form** adds a Server Group picker (including inline “add group”), reorganized into three columns (source/auth, connection limits, sync/content), and a Manage server groups shortcut.
- **VOD profile selection** uses `pool_has_capacity_for_profile()` so grouped accounts respect shared credential limits before a profile is chosen (non-grouped accounts behave as before).
- **Live profile switches** move the shared credential counter when the new profile uses a different provider login; same-login switches leave the credential counter unchanged.
- **Credential release keys** stored at reserve time allow counters to be released even if the M3U profile row is deleted afterward.
### Changed
- **Live and VOD connection slot logic now routes through `connection_pool`.** `Channel.get_stream()`, direct `Stream.get_stream()`, VOD profile selection, and the multi-worker VOD connection manager all call `reserve_profile_slot()` / `release_profile_slot()` instead of inline Redis INCR/DECR. VOD profile selection also checks `pool_has_capacity_for_profile()` before choosing a profile.
- **Live XC upstream URLs use current credentials.** `_resolve_live_stream_url()` builds `/live/{user}/{pass}/{stream_id}.ts` from transformed account credentials and the stream's provider `stream_id`, so playback stays aligned with the active login after credential or profile changes instead of reusing a stale `stream.url` from sync.
- **Channel stream switches check pooled capacity.** `get_stream_info_for_switch()` uses `profile_available_for_channel_switch()` when targeting a specific stream, and releases a reserved slot if URL assembly fails after `get_stream()` allocated one.
- **Live proxy init reads Redis assignment once.** After `generate_stream_url()` reserves a slot, the stream handler reads `channel_stream` / `stream_profile` from Redis instead of calling `get_stream()` again, avoiding double INCR under concurrent release.
- **EPG auto-match overhaul** — matching logic moved to `apps/channels/epg_matching.py`; Celery tasks in `tasks.py` are thin wrappers.
- Single-channel auto-match is now asynchronous: the API returns `202 Accepted` and pushes the result over WebSocket (`single_channel_epg_match`), so large EPG libraries no longer hit the previous 30-second HTTP timeout.
- Progress, bulk completion, and single-channel results use `send_websocket_update` instead of `async_to_sync(channel_layer.group_send)`, so notifications work reliably under gevent-patched uWSGI and Celery workers.
@ -27,6 +40,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Fixed
- **Stale `channel_stream` Redis keys after a channel stopped could skip connection accounting on retune.** On `dev`, `get_stream()` reused any existing `channel_stream` / `stream_profile` assignment without reserving a new slot. If those keys were left behind after a stop, the next tune-in could reach the provider without incrementing Redis counters. `get_stream()` now releases stale assignments when proxy metadata shows the channel is inactive, then reserves fresh slots.
- **EPG auto-match reliability fixes.**
- Memory could spike to multiple GB on large EPG sources when building a full in-memory catalog before fuzzy matching; single-channel matching now streams rows and bounds ML work to a small candidate set.
- Wrong channel assignments from global ML similarity; ML validation now checks the fuzzy best match (or top fuzzy candidates as a last resort) instead of scoring the entire catalog.

216
apps/channels/models.py
View file

@ -4,7 +4,7 @@ from django.conf import settings
from core.models import StreamProfile, CoreSettings
from core.utils import RedisClient
from apps.proxy.live_proxy.redis_keys import RedisKeys
from apps.proxy.live_proxy.constants import ChannelMetadataField
from apps.proxy.live_proxy.constants import ChannelMetadataField, ChannelState
import logging
import uuid
from django.utils import timezone
@ -17,6 +17,7 @@ logger = logging.getLogger(__name__)
# If you have an M3UAccount model in apps.m3u, you can still import it:
from apps.m3u.models import M3UAccount
from apps.m3u.connection_pool import reserve_profile_slot, release_profile_slot
# Add fallback functions if Redis isn't available
@ -209,13 +210,14 @@ class Stream(models.Model):
Finds an available profile for this stream and reserves a connection slot.
Returns:
Tuple[Optional[int], Optional[int], Optional[str]]: (stream_id, profile_id, error_reason)
Tuple[Optional[int], Optional[int], Optional[str], bool]:
(stream_id, profile_id, error_reason, slot_reserved)
"""
redis_client = RedisClient.get_client()
profile_id = redis_client.get(f"stream_profile:{self.id}")
if profile_id:
profile_id = int(profile_id)
return self.id, profile_id, None
return self.id, profile_id, None, False
# Retrieve the M3U account associated with the stream.
m3u_account = self.m3u_account
@ -226,29 +228,22 @@ class Stream(models.Model):
]
for profile in profiles:
logger.info(profile)
logger.debug("Evaluating profile %s for stream %s", profile.id, self.id)
# Skip inactive profiles
if profile.is_active == False:
continue
# Atomic slot reservation: INCR first, check, rollback if over capacity
if profile.max_streams == 0:
reserved = True
else:
profile_connections_key = f"profile_connections:{profile.id}"
new_count = redis_client.incr(profile_connections_key)
if new_count <= profile.max_streams:
reserved = True
else:
redis_client.decr(profile_connections_key)
reserved = False
# Atomic slot reservation via shared connection pool helper
reserved, _count, _failure_reason = reserve_profile_slot(
profile, redis_client
)
if reserved:
redis_client.set(f"channel_stream:{self.id}", self.id)
redis_client.set(f"stream_profile:{self.id}", profile.id)
return self.id, profile.id, None
return self.id, profile.id, None, True
return None, None, "All active M3U profiles have reached maximum connection limits"
return None, None, "All active M3U profiles have reached maximum connection limits", False
def release_stream(self):
"""
@ -277,12 +272,7 @@ class Stream(models.Model):
f"Stream {stream_id}: found profile_id={profile_id}"
)
profile_connections_key = f"profile_connections:{profile_id}"
# Only decrement if the profile had a max_connections limit
current_count = int(redis_client.get(profile_connections_key) or 0)
if current_count > 0:
redis_client.decr(profile_connections_key)
release_profile_slot(profile_id, redis_client)
return True
@ -599,44 +589,65 @@ class Channel(models.Model):
return victim_id
def _check_and_reserve_profile_slot(self, profile, redis_client):
def _channel_proxy_is_active(self, redis_client) -> bool:
"""True when live proxy metadata shows this channel is still running."""
metadata_key = RedisKeys.channel_metadata(str(self.uuid))
if not redis_client.exists(metadata_key):
return False
state = redis_client.hget(metadata_key, ChannelMetadataField.STATE)
if state is None:
return False
if isinstance(state, bytes):
state = state.decode()
return state in (
ChannelState.ACTIVE,
ChannelState.WAITING_FOR_CLIENTS,
ChannelState.BUFFERING,
ChannelState.INITIALIZING,
ChannelState.CONNECTING,
)
def _stream_assignment_is_reusable(self, redis_client, stream_id: int) -> bool:
"""
Atomically check and reserve a connection slot for the given profile.
Return True when an existing channel_stream assignment should be reused.
Uses an INCR-first-then-check pattern to eliminate the TOCTOU race
condition where separate GET + check + INCR operations could allow
concurrent requests to both pass the capacity check.
For profiles with max_streams=0 (unlimited), no reservation is needed.
Args:
profile: M3UAccountProfile instance
redis_client: Redis client instance
Returns:
tuple: (reserved: bool, current_count: int)
Reuse when the proxy is active, or when metadata is not written yet
(between get_stream() reserving slots and initialize_channel() starting).
When metadata exists but the proxy is inactive, the assignment is stale.
"""
if profile.max_streams == 0:
return (True, 0)
if self._channel_proxy_is_active(redis_client):
return True
profile_connections_key = f"profile_connections:{profile.id}"
metadata_key = RedisKeys.channel_metadata(str(self.uuid))
if not redis_client.exists(metadata_key):
return redis_client.get(f"stream_profile:{stream_id}") is not None
# Atomically increment first — this is a single Redis command
new_count = redis_client.incr(profile_connections_key)
return False
if new_count <= profile.max_streams:
return (True, new_count)
def _release_stale_stream_assignment(self, redis_client, stream_id: int) -> None:
"""Release pool counters and remove stale channel/stream assignment keys."""
profile_id_bytes = redis_client.get(f"stream_profile:{stream_id}")
if profile_id_bytes:
try:
profile_id = int(profile_id_bytes)
release_profile_slot(profile_id, redis_client)
except (ValueError, TypeError):
logger.debug(
"Invalid profile ID for stale assignment on stream %s: %s",
stream_id,
profile_id_bytes,
)
# Over capacity — roll back the increment
redis_client.decr(profile_connections_key)
return (False, new_count - 1)
redis_client.delete(f"channel_stream:{self.id}")
redis_client.delete(f"stream_profile:{stream_id}")
def get_stream(self, requester=None):
"""
Finds an available stream for the requested channel and returns the selected stream and profile.
Returns:
Tuple[Optional[int], Optional[int], Optional[str]]: (stream_id, profile_id, error_reason)
Tuple[Optional[int], Optional[int], Optional[str], bool]:
(stream_id, profile_id, error_reason, slot_reserved)
"""
redis_client = RedisClient.get_client()
error_reason = None
@ -644,26 +655,42 @@ class Channel(models.Model):
# Check if this channel has any streams
if not self.streams.exists():
error_reason = "No streams assigned to channel"
return None, None, error_reason
return None, None, error_reason, False
# Check if a stream is already active for this channel
# Reuse assignment only when this channel is still active in the proxy.
# Stale channel_stream keys after stop/disconnect skip INCR and break pool
# accounting, which lets a second stream reach the provider and fail validation.
stream_id_bytes = redis_client.get(f"channel_stream:{self.id}")
if stream_id_bytes:
try:
stream_id = int(stream_id_bytes)
profile_id_bytes = redis_client.get(f"stream_profile:{stream_id}")
if profile_id_bytes:
try:
profile_id = int(profile_id_bytes)
return stream_id, profile_id, None
except (ValueError, TypeError):
logger.debug(
f"Invalid profile ID retrieved from Redis: {profile_id_bytes}"
)
except (ValueError, TypeError):
logger.debug(
f"Invalid stream ID retrieved from Redis: {stream_id_bytes}"
)
stream_id = None
if stream_id is not None:
if self._stream_assignment_is_reusable(redis_client, stream_id):
profile_id_bytes = redis_client.get(f"stream_profile:{stream_id}")
if profile_id_bytes:
try:
profile_id = int(profile_id_bytes)
logger.debug(
f"Channel {self.uuid}: reusing stream assignment "
f"stream={stream_id} profile={profile_id}"
)
return stream_id, profile_id, None, False
except (ValueError, TypeError):
logger.debug(
f"Invalid profile ID retrieved from Redis: {profile_id_bytes}"
)
else:
logger.info(
f"Channel {self.uuid}: releasing stale stream assignment "
f"(stream={stream_id}, proxy not active)"
)
self._release_stale_stream_assignment(redis_client, stream_id)
# No existing active stream, attempt to assign a new one
has_streams_but_maxed_out = False
@ -697,7 +724,7 @@ class Channel(models.Model):
has_active_profiles = True
# Atomically check and reserve a slot (INCR-first pattern)
reserved, current_count = self._check_and_reserve_profile_slot(
reserved, current_count, failure_reason = reserve_profile_slot(
profile, redis_client
)
@ -705,11 +732,16 @@ class Channel(models.Model):
# Slot reserved — assign stream to this channel
redis_client.set(f"channel_stream:{self.id}", stream.id)
redis_client.set(f"stream_profile:{stream.id}", profile.id)
logger.info(
f"Channel {self.uuid}: assigned stream {stream.id} "
f"profile {profile.id} ({profile.name})"
)
return (
stream.id,
profile.id,
None,
True,
) # Return newly assigned stream and matched profile
else:
# At capacity: try to preempt a lower-impact channel on this profile
@ -723,12 +755,21 @@ class Channel(models.Model):
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(
f"Profile {profile.id} at max connections: "
f"{current_count}/{profile.max_streams}"
)
if failure_reason == "profile_full":
logger.info(
f"Profile {profile.id} at max connections: "
f"{current_count}/{profile.max_streams}, trying next profile"
)
elif failure_reason == "credential_full":
logger.info(
f"Profile {profile.id} shared login pool full, trying next profile"
)
else:
logger.debug(
f"Profile {profile.id} reservation failed: "
f"{current_count}/{profile.max_streams}"
)
# No available streams - determine specific reason
if has_streams_but_maxed_out:
@ -738,7 +779,7 @@ class Channel(models.Model):
else:
error_reason = "No active profiles found for any assigned stream"
return None, None, error_reason
return None, None, error_reason, False
def release_stream(self):
"""
@ -782,12 +823,7 @@ class Channel(models.Model):
ChannelMetadataField.M3U_PROFILE,
)
profile_connections_key = f"profile_connections:{profile_id}"
current_count = int(
redis_client.get(profile_connections_key) or 0
)
if current_count > 0:
redis_client.decr(profile_connections_key)
release_profile_slot(profile_id, redis_client)
return True
logger.debug(
@ -832,12 +868,7 @@ class Channel(models.Model):
f"stream {stream_id}"
)
profile_connections_key = f"profile_connections:{profile_id}"
# Only decrement if the profile had a max_connections limit
current_count = int(redis_client.get(profile_connections_key) or 0)
if current_count > 0:
redis_client.decr(profile_connections_key)
release_profile_slot(profile_id, redis_client)
# Clear metadata fields so duplicate release_stream() calls
# (e.g. from _clean_redis_keys or ChannelService.stop_channel)
@ -883,10 +914,31 @@ class Channel(models.Model):
if current_profile_id == new_profile_id:
return True
# Use pipeline for atomic profile switch to prevent counter drift
# if an exception occurs between DECR and INCR
old_profile_connections_key = f"profile_connections:{current_profile_id}"
new_profile_connections_key = f"profile_connections:{new_profile_id}"
from apps.m3u.connection_pool import (
move_credential_slot_on_profile_switch,
profile_connections_key,
)
from apps.m3u.models import M3UAccountProfile
old_profile = M3UAccountProfile.objects.select_related(
"m3u_account__server_group"
).get(id=current_profile_id)
new_profile = M3UAccountProfile.objects.select_related(
"m3u_account__server_group"
).get(id=new_profile_id)
if not move_credential_slot_on_profile_switch(
old_profile, new_profile, redis_client
):
logger.warning(
"Shared login pool full for profile %s during stream profile switch",
new_profile_id,
)
return False
# Profile counters always move on switch; credential totals move only when login changes.
old_profile_connections_key = profile_connections_key(current_profile_id)
new_profile_connections_key = profile_connections_key(new_profile_id)
old_count = int(redis_client.get(old_profile_connections_key) or 0)
pipe = redis_client.pipeline()

173
apps/channels/tests/test_get_stream_assignment.py Normal file
View file

@ -0,0 +1,173 @@
"""Tests for Channel.get_stream() assignment reuse and stale cleanup."""
from unittest.mock import patch
from django.test import TestCase
from apps.channels.models import Channel, ChannelStream, Stream
from apps.m3u.models import M3UAccount, M3UAccountProfile
from apps.proxy.live_proxy.constants import ChannelMetadataField, ChannelState
from apps.proxy.live_proxy.redis_keys import RedisKeys
class FakeAssignmentRedis:
"""In-memory Redis for channel_stream assignment tests."""
def __init__(self):
self._strings = {}
self._hashes = {}
def _decode(self, value):
if isinstance(value, bytes):
return value.decode()
return value
def get(self, key):
value = self._strings.get(key)
if value is None:
return None
if isinstance(value, int):
return str(value).encode()
return str(value).encode()
def set(self, key, value):
self._strings[key] = value
def delete(self, key):
self._strings.pop(key, None)
self._hashes.pop(key, None)
def exists(self, key):
return key in self._strings or key in self._hashes
def hget(self, key, field):
return self._hashes.get(key, {}).get(field)
def hset(self, key, mapping=None, **kwargs):
bucket = self._hashes.setdefault(key, {})
if mapping:
bucket.update(mapping)
bucket.update(kwargs)
def incr(self, key):
current = int(self._decode(self.get(key)) or 0)
current += 1
self._strings[key] = current
return current
def decr(self, key):
current = int(self._decode(self.get(key)) or 0)
current -= 1
self._strings[key] = current
return current
class ChannelGetStreamAssignmentTests(TestCase):
def setUp(self):
self.redis = FakeAssignmentRedis()
self.account = M3UAccount.objects.create(
name="assignment-test",
account_type="XC",
username="user",
password="pass",
max_streams=5,
)
self.profile = M3UAccountProfile.objects.get(
m3u_account=self.account, is_default=True
)
self.profile.max_streams = 2
self.profile.save()
self.stream = Stream.objects.create(
name="Test Stream",
url="http://example.com/live/user/pass/1.ts",
m3u_account=self.account,
)
self.channel = Channel.objects.create(channel_number=501, name="Assignment Ch")
ChannelStream.objects.create(channel=self.channel, stream=self.stream, order=0)
self.metadata_key = RedisKeys.channel_metadata(str(self.channel.uuid))
def _seed_assignment(self):
self.redis.set(f"channel_stream:{self.channel.id}", self.stream.id)
self.redis.set(f"stream_profile:{self.stream.id}", self.profile.id)
@patch("apps.channels.models.RedisClient.get_client")
@patch("apps.channels.models.reserve_profile_slot")
def test_reuses_assignment_when_proxy_active(
self, mock_reserve, mock_get_client
):
mock_get_client.return_value = self.redis
self._seed_assignment()
self.redis.hset(
self.metadata_key,
{ChannelMetadataField.STATE: ChannelState.ACTIVE},
)
stream_id, profile_id, error, slot_reserved = self.channel.get_stream()
self.assertEqual(stream_id, self.stream.id)
self.assertEqual(profile_id, self.profile.id)
self.assertIsNone(error)
self.assertFalse(slot_reserved)
mock_reserve.assert_not_called()
@patch("apps.channels.models.RedisClient.get_client")
@patch("apps.channels.models.reserve_profile_slot")
def test_reuses_assignment_during_init_before_metadata(
self, mock_reserve, mock_get_client
):
mock_get_client.return_value = self.redis
self._seed_assignment()
stream_id, profile_id, error, slot_reserved = self.channel.get_stream()
self.assertEqual(stream_id, self.stream.id)
self.assertEqual(profile_id, self.profile.id)
self.assertIsNone(error)
self.assertFalse(slot_reserved)
mock_reserve.assert_not_called()
@patch("apps.channels.models.RedisClient.get_client")
@patch("apps.channels.models.release_profile_slot")
@patch("apps.channels.models.reserve_profile_slot")
def test_releases_stale_assignment_when_proxy_stopped(
self, mock_reserve, mock_release, mock_get_client
):
mock_get_client.return_value = self.redis
mock_reserve.return_value = (True, 1, None)
self._seed_assignment()
self.redis.hset(
self.metadata_key,
{ChannelMetadataField.STATE: ChannelState.STOPPED},
)
stream_id, profile_id, error, slot_reserved = self.channel.get_stream()
mock_release.assert_called_once_with(self.profile.id, self.redis)
mock_reserve.assert_called_once()
self.assertEqual(stream_id, self.stream.id)
self.assertEqual(profile_id, self.profile.id)
self.assertTrue(slot_reserved)
@patch("apps.channels.models.RedisClient.get_client")
def test_stream_assignment_is_reusable_during_init_pending(self, mock_get_client):
mock_get_client.return_value = self.redis
self._seed_assignment()
self.assertTrue(
self.channel._stream_assignment_is_reusable(self.redis, self.stream.id)
)
@patch("apps.channels.models.RedisClient.get_client")
def test_stream_assignment_not_reusable_when_stopped(self, mock_get_client):
mock_get_client.return_value = self.redis
self._seed_assignment()
self.redis.hset(
self.metadata_key,
{ChannelMetadataField.STATE: ChannelState.STOPPED},
)
self.assertFalse(
self.channel._stream_assignment_is_reusable(self.redis, self.stream.id)
)

323
apps/m3u/connection_pool.py Normal file
View file

@ -0,0 +1,323 @@
"""
Shared connection pool enforcement for M3U accounts in the same ServerGroup.
Profile selection rotates across M3UAccountProfile rows using each profile's own
Redis counter (the pre-pool behavior). When an account belongs to a ServerGroup, a credential-scoped counter is checked on reserve/release
so accounts sharing the same provider login share one limit without blocking
unrelated logins on the same group. Account profiles with max_streams=0 skip
credential enforcement for that profile.
"""
from __future__ import annotations
import hashlib
import logging
import re
from typing import Literal, Optional, Tuple
logger = logging.getLogger(__name__)
ReserveFailureReason = Literal["profile_full", "credential_full"]
PROFILE_CONNECTIONS_KEY = "profile_connections:{profile_id}"
PROFILE_CREDENTIAL_RELEASE_KEY = "profile_credential_release:{profile_id}"
SERVER_GROUP_CONNECTIONS_KEY = "server_group_connections:{group_id}:{fingerprint}"
_XC_URL_CREDENTIALS_RE = re.compile(
r"/(?:live|movie|series)/([^/]+)/([^/]+)/",
re.IGNORECASE,
)
def profile_connections_key(profile_id: int) -> str:
return PROFILE_CONNECTIONS_KEY.format(profile_id=profile_id)
def profile_credential_release_key(profile_id: int) -> str:
"""Redis key storing the credential counter to release when the profile row is gone."""
return PROFILE_CREDENTIAL_RELEASE_KEY.format(profile_id=profile_id)
def server_group_connections_key(group_id: int, fingerprint: str) -> str:
"""Redis key for per-credential usage within a ServerGroup."""
return SERVER_GROUP_CONNECTIONS_KEY.format(
group_id=group_id,
fingerprint=fingerprint[:16],
)
def compute_credential_fingerprint(username: str, password: str) -> Optional[str]:
"""Return a stable hash for grouping accounts with the same IPTV login."""
if not username or not password:
return None
normalized = f"{username.strip().lower()}\0{password.strip()}"
return hashlib.sha256(normalized.encode("utf-8")).hexdigest()
def extract_credentials_from_stream_url(url: str) -> Tuple[Optional[str], Optional[str]]:
"""Parse username/password embedded in an Xtream-style stream URL."""
if not url:
return None, None
match = _XC_URL_CREDENTIALS_RE.search(url)
if not match:
return None, None
return match.group(1), match.group(2)
def _fingerprint_from_profile_stream_url(profile) -> Optional[str]:
"""STD/M3U: fingerprint from a sample stream URL after profile rewrite."""
from apps.channels.models import Stream
sample_url = (
Stream.objects.filter(m3u_account=profile.m3u_account)
.exclude(url="")
.values_list("url", flat=True)
.first()
)
if not sample_url:
return None
try:
from apps.proxy.live_proxy.url_utils import transform_url
transformed = transform_url(
sample_url,
profile.search_pattern or "",
profile.replace_pattern or "",
)
url_user, url_pass = extract_credentials_from_stream_url(
transformed or sample_url
)
return compute_credential_fingerprint(url_user or "", url_pass or "")
except Exception as exc:
logger.debug(
"Could not derive profile %s fingerprint from stream URL: %s",
profile.pk,
exc,
)
return None
def get_profile_credential_fingerprint(profile) -> Optional[str]:
"""Fingerprint for credentials this profile uses at playback time."""
m3u_account = profile.m3u_account
if m3u_account.account_type == "XC":
try:
from apps.m3u.tasks import get_transformed_credentials
_url, username, password = get_transformed_credentials(m3u_account, profile)
fingerprint = compute_credential_fingerprint(username or "", password or "")
if fingerprint:
return fingerprint
except Exception as exc:
logger.debug(
"Could not resolve transformed credentials for profile %s: %s",
profile.pk,
exc,
)
fingerprint = _fingerprint_from_profile_stream_url(profile)
if fingerprint:
return fingerprint
return compute_credential_fingerprint(
m3u_account.username or "",
m3u_account.password or "",
)
def get_enforced_server_group_for_profile(profile):
"""Return the ServerGroup for credential pooling when the account is assigned to one."""
group = profile.m3u_account.server_group
if group:
return group
return None
def _credential_counter_key(profile, group) -> Optional[str]:
fingerprint = get_profile_credential_fingerprint(profile)
if not fingerprint:
return None
return server_group_connections_key(group.id, fingerprint)
def get_profile_connection_count(profile, redis_client) -> int:
return int(redis_client.get(profile_connections_key(profile.id)) or 0)
def get_credential_connection_count(profile, redis_client) -> int:
group = get_enforced_server_group_for_profile(profile)
if not group:
return 0
cred_key = _credential_counter_key(profile, group)
if not cred_key:
return 0
return int(redis_client.get(cred_key) or 0)
def profile_has_capacity_for_selection(profile, redis_client) -> bool:
"""Per-profile capacity check used when rotating across profiles on one account."""
if profile.max_streams == 0:
return True
return get_profile_connection_count(profile, redis_client) < profile.max_streams
def group_has_capacity_for_profile(profile, redis_client) -> bool:
# Profiles with max_streams=0 skip credential enforcement entirely. An unlimited
# profile in a pooled group can still stream while other accounts share the login.
group = get_enforced_server_group_for_profile(profile)
if not group or profile.max_streams == 0:
return True
cred_key = _credential_counter_key(profile, group)
if not cred_key:
return True
return int(redis_client.get(cred_key) or 0) < profile.max_streams
def pool_has_capacity_for_profile(profile, redis_client) -> bool:
"""Non-mutating check before reserve: profile slot and credential slot if applicable."""
return profile_has_capacity_for_selection(profile, redis_client) and group_has_capacity_for_profile(
profile, redis_client
)
def profile_available_for_channel_switch(
profile, redis_client, *, channel_already_on_profile: bool
) -> bool:
"""
Non-mutating capacity check when selecting a profile for an in-flight channel.
If the channel already holds this profile's slots, skip re-checking capacity.
"""
if channel_already_on_profile:
return True
return pool_has_capacity_for_profile(profile, redis_client)
def move_credential_slot_on_profile_switch(
old_profile, new_profile, redis_client
) -> bool:
"""
Move the shared credential counter when switching to a different provider login.
Profile counters are managed separately by Channel.update_stream_profile().
Returns False when the new profile's credential pool is full.
"""
old_fp = get_profile_credential_fingerprint(old_profile)
new_fp = get_profile_credential_fingerprint(new_profile)
if old_fp == new_fp:
return True
_release_credential_slot_by_profile_id(old_profile.id, redis_client)
cred_reserved, cred_key = _reserve_server_group_slot_for_profile(
new_profile, redis_client
)
if not cred_reserved:
restore_reserved, restore_key = _reserve_server_group_slot_for_profile(
old_profile, redis_client
)
if restore_reserved and restore_key:
_remember_credential_release_key(
old_profile.id, restore_key, redis_client
)
return False
if cred_key:
_remember_credential_release_key(new_profile.id, cred_key, redis_client)
return True
def _safe_decr(redis_client, key: str) -> None:
current = int(redis_client.get(key) or 0)
if current <= 0:
return
new_count = redis_client.decr(key)
if new_count < 0:
redis_client.set(key, 0)
def _remember_credential_release_key(
profile_id: int, cred_key: str, redis_client
) -> None:
redis_client.set(profile_credential_release_key(profile_id), cred_key)
def _release_credential_slot_by_profile_id(profile_id: int, redis_client) -> bool:
"""Release a reserved credential counter using the key stored at reserve time."""
release_key = profile_credential_release_key(profile_id)
cred_key = redis_client.get(release_key)
if not cred_key:
return False
if isinstance(cred_key, bytes):
cred_key = cred_key.decode()
_safe_decr(redis_client, cred_key)
redis_client.delete(release_key)
return True
def _reserve_server_group_slot_for_profile(
profile, redis_client
) -> Tuple[bool, Optional[str]]:
group = get_enforced_server_group_for_profile(profile)
if not group or profile.max_streams == 0:
return True, None
cred_key = _credential_counter_key(profile, group)
if not cred_key:
return True, None
cred_count = redis_client.incr(cred_key)
if cred_count <= profile.max_streams:
return True, cred_key
redis_client.decr(cred_key)
return False, None
def reserve_profile_slot(
profile, redis_client
) -> Tuple[bool, int, Optional[ReserveFailureReason]]:
"""
Atomically reserve profile + optional credential slots (INCR-first).
Returns (reserved, profile_count_after_attempt, failure_reason).
failure_reason is set when reserved is False.
"""
profile_key = profile_connections_key(profile.id)
profile_count = 0
if profile.max_streams > 0:
profile_count = redis_client.incr(profile_key)
if profile_count > profile.max_streams:
redis_client.decr(profile_key)
return False, profile_count - 1, "profile_full"
cred_reserved, cred_key = _reserve_server_group_slot_for_profile(
profile, redis_client
)
if not cred_reserved:
if profile.max_streams > 0:
redis_client.decr(profile_key)
return (
False,
profile_count - 1 if profile.max_streams > 0 else 0,
"credential_full",
)
if cred_key:
_remember_credential_release_key(profile.id, cred_key, redis_client)
return True, profile_count, None
def release_profile_slot(profile_id: int, redis_client) -> None:
"""Release profile and shared credential slots after a stream end."""
_release_credential_slot_by_profile_id(profile_id, redis_client)
profile_key = profile_connections_key(profile_id)
current = int(redis_client.get(profile_key) or 0)
if current > 0:
redis_client.decr(profile_key)

8
apps/m3u/models.py
View file

@ -194,7 +194,13 @@ class M3UFilter(models.Model):
class ServerGroup(models.Model):
"""Represents a logical grouping of servers or channels."""
"""
Groups M3U accounts that share provider credentials.
Accounts assigned to the same server group share credential-scoped connection
counters when their logins match. Limits come from each account profile's
max_streams, not from the group itself.
"""
name = models.CharField(
max_length=100, unique=True, help_text="Unique name for this server group."

594
apps/m3u/tests/test_connection_pool.py Normal file
View file

@ -0,0 +1,594 @@
"""Tests for shared ServerGroup connection pools (#1137)."""
from django.test import TestCase
from unittest.mock import patch
from apps.m3u.connection_pool import (
extract_credentials_from_stream_url,
get_credential_connection_count,
get_enforced_server_group_for_profile,
get_profile_connection_count,
get_profile_credential_fingerprint,
group_has_capacity_for_profile,
pool_has_capacity_for_profile,
profile_has_capacity_for_selection,
profile_connections_key,
profile_credential_release_key,
release_profile_slot,
reserve_profile_slot,
server_group_connections_key,
)
from apps.m3u.models import M3UAccount, M3UAccountProfile, ServerGroup
class FakeRedis:
"""Minimal in-memory Redis stand-in for counter tests."""
def __init__(self):
self._data = {}
def get(self, key):
val = self._data.get(key)
if val is None:
return None
if isinstance(val, str):
return val.encode()
return str(val).encode()
def set(self, key, value, ex=None):
try:
self._data[key] = int(value)
except (ValueError, TypeError):
self._data[key] = value
def incr(self, key):
self._data[key] = self._data.get(key, 0) + 1
return self._data[key]
def decr(self, key):
self._data[key] = self._data.get(key, 0) - 1
return self._data[key]
def delete(self, key):
self._data.pop(key, None)
def pipeline(self):
return FakeRedisPipeline(self)
class FakeRedisPipeline:
def __init__(self, redis):
self.redis = redis
self._ops = []
def decr(self, key):
self._ops.append(("decr", key))
return self
def incr(self, key):
self._ops.append(("incr", key))
return self
def set(self, key, value):
self._ops.append(("set", key, value))
return self
def execute(self):
for op in self._ops:
if op[0] == "decr":
self.redis.decr(op[1])
elif op[0] == "incr":
self.redis.incr(op[1])
elif op[0] == "set":
self.redis.set(op[1], op[2])
self._ops = []
class ExtractCredentialsTests(TestCase):
def test_extract_credentials_from_xc_style_url(self):
url = "http://example.com/live/alice/secret123/99999.ts"
user, password = extract_credentials_from_stream_url(url)
self.assertEqual(user, "alice")
self.assertEqual(password, "secret123")
class ManualServerGroupTests(TestCase):
def test_group_enforced_when_account_assigned(self):
group = ServerGroup.objects.create(name="provider-a")
account = M3UAccount.objects.create(
name="Account A",
account_type="XC",
username="user",
password="pass",
server_group=group,
)
profile = M3UAccountProfile.objects.get(m3u_account=account, is_default=True)
self.assertEqual(get_enforced_server_group_for_profile(profile), group)
def test_accounts_in_same_group_share_credential_counter(self):
group = ServerGroup.objects.create(name="shared")
account1 = M3UAccount.objects.create(
name="XC Account",
account_type="XC",
username="user",
password="pass",
server_url="http://xc.example.com",
server_group=group,
max_streams=5,
)
account2 = M3UAccount.objects.create(
name="M3U Account",
account_type="STD",
username="user",
password="pass",
server_group=group,
max_streams=5,
)
profile1 = M3UAccountProfile.objects.get(m3u_account=account1, is_default=True)
profile2 = M3UAccountProfile.objects.get(m3u_account=account2, is_default=True)
profile1.max_streams = 1
profile1.save()
profile2.max_streams = 1
profile2.save()
redis = FakeRedis()
reserved1, _, _ = reserve_profile_slot(profile1, redis)
self.assertTrue(reserved1)
reserved2, _, _ = reserve_profile_slot(profile2, redis)
self.assertFalse(reserved2)
self.assertFalse(group_has_capacity_for_profile(profile2, redis))
def test_profile_rotation_when_default_profile_full(self):
account = M3UAccount.objects.create(
name="Multi-profile",
account_type="XC",
max_streams=1,
)
default = M3UAccountProfile.objects.get(m3u_account=account, is_default=True)
default.max_streams = 1
default.save()
alt = M3UAccountProfile.objects.create(
m3u_account=account,
name="alt_profile",
is_default=False,
is_active=True,
max_streams=1,
search_pattern="",
replace_pattern="",
)
redis = FakeRedis()
reserved, _, _ = reserve_profile_slot(default, redis)
self.assertTrue(reserved)
self.assertFalse(profile_has_capacity_for_selection(default, redis))
self.assertTrue(profile_has_capacity_for_selection(alt, redis))
class PoolEnforcementTests(TestCase):
def setUp(self):
self.redis = FakeRedis()
self.group = ServerGroup.objects.create(name="test-pool")
self.account = M3UAccount.objects.create(
name="Test Account",
account_type="XC",
username="user",
password="pass",
server_url="http://xc.example.com",
server_group=self.group,
max_streams=5,
)
self.profile = M3UAccountProfile.objects.get(
m3u_account=self.account, is_default=True
)
self.profile.max_streams = 1
self.profile.save()
def test_group_has_capacity_reads_credential_counter_directly(self):
cred_key = server_group_connections_key(
self.group.id,
get_profile_credential_fingerprint(self.profile),
)
self.redis.set(cred_key, 1)
self.assertFalse(group_has_capacity_for_profile(self.profile, self.redis))
self.redis.set(cred_key, 0)
self.assertTrue(group_has_capacity_for_profile(self.profile, self.redis))
def test_reserve_and_release_both_counters(self):
reserved, count, _ = reserve_profile_slot(self.profile, self.redis)
self.assertTrue(reserved)
self.assertEqual(count, 1)
cred_key = server_group_connections_key(
self.group.id,
get_profile_credential_fingerprint(self.profile),
)
profile_key = profile_connections_key(self.profile.id)
self.assertEqual(self.redis._data[cred_key], 1)
self.assertEqual(self.redis._data[profile_key], 1)
release_profile_slot(self.profile.id, self.redis)
self.assertEqual(self.redis._data[cred_key], 0)
self.assertEqual(self.redis._data[profile_key], 0)
def test_same_credential_capped_at_profile_max(self):
"""Shared credential counter is capped by each profile's max_streams."""
account2 = M3UAccount.objects.create(
name="Second Account",
account_type="XC",
username="user",
password="pass",
server_url="http://xc.example.com",
server_group=self.group,
max_streams=5,
)
profile2 = M3UAccountProfile.objects.get(m3u_account=account2, is_default=True)
profile2.max_streams = 1
profile2.save()
self.assertTrue(reserve_profile_slot(self.profile, self.redis)[0])
self.assertFalse(reserve_profile_slot(profile2, self.redis)[0])
fp = get_profile_credential_fingerprint(self.profile)
cred_key = server_group_connections_key(self.group.id, fp)
self.assertEqual(self.redis._data[cred_key], 1)
def test_different_logins_both_stream_in_same_group(self):
"""Different provider logins keep separate credential counters in one group."""
account = M3UAccount.objects.create(
name="Grouped multi-login",
account_type="XC",
username="login_a",
password="pass_a",
server_group=self.group,
max_streams=5,
)
default = M3UAccountProfile.objects.get(m3u_account=account, is_default=True)
default.max_streams = 1
default.save()
alt = M3UAccountProfile.objects.create(
m3u_account=account,
name="alt_login",
is_default=False,
is_active=True,
max_streams=1,
search_pattern="",
replace_pattern="",
)
fp_a = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
fp_b = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
with patch(
"apps.m3u.connection_pool.get_profile_credential_fingerprint",
side_effect=lambda profile: fp_a if profile.id == default.id else fp_b,
):
self.assertTrue(reserve_profile_slot(default, self.redis)[0])
self.assertTrue(reserve_profile_slot(alt, self.redis)[0])
key_a = server_group_connections_key(self.group.id, fp_a)
key_b = server_group_connections_key(self.group.id, fp_b)
self.assertEqual(self.redis._data[key_a], 1)
self.assertEqual(self.redis._data[key_b], 1)
def test_no_fingerprint_skips_credential_counter(self):
account = M3UAccount.objects.create(
name="No creds",
account_type="STD",
server_group=self.group,
max_streams=5,
)
profile = M3UAccountProfile.objects.get(m3u_account=account, is_default=True)
profile.max_streams = 1
profile.save()
with patch(
"apps.m3u.connection_pool.get_profile_credential_fingerprint",
return_value=None,
):
self.assertTrue(reserve_profile_slot(profile, self.redis)[0])
self.assertEqual(get_credential_connection_count(profile, self.redis), 0)
self.assertEqual(get_credential_connection_count(profile, self.redis), 0)
def test_release_when_profile_row_deleted(self):
profile_id = self.profile.id
fp = get_profile_credential_fingerprint(self.profile)
cred_key = server_group_connections_key(self.group.id, fp)
reserved, _, failure_reason = reserve_profile_slot(
self.profile, self.redis
)
self.assertTrue(reserved)
self.assertIsNone(failure_reason)
self.profile.delete()
release_profile_slot(profile_id, self.redis)
self.assertEqual(self.redis._data[profile_connections_key(profile_id)], 0)
self.assertEqual(self.redis._data[cred_key], 0)
self.assertNotIn(profile_credential_release_key(profile_id), self.redis._data)
def test_release_uses_stored_credential_key_without_db_lookup(self):
profile_id = self.profile.id
cred_key = server_group_connections_key(
self.group.id,
get_profile_credential_fingerprint(self.profile),
)
self.assertTrue(reserve_profile_slot(self.profile, self.redis)[0])
with patch("apps.m3u.models.M3UAccountProfile.objects.get") as mock_get:
release_profile_slot(profile_id, self.redis)
mock_get.assert_not_called()
self.assertEqual(self.redis._data[profile_connections_key(profile_id)], 0)
self.assertEqual(self.redis._data[cred_key], 0)
def test_reserve_returns_failure_reason_without_extra_checks(self):
account2 = M3UAccount.objects.create(
name="Reason Account",
account_type="XC",
username="user",
password="pass",
server_url="http://xc.example.com",
server_group=self.group,
max_streams=5,
)
profile2 = M3UAccountProfile.objects.get(m3u_account=account2, is_default=True)
profile2.max_streams = 1
profile2.save()
self.assertTrue(reserve_profile_slot(self.profile, self.redis)[0])
reserved, _count, reason = reserve_profile_slot(profile2, self.redis)
self.assertFalse(reserved)
self.assertEqual(reason, "credential_full")
class StaleAssignmentTests(TestCase):
def test_stale_assignment_releases_counters(self):
from apps.channels.models import Channel, Stream
redis = FakeRedis()
group = ServerGroup.objects.create(name="stale-group")
account = M3UAccount.objects.create(
name="Stale Account",
account_type="XC",
username="user",
password="pass",
server_url="http://xc.example.com",
server_group=group,
max_streams=5,
)
profile = M3UAccountProfile.objects.get(m3u_account=account, is_default=True)
profile.max_streams = 1
profile.save()
stream = Stream.objects.create(name="Stale Stream", m3u_account=account)
channel = Channel.objects.create(channel_number=502, name="Stale Channel")
channel.streams.add(stream)
reserve_profile_slot(profile, redis)
redis.set(f"channel_stream:{channel.id}", stream.id)
redis.set(f"stream_profile:{stream.id}", profile.id)
profile_key = profile_connections_key(profile.id)
cred_key = server_group_connections_key(
group.id, get_profile_credential_fingerprint(profile)
)
self.assertEqual(redis._data[profile_key], 1)
self.assertEqual(redis._data[cred_key], 1)
with patch("core.utils.RedisClient.get_client", return_value=redis):
channel._release_stale_stream_assignment(redis, stream.id)
self.assertNotIn(f"channel_stream:{channel.id}", redis._data)
self.assertNotIn(f"stream_profile:{stream.id}", redis._data)
self.assertEqual(redis._data[profile_key], 0)
self.assertEqual(redis._data[cred_key], 0)
class UpdateStreamProfileTests(TestCase):
def test_switch_updates_profile_counters_when_group_assigned(self):
from apps.channels.models import Channel, Stream
redis = FakeRedis()
group = ServerGroup.objects.create(name="switch-group")
account = M3UAccount.objects.create(
name="Switch Account",
account_type="XC",
username="user",
password="pass",
server_url="http://xc.example.com",
server_group=group,
max_streams=5,
)
profile_a = M3UAccountProfile.objects.get(m3u_account=account, is_default=True)
profile_a.max_streams = 1
profile_a.save()
profile_b = M3UAccountProfile.objects.create(
m3u_account=account,
name="alt",
is_default=False,
is_active=True,
max_streams=1,
search_pattern="",
replace_pattern="",
)
stream = Stream.objects.create(name="Test Stream", m3u_account=account)
channel = Channel.objects.create(channel_number=501, name="Switch Channel")
channel.streams.add(stream)
reserve_profile_slot(profile_a, redis)
redis.set(f"channel_stream:{channel.id}", stream.id)
redis.set(f"stream_profile:{stream.id}", profile_a.id)
cred_key = server_group_connections_key(
group.id, get_profile_credential_fingerprint(profile_a)
)
cred_before = redis._data[cred_key]
with patch("core.utils.RedisClient.get_client", return_value=redis):
self.assertTrue(channel.update_stream_profile(profile_b.id))
self.assertEqual(int(redis.get(f"stream_profile:{stream.id}")), profile_b.id)
self.assertEqual(redis._data[profile_connections_key(profile_a.id)], 0)
self.assertEqual(redis._data[profile_connections_key(profile_b.id)], 1)
self.assertEqual(redis._data[cred_key], cred_before)
def test_switch_moves_credential_counter_when_login_changes(self):
from apps.channels.models import Channel, Stream
redis = FakeRedis()
group = ServerGroup.objects.create(name="cred-switch-group")
account = M3UAccount.objects.create(
name="Cred Switch Account",
account_type="XC",
username="login_a",
password="pass_a",
server_url="http://xc.example.com",
server_group=group,
max_streams=5,
)
profile_a = M3UAccountProfile.objects.get(m3u_account=account, is_default=True)
profile_a.max_streams = 1
profile_a.save()
profile_b = M3UAccountProfile.objects.create(
m3u_account=account,
name="alt_login",
is_default=False,
is_active=True,
max_streams=1,
search_pattern="",
replace_pattern="",
)
fp_a = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
fp_b = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
key_a = server_group_connections_key(group.id, fp_a)
key_b = server_group_connections_key(group.id, fp_b)
stream = Stream.objects.create(name="Cred Switch Stream", m3u_account=account)
channel = Channel.objects.create(channel_number=503, name="Cred Switch Channel")
channel.streams.add(stream)
with patch(
"apps.m3u.connection_pool.get_profile_credential_fingerprint",
side_effect=lambda profile: fp_a if profile.id == profile_a.id else fp_b,
):
reserve_profile_slot(profile_a, redis)
redis.set(f"channel_stream:{channel.id}", stream.id)
redis.set(f"stream_profile:{stream.id}", profile_a.id)
self.assertEqual(redis._data[key_a], 1)
self.assertNotIn(key_b, redis._data)
with patch("core.utils.RedisClient.get_client", return_value=redis):
self.assertTrue(channel.update_stream_profile(profile_b.id))
self.assertEqual(redis._data[key_a], 0)
self.assertEqual(redis._data[key_b], 1)
class VodProfileSelectionTests(TestCase):
def test_get_m3u_profile_skips_default_when_profile_full(self):
from apps.proxy.vod_proxy.views import _get_m3u_profile
account = M3UAccount.objects.create(
name="VOD multi-profile",
account_type="XC",
username="xc_user_a",
password="xc_pass_a",
server_url="http://xc.example.com",
max_streams=1,
)
default = M3UAccountProfile.objects.get(m3u_account=account, is_default=True)
default.max_streams = 1
default.save()
alt = M3UAccountProfile.objects.create(
m3u_account=account,
name="alt_profile",
is_default=False,
is_active=True,
max_streams=1,
search_pattern="",
replace_pattern="",
)
redis = FakeRedis()
self.assertTrue(reserve_profile_slot(default, redis)[0])
with patch("core.utils.RedisClient.get_client", return_value=redis):
result = _get_m3u_profile(account, None, None)
self.assertIsNotNone(result)
selected, _connections = result
self.assertEqual(selected.id, alt.id)
def test_get_m3u_profile_skips_default_when_credential_pool_full(self):
from apps.proxy.vod_proxy.views import _get_m3u_profile
group = ServerGroup.objects.create(name="vod-cred-pool")
account = M3UAccount.objects.create(
name="VOD pooled",
account_type="XC",
username="shared_user",
password="shared_pass",
server_url="http://xc.example.com",
server_group=group,
max_streams=5,
)
default = M3UAccountProfile.objects.get(m3u_account=account, is_default=True)
default.max_streams = 1
default.save()
alt = M3UAccountProfile.objects.create(
m3u_account=account,
name="alt_login",
is_default=False,
is_active=True,
max_streams=1,
search_pattern="",
replace_pattern="",
)
other_account = M3UAccount.objects.create(
name="Other pooled account",
account_type="XC",
username="shared_user",
password="shared_pass",
server_url="http://xc.example.com",
server_group=group,
max_streams=5,
)
other_profile = M3UAccountProfile.objects.get(
m3u_account=other_account, is_default=True
)
other_profile.max_streams = 1
other_profile.save()
fp_shared = "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc"
fp_alt = "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"
redis = FakeRedis()
with patch(
"apps.m3u.connection_pool.get_profile_credential_fingerprint",
side_effect=lambda profile: (
fp_shared
if profile.id in (default.id, other_profile.id)
else fp_alt
),
):
self.assertTrue(reserve_profile_slot(other_profile, redis)[0])
with patch("core.utils.RedisClient.get_client", return_value=redis):
result = _get_m3u_profile(account, None, None)
self.assertIsNotNone(result)
selected, _connections = result
self.assertEqual(selected.id, alt.id)

176
apps/proxy/live_proxy/url_utils.py
View file

@ -8,6 +8,10 @@ from typing import Optional, Tuple, List
from django.shortcuts import get_object_or_404
from apps.channels.models import Channel, Stream
from apps.m3u.models import M3UAccount, M3UAccountProfile
from apps.m3u.connection_pool import (
get_profile_connection_count,
profile_available_for_channel_switch,
)
from core.models import UserAgent, CoreSettings, StreamProfile
from .utils import get_logger
from uuid import UUID
@ -15,6 +19,35 @@ import requests
logger = get_logger()
def _resolve_live_stream_url(stream, m3u_account, m3u_profile):
"""
Build the upstream URL for live playback.
XC accounts use current transformed credentials plus provider stream_id so
playback matches the account login (not a stale stream.url from an old sync).
STD/M3U accounts keep using the URL stored on the stream row.
"""
if (
m3u_account.account_type == M3UAccount.Types.XC
and stream.stream_id
):
from apps.m3u.tasks import get_transformed_credentials
server_url, username, password = get_transformed_credentials(
m3u_account, m3u_profile
)
if server_url and username and password:
base = server_url.rstrip("/")
return f"{base}/live/{username}/{password}/{stream.stream_id}.ts"
return transform_url(
stream.url or "",
m3u_profile.search_pattern,
m3u_profile.replace_pattern,
)
def get_stream_object(id: str):
try:
logger.info(f"Fetching channel ID {id}")
@ -24,15 +57,14 @@ def get_stream_object(id: str):
logger.info(f"Fetching stream hash {id}")
return get_object_or_404(Stream, stream_hash=id)
def generate_stream_url(channel_id: str) -> Tuple[str, str, bool, Optional[int]]:
def generate_stream_url(
channel_id: str,
) -> Tuple[str, str, bool, Optional[int], bool, Optional[str]]:
"""
Generate the appropriate stream URL for a channel or stream based on its profile settings.
Args:
channel_id: The UUID of the channel or stream hash
Returns:
Tuple[str, str, bool, Optional[int]]: (stream_url, user_agent, transcode_flag, profile_id)
Tuple: (stream_url, user_agent, transcode_flag, profile_id, slot_reserved, error_reason)
"""
try:
channel_or_stream = get_stream_object(channel_id)
@ -44,15 +76,12 @@ def generate_stream_url(channel_id: str) -> Tuple[str, str, bool, Optional[int]]
if not stream.m3u_account:
logger.error(f"Stream {stream.id} has no M3U account")
return None, None, False, None
return None, None, False, None, False, "Stream has no M3U account"
# Use get_stream() to atomically reserve a slot and write the
# channel_stream / stream_profile Redis keys, matching the channel
# path so stream_name and stream_stats work correctly.
stream_id, profile_id, error_reason = stream.get_stream()
stream_id, profile_id, error_reason, slot_reserved = stream.get_stream()
if not stream_id or not profile_id:
logger.error(f"No profile available for stream {stream.id}: {error_reason}")
return None, None, False, None
return None, None, False, None, False, error_reason
try:
profile = M3UAccountProfile.objects.get(id=profile_id)
@ -63,7 +92,7 @@ def generate_stream_url(channel_id: str) -> Tuple[str, str, bool, Optional[int]]
stream_user_agent = UserAgent.objects.get(id=CoreSettings.get_default_user_agent_id())
logger.debug(f"No user agent found for account, using default: {stream_user_agent}")
stream_url = transform_url(stream.url, profile.search_pattern, profile.replace_pattern)
stream_url = _resolve_live_stream_url(stream, m3u_account, profile)
stream_profile = stream.get_stream_profile()
logger.debug(f"Using stream profile: {stream_profile.name}")
@ -71,23 +100,23 @@ def generate_stream_url(channel_id: str) -> Tuple[str, str, bool, Optional[int]]
transcode = not stream_profile.is_proxy()
stream_profile_id = stream_profile.id
return stream_url, stream_user_agent, transcode, stream_profile_id
return stream_url, stream_user_agent, transcode, stream_profile_id, slot_reserved, None
except Exception as e:
logger.error(f"Error generating stream URL for stream {stream.id}: {e}")
stream.release_stream()
return None, None, False, None
if slot_reserved:
stream.release_stream()
return None, None, False, None, False, str(e)
# Handle channel preview (existing logic)
channel = channel_or_stream
# Get stream and profile for this channel
# Note: get_stream now returns 3 values (stream_id, profile_id, error_reason)
stream_id, profile_id, error_reason = channel.get_stream()
stream_id, profile_id, error_reason, slot_reserved = channel.get_stream()
if not stream_id or not profile_id:
logger.error(f"No stream available for channel {channel_id}: {error_reason}")
return None, None, False, None
return None, None, False, None, False, error_reason
# get_stream() allocated a connection slot - ensure it's released on any error
try:
@ -106,9 +135,7 @@ def generate_stream_url(channel_id: str) -> Tuple[str, str, bool, Optional[int]]
stream_user_agent = UserAgent.objects.get(id=CoreSettings.get_default_user_agent_id())
logger.debug(f"No user agent found for account, using default: {stream_user_agent}")
# Generate stream URL based on the selected profile
input_url = stream.url
stream_url = transform_url(input_url, m3u_profile.search_pattern, m3u_profile.replace_pattern)
stream_url = _resolve_live_stream_url(stream, m3u_account, m3u_profile)
# Check if transcoding is needed
stream_profile = channel.get_stream_profile()
@ -119,15 +146,16 @@ def generate_stream_url(channel_id: str) -> Tuple[str, str, bool, Optional[int]]
stream_profile_id = stream_profile.id
return stream_url, stream_user_agent, transcode, stream_profile_id
return stream_url, stream_user_agent, transcode, stream_profile_id, slot_reserved, None
except Exception as e:
logger.error(f"Error generating stream URL for channel {channel_id}: {e}")
if not channel.release_stream():
logger.warning(f"Failed to release stream for channel {channel_id} after URL generation error")
return None, None, False, None
if slot_reserved:
if not channel.release_stream():
logger.warning(f"Failed to release stream for channel {channel_id} after URL generation error")
return None, None, False, None, False, str(e)
except Exception as e:
logger.error(f"Error generating stream URL: {e}")
return None, None, False, None
return None, None, False, None, False, str(e)
def transform_url(input_url: str, search_pattern: str, replace_pattern: str) -> str:
"""
@ -175,6 +203,8 @@ def get_stream_info_for_switch(channel_id: str, target_stream_id: Optional[int]
Returns:
dict: Stream information including URL, user agent and transcode flag
"""
slot_reserved = False
channel = None
try:
from core.utils import RedisClient
@ -204,35 +234,34 @@ def get_stream_info_for_switch(channel_id: str, target_stream_id: Optional[int]
selected_profile = None
for profile in profiles:
# Check connection availability
if redis_client:
profile_connections_key = f"profile_connections:{profile.id}"
current_connections = int(redis_client.get(profile_connections_key) or 0)
# Check if this channel is already using this profile
channel_using_profile = False
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
existing_profile_id = redis_client.get(f"stream_profile:{existing_stream_id}")
existing_profile_id = redis_client.get(
f"stream_profile:{existing_stream_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}")
# Calculate effective connections (subtract 1 if channel already using this profile)
effective_connections = current_connections - (1 if channel_using_profile else 0)
# Check if profile has available slots
if profile.max_streams == 0 or effective_connections < profile.max_streams:
if profile_available_for_channel_switch(
profile,
redis_client,
channel_already_on_profile=channel_using_profile,
):
current_connections = get_profile_connection_count(
profile, redis_client
)
selected_profile = profile
logger.debug(f"Selected profile {profile.id} with {effective_connections}/{profile.max_streams} effective connections (current: {current_connections}, already using: {channel_using_profile})")
logger.debug(
f"Selected profile {profile.id} with "
f"{current_connections}/{profile.max_streams} connections"
)
break
else:
logger.debug(f"Profile {profile.id} at max connections: {effective_connections}/{profile.max_streams} (current: {current_connections}, already using: {channel_using_profile})")
logger.debug(
f"Profile {profile.id} unavailable for channel switch"
)
else:
# No Redis available, assume first active profile is okay
selected_profile = profile
break
@ -241,29 +270,18 @@ def get_stream_info_for_switch(channel_id: str, target_stream_id: Optional[int]
m3u_profile_id = selected_profile.id
else:
stream_id, m3u_profile_id, error_reason = channel.get_stream()
stream_id, m3u_profile_id, error_reason, slot_reserved = channel.get_stream()
if stream_id is None or m3u_profile_id is None:
return {'error': error_reason or 'No stream assigned to channel'}
# Get the stream and profile objects directly
stream = get_object_or_404(Stream, pk=stream_id)
profile = get_object_or_404(M3UAccountProfile, pk=m3u_profile_id)
# Check connections left
m3u_account = M3UAccount.objects.get(id=profile.m3u_account.id)
#connections_left = get_connections_left(m3u_profile_id)
#if connections_left <= 0:
#logger.warning(f"No connections left for M3U account {m3u_account.id}")
#return {'error': 'No connections left'}
# Get the user agent from the M3U account
user_agent = m3u_account.get_user_agent().user_agent
# Generate URL using the transform function directly
stream_url = transform_url(stream.url, profile.search_pattern, profile.replace_pattern)
stream_url = _resolve_live_stream_url(stream, m3u_account, profile)
# Get transcode info from the channel's stream profile
stream_profile = channel.get_stream_profile()
transcode = not (stream_profile.is_proxy() or stream_profile is None)
profile_value = stream_profile.id
@ -278,6 +296,8 @@ def get_stream_info_for_switch(channel_id: str, target_stream_id: Optional[int]
'stream_name': stream.name,
}
except Exception as e:
if slot_reserved and channel is not None:
channel.release_stream()
logger.error(f"Error getting stream info for switch: {e}", exc_info=True)
return {'error': f'Error: {str(e)}'}
@ -344,34 +364,38 @@ def get_alternate_streams(channel_id: str, current_stream_id: Optional[int] = No
selected_profile = None
for profile in profiles:
# Check connection availability
if redis_client:
profile_connections_key = f"profile_connections:{profile.id}"
current_connections = int(redis_client.get(profile_connections_key) or 0)
# Check if this channel is already using this profile
channel_using_profile = False
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
existing_profile_id = redis_client.get(f"stream_profile:{existing_stream_id}")
existing_profile_id = redis_client.get(
f"stream_profile:{existing_stream_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}")
logger.debug(
f"Channel {channel.id} already using profile {profile.id}"
)
# Calculate effective connections (subtract 1 if channel already using this profile)
effective_connections = current_connections - (1 if channel_using_profile else 0)
# Check if profile has available slots
if profile.max_streams == 0 or effective_connections < profile.max_streams:
if profile_available_for_channel_switch(
profile,
redis_client,
channel_already_on_profile=channel_using_profile,
):
current_connections = get_profile_connection_count(
profile, redis_client
)
selected_profile = profile
logger.debug(f"Found available profile {profile.id} for stream {stream.id}: {effective_connections}/{profile.max_streams} effective (current: {current_connections}, already using: {channel_using_profile})")
logger.debug(
f"Found available profile {profile.id} for stream {stream.id}: "
f"{current_connections}/{profile.max_streams} "
f"(already using: {channel_using_profile})"
)
break
else:
logger.debug(f"Profile {profile.id} at max connections: {effective_connections}/{profile.max_streams} (current: {current_connections}, already using: {channel_using_profile})")
logger.debug(
f"Profile {profile.id} unavailable for alternate stream {stream.id}"
)
else:
# No Redis available, assume first active profile is okay
selected_profile = profile
break

34
apps/proxy/live_proxy/views.py
View file

@ -213,6 +213,7 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
stream_user_agent = None
transcode = False
profile_value = None
slot_reserved = False
error_reason = None
attempt = 0
should_retry = True
@ -220,9 +221,14 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
# Try to get a stream with fixed interval retries
while should_retry and time.time() - wait_start_time < retry_timeout:
attempt += 1
stream_url, stream_user_agent, transcode, profile_value = (
generate_stream_url(channel_id)
)
(
stream_url,
stream_user_agent,
transcode,
profile_value,
slot_reserved,
error_reason,
) = generate_stream_url(channel_id)
if stream_url is not None:
logger.info(
@ -232,7 +238,6 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
# On first failure, check if the error is retryable
if attempt == 1:
_, _, error_reason = channel.get_stream()
if error_reason and "maximum connection limits" not in error_reason:
logger.warning(
f"[{client_id}] Can't retry - error not related to connection limits: {error_reason}"
@ -265,18 +270,21 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
logger.info(
f"[{client_id}] Making final attempt {attempt} at timeout boundary"
)
stream_url, stream_user_agent, transcode, profile_value = (
generate_stream_url(channel_id)
)
(
stream_url,
stream_user_agent,
transcode,
profile_value,
slot_reserved,
error_reason,
) = generate_stream_url(channel_id)
if stream_url is not None:
logger.info(
f"[{client_id}] Successfully obtained stream on final attempt for channel {channel_id}"
)
if stream_url is None:
# Release any connection slot that may have been allocated
# by the error-checking get_stream() call during retries
if not channel.release_stream():
if slot_reserved and not channel.release_stream():
logger.debug(f"[{client_id}] release_stream found no keys during failed init cleanup")
# Get the specific error message if available
@ -295,7 +303,7 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
# generate_stream_url() called get_stream() which allocated a connection
# slot (INCR'd profile_connections) - track this for cleanup on error
if needs_initialization:
if needs_initialization and slot_reserved:
connection_allocated = True
# Read stream assignment from Redis (already set by generate_stream_url → get_stream).
@ -378,8 +386,8 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
logger.warning(
f"[{client_id}] Alternate stream #{alt['stream_id']} failed validation: {message}"
)
# Release stream lock before redirecting
if not channel.release_stream():
# Release stream lock before redirecting only if we reserved a slot
if connection_allocated and not channel.release_stream():
logger.warning(f"[{client_id}] Failed to release stream before redirect")
connection_allocated = False
# Final decision based on validation results

82
apps/proxy/vod_proxy/multi_worker_connection_manager.py
View file

@ -735,68 +735,35 @@ class MultiWorkerVODConnectionManager:
"""Get Redis key for tracking connections per profile - STANDARDIZED with TS proxy"""
return f"profile_connections:{profile_id}"
def _check_profile_limits(self, m3u_profile) -> bool:
"""Check if profile has available connection slots"""
if m3u_profile.max_streams == 0: # Unlimited
return True
try:
profile_connections_key = self._get_profile_connections_key(m3u_profile.id)
current_connections = int(self.redis_client.get(profile_connections_key) or 0)
logger.info(f"[PROFILE-CHECK] Profile {m3u_profile.id} has {current_connections}/{m3u_profile.max_streams} connections")
return current_connections < m3u_profile.max_streams
except Exception as e:
logger.error(f"Error checking profile limits: {e}")
return False
def _check_and_reserve_profile_slot(self, m3u_profile) -> bool:
"""
Atomically check and reserve a connection slot for the given profile.
Uses an INCR-first-then-check pattern to eliminate the TOCTOU race
condition where separate GET > check > INCR operations could allow
concurrent requests to both pass the capacity check.
For profiles with max_streams=0 (unlimited), no reservation is needed.
Returns:
bool: True if slot was reserved (or unlimited), False if at capacity
"""
if m3u_profile.max_streams == 0: # Unlimited
return True
from apps.m3u.connection_pool import reserve_profile_slot
try:
profile_connections_key = self._get_profile_connections_key(m3u_profile.id)
# Atomically increment first — single Redis command eliminates race window
new_count = self.redis_client.incr(profile_connections_key)
if new_count <= m3u_profile.max_streams:
logger.info(f"[PROFILE-RESERVE] Profile {m3u_profile.id} slot reserved: {new_count}/{m3u_profile.max_streams}")
return True
# Over capacity — roll back the increment
self.redis_client.decr(profile_connections_key)
logger.info(f"[PROFILE-RESERVE] Profile {m3u_profile.id} at capacity: {new_count - 1}/{m3u_profile.max_streams}")
return False
reserved, new_count, _failure_reason = reserve_profile_slot(
m3u_profile, self.redis_client
)
if reserved:
logger.info(
f"[PROFILE-RESERVE] Profile {m3u_profile.id} slot reserved: "
f"{new_count}/{m3u_profile.max_streams}"
)
else:
logger.info(
f"[PROFILE-RESERVE] Profile {m3u_profile.id} at capacity: "
f"{new_count}/{m3u_profile.max_streams}"
)
return reserved
except Exception as e:
logger.error(f"Error reserving profile slot: {e}")
return False
def _increment_profile_connections(self, m3u_profile):
"""Increment profile connection count"""
try:
profile_connections_key = self._get_profile_connections_key(m3u_profile.id)
new_count = self.redis_client.incr(profile_connections_key)
logger.info(f"[PROFILE-INCR] Profile {m3u_profile.id} connections: {new_count}")
return new_count
except Exception as e:
logger.error(f"Error incrementing profile connections: {e}")
return None
def _trigger_vod_stats_update(self):
"""Trigger a VOD stats WebSocket update in a background thread."""
threading.Thread(target=self._do_vod_stats_update, daemon=True).start()
@ -859,21 +826,14 @@ class MultiWorkerVODConnectionManager:
logger.error(f"Failed to trigger VOD stats update: {e}")
def _decrement_profile_connections(self, m3u_profile_id: int):
"""Decrement profile connection count.
"""Decrement profile and shared pool connection counters."""
from apps.m3u.connection_pool import release_profile_slot
Uses a single atomic DECR (no GET-before-DECR) to avoid the race condition
where two concurrent decrements both pass a >0 guard and both fire, sending
the counter negative. If the counter would go below zero it is clamped to 0.
"""
try:
profile_connections_key = self._get_profile_connections_key(m3u_profile_id)
new_count = self.redis_client.decr(profile_connections_key)
if new_count < 0:
self.redis_client.set(profile_connections_key, 0)
new_count = 0
logger.warning(f"[PROFILE-DECR] Profile {m3u_profile_id} counter went negative, clamped to 0")
else:
logger.info(f"[PROFILE-DECR] Profile {m3u_profile_id} connections: {new_count}")
release_profile_slot(m3u_profile_id, self.redis_client)
profile_key = self._get_profile_connections_key(m3u_profile_id)
new_count = int(self.redis_client.get(profile_key) or 0)
logger.info(f"[PROFILE-DECR] Profile {m3u_profile_id} connections: {new_count}")
return new_count
except Exception as e:
logger.error(f"Error decrementing profile connections: {e}")

31
apps/proxy/vod_proxy/views.py
View file

@ -260,6 +260,10 @@ def _get_m3u_profile(m3u_account, profile_id, session_id=None):
"""
try:
from core.utils import RedisClient
from apps.m3u.connection_pool import (
get_profile_connection_count,
pool_has_capacity_for_profile,
)
redis_client = RedisClient.get_client()
if not redis_client:
@ -296,7 +300,11 @@ def _get_m3u_profile(m3u_account, profile_id, session_id=None):
except Exception as e:
logger.warning(f"[PROFILE-SELECTION] Error checking existing profile for session {session_id}: {e}")
else:
logger.debug(f"[PROFILE-SELECTION] Session {session_id} exists but has no profile ID stored") # If specific profile requested, try to use it
logger.debug(
f"[PROFILE-SELECTION] Session {session_id} exists but has no profile ID stored"
)
# If specific profile requested, try to use it
if profile_id:
try:
profile = M3UAccountProfile.objects.get(
@ -306,13 +314,12 @@ def _get_m3u_profile(m3u_account, profile_id, session_id=None):
)
# Check Redis-based current connections
profile_connections_key = f"profile_connections:{profile.id}"
current_connections = int(redis_client.get(profile_connections_key) or 0)
current_connections = get_profile_connection_count(profile, redis_client)
if profile.max_streams == 0 or current_connections < profile.max_streams:
if pool_has_capacity_for_profile(profile, redis_client):
logger.info(f"[PROFILE-SELECTION] Using requested profile {profile.id}: {current_connections}/{profile.max_streams} connections")
return (profile, current_connections)
else:
logger.warning(f"[PROFILE-SELECTION] Requested profile {profile.id} is at capacity: {current_connections}/{profile.max_streams}")
logger.warning(f"[PROFILE-SELECTION] Requested profile {profile.id} is at capacity: {current_connections}/{profile.max_streams}")
except M3UAccountProfile.DoesNotExist:
logger.warning(f"[PROFILE-SELECTION] Requested profile {profile_id} not found")
@ -331,15 +338,16 @@ def _get_m3u_profile(m3u_account, profile_id, session_id=None):
profiles = [default_profile] + list(m3u_profiles.filter(is_default=False))
for profile in profiles:
profile_connections_key = f"profile_connections:{profile.id}"
current_connections = int(redis_client.get(profile_connections_key) or 0)
current_connections = get_profile_connection_count(profile, redis_client)
# Check if profile has available connection slots
if profile.max_streams == 0 or current_connections < profile.max_streams:
if pool_has_capacity_for_profile(profile, redis_client):
logger.info(f"[PROFILE-SELECTION] Selected profile {profile.id} ({profile.name}): {current_connections}/{profile.max_streams} connections")
return (profile, current_connections)
else:
logger.debug(f"[PROFILE-SELECTION] Profile {profile.id} at capacity: {current_connections}/{profile.max_streams}")
logger.debug(
f"[PROFILE-SELECTION] Profile {profile.id} unavailable "
f"(profile={current_connections}/{profile.max_streams})"
)
# All profiles are at capacity - return None to trigger error response
logger.error(f"[PROFILE-SELECTION] All profiles at capacity for M3U account {m3u_account.id}, rejecting request")
@ -906,13 +914,14 @@ def build_vod_stats_data(redis_client):
if m3u_profile_id:
try:
from apps.m3u.models import M3UAccountProfile
profile = M3UAccountProfile.objects.select_related('m3u_account').get(id=m3u_profile_id)
m3u_profile_info = {
'profile_name': profile.name,
'account_name': profile.m3u_account.name,
'account_id': profile.m3u_account.id,
'max_streams': profile.m3u_account.max_streams,
'm3u_profile_id': int(m3u_profile_id)
'm3u_profile_id': int(m3u_profile_id),
}
except Exception as e:
logger.warning(f"Could not fetch M3U profile {m3u_profile_id}: {e}")

54
frontend/src/api.js
View file

@ -3,6 +3,7 @@ import useAuthStore from './store/auth';
import useChannelsStore from './store/channels';
import useLogosStore from './store/logos';
import useUserAgentsStore from './store/userAgents';
import useServerGroupsStore from './store/serverGroups';
import usePlaylistsStore from './store/playlists';
import useEPGsStore from './store/epgs';
import useStreamsStore from './store/streams';
@ -1305,6 +1306,59 @@ export default class API {
}
}
static async getServerGroups() {
try {
const response = await request(`${host}/api/m3u/server-groups/`);
return response;
} catch (e) {
errorNotification('Failed to retrieve server groups', e);
}
}
static async addServerGroup(values) {
try {
const response = await request(`${host}/api/m3u/server-groups/`, {
method: 'POST',
body: values,
});
useServerGroupsStore.getState().addServerGroup(response);
return response;
} catch (e) {
errorNotification('Failed to create server group', e);
}
}
static async updateServerGroup(values) {
try {
const { id, ...payload } = values;
const response = await request(`${host}/api/m3u/server-groups/${id}/`, {
method: 'PUT',
body: payload,
});
useServerGroupsStore.getState().updateServerGroup(response);
return response;
} catch (e) {
errorNotification('Failed to update server group', e);
}
}
static async deleteServerGroup(id) {
try {
await request(`${host}/api/m3u/server-groups/${id}/`, {
method: 'DELETE',
});
useServerGroupsStore.getState().removeServerGroups([id]);
} catch (e) {
errorNotification('Failed to delete server group', e);
}
}
static async getPlaylist(id) {
try {
const response = await request(`${host}/api/m3u/accounts/${id}/`);

28
frontend/src/components/ServerGroupsManagerModal.jsx Normal file
View file

@ -0,0 +1,28 @@
import { Modal } from '@mantine/core';
import ServerGroupsTable from './tables/ServerGroupsTable';
const ServerGroupsManagerModal = ({
isOpen,
onClose,
onGroupCreated,
openCreateOnMount = false,
}) => {
return (
<Modal
opened={isOpen}
onClose={onClose}
title="Server Groups"
size="md"
centered
>
{isOpen ? (
<ServerGroupsTable
onGroupCreated={onGroupCreated}
openCreateOnMount={openCreateOnMount}
/>
) : null}
</Modal>
);
};
export default ServerGroupsManagerModal;

61
frontend/src/components/__tests__/ServerGroupsManagerModal.test.jsx Normal file
View file

@ -0,0 +1,61 @@
import { describe, it, expect, vi, beforeEach } from 'vitest';
import { render, screen, waitFor } from '@testing-library/react';
import { MantineProvider } from '@mantine/core';
import ServerGroupsManagerModal from '../ServerGroupsManagerModal';
vi.mock('../../api', () => ({
default: {
deleteServerGroup: vi.fn(),
},
}));
vi.mock('../../store/serverGroups', () => ({
default: (selector) =>
selector({
serverGroups: [
{ id: 1, name: 'Pool A' },
{ id: 2, name: 'Pool B' },
],
isLoading: false,
error: null,
fetchServerGroups: vi.fn().mockResolvedValue(undefined),
}),
}));
vi.mock('../../store/playlists', () => ({
default: (selector) =>
selector({
playlists: [{ id: 10, server_group: 2 }],
}),
}));
vi.mock('../forms/ServerGroup', () => ({
default: () => null,
}));
vi.mock('../../hooks/useLocalStorage', () => ({
default: () => ['default', vi.fn()],
}));
const renderModal = (props = {}) =>
render(
<MantineProvider>
<ServerGroupsManagerModal isOpen onClose={vi.fn()} {...props} />
</MantineProvider>
);
describe('ServerGroupsManagerModal', () => {
beforeEach(() => {
vi.clearAllMocks();
});
it('opens without Mantine Tooltip errors', async () => {
expect(() => renderModal()).not.toThrow();
await waitFor(() => {
expect(screen.getByText('Pool A')).toBeInTheDocument();
expect(screen.getByText('Pool B')).toBeInTheDocument();
expect(screen.getByText('Accounts')).toBeInTheDocument();
});
});
});

276
frontend/src/components/forms/M3U.jsx
View file

@ -1,11 +1,11 @@
// Modal.js
import React, { useEffect, useState } from 'react';
import useUserAgentsStore from '../../store/userAgents';
import useServerGroupsStore from '../../store/serverGroups';
import M3UProfiles from './M3UProfiles';
import {
Box,
Button,
Checkbox,
Divider,
FileInput,
Flex,
@ -35,6 +35,7 @@ import {
prepareSubmitValues,
updatePlaylist,
} from '../../utils/forms/M3uUtils.js';
import ServerGroupsManagerModal from '../ServerGroupsManagerModal';
const M3U = ({
m3uAccount = null,
@ -43,6 +44,7 @@ const M3U = ({
playlistCreated = false,
}) => {
const userAgents = useUserAgentsStore((s) => s.userAgents);
const serverGroups = useServerGroupsStore((s) => s.serverGroups);
const fetchChannelGroups = useChannelsStore((s) => s.fetchChannelGroups);
const fetchEPGs = useEPGsStore((s) => s.fetchEPGs);
const fetchCategories = useVODStore((s) => s.fetchCategories);
@ -54,6 +56,9 @@ const M3U = ({
const [groupFilterModalOpen, setGroupFilterModalOpen] = useState(false);
const [filterModalOpen, setFilterModalOpen] = useState(false);
const [scheduleType, setScheduleType] = useState('interval');
const [serverGroupsManagerOpen, setServerGroupsManagerOpen] = useState(false);
const [serverGroupsCreateOnOpen, setServerGroupsCreateOnOpen] =
useState(false);
const form = useForm({
mode: 'uncontrolled',
@ -61,6 +66,7 @@ const M3U = ({
name: '',
server_url: '',
user_agent: '0',
server_group: '0',
is_active: true,
max_streams: 0,
refresh_interval: 24,
@ -88,6 +94,9 @@ const M3U = ({
server_url: m3uAccount.server_url,
max_streams: m3uAccount.max_streams,
user_agent: m3uAccount.user_agent ? `${m3uAccount.user_agent}` : '0',
server_group: m3uAccount.server_group
? `${m3uAccount.server_group}`
: '0',
is_active: m3uAccount.is_active,
refresh_interval: m3uAccount.refresh_interval,
cron_expression: m3uAccount.cron_expression || '',
@ -203,7 +212,7 @@ const M3U = ({
return (
<>
<Modal
size={700}
size={960}
opened={isOpen}
onClose={close}
title="M3U Account"
@ -216,10 +225,9 @@ const M3U = ({
<LoadingOverlay visible={form.submitting} overlayBlur={2} />
<form onSubmit={form.onSubmit(onSubmit)}>
<Group justify="space-between" align="top">
<Stack gap="5" style={{ flex: 1, minWidth: 0 }}>
<Group align="flex-start" gap="md" wrap="nowrap">
<Stack gap="xs" style={{ flex: 1, minWidth: 0 }}>
<TextInput
style={{ width: '100%' }}
id="name"
name="name"
label="Name"
@ -228,7 +236,6 @@ const M3U = ({
key={form.key('name')}
/>
<TextInput
style={{ width: '100%' }}
id="server_url"
name="server_url"
label="URL"
@ -236,7 +243,6 @@ const M3U = ({
{...form.getInputProps('server_url')}
key={form.key('server_url')}
/>
<Select
id="account_type"
name="account_type"
@ -248,49 +254,15 @@ const M3U = ({
</>
}
data={[
{
value: 'STD',
label: 'Standard',
},
{
value: 'XC',
label: 'Xtream Codes',
},
{ value: 'STD', label: 'Standard' },
{ value: 'XC', label: 'Xtream Codes' },
]}
key={form.key('account_type')}
{...form.getInputProps('account_type')}
/>
{form.getValues().account_type == 'XC' && (
<Box>
{!m3uAccount && (
<Group justify="space-between">
<Box>Create EPG</Box>
<Switch
id="create_epg"
name="create_epg"
description="Automatically create matching EPG source for this Xtream account"
key={form.key('create_epg')}
{...form.getInputProps('create_epg', {
type: 'checkbox',
})}
/>
</Group>
)}
<Group justify="space-between">
<Box>Enable VOD Scanning</Box>
<Switch
id="enable_vod"
name="enable_vod"
description="Scan and import VOD content (movies/series) from this Xtream account"
key={form.key('enable_vod')}
{...form.getInputProps('enable_vod', {
type: 'checkbox',
})}
/>
</Group>
<>
<TextInput
id="username"
name="username"
@ -298,7 +270,6 @@ const M3U = ({
description="Username for Xtream Codes authentication"
{...form.getInputProps('username')}
/>
<PasswordInput
id="password"
name="password"
@ -306,7 +277,7 @@ const M3U = ({
description="Password for Xtream Codes authentication (leave empty to keep existing)"
{...form.getInputProps('password')}
/>
</Box>
</>
)}
{form.getValues().account_type != 'XC' && (
@ -326,7 +297,6 @@ const M3U = ({
},
}}
/>
<DateTimePicker
label="Expiration Date"
description="Set an expiration date to receive a warning notification"
@ -342,9 +312,8 @@ const M3U = ({
<Divider size="sm" orientation="vertical" />
<Stack gap="5" style={{ flex: 1 }}>
<Stack gap="xs" style={{ flex: 1, minWidth: 0 }}>
<NumberInput
style={{ width: '100%' }}
id="max_streams"
name="max_streams"
label="Max Streams"
@ -354,7 +323,41 @@ const M3U = ({
{...form.getInputProps('max_streams')}
key={form.key('max_streams')}
/>
<Select
id="server_group"
name="server_group"
label="Server Group"
description="Share login limits across accounts in a server group. Set max streams on each profile (unlimited profiles skip group enforcement)."
key={form.key('server_group')}
value={form.getValues().server_group}
onChange={(value) => {
if (value === '__new__') {
setServerGroupsCreateOnOpen(true);
setServerGroupsManagerOpen(true);
return;
}
form.setFieldValue('server_group', value);
}}
data={[
{ value: '0', label: '(None)' },
...serverGroups.map((group) => ({
label: group.name,
value: `${group.id}`,
})),
{ value: '__new__', label: '+ Add server group...' },
]}
/>
<Button
variant="subtle"
size="compact-xs"
onClick={() => {
setServerGroupsCreateOnOpen(false);
setServerGroupsManagerOpen(true);
}}
style={{ alignSelf: 'flex-start' }}
>
Manage server groups
</Button>
<Select
id="user_agent"
name="user_agent"
@ -369,7 +372,11 @@ const M3U = ({
}))
)}
/>
</Stack>
<Divider size="sm" orientation="vertical" />
<Stack gap="xs" style={{ flex: 1, minWidth: 0 }}>
<ScheduleInput
scheduleType={scheduleType}
onScheduleTypeChange={setScheduleType}
@ -390,7 +397,6 @@ const M3U = ({
</>
}
/>
<NumberInput
min={0}
max={365}
@ -399,74 +405,114 @@ const M3U = ({
{...form.getInputProps('stale_stream_days')}
/>
<NumberInput
min={0}
max={999}
label="VOD Priority"
description="Priority for VOD provider selection (higher numbers = higher priority). Used when multiple providers offer the same content."
{...form.getInputProps('priority')}
key={form.key('priority')}
/>
{form.getValues().account_type == 'XC' && (
<Box>
<NumberInput
min={0}
max={999}
label="VOD Priority"
description="Priority for VOD provider selection (higher numbers = higher priority). Used when multiple providers offer the same content."
{...form.getInputProps('priority')}
key={form.key('priority')}
/>
<Group justify="space-between">
<Box>Enable VOD Scanning</Box>
<Switch
id="enable_vod"
name="enable_vod"
description="Scan and import VOD content (movies/series) from this Xtream account"
key={form.key('enable_vod')}
{...form.getInputProps('enable_vod', {
type: 'checkbox',
})}
/>
</Group>
{!m3uAccount && (
<Group justify="space-between">
<Box>Create EPG</Box>
<Switch
id="create_epg"
name="create_epg"
description="Automatically create matching EPG source for this Xtream account"
key={form.key('create_epg')}
{...form.getInputProps('create_epg', {
type: 'checkbox',
})}
/>
</Group>
)}
</Box>
)}
</Stack>
</Group>
<Divider my="md" />
<Flex gap="xl" wrap="wrap">
<Checkbox
<Flex
gap="md"
justify="space-between"
align="center"
wrap="wrap"
mih={50}
>
<Switch
id="is_active"
name="is_active"
label="Is Active"
description="Enable or disable this M3U account"
{...form.getInputProps('is_active', { type: 'checkbox' })}
key={form.key('is_active')}
{...form.getInputProps('is_active', { type: 'checkbox' })}
/>
</Flex>
<Flex mih={50} gap="xs" justify="flex-end" align="flex-end">
{playlist && (
<>
<Button
variant="filled"
size="sm"
onClick={() => setFilterModalOpen(true)}
>
Filters
</Button>
<Button
variant="filled"
// color={theme.custom.colors.buttonPrimary}
size="sm"
onClick={() => {
// If this is an XC account with VOD enabled, fetch VOD categories
if (
m3uAccount?.account_type === 'XC' &&
m3uAccount?.enable_vod
) {
fetchCategories();
}
setGroupFilterModalOpen(true);
}}
>
Groups
</Button>
<Button
variant="filled"
// color={theme.custom.colors.buttonPrimary}
size="sm"
onClick={() => setProfileModalOpen(true)}
>
Profiles
</Button>
</>
)}
<Flex gap="xs" align="center">
{playlist && (
<>
<Button
variant="filled"
size="sm"
onClick={() => setFilterModalOpen(true)}
>
Filters
</Button>
<Button
variant="filled"
// color={theme.custom.colors.buttonPrimary}
size="sm"
onClick={() => {
// If this is an XC account with VOD enabled, fetch VOD categories
if (
m3uAccount?.account_type === 'XC' &&
m3uAccount?.enable_vod
) {
fetchCategories();
}
setGroupFilterModalOpen(true);
}}
>
Groups
</Button>
<Button
variant="filled"
// color={theme.custom.colors.buttonPrimary}
size="sm"
onClick={() => setProfileModalOpen(true)}
>
Profiles
</Button>
</>
)}
<Button
type="submit"
variant="filled"
disabled={form.submitting}
size="sm"
>
Save
</Button>
<Button
type="submit"
variant="filled"
disabled={form.submitting}
size="sm"
>
Save
</Button>
</Flex>
</Flex>
</form>
</Modal>
@ -489,6 +535,20 @@ const M3U = ({
/>
</>
)}
<ServerGroupsManagerModal
isOpen={serverGroupsManagerOpen}
onClose={() => {
setServerGroupsManagerOpen(false);
setServerGroupsCreateOnOpen(false);
}}
openCreateOnMount={serverGroupsCreateOnOpen}
onGroupCreated={(group) => {
if (group?.id) {
form.setFieldValue('server_group', `${group.id}`);
}
}}
/>
</>
);
};

86
frontend/src/components/forms/ServerGroup.jsx Normal file
View file

@ -0,0 +1,86 @@
import React, { useEffect, useMemo } from 'react';
import { useForm } from 'react-hook-form';
import { yupResolver } from '@hookform/resolvers/yup';
import * as Yup from 'yup';
import API from '../../api';
import { Button, Flex, Modal, TextInput } from '@mantine/core';
const schema = Yup.object({
name: Yup.string().required('Name is required'),
});
const ServerGroupForm = ({
serverGroup = null,
isOpen,
onClose,
onSaved,
}) => {
const defaultValues = useMemo(
() => ({
name: serverGroup?.name || '',
}),
[serverGroup]
);
const {
register,
handleSubmit,
formState: { errors, isSubmitting },
reset,
} = useForm({
defaultValues,
resolver: yupResolver(schema),
});
const onSubmit = async (values) => {
let response;
if (serverGroup?.id) {
response = await API.updateServerGroup({ id: serverGroup.id, ...values });
} else {
response = await API.addServerGroup(values);
}
if (response) {
onSaved?.(response);
}
reset();
onClose();
};
useEffect(() => {
reset(defaultValues);
}, [defaultValues, reset]);
if (!isOpen) {
return null;
}
return (
<Modal
opened={isOpen}
onClose={onClose}
title="Server Group"
centered
withinPortal
zIndex={400}
>
<form onSubmit={handleSubmit(onSubmit)}>
<TextInput
label="Name"
description="Accounts in this group share connection limits when they use the same provider login. Limits come from each account profile's max streams."
{...register('name')}
error={errors.name?.message}
/>
<Flex mih={50} gap="xs" justify="flex-end" align="flex-end">
<Button size="small" type="submit" disabled={isSubmitting}>
Submit
</Button>
</Flex>
</form>
</Modal>
);
};
export default ServerGroupForm;

47
frontend/src/components/tables/M3UsTable.jsx
View file

@ -8,6 +8,7 @@ import React, {
import API from '../../api';
import usePlaylistsStore from '../../store/playlists';
import M3UForm from '../forms/M3U';
import ServerGroupsManagerModal from '../ServerGroupsManagerModal';
import { TableHelper } from '../../helpers';
import {
useMantineTheme,
@ -144,6 +145,7 @@ const M3UTable = () => {
const [data, setData] = useState([]);
const [sorting, setSorting] = useState([{ id: 'name', desc: '' }]);
const [deleting, setDeleting] = useState(false);
const [serverGroupsManagerOpen, setServerGroupsManagerOpen] = useState(false);
const playlists = usePlaylistsStore((s) => s.playlists);
const refreshProgress = usePlaylistsStore((s) => s.refreshProgress);
@ -988,21 +990,31 @@ const M3UTable = () => {
>
M3U Accounts
</Text>
<Button
leftSection={<SquarePlus size={14} />}
variant="light"
size="xs"
onClick={() => editPlaylist()}
p={5}
color="green"
style={{
borderWidth: '1px',
borderColor: 'green',
color: 'white',
}}
>
Add M3U
</Button>
<Flex gap={6}>
<Button
variant="light"
size="xs"
onClick={() => setServerGroupsManagerOpen(true)}
p={5}
>
Server Groups
</Button>
<Button
leftSection={<SquarePlus size={14} />}
variant="light"
size="xs"
onClick={() => editPlaylist()}
p={5}
color="green"
style={{
borderWidth: '1px',
borderColor: 'green',
color: 'white',
}}
>
Add M3U
</Button>
</Flex>
</Flex>
<Paper
@ -1051,6 +1063,11 @@ const M3UTable = () => {
playlistCreated={playlistCreated}
/>
<ServerGroupsManagerModal
isOpen={serverGroupsManagerOpen}
onClose={() => setServerGroupsManagerOpen(false)}
/>
<ConfirmationDialog
opened={confirmDeleteOpen}
onClose={() => setConfirmDeleteOpen(false)}

298
frontend/src/components/tables/ServerGroupsTable.jsx Normal file
View file

@ -0,0 +1,298 @@
import { useEffect, useMemo, useRef, useState } from 'react';
import API from '../../api';
import useServerGroupsStore from '../../store/serverGroups';
import usePlaylistsStore from '../../store/playlists';
import useWarningsStore from '../../store/warnings';
import ServerGroupForm from '../forms/ServerGroup';
import ConfirmationDialog from '../ConfirmationDialog';
import {
ActionIcon,
Box,
Button,
Center,
Flex,
Stack,
Text,
} from '@mantine/core';
import { SquareMinus, SquarePen, SquarePlus } from 'lucide-react';
import { CustomTable, useTable } from './CustomTable';
import './table.css';
const TABLE_WIDTH = 360;
const ACTIONS_COLUMN_SIZE = 76;
const ACCOUNTS_COLUMN_SIZE = 72;
const RowActions = ({ row, editServerGroup, deleteServerGroup }) => (
<Flex gap={4} justify="center" w="100%">
<ActionIcon
variant="transparent"
size="sm"
color="yellow.5"
onClick={() => editServerGroup(row.original)}
>
<SquarePen size="18" />
</ActionIcon>
<ActionIcon
variant="transparent"
size="sm"
color="red.9"
onClick={() => deleteServerGroup(row.original.id)}
>
<SquareMinus size="18" />
</ActionIcon>
</Flex>
);
const ServerGroupsTable = ({ onGroupCreated, openCreateOnMount = false }) => {
const [serverGroup, setServerGroup] = useState(null);
const [serverGroupModalOpen, setServerGroupModalOpen] = useState(false);
const [confirmDeleteOpen, setConfirmDeleteOpen] = useState(false);
const [deleteTarget, setDeleteTarget] = useState(null);
const [groupToDelete, setGroupToDelete] = useState(null);
const [deleting, setDeleting] = useState(false);
const openedCreateOnMount = useRef(false);
const isWarningSuppressed = useWarningsStore((s) => s.isWarningSuppressed);
const suppressWarning = useWarningsStore((s) => s.suppressWarning);
const serverGroups = useServerGroupsStore((state) => state.serverGroups);
const isLoading = useServerGroupsStore((state) => state.isLoading);
const error = useServerGroupsStore((state) => state.error);
const fetchServerGroups = useServerGroupsStore(
(state) => state.fetchServerGroups
);
const playlists = usePlaylistsStore((state) => state.playlists);
const tableData = useMemo(
() =>
serverGroups.map((group) => ({
...group,
accountCount: playlists.filter(
(playlist) => playlist.server_group === group.id
).length,
})),
[serverGroups, playlists]
);
const columns = useMemo(
() => [
{
header: 'Name',
accessorKey: 'name',
grow: true,
minSize: 100,
cell: ({ cell }) => (
<Text size="sm" truncate>
{cell.getValue()}
</Text>
),
},
{
header: 'Accounts',
accessorKey: 'accountCount',
size: ACCOUNTS_COLUMN_SIZE,
minSize: 65,
cell: ({ cell }) => (
<Center w="100%">
<Text size="sm">{cell.getValue() ?? 0}</Text>
</Center>
),
},
{
id: 'actions',
header: 'Actions',
size: ACTIONS_COLUMN_SIZE,
minSize: 65,
},
],
[]
);
const editServerGroup = (group = null) => {
setServerGroup(group);
setServerGroupModalOpen(true);
};
const executeDeleteServerGroup = async (id) => {
setDeleting(true);
try {
await API.deleteServerGroup(id);
} finally {
setDeleting(false);
setConfirmDeleteOpen(false);
}
};
const deleteServerGroup = (id) => {
const group = tableData.find((item) => item.id === id);
setGroupToDelete(group);
setDeleteTarget(id);
if (isWarningSuppressed('delete-server-group')) {
return executeDeleteServerGroup(id);
}
setConfirmDeleteOpen(true);
};
const closeServerGroupForm = () => {
setServerGroup(null);
setServerGroupModalOpen(false);
};
const handleServerGroupSaved = (savedGroup) => {
if (!serverGroup?.id) {
onGroupCreated?.(savedGroup);
}
};
useEffect(() => {
fetchServerGroups();
}, [fetchServerGroups]);
useEffect(() => {
if (!openCreateOnMount) {
openedCreateOnMount.current = false;
return;
}
if (!isLoading && !openedCreateOnMount.current) {
openedCreateOnMount.current = true;
editServerGroup();
}
}, [openCreateOnMount, isLoading]);
const renderHeaderCell = (header) => (
<Text size="sm" name={header.id}>
{header.column.columnDef.header}
</Text>
);
const renderBodyCell = ({ row }) => (
<RowActions
row={row}
editServerGroup={editServerGroup}
deleteServerGroup={deleteServerGroup}
/>
);
const table = useTable({
columns,
data: tableData,
allRowIds: tableData.map((group) => group.id),
bodyCellRenderFns: {
actions: renderBodyCell,
},
headerCellRenderFns: {
name: renderHeaderCell,
accountCount: renderHeaderCell,
actions: renderHeaderCell,
},
});
if (isLoading) {
return (
<Center py="md">
<Text size="sm">Loading server groups...</Text>
</Center>
);
}
if (error) {
return (
<Center py="md">
<Text size="sm" c="red">
{error}
</Text>
</Center>
);
}
return (
<Stack gap="md">
<Text size="sm" c="dimmed">
Group accounts that share the same provider login so their connection
limits are enforced together. Assign a group when editing an M3U
account.
</Text>
<Flex justify="center">
<Stack gap="xs" w={TABLE_WIDTH} maw="100%">
<Flex justify="flex-end">
<Button
leftSection={<SquarePlus size={18} />}
variant="light"
size="xs"
onClick={() => editServerGroup()}
p={5}
color="green"
title="Create a shared connection pool for multiple accounts"
style={{
borderWidth: '1px',
borderColor: 'green',
color: 'white',
}}
>
Add Server Group
</Button>
</Flex>
<Box
style={{
maxHeight: 280,
overflowY: 'auto',
overflowX: 'auto',
border: 'solid 1px rgb(68,68,68)',
borderRadius: 'var(--mantine-radius-default)',
}}
>
{tableData.length === 0 ? (
<Center py="lg" px="xl">
<Text size="sm" c="dimmed">
No server groups yet.
</Text>
</Center>
) : (
<CustomTable table={table} />
)}
</Box>
</Stack>
</Flex>
<ServerGroupForm
serverGroup={serverGroup}
isOpen={serverGroupModalOpen}
onClose={closeServerGroupForm}
onSaved={handleServerGroupSaved}
/>
<ConfirmationDialog
opened={confirmDeleteOpen}
onClose={() => setConfirmDeleteOpen(false)}
onConfirm={() => executeDeleteServerGroup(deleteTarget)}
loading={deleting}
title="Confirm Server Group Deletion"
message={
groupToDelete ? (
<div style={{ whiteSpace: 'pre-line' }}>
{`Are you sure you want to delete the following server group?
Name: ${groupToDelete.name}
Accounts: ${groupToDelete.accountCount ?? 0}
Accounts in this group will no longer share connection limits. This action cannot be undone.`}
</div>
) : (
'Are you sure you want to delete this server group? This action cannot be undone.'
)
}
confirmLabel="Delete"
cancelLabel="Cancel"
actionKey="delete-server-group"
onSuppressChange={suppressWarning}
zIndex={401}
/>
</Stack>
);
};
export default ServerGroupsTable;

83
frontend/src/components/tables/__tests__/ServerGroupsTable.test.jsx Normal file
View file

@ -0,0 +1,83 @@
import { describe, it, expect, vi, beforeEach } from 'vitest';
import { render, screen, waitFor } from '@testing-library/react';
import { MantineProvider } from '@mantine/core';
import ServerGroupsTable from '../ServerGroupsTable';
const mockServerGroupsState = vi.hoisted(() => ({
serverGroups: [
{ id: 1, name: 'Pool A' },
{ id: 2, name: 'Pool B' },
],
isLoading: false,
error: null,
fetchServerGroups: vi.fn().mockResolvedValue(undefined),
}));
const renderTable = () =>
render(
<MantineProvider>
<ServerGroupsTable />
</MantineProvider>
);
vi.mock('../../../api', () => ({
default: {
deleteServerGroup: vi.fn(),
},
}));
vi.mock('../../../store/serverGroups', () => ({
default: (selector) => selector(mockServerGroupsState),
}));
vi.mock('../../../store/playlists', () => ({
default: (selector) =>
selector({
playlists: [
{ id: 10, server_group: 2 },
{ id: 11, server_group: 1 },
],
}),
}));
vi.mock('../../forms/ServerGroup', () => ({
default: () => null,
}));
describe('ServerGroupsTable', () => {
beforeEach(() => {
vi.clearAllMocks();
mockServerGroupsState.serverGroups = [
{ id: 1, name: 'Pool A' },
{ id: 2, name: 'Pool B' },
];
mockServerGroupsState.isLoading = false;
mockServerGroupsState.error = null;
mockServerGroupsState.fetchServerGroups = vi
.fn()
.mockResolvedValue(undefined);
});
it('renders server groups without tooltip errors', async () => {
expect(() => renderTable()).not.toThrow();
await waitFor(() => {
expect(screen.getByText('Pool A')).toBeInTheDocument();
expect(screen.getByText('Pool B')).toBeInTheDocument();
expect(screen.getByText('Accounts')).toBeInTheDocument();
});
});
it('shows store error message when fetch fails', async () => {
mockServerGroupsState.serverGroups = [];
mockServerGroupsState.error = 'Failed to load server groups.';
renderTable();
await waitFor(() => {
expect(
screen.getByText('Failed to load server groups.')
).toBeInTheDocument();
});
});
});

2
frontend/src/store/auth.jsx
View file

@ -6,6 +6,7 @@ import useEPGsStore from './epgs';
import useStreamProfilesStore from './streamProfiles';
import useOutputProfilesStore from './outputProfiles';
import useUserAgentsStore from './userAgents';
import useServerGroupsStore from './serverGroups';
import useUsersStore from './users';
import API from '../api';
import { USER_LEVELS } from '../constants';
@ -124,6 +125,7 @@ const useAuthStore = create((set, get) => ({
useStreamProfilesStore.getState().fetchProfiles(),
useOutputProfilesStore.getState().fetchProfiles(),
useUserAgentsStore.getState().fetchUserAgents(),
useServerGroupsStore.getState().fetchServerGroups(),
useChannelsStore.getState().fetchChannelIds(),
]);

40
frontend/src/store/serverGroups.jsx Normal file
View file

@ -0,0 +1,40 @@
import { create } from 'zustand';
import api from '../api';
const useServerGroupsStore = create((set) => ({
serverGroups: [],
isLoading: false,
error: null,
fetchServerGroups: async () => {
set({ isLoading: true, error: null });
try {
const serverGroups = await api.getServerGroups();
set({ serverGroups: serverGroups || [], isLoading: false });
} catch (error) {
console.error('Failed to fetch server groups:', error);
set({ error: 'Failed to load server groups.', isLoading: false });
}
},
addServerGroup: (serverGroup) =>
set((state) => ({
serverGroups: [...state.serverGroups, serverGroup],
})),
updateServerGroup: (serverGroup) =>
set((state) => ({
serverGroups: state.serverGroups.map((group) =>
group.id === serverGroup.id ? serverGroup : group
),
})),
removeServerGroups: (serverGroupIds) =>
set((state) => ({
serverGroups: state.serverGroups.filter(
(group) => !serverGroupIds.includes(group.id)
),
})),
}));
export default useServerGroupsStore;

4
frontend/src/utils/forms/M3uUtils.js
View file

@ -50,5 +50,9 @@ export const prepareSubmitValues = (values, expDate) => {
prepared.user_agent = null;
}
if (prepared.server_group == '0') {
prepared.server_group = null;
}
return prepared;
};