Dispatcharr/core/utils.py

import redis
import logging
import time
import os
import threading
from pathlib import Path
import re
from django.conf import settings
from redis.exceptions import ConnectionError, TimeoutError
from django.core.cache import cache
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
from django.core.validators import URLValidator
from django.core.exceptions import ValidationError
import gc

_REDIS_TLS_HINT = " (TLS is enabled — verify certificate paths and that Redis is configured for TLS)"

logger = logging.getLogger(__name__)

# Import the command detector
from .command_utils import is_management_command


def dispatcharr_user_agent():
    """Return the standard Dispatcharr User-Agent string (Dispatcharr/{version})."""
    from version import __version__
    return f'Dispatcharr/{__version__}'


def dispatcharr_dvr_user_agent(recording_id):
    """Return the User-Agent string used by DVR FFmpeg clients for a recording."""
    return f'Dispatcharr-DVR/recording-{recording_id}'


def dispatcharr_http_headers(*, token=None, content_type='application/json'):
    """
    Build HTTP headers for outbound Dispatcharr requests.

    content_type=None omits Content-Type (e.g. simple GET proxies).
    token is included when authenticating with Schedules Direct.
    """
    headers = {'User-Agent': dispatcharr_user_agent()}
    if content_type:
        headers['Content-Type'] = content_type
    if token:
        headers['token'] = token
    return headers


def natural_sort_key(text):
    """
    Convert a string into a list of string and number chunks for natural sorting.
    "PPV 10" becomes ['PPV ', 10] so it sorts correctly with "PPV 2".

    This function enables natural/alphanumeric sorting where numbers within strings
    are treated as actual numbers rather than strings.

    Args:
        text (str): The text to convert for sorting

    Returns:
        list: A list of strings and integers for proper sorting

    Example:
        >>> sorted(['PPV 1', 'PPV 10', 'PPV 2'], key=natural_sort_key)
        ['PPV 1', 'PPV 2', 'PPV 10']
    """
    def convert(chunk):
        return int(chunk) if chunk.isdigit() else chunk.lower()

    return [convert(c) for c in re.split('([0-9]+)', text)]

