diff --git a/CHANGELOG.md b/CHANGELOG.md index 22b27745..19ff2f53 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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. diff --git a/apps/channels/models.py b/apps/channels/models.py index 320470bb..f900a817 100644 --- a/apps/channels/models.py +++ b/apps/channels/models.py @@ -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() diff --git a/apps/channels/tests/test_get_stream_assignment.py b/apps/channels/tests/test_get_stream_assignment.py new file mode 100644 index 00000000..bd5fea2e --- /dev/null +++ b/apps/channels/tests/test_get_stream_assignment.py @@ -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) + ) diff --git a/apps/m3u/connection_pool.py b/apps/m3u/connection_pool.py new file mode 100644 index 00000000..375714ee --- /dev/null +++ b/apps/m3u/connection_pool.py @@ -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) diff --git a/apps/m3u/models.py b/apps/m3u/models.py index 6fc48bec..53c18d38 100644 --- a/apps/m3u/models.py +++ b/apps/m3u/models.py @@ -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." diff --git a/apps/m3u/tests/test_connection_pool.py b/apps/m3u/tests/test_connection_pool.py new file mode 100644 index 00000000..443c5bd8 --- /dev/null +++ b/apps/m3u/tests/test_connection_pool.py @@ -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) diff --git a/apps/proxy/live_proxy/url_utils.py b/apps/proxy/live_proxy/url_utils.py index 5227a3fa..7cdc92ef 100644 --- a/apps/proxy/live_proxy/url_utils.py +++ b/apps/proxy/live_proxy/url_utils.py @@ -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 diff --git a/apps/proxy/live_proxy/views.py b/apps/proxy/live_proxy/views.py index d882bfca..d60d9703 100644 --- a/apps/proxy/live_proxy/views.py +++ b/apps/proxy/live_proxy/views.py @@ -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 diff --git a/apps/proxy/vod_proxy/multi_worker_connection_manager.py b/apps/proxy/vod_proxy/multi_worker_connection_manager.py index 7d367531..8d803dcf 100644 --- a/apps/proxy/vod_proxy/multi_worker_connection_manager.py +++ b/apps/proxy/vod_proxy/multi_worker_connection_manager.py @@ -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}") diff --git a/apps/proxy/vod_proxy/views.py b/apps/proxy/vod_proxy/views.py index 9b330197..27208ec9 100644 --- a/apps/proxy/vod_proxy/views.py +++ b/apps/proxy/vod_proxy/views.py @@ -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}") diff --git a/frontend/src/api.js b/frontend/src/api.js index 8fd95fd9..1c826465 100644 --- a/frontend/src/api.js +++ b/frontend/src/api.js @@ -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}/`); diff --git a/frontend/src/components/ServerGroupsManagerModal.jsx b/frontend/src/components/ServerGroupsManagerModal.jsx new file mode 100644 index 00000000..2d403aa1 --- /dev/null +++ b/frontend/src/components/ServerGroupsManagerModal.jsx @@ -0,0 +1,28 @@ +import { Modal } from '@mantine/core'; +import ServerGroupsTable from './tables/ServerGroupsTable'; + +const ServerGroupsManagerModal = ({ + isOpen, + onClose, + onGroupCreated, + openCreateOnMount = false, +}) => { + return ( + + {isOpen ? ( + + ) : null} + + ); +}; + +export default ServerGroupsManagerModal; diff --git a/frontend/src/components/__tests__/ServerGroupsManagerModal.test.jsx b/frontend/src/components/__tests__/ServerGroupsManagerModal.test.jsx new file mode 100644 index 00000000..c35b1be1 --- /dev/null +++ b/frontend/src/components/__tests__/ServerGroupsManagerModal.test.jsx @@ -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( + + + + ); + +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(); + }); + }); +}); diff --git a/frontend/src/components/forms/M3U.jsx b/frontend/src/components/forms/M3U.jsx index e27570db..07b430ef 100644 --- a/frontend/src/components/forms/M3U.jsx +++ b/frontend/src/components/forms/M3U.jsx @@ -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 ( <>
- - + + - { + 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...' }, + ]} + /> +