class RedisClient:
    _client = None
    _buffer = None
    _pubsub_client = None

    @classmethod
    def _init_client(cls, decode_responses=True, max_retries=5, retry_interval=1):
        retry_count = 0
        while retry_count < max_retries:
            try:
                # Get connection parameters from settings or environment
                redis_host = os.environ.get("REDIS_HOST", getattr(settings, 'REDIS_HOST', 'localhost'))
                redis_port = int(os.environ.get("REDIS_PORT", getattr(settings, 'REDIS_PORT', 6379)))
                redis_db = int(os.environ.get("REDIS_DB", getattr(settings, 'REDIS_DB', 0)))
                redis_password = os.environ.get("REDIS_PASSWORD", getattr(settings, 'REDIS_PASSWORD', ''))
                redis_user = os.environ.get("REDIS_USER", getattr(settings, 'REDIS_USER', ''))

                # Use standardized settings
                socket_timeout = getattr(settings, 'REDIS_SOCKET_TIMEOUT', 5)
                socket_connect_timeout = getattr(settings, 'REDIS_SOCKET_CONNECT_TIMEOUT', 5)
                health_check_interval = getattr(settings, 'REDIS_HEALTH_CHECK_INTERVAL', 30)
                socket_keepalive = getattr(settings, 'REDIS_SOCKET_KEEPALIVE', True)
                retry_on_timeout = getattr(settings, 'REDIS_RETRY_ON_TIMEOUT', True)

                # TLS params from settings (empty dict when TLS is disabled)
                ssl_params = getattr(settings, 'REDIS_SSL_PARAMS', {})

                # Create Redis client with better defaults
                client = redis.Redis(
                    host=redis_host,
                    port=redis_port,
                    db=redis_db,
                    password=redis_password if redis_password else None,
                    username=redis_user if redis_user else None,
                    socket_timeout=socket_timeout,
                    socket_connect_timeout=socket_connect_timeout,
                    socket_keepalive=socket_keepalive,
                    health_check_interval=health_check_interval,
                    retry_on_timeout=retry_on_timeout,
                    decode_responses=decode_responses,
                    **ssl_params
                )

                # Validate connection with ping
                client.ping()

                # Disable persistence on first connection - improves performance
                # Only try to disable if not in a read-only environment
                try:
                    client.config_set('save', '')  # Disable RDB snapshots
                    client.config_set('appendonly', 'no')  # Disable AOF logging

                    # Disable protected mode when in debug mode
                    if os.environ.get('DISPATCHARR_DEBUG', '').lower() == 'true':
                        client.config_set('protected-mode', 'no')  # Disable protected mode in debug
                        logger.warning("Redis protected mode disabled for debug environment")

                    logger.trace("Redis persistence disabled for better performance")
                except redis.exceptions.ResponseError as e:
                    # Improve error handling for Redis configuration errors
                    if "OOM" in str(e):
                        logger.error(f"Redis OOM during configuration: {e}")
                        # Try to increase maxmemory as an emergency measure
                        try:
                            client.config_set('maxmemory', '768mb')
                            logger.warning("Applied emergency Redis memory increase to 768MB")
                        except:
                            pass
                    else:
                        logger.error(f"Redis configuration error: {e}")

                logger.info(f"Connected to Redis at {redis_host}:{redis_port}/{redis_db}")

                return client

            except (ConnectionError, TimeoutError) as e:
                retry_count += 1
                _tls_hint = _REDIS_TLS_HINT if ssl_params else ""
                if retry_count >= max_retries:
                    logger.error(f"Failed to connect to Redis after {max_retries} attempts: {e}{_tls_hint}")
                    return None
                else:
                    # Use exponential backoff for retries
                    wait_time = retry_interval * (2 ** (retry_count - 1))
                    logger.warning(f"Redis connection failed. Retrying in {wait_time}s... ({retry_count}/{max_retries})")
                    time.sleep(wait_time)

            except Exception as e:
                _tls_hint = ""
                try:
                    _tls_hint = _REDIS_TLS_HINT if ssl_params else ""
                except NameError:
                    pass
                logger.error(f"Unexpected error connecting to Redis: {e}{_tls_hint}")
                return None

        return None

    @classmethod
    def get_client(cls, max_retries=5, retry_interval=1):
        """Get Redis client optimized for non-binary data (decoded responses)"""
        if cls._client is None:
            cls._client = cls._init_client(decode_responses=True, max_retries=max_retries, retry_interval=retry_interval)
        return cls._client

    @classmethod
    def get_buffer(cls, max_retries=5, retry_interval=1):
        """Get Redis client optimized for binary data (no decoding)"""
        if cls._buffer is None:
            cls._buffer = cls._init_client(decode_responses=False, max_retries=max_retries, retry_interval=retry_interval)
        return cls._buffer

    @classmethod
    def get_pubsub_client(cls, max_retries=5, retry_interval=1):
        """Get Redis client optimized for PubSub operations"""
        if cls._pubsub_client is None:
            retry_count = 0
            while retry_count < max_retries:
                try:
                    # Get connection parameters from settings or environment
                    redis_host = os.environ.get("REDIS_HOST", getattr(settings, 'REDIS_HOST', 'localhost'))
                    redis_port = int(os.environ.get("REDIS_PORT", getattr(settings, 'REDIS_PORT', 6379)))
                    redis_db = int(os.environ.get("REDIS_DB", getattr(settings, 'REDIS_DB', 0)))
                    redis_password = os.environ.get("REDIS_PASSWORD", getattr(settings, 'REDIS_PASSWORD', ''))
                    redis_user = os.environ.get("REDIS_USER", getattr(settings, 'REDIS_USER', ''))

                    # Use standardized settings but without socket timeouts for PubSub
                    # Important: socket_timeout is None for PubSub operations
                    socket_connect_timeout = getattr(settings, 'REDIS_SOCKET_CONNECT_TIMEOUT', 5)
                    socket_keepalive = getattr(settings, 'REDIS_SOCKET_KEEPALIVE', True)
                    health_check_interval = getattr(settings, 'REDIS_HEALTH_CHECK_INTERVAL', 30)
                    retry_on_timeout = getattr(settings, 'REDIS_RETRY_ON_TIMEOUT', True)

                    ssl_params = getattr(settings, 'REDIS_SSL_PARAMS', {})

                    # Create Redis client with PubSub-optimized settings - no timeout
                    client = redis.Redis(
                        host=redis_host,
                        port=redis_port,
                        db=redis_db,
                        password=redis_password if redis_password else None,
                        username=redis_user if redis_user else None,
                        socket_timeout=None,  # Critical: No timeout for PubSub operations
                        socket_connect_timeout=socket_connect_timeout,
                        socket_keepalive=socket_keepalive,
                        health_check_interval=health_check_interval,
                        retry_on_timeout=retry_on_timeout,
                        decode_responses=True,
                        **ssl_params
                    )

                    # Validate connection with ping
                    client.ping()
                    logger.info(f"Connected to Redis for PubSub at {redis_host}:{redis_port}/{redis_db}")

                    # We don't need the keepalive thread anymore since we're using proper PubSub handling
                    cls._pubsub_client = client
                    break

                except (ConnectionError, TimeoutError) as e:
                    retry_count += 1
                    _tls_hint = _REDIS_TLS_HINT if ssl_params else ""
                    if retry_count >= max_retries:
                        logger.error(f"Failed to connect to Redis for PubSub after {max_retries} attempts: {e}{_tls_hint}")
                        return None
                    else:
                        # Use exponential backoff for retries
                        wait_time = retry_interval * (2 ** (retry_count - 1))
                        logger.warning(f"Redis PubSub connection failed. Retrying in {wait_time}s... ({retry_count}/{max_retries})")
                        time.sleep(wait_time)

                except Exception as e:
                    _tls_hint = _REDIS_TLS_HINT if ssl_params else ""
                    logger.error(f"Unexpected error connecting to Redis for PubSub: {e}{_tls_hint}")
                    return None

        return cls._pubsub_client

def acquire_task_lock(task_name, id):
    """Acquire a lock to prevent concurrent task execution."""
    redis_client = RedisClient.get_client()
    lock_id = f"task_lock_{task_name}_{id}"

    # Use the Redis SET command with NX (only set if not exists) and EX (set expiration)
    lock_acquired = redis_client.set(lock_id, "locked", ex=300, nx=True)

    if not lock_acquired:
        logger.warning(f"Lock for {task_name} and id={id} already acquired. Task will not proceed.")

    return lock_acquired

def release_task_lock(task_name, id):
    """Release the lock after task execution."""
    redis_client = RedisClient.get_client()
    lock_id = f"task_lock_{task_name}_{id}"

    # Remove the lock
    redis_client.delete(lock_id)


def is_task_lock_held(task_name, id):
    """Return True when another worker holds the task lock (read-only check)."""
    redis_client = RedisClient.get_client()
    if redis_client is None:
        return False
    lock_id = f"task_lock_{task_name}_{id}"
    return bool(redis_client.exists(lock_id))


class TaskLockRenewer:
    """Periodically renews a Redis task lock to prevent expiry during long-running tasks.

    Use as a context manager after acquiring a lock:

        if acquire_task_lock("my_task", task_id):
            with TaskLockRenewer("my_task", task_id):
                # ... long-running work ...
            release_task_lock("my_task", task_id)

    A daemon thread extends the lock TTL at regular intervals so that
    slow downloads or large parsing jobs don't lose their lock mid-operation.
    """

    def __init__(self, task_name, id, ttl=300, renewal_interval=120):
        self.task_name = task_name
        self.id = id
        self.ttl = ttl
        self.renewal_interval = renewal_interval
        self.lock_id = f"task_lock_{task_name}_{id}"
        self._stop_event = threading.Event()
        self._thread = None

    def _renew_loop(self):
        """Background loop that extends the lock TTL until stopped."""
        while not self._stop_event.wait(self.renewal_interval):
            try:
                redis_client = RedisClient.get_client()
                if redis_client.exists(self.lock_id):
                    redis_client.expire(self.lock_id, self.ttl)
                    logger.debug(
                        f"Renewed lock {self.lock_id} TTL to {self.ttl}s"
                    )
                else:
                    # Lock was deleted externally (e.g. manual release) — stop renewing
                    logger.warning(
                        f"Lock {self.lock_id} no longer exists, stopping renewal"
                    )
                    break
            except Exception as e:
                logger.error(f"Error renewing lock {self.lock_id}: {e}")

    def start(self):
        """Start the background renewal thread."""
        self._stop_event.clear()
        self._thread = threading.Thread(
            target=self._renew_loop, daemon=True,
            name=f"lock-renew-{self.task_name}-{self.id}"
        )
        self._thread.start()
        return self

    def stop(self):
        """Stop the renewal thread."""
        self._stop_event.set()
        if self._thread and self._thread.is_alive():
            self._thread.join(timeout=5)
        self._thread = None

    def __enter__(self):
        self.start()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.stop()
        return False


def _is_gevent_monkey_patched():
    try:
        import gevent.monkey
        return gevent.monkey.is_module_patched('threading')
    except Exception:
        return False


def _is_celery_worker_context():
    """True when executing inside an active Celery task (prefork worker)."""
    try:
        from celery import current_task
        request = getattr(current_task, 'request', None)
        return bool(request and getattr(request, 'id', None))
    except Exception:
        return False


def _should_use_sync_websocket_send():
    """
    Use synchronous Redis delivery when gevent is monkey-patched but no gevent
    hub is driving the process — e.g. Celery prefork workers that inherit
    gevent patching from uWSGI imports. gevent.spawn in that context schedules
    coroutines that never run.
    """
    return _is_gevent_monkey_patched() and _is_celery_worker_context()


def _gevent_ws_send(group_name, message):
    """
    Publishes a WebSocket group message synchronously through Redis.

    gevent's monkey-patching removes select.epoll, which breaks asyncio event
    loop creation in threadpool threads. This function replicates channels_redis
    4.x group_send directly via a sync Redis client, avoiding asyncio entirely.

    Matches channels_redis 4.x defaults: prefix="asgi", expiry=60,
    group_expiry=86400, msgpack serializer with 12-byte random prefix.
    """
    try:
        import msgpack
        redis = RedisClient.get_buffer()  # decode_responses=False for binary values

        prefix = "asgi"
        group_expiry = 86400
        channel_expiry = 60
        rand_len = 12

        group_key = f"{prefix}:group:{group_name}"
        now = time.time()

        redis.zremrangebyscore(group_key, 0, now - group_expiry)
        raw = redis.zrange(group_key, 0, -1)
        if not raw:
            return

        channels = [m.decode('utf-8') if isinstance(m, bytes) else m for m in raw]

        # Group channels by non-local name (prefix up to and including "!") so
        # specific channels sharing a prefix share one Redis sorted-set key.
        nonlocal_map = {}
        for ch in channels:
            pos = ch.find("!")
            nl = ch[:pos + 1] if pos >= 0 else ch
            nonlocal_map.setdefault(nl, []).append(ch)

        pipe = redis.pipeline(transaction=False)
        for nl, chs in nonlocal_map.items():
            channel_key = prefix + nl
            msg = dict(message)
            msg["__asgi_channel__"] = chs
            serialized = os.urandom(rand_len) + msgpack.packb(msg)
            pipe.zadd(channel_key, {serialized: now})
            pipe.expire(channel_key, channel_expiry)
        pipe.execute()
    except Exception as e:
        logger.warning(f"Failed to send WebSocket update: {e}")


def send_websocket_update_sync(group_name, event_type, data):
    """Send a WebSocket group message synchronously via Redis (channels_redis wire format)."""
    message = {'type': event_type, 'data': data}
    _gevent_ws_send(group_name, message)


def send_websocket_update(group_name, event_type, data, collect_garbage=False):
    """
    Sends a WebSocket group message.

    In gevent-patched uWSGI workers, asyncio event loop creation fails because
    monkey-patching removes select.epoll. For those contexts a synchronous Redis
    path is used instead, matching the channels_redis 4.x wire format.

    Celery prefork workers may inherit gevent monkey-patching without a running
    gevent hub; in that case gevent.spawn would never execute, so delivery is
    synchronous via Redis instead.
    """
    message = {'type': event_type, 'data': data}

    if _should_use_sync_websocket_send():
        _gevent_ws_send(group_name, message)
    elif _is_gevent_monkey_patched():
        import gevent
        gevent.spawn(_gevent_ws_send, group_name, message)
    else:
        # Not gevent-patched (plain Celery, tests) — use asyncio channel layer
        try:
            async_to_sync(get_channel_layer().group_send)(group_name, message)
        except Exception as e:
            logger.warning(f"Failed to send WebSocket update: {e}")

    if collect_garbage:
        gc.collect()

def send_websocket_event(event, success, data):
    """Acquire a lock to prevent concurrent task execution."""
    data_payload = {"success": success, "type": event}
    if data:
        # Make a copy to avoid modifying the original
        data_payload.update(data)

    # Use the standardized function
    send_websocket_update('updates', 'update', data_payload)

    # Help garbage collection by clearing references
    data_payload = None

# Add memory monitoring utilities
def get_memory_usage():
    """Returns current memory usage in MB"""
    import psutil
    process = psutil.Process(os.getpid())
    return process.memory_info().rss / (1024 * 1024)

def monitor_memory_usage(func):
    """Decorator to monitor memory usage before and after function execution"""
    def wrapper(*args, **kwargs):
        import gc
        # Force garbage collection before measuring
        gc.collect()

        # Get initial memory usage
        start_mem = get_memory_usage()
        logger.debug(f"Memory usage before {func.__name__}: {start_mem:.2f} MB")

        # Call the original function
        result = func(*args, **kwargs)

        # Force garbage collection before measuring again
        gc.collect()

        # Get final memory usage
        end_mem = get_memory_usage()
        logger.debug(f"Memory usage after {func.__name__}: {end_mem:.2f} MB (Change: {end_mem - start_mem:.2f} MB)")

        return result
    return wrapper

def cleanup_memory(log_usage=False, force_collection=True):
    """
    Comprehensive memory cleanup function to reduce memory footprint

    Args:
        log_usage: Whether to log memory usage before and after cleanup
        force_collection: Whether to force garbage collection
    """
    logger.trace("Starting memory cleanup django memory cleanup")
    # Skip logging if log level is not set to debug or more verbose (like trace)
    current_log_level = logger.getEffectiveLevel()
    if not current_log_level <= logging.DEBUG:
        log_usage = False
    if log_usage:
        try:
            import psutil
            process = psutil.Process()
            before_mem = process.memory_info().rss / (1024 * 1024)
            logger.debug(f"Memory before cleanup: {before_mem:.2f} MB")
        except (ImportError, Exception) as e:
            logger.debug(f"Error getting memory usage: {e}")

    # Clear any object caches from Django ORM
    from django.db import connection, reset_queries
    reset_queries()

    # Force garbage collection
    if force_collection:
        # Run full collection
        gc.collect(generation=2)
        # Clear cyclic references
        gc.collect(generation=0)

    if log_usage:
        try:
            import psutil
            process = psutil.Process()
            after_mem = process.memory_info().rss / (1024 * 1024)
            logger.debug(f"Memory after cleanup: {after_mem:.2f} MB (change: {after_mem-before_mem:.2f} MB)")
        except (ImportError, Exception):
            pass
    logger.trace("Memory cleanup complete for django")

def safe_upload_path(filename: str, base_dir) -> str:
    """Return a safe absolute path for an uploaded file within base_dir.

    Strips all directory components from *filename* and verifies the resolved
    path stays inside *base_dir*.  Raises ValueError on path traversal attempts.
    """
    safe_name = Path(filename).name
    base = Path(base_dir).resolve()
    file_path = (base / safe_name).resolve()
    if not file_path.is_relative_to(base):
        raise ValueError("Invalid filename.")
    return str(file_path)


def is_protected_path(file_path):
    """
    Determine if a file path is in a protected directory that shouldn't be deleted.

    Args:
        file_path (str): The file path to check

    Returns:
        bool: True if the path is protected, False otherwise
    """
    if not file_path:
        return False

    # List of protected directory prefixes
    protected_dirs = [
        '/data/epgs',     # EPG files mapped from host
        '/data/uploads',   # User uploaded files
        '/data/m3us'       # M3U files mapped from host
    ]

    # Check if the path starts with any protected directory
    for protected_dir in protected_dirs:
        if file_path.startswith(protected_dir):
            return True

    return False

def validate_flexible_url(value):
    """
    Custom URL validator that accepts URLs with hostnames that aren't FQDNs.
    This allows URLs like "http://hostname/" which
    Django's standard URLValidator rejects.
    """
    if not value:
        return  # Allow empty values since the field is nullable

    # Create a standard Django URL validator
    url_validator = URLValidator()

    try:
        # First try the standard validation
        url_validator(value)
    except ValidationError as e:
        # If standard validation fails, check if it's a non-FQDN hostname
        import re

        # More flexible pattern for non-FQDN hostnames with paths
        # Matches: http://hostname, https://hostname/, http://hostname:port/path/to/file.xml, rtp://192.168.2.1,  rtsp://192.168.178.1, udp://239.0.0.1:1234
        # Also matches FQDNs for rtsp/rtp/udp protocols: rtsp://FQDN/path?query=value
        # Also supports authentication: rtsp://user:pass@hostname/path
        non_fqdn_pattern = r'^(rts?p|https?|udp)://([a-zA-Z0-9_\-\.]+:[^\s@]+@)?([a-zA-Z0-9]([a-zA-Z0-9\-\.]{0,61}[a-zA-Z0-9])?|[0-9.]+)?(\:[0-9]+)?(/[^\s]*)?$'
        non_fqdn_match = re.match(non_fqdn_pattern, value)

        if non_fqdn_match:
            return  # Accept non-FQDN hostnames and rtsp/rtp/udp URLs with optional authentication

        # If it doesn't match our flexible patterns, raise the original error
        raise ValidationError("Enter a valid URL.")

def dispatch_event_system(event_type, channel_id=None, channel_name=None, **details):
    try:
        from apps.connect.utils import trigger_event
        from apps.channels.models import Channel, Stream
        from core.models import StreamProfile
        from core.utils import RedisClient

        payload = dict(details)

        channel_obj = None
        if channel_id:
            try:
                channel_obj = Channel.objects.get(uuid=channel_id)
                payload["channel_name"] = channel_obj.name
            except Exception:
                payload["channel_name"] = channel_name or None
        else:
            payload["channel_name"] = channel_name or None

        # Resolve current stream info
        stream_id = details.get("stream_id")
        stream_obj = None
        if not stream_id and channel_obj:
            try:
                redis = RedisClient.get_client()
                sid = redis.get(f"channel_stream:{channel_obj.id}")
                if sid:
                    stream_id = int(sid)
            except Exception:
                stream_id = None

        if stream_id:
            try:
                stream_obj = Stream.objects.get(id=stream_id)
            except Exception:
                stream_obj = None

        # Populate stream details
        payload["stream_name"] = getattr(stream_obj, "name", None)
        payload["stream_url"] = getattr(stream_obj, "url", None)

        # Channel URL: use stream URL as best-effort
        payload["channel_url"] = payload.get("stream_url")

        # Provider name from M3U account
        provider_name = None
        try:
            if stream_obj and stream_obj.m3u_account:
                provider_name = stream_obj.m3u_account.name
        except Exception:
            provider_name = None
        payload["provider_name"] = provider_name

        # Profile used
        profile_used = None
        try:
            if stream_id:
                redis = RedisClient.get_client()
                pid = redis.get(f"stream_profile:{stream_id}")
                if pid:
                    profile = StreamProfile.objects.filter(id=int(pid)).first()
                    profile_used = profile.name if profile else None
        except Exception:
            profile_used = None

        payload["profile_used"] = profile_used

        # remove empty keys
        for k in list(payload.keys()):
            if not payload[k]:
                del payload[k]

        trigger_event(event_type, payload)

    except Exception as e:
        # Don't fail main path if connect dispatch fails
        pass

def log_system_event(event_type, channel_id=None, channel_name=None, **details):
    """
    Log a system event and maintain the configured max history.

    Args:
        event_type: Type of event (e.g., 'channel_start', 'client_connect')
        channel_id: Optional UUID of the channel
        channel_name: Optional name of the channel
        **details: Additional details to store in the event (stored as JSON)

    Example:
        log_system_event('channel_start', channel_id=uuid, channel_name='CNN',
                        stream_url='http://...', user='admin')
    """
    from core.models import SystemEvent, CoreSettings

    try:
        # Create the event
        SystemEvent.objects.create(
            event_type=event_type,
            channel_id=channel_id,
            channel_name=channel_name,
            details=details
        )

        # Trigger connect integrations for specific events
        dispatch_event_system(event_type, channel_id=channel_id, channel_name=channel_name, **details)

        # Get max events from settings (default 100)
        try:
            from .models import CoreSettings
            system_settings = CoreSettings.objects.filter(key='system_settings').first()
            if system_settings and isinstance(system_settings.value, dict):
                max_events = int(system_settings.value.get('max_system_events', 100))
            else:
                max_events = 100
        except Exception:
            max_events = 100

        # Delete old events beyond the limit (keep it efficient with a single query)
        total_count = SystemEvent.objects.count()
        if total_count > max_events:
            # Get the ID of the event at the cutoff point
            cutoff_event = SystemEvent.objects.values_list('id', flat=True)[max_events]
            # Delete all events with ID less than cutoff (older events)
            SystemEvent.objects.filter(id__lt=cutoff_event).delete()

    except Exception as e:
        # Don't let event logging break the main application
        logger.error(f"Failed to log system event {event_type}: {e}")


def _send_async(channel_layer, group, message):
    """Send a channel layer group message without blocking the gevent hub."""
    def _do():
        try:
            async_to_sync(channel_layer.group_send)(group, message)
        except Exception as e:
            logger.warning(f"Failed WebSocket group_send to '{group}': {e}")

    try:
        import gevent.monkey
        if gevent.monkey.is_module_patched("threading"):
            import gevent
            gevent.spawn(_gevent_ws_send, group, message)
            return
    except Exception:
        pass
    _do()


def send_websocket_notification(notification):
    """
    Send a system notification to all connected WebSocket clients.

    Args:
        notification: A SystemNotification model instance or dict with notification data

    Example:
        from core.models import SystemNotification
        notification = SystemNotification.create_version_notification('0.19.0', 'https://...')
        send_websocket_notification(notification)
    """
    try:
        channel_layer = get_channel_layer()

        # Convert model instance to dict if needed
        if hasattr(notification, 'id'):
            notification_data = {
                'id': notification.id,
                'notification_key': notification.notification_key,
                'notification_type': notification.notification_type,
                'priority': notification.priority,
                'title': notification.title,
                'message': notification.message,
                'action_data': notification.action_data,
                'is_active': notification.is_active,
                'admin_only': notification.admin_only,
                'created_at': notification.created_at.isoformat() if notification.created_at else None,
            }
        else:
            notification_data = notification

        _send_async(
            channel_layer,
            'updates',
            {
                'type': 'update',
                'data': {
                    'type': 'system_notification',
                    'notification': notification_data,
                }
            }
        )
        logger.debug(f"Sent WebSocket notification: {notification_data.get('title', 'Unknown')}")
    except Exception as e:
        logger.error(f"Failed to send WebSocket notification: {e}")


def get_host_and_port(request):
    """
    Returns (host, port) for building absolute URIs.
    - Prefers X-Forwarded-Host/X-Forwarded-Port (nginx).
    - Falls back to Host header.
    - Returns None for port if using standard ports (80/443) to omit from URLs.
    - In dev, uses 5656 as a guess if port cannot be determined.
    """
    scheme = request.META.get("HTTP_X_FORWARDED_PROTO", request.scheme)
    standard_port = "443" if scheme == "https" else "80"

    # 1. Try X-Forwarded-Host (may include port) - set by our nginx
    xfh = request.META.get("HTTP_X_FORWARDED_HOST")
    if xfh:
        if ":" in xfh:
            host, port = xfh.split(":", 1)
            if port == standard_port:
                return host, None
            return host, port
        else:
            host = xfh

        port = request.META.get("HTTP_X_FORWARDED_PORT")
        if port:
            return host, None if port == standard_port else port
        if request.META.get("HTTP_X_FORWARDED_PROTO"):
            return host, None

    # 2. Try Host header
    raw_host = request.get_host()
    if ":" in raw_host:
        host, port = raw_host.split(":", 1)
        return host, None if port == standard_port else port
    else:
        host = raw_host

    # 3. Check for X-Forwarded-Port (when Host header has no port but we're behind a reverse proxy)
    port = request.META.get("HTTP_X_FORWARDED_PORT")
    if port:
        return host, None if port == standard_port else port

    # 4. Behind a reverse proxy with no port info - assume standard port
    if request.META.get("HTTP_X_FORWARDED_PROTO") or request.META.get("HTTP_X_FORWARDED_FOR"):
        return host, None

    # 5. Try SERVER_PORT from META (only if NOT behind reverse proxy)
    port = request.META.get("SERVER_PORT")
    if port:
        return host, None if port == standard_port else port

    # 6. Dev fallback
    if os.environ.get("DISPATCHARR_ENV") == "dev" or host in ("localhost", "127.0.0.1"):
        return host, "5656"

    # 7. Final fallback: assume standard port for scheme
    return host, None


def build_absolute_uri_with_port(request, path):
    """
    Build an absolute URI with optional port.
    Port is omitted from URL if None (standard port for scheme).
    """
    host, port = get_host_and_port(request)
    scheme = request.META.get("HTTP_X_FORWARDED_PROTO", request.scheme)
    if port:
        return f"{scheme}://{host}:{port}{path}"
    return f"{scheme}://{host}{path}"


def send_notification_dismissed(notification_key):
    """
    Notify all connected clients that a notification was dismissed.
    Useful for syncing dismissal state across multiple browser tabs/sessions.

    Args:
        notification_key: The unique key of the dismissed notification
    """
    try:
        channel_layer = get_channel_layer()

        _send_async(
            channel_layer,
            'updates',
            {
                'type': 'update',
                'data': {
                    'type': 'notification_dismissed',
                    'notification_key': notification_key,
                }
            }
        )
    except Exception as e:
        logger.error(f"Failed to send notification dismissed event: {e}